Add CA admin UI and extend admin API for the unified platform.

Deliver parsers, events, analytics, and PI management in Vue; fix Telegram session mount and map navigation to events by eventId.

Co-authored-by: Cursor <cursoragent@cursor.com>

centers/analytics/api/app/routers/admin.py

@@ -1,16 +1,36 @@
+from datetime import datetime
+
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 
 from ..database import get_db
-from ..models import Event, ParseJob
+from ..models import Consumer, Event, ParseJob
 from ..schemas import (
+    AnalyticsSummary,
     ConsumerCreate,
     ConsumerRead,
+    ConsumerUpdate,
+    EventListResponse,
     EventRead,
     ParseJobCreate,
     ParseJobRead,
+    TimelinePoint,
+    TopItem,
+)
+from ..services.analytics import (
+    get_analytics_summary,
+    get_timeline,
+    get_top_localities,
+    get_top_regions,
+)
+from ..services.events_query import build_events_query
+from ..services.filtering import (
+    consumer_to_read,
+    create_consumer,
+    list_consumers,
+    rotate_consumer_key,
+    update_consumer,
 )
-from ..services.filtering import create_consumer
 from ..services.jobs import enqueue_job
 
 router = APIRouter(prefix="/admin", tags=["admin"])
@@ -45,19 +65,81 @@ def get_parse_job(job_id: int, db: Session = Depends(get_db)):
     return job
 
 
