You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
133 lines
3.6 KiB
Python
133 lines
3.6 KiB
Python
"""
|
|
Overlay API — headline ingestion only.
|
|
|
|
Scope was reduced: wikimedia / youtube / google_trends / finra / feature_build
|
|
were removed. Only the Yahoo RSS headline collector remains.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, desc
|
|
|
|
from app.core.database import get_db
|
|
from app.models.overlay_feature import OverlayJobLog
|
|
from app.models.overlay_raw_event import OverlayHeadlineEvent
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class HeadlineItem(BaseModel):
|
|
title: str
|
|
publisher: Optional[str] = None
|
|
published_at: datetime
|
|
article_guid: str
|
|
|
|
|
|
class HeadlinesResponse(BaseModel):
|
|
symbol: str
|
|
headlines: List[HeadlineItem]
|
|
headline_count_6h: int
|
|
headline_count_24h: int
|
|
publisher_breadth_24h: int
|
|
|
|
|
|
class JobLogEntry(BaseModel):
|
|
id: str
|
|
job_type: str
|
|
status: str
|
|
started_at: datetime
|
|
completed_at: Optional[datetime] = None
|
|
records_processed: int = 0
|
|
error_message: Optional[str] = None
|
|
|
|
|
|
class JobLogResponse(BaseModel):
|
|
logs: List[JobLogEntry]
|
|
total_count: int
|
|
|
|
|
|
def _utc(dt):
|
|
if isinstance(dt, datetime) and dt.tzinfo is None:
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
|
|
|
|
@router.get(
|
|
"/admin/job-log",
|
|
response_model=JobLogResponse,
|
|
summary="Overlay job log",
|
|
tags=["overlay-admin"],
|
|
)
|
|
async def get_job_log(
|
|
limit: int = Query(50, ge=1, le=500),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(OverlayJobLog).order_by(desc(OverlayJobLog.started_at)).limit(limit)
|
|
)
|
|
logs = result.scalars().all()
|
|
entries = [
|
|
JobLogEntry(
|
|
id=str(log.id),
|
|
job_type=log.job_type,
|
|
status=log.status,
|
|
started_at=log.started_at,
|
|
completed_at=log.completed_at,
|
|
records_processed=log.records_processed or 0,
|
|
error_message=log.error_message,
|
|
)
|
|
for log in logs
|
|
]
|
|
return JobLogResponse(logs=entries, total_count=len(entries))
|
|
|
|
|
|
@router.get(
|
|
"/{symbol}/headlines",
|
|
response_model=HeadlinesResponse,
|
|
summary="Recent headlines for a symbol",
|
|
)
|
|
async def get_headlines(
|
|
symbol: str,
|
|
hours: int = Query(24, ge=1, le=168),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
symbol = symbol.upper()
|
|
cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
|
|
cutoff_6h = datetime.now(timezone.utc) - timedelta(hours=6)
|
|
|
|
result = await db.execute(
|
|
select(OverlayHeadlineEvent)
|
|
.where(OverlayHeadlineEvent.published_at >= cutoff)
|
|
.order_by(desc(OverlayHeadlineEvent.published_at))
|
|
.limit(500)
|
|
)
|
|
all_events = result.scalars().all()
|
|
sym_events = [e for e in all_events if symbol in (e.matched_symbols or [])]
|
|
|
|
if not sym_events:
|
|
raise HTTPException(status_code=404, detail=f"No headlines for {symbol} in last {hours}h")
|
|
|
|
headlines = [
|
|
HeadlineItem(
|
|
title=e.title,
|
|
publisher=e.publisher,
|
|
published_at=e.published_at,
|
|
article_guid=e.article_guid,
|
|
)
|
|
for e in sym_events
|
|
]
|
|
publishers = {e.publisher for e in sym_events if e.publisher}
|
|
count_6h = sum(1 for e in sym_events if _utc(e.published_at) >= cutoff_6h)
|
|
|
|
return HeadlinesResponse(
|
|
symbol=symbol,
|
|
headlines=headlines,
|
|
headline_count_6h=count_6h,
|
|
headline_count_24h=len(sym_events),
|
|
publisher_breadth_24h=len(publishers),
|
|
)
|