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>
64 lines
1.6 KiB
Python
64 lines
1.6 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from ..models import Event, ParseJob
|
|
from ..schemas import IngestEventItem
|
|
from .filtering import sync_event_to_map_object
|
|
|
|
|
|
def ingest_events(
|
|
db: Session,
|
|
items: list[IngestEventItem],
|
|
job_id: int | None = None,
|
|
*,
|
|
touch_job: bool = True,
|
|
) -> tuple[int, int, int, int]:
|
|
ingested = 0
|
|
updated = 0
|
|
skipped = 0
|
|
map_synced = 0
|
|
|
|
for item in items:
|
|
existing = (
|
|
db.query(Event)
|
|
.filter(Event.source_url == item.source_url)
|
|
.first()
|
|
)
|
|
|
|
if existing:
|
|
skipped += 1
|
|
continue
|
|
|
|
event = Event(
|
|
source_type=item.source_type,
|
|
source_url=item.source_url,
|
|
raw_text=item.raw_text,
|
|
title=item.title,
|
|
description=item.description,
|
|
locality=item.locality,
|
|
latitude=item.latitude,
|
|
longitude=item.longitude,
|
|
event_date=item.event_date,
|
|
region=item.region,
|
|
topic=item.topic,
|
|
tags=item.tags,
|
|
metadata_=item.metadata,
|
|
)
|
|
db.add(event)
|
|
ingested += 1
|
|
|
|
db.flush()
|
|
if sync_event_to_map_object(db, event):
|
|
map_synced += 1
|
|
|
|
if touch_job and job_id is not None:
|
|
job = db.query(ParseJob).filter(ParseJob.id == job_id).first()
|
|
if job:
|
|
from datetime import datetime, timezone
|
|
|
|
job.status = "completed"
|
|
job.last_run_at = datetime.now(timezone.utc)
|
|
job.last_error = None
|
|
|
|
db.commit()
|
|
return ingested, updated, skipped, map_synced
|