-@router.get("/events", response_model=list[EventRead])
+@router.post("/jobs/{job_id}/retry", response_model=ParseJobRead)
+def retry_parse_job(job_id: int, 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 = "queued"
+    job.last_error = None
+    db.commit()
+    db.refresh(job)
+
+    enqueue_job(job.id, job.source_type, job.source_config)
+    return job
+
+
+@router.get("/events", response_model=EventListResponse)
 def list_events(
-    limit: int = Query(default=100, ge=1, le=1000),
+    limit: int = Query(default=50, ge=1, le=1000),
     offset: int = Query(default=0, ge=0),
+    source_type: str | None = None,
+    region: str | None = None,
+    topic: str | None = None,
+    locality: str | None = None,
+    date_from: datetime | None = None,
+    date_to: datetime | None = None,
+    search: str | None = None,
     db: Session = Depends(get_db),
 ):
-    return (
-        db.query(Event)
-        .order_by(Event.ingested_at.desc())
-        .offset(offset)
-        .limit(limit)
-        .all()
+    query = build_events_query(
+        db,
+        source_type=source_type,
+        region=region,
+        topic=topic,
+        locality=locality,
+        date_from=date_from,
+        date_to=date_to,
+        search=search,
     )
+    total = query.count()
+    items = query.offset(offset).limit(limit).all()
+    return EventListResponse(items=items, total=total)
+
+
+@router.get("/analytics/summary", response_model=AnalyticsSummary)
+def analytics_summary(db: Session = Depends(get_db)):
+    return get_analytics_summary(db)
+
+
+@router.get("/analytics/timeline", response_model=list[TimelinePoint])
+def analytics_timeline(
+    days: int = Query(default=30, ge=1, le=365),
+    db: Session = Depends(get_db),
+):
+    return get_timeline(db, days=days)
+
+
+@router.get("/analytics/top-localities", response_model=list[TopItem])
+def analytics_top_localities(
+    limit: int = Query(default=10, ge=1, le=100),
+    db: Session = Depends(get_db),
+):
+    return get_top_localities(db, limit=limit)
+
+
+@router.get("/analytics/top-regions", response_model=list[TopItem])
+def analytics_top_regions(
+    limit: int = Query(default=10, ge=1, le=100),
+    db: Session = Depends(get_db),
+):
+    return get_top_regions(db, limit=limit)
+
+
+@router.get("/consumers", response_model=list[ConsumerRead])
+def list_pi_consumers(db: Session = Depends(get_db)):
+    return [consumer_to_read(c) for c in list_consumers(db)]
 
 
 @router.post("/consumers", response_model=ConsumerRead, status_code=201)
@@ -69,10 +151,36 @@ def create_pi_consumer(payload: ConsumerCreate, db: Session = Depends(get_db)):
         topics=payload.topics,
         date_from=payload.date_from,
     )
-    return ConsumerRead(
-        id=consumer.id,
-        name=consumer.name,
-        is_active=consumer.is_active,
-        created_at=consumer.created_at,
-        api_key=api_key,
+    return consumer_to_read(consumer, api_key=api_key)
+
+
+@router.patch("/consumers/{consumer_id}", response_model=ConsumerRead)
+def patch_pi_consumer(
+    consumer_id: int,
+    payload: ConsumerUpdate,
+    db: Session = Depends(get_db),
+):
+    consumer = db.query(Consumer).filter(Consumer.id == consumer_id).first()
+    if not consumer:
+        raise HTTPException(status_code=404, detail="Consumer not found")
+
+    consumer = update_consumer(
+        db,
+        consumer,
+        name=payload.name,
+        is_active=payload.is_active,
+        regions=payload.regions,
+        topics=payload.topics,
+        date_from=payload.date_from,
     )
+    return consumer_to_read(consumer)
+
+
+@router.post("/consumers/{consumer_id}/rotate-key", response_model=ConsumerRead)
+def rotate_pi_consumer_key(consumer_id: int, db: Session = Depends(get_db)):
+    consumer = db.query(Consumer).filter(Consumer.id == consumer_id).first()
+    if not consumer:
+        raise HTTPException(status_code=404, detail="Consumer not found")
+
+    consumer, api_key = rotate_consumer_key(db, consumer)
+    return consumer_to_read(consumer, api_key=api_key)

centers/analytics/api/app/schemas.py

@@ -122,6 +122,14 @@ class ConsumerCreate(BaseModel):
     date_from: datetime | None = None
 
 
+class ConsumerUpdate(BaseModel):
+    name: str | None = None
+    is_active: bool | None = None
+    regions: list[str] | None = None
+    topics: list[str] | None = None
+    date_from: datetime | None = None
+
+
 class ConsumerRead(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
@@ -130,3 +138,31 @@ class ConsumerRead(BaseModel):
     is_active: bool
     created_at: datetime
     api_key: str | None = None
+    regions: list[str] | None = None
+    topics: list[str] | None = None
+    date_from: datetime | None = None
+
+
+class EventListResponse(BaseModel):
+    items: list[EventRead]
+    total: int
+
+
+class AnalyticsSummary(BaseModel):
+    total_events: int
+    events_with_coords: int
+    events_last_24h: int
+    total_consumers: int
+    active_consumers: int
+    total_jobs: int
+    pending_jobs: int
+
+
+class TimelinePoint(BaseModel):
+    date: str
+    count: int
+
+
+class TopItem(BaseModel):
+    name: str
+    count: int

centers/analytics/api/app/services/analytics.py

@@ -0,0 +1,92 @@
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from ..models import Consumer, Event, ParseJob
+from ..schemas import AnalyticsSummary, TimelinePoint, TopItem
+
+
+def get_analytics_summary(db: Session) -> AnalyticsSummary:
+    now = datetime.now(timezone.utc)
+    day_ago = now - timedelta(days=1)
+
+    total_events = db.query(func.count(Event.id)).scalar() or 0
+    events_with_coords = (
+        db.query(func.count(Event.id))
+        .filter(Event.latitude.isnot(None), Event.longitude.isnot(None))
+        .scalar()
+        or 0
+    )
+    events_last_24h = (
+        db.query(func.count(Event.id))
+        .filter(Event.ingested_at >= day_ago)
+        .scalar()
+        or 0
+    )
+    total_consumers = db.query(func.count(Consumer.id)).scalar() or 0
+    active_consumers = (
+        db.query(func.count(Consumer.id))
+        .filter(Consumer.is_active.is_(True))
+        .scalar()
+        or 0
+    )
+    total_jobs = db.query(func.count(ParseJob.id)).scalar() or 0
+    pending_jobs = (
+        db.query(func.count(ParseJob.id))
+        .filter(ParseJob.status.in_(["queued", "running", "pending"]))
+        .scalar()
+        or 0
+    )
+
+    return AnalyticsSummary(
+        total_events=total_events,
+        events_with_coords=events_with_coords,
+        events_last_24h=events_last_24h,
+        total_consumers=total_consumers,
+        active_consumers=active_consumers,
+        total_jobs=total_jobs,
+        pending_jobs=pending_jobs,
+    )
+
+
+def get_timeline(db: Session, days: int = 30) -> list[TimelinePoint]:
+    since = datetime.now(timezone.utc) - timedelta(days=days)
+    day_col = func.date_trunc("day", Event.ingested_at).label("day")
+
+    rows = (
+        db.query(day_col, func.count(Event.id).label("count"))
+        .filter(Event.ingested_at >= since)
+        .group_by(day_col)
+        .order_by(day_col)
+        .all()
+    )
+
+    return [
+        TimelinePoint(date=row.day.date().isoformat(), count=row.count)
+        for row in rows
+    ]
+
+
+def get_top_localities(db: Session, limit: int = 10) -> list[TopItem]:
+    rows = (
+        db.query(Event.locality, func.count(Event.id).label("count"))
+        .filter(Event.locality != "", Event.locality.isnot(None))
+        .group_by(Event.locality)
+        .order_by(func.count(Event.id).desc())
+        .limit(limit)
+        .all()
+    )
+    return [TopItem(name=row.locality, count=row.count) for row in rows]
+
+
+def get_top_regions(db: Session, limit: int = 10) -> list[TopItem]:
+    rows = (
+        db.query(Event.region, func.count(Event.id).label("count"))
+        .filter(Event.region.isnot(None), Event.region != "")
+        .group_by(Event.region)
+        .order_by(func.count(Event.id).desc())
+        .limit(limit)
+        .all()
+    )
+    return [TopItem(name=row.region, count=row.count) for row in rows]

centers/analytics/api/app/services/events_query.py

@@ -0,0 +1,42 @@
+from datetime import datetime
+
+from sqlalchemy.orm import Query, Session
+
+from ..models import Event
+
+
+def build_events_query(
+    db: Session,
+    *,
+    source_type: str | None = None,
+    region: str | None = None,
+    topic: str | None = None,
+    locality: str | None = None,
+    date_from: datetime | None = None,
+    date_to: datetime | None = None,
+    search: str | None = None,
+) -> Query:
+    query = db.query(Event)
+
+    if source_type:
+        query = query.filter(Event.source_type == source_type)
+    if region:
+        query = query.filter(Event.region == region)
+    if topic:
+        query = query.filter(Event.topic == topic)
+    if locality:
+        query = query.filter(Event.locality.ilike(f"%{locality}%"))
+    if date_from:
+        query = query.filter(Event.event_date >= date_from)
+    if date_to:
+        query = query.filter(Event.event_date <= date_to)
+    if search:
+        pattern = f"%{search}%"
+        query = query.filter(
+            Event.title.ilike(pattern)
+            | Event.description.ilike(pattern)
+            | Event.raw_text.ilike(pattern)
+            | Event.locality.ilike(pattern)
+        )
+
+    return query.order_by(Event.ingested_at.desc())

centers/analytics/api/app/services/filtering.py

@@ -5,6 +5,7 @@ from datetime import datetime
 from sqlalchemy.orm import Session
 
 from ..models import Consumer, ConsumerFilter, Event, MapObject
+from ..schemas import ConsumerRead
 
 
 def hash_api_key(api_key: str) -> str:
@@ -15,6 +16,66 @@ def generate_api_key() -> str:
     return secrets.token_urlsafe(32)
 
 
+def consumer_to_read(consumer: Consumer, api_key: str | None = None) -> ConsumerRead:
+    consumer_filter = consumer.filter
+    return ConsumerRead(
+        id=consumer.id,
+        name=consumer.name,
+        is_active=consumer.is_active,
+        created_at=consumer.created_at,
+        api_key=api_key,
+        regions=consumer_filter.regions if consumer_filter else None,
+        topics=consumer_filter.topics if consumer_filter else None,
+        date_from=consumer_filter.date_from if consumer_filter else None,
+    )
+
+
+def list_consumers(db: Session) -> list[Consumer]:
+    return db.query(Consumer).order_by(Consumer.id).all()
+
+
+def update_consumer(
+    db: Session,
+    consumer: Consumer,
+    *,
+    name: str | None = None,
+    is_active: bool | None = None,
+    regions: list[str] | None = None,
+    topics: list[str] | None = None,
+    date_from: datetime | None = None,
+) -> Consumer:
+    if name is not None:
+        consumer.name = name
+    if is_active is not None:
+        consumer.is_active = is_active
+
+    if any(v is not None for v in (regions, topics, date_from)):
+        consumer_filter = consumer.filter
+        if consumer_filter is None:
+            consumer_filter = ConsumerFilter(consumer_id=consumer.id)
+            db.add(consumer_filter)
+            consumer.filter = consumer_filter
+
+        if regions is not None:
+            consumer_filter.regions = regions
+        if topics is not None:
+            consumer_filter.topics = topics
+        if date_from is not None:
+            consumer_filter.date_from = date_from
+
+    db.commit()
+    db.refresh(consumer)
+    return consumer
+
+
+def rotate_consumer_key(db: Session, consumer: Consumer) -> tuple[Consumer, str]:
+    api_key = generate_api_key()
+    consumer.api_key_hash = hash_api_key(api_key)
+    db.commit()
+    db.refresh(consumer)
+    return consumer, api_key
+
+
 def create_consumer(
     db: Session,
     name: str,