gitrusprusandCursor
1492576fd9
Introduce background scheduling and migrations for analytics ingest, expand parser management and map toolbar UX, and ignore local data/session files from version control. Co-authored-by: Cursor <cursoragent@cursor.com>
76 lines
2.1 KiB
Python
76 lines
2.1 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..database import get_db
|
|
from ..deps import verify_internal_token
|
|
from ..models import ParseJob
|
|
from ..schemas import IngestRequest, IngestResponse, ListenerSubscription
|
|
from ..services.ingest import ingest_events
|
|
|
|
router = APIRouter(prefix="/internal", tags=["internal"])
|
|
|
|
|
|
@router.post("/ingest", response_model=IngestResponse)
|
|
def internal_ingest(
|
|
payload: IngestRequest,
|
|
_: None = Depends(verify_internal_token),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
ingested, updated, skipped, map_synced = ingest_events(
|
|
db,
|
|
payload.events,
|
|
payload.job_id,
|
|
touch_job=not payload.listener,
|
|
)
|
|
return IngestResponse(
|
|
ingested=ingested,
|
|
updated=updated,
|
|
skipped=skipped,
|
|
map_objects_synced=map_synced,
|
|
)
|
|
|
|
|
|
@router.patch("/jobs/{job_id}")
|
|
def update_job_status(
|
|
job_id: int,
|
|
status: str,
|
|
error: str | None = None,
|
|
_: None = Depends(verify_internal_token),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
job = db.query(ParseJob).filter(ParseJob.id == job_id).first()
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
|
|
job.status = status
|
|
if status in ("completed", "failed"):
|
|
job.last_run_at = datetime.now(timezone.utc)
|
|
job.last_error = error
|
|
db.commit()
|
|
return {"id": job.id, "status": job.status}
|
|
|
|
|
|
@router.get("/listener/subscriptions", response_model=list[ListenerSubscription])
|
|
def listener_subscriptions(
|
|
_: None = Depends(verify_internal_token),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
jobs = (
|
|
db.query(ParseJob)
|
|
.filter(
|
|
ParseJob.is_active.is_(True),
|
|
ParseJob.source_type == "telegram",
|
|
)
|
|
.order_by(ParseJob.id.asc())
|
|
.all()
|
|
)
|
|
result: list[ListenerSubscription] = []
|
|
for job in jobs:
|
|
channel = job.source_config.get("channel") if job.source_config else None
|
|
if not channel:
|
|
continue
|
|
result.append(ListenerSubscription(job_id=job.id, channel=str(channel)))
|
|
return result
|