signalscout-backend/app/api/routes.py

from __future__ import annotations

from datetime import datetime, timedelta
from typing import Literal, Optional

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import HTMLResponse
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.agent.scheduler import run_scheduled_job
from app.core.config import Settings, get_settings
from app.db.models import AgentRun, Report, Signal, Topic
from app.db.session import get_db
from app.schemas.agent import (
    DashboardRead,
    ReportRead,
    RunRead,
    RunTriggerRead,
    SignalRead,
    TopicCreate,
    TopicRead,
)
from app.ui.home import HOME_PAGE_HTML

router = APIRouter()


@router.get("/", response_class=HTMLResponse)
def home() -> HTMLResponse:
    # 根路径返回业务工作台，避免用户直接打开域名时看到接口404。
    return HTMLResponse(HOME_PAGE_HTML)


@router.get("/health")
def health(settings: Settings = Depends(get_settings)) -> dict:
    # 健康检查只返回服务身份，避免暴露密钥和内部地址。
    return {"status": "ok", "app": settings.app_name, "env": settings.env}


@router.get("/agent/runs", response_model=list[RunRead])
def list_runs(db: Session = Depends(get_db)) -> list[AgentRun]:
    # 运行列表保持紧凑，方便仪表盘和命令行查看最近状态。
    return list(db.scalars(select(AgentRun).order_by(AgentRun.id.desc()).limit(20)).all())


@router.post("/agent/runs/daily", response_model=RunTriggerRead)
def trigger_daily_run(
    request: Request,
    settings: Settings = Depends(get_settings),
    db: Session = Depends(get_db),
) -> RunTriggerRead:
    # 暂停期间拒绝所有入口的采集请求，保证不会产生外部模型调用。
    if not settings.agent_enabled:
        raise HTTPException(status_code=503, detail="情报采集当前已暂停")
    # 手动搜索投递到调度器线程，避免HTTP请求承载长时间Grok搜索任务。
    running = db.scalar(
        select(AgentRun).where(AgentRun.status == "running").order_by(AgentRun.id.desc()).limit(1)
    )
    if running is not None:
        return RunTriggerRead(status="running", run_id=running.id)
    scheduler = getattr(request.app.state, "scheduler", None)
    if scheduler is not None and scheduler.running:
        scheduler.add_job(
            run_scheduled_job,
            trigger="date",
            run_date=datetime.now(scheduler.timezone) + timedelta(seconds=1),
            args=[settings],
            id=f"manual-intelligence-{datetime.now().timestamp()}",
            max_instances=1,
        )
        return RunTriggerRead(status="started")
    run_scheduled_job(settings)
    return RunTriggerRead(status="started")


@router.get("/signals", response_model=list[SignalRead])
def list_signals(
    limit: int = 100,
    source_type: Optional[Literal["news", "github"]] = None,
    db: Session = Depends(get_db),
) -> list[Signal]:
    # 情报列表按采集时间倒序返回，是仪表盘的主要阅读入口。
    safe_limit = max(1, min(limit, 200))
    return list(
        db.scalars(
            _visible_signals_query(source_type=source_type)
            .order_by(Signal.id.desc())
            .limit(safe_limit)
        ).all()
    )


@router.get("/reports/latest", response_model=Optional[ReportRead])
def latest_report(db: Session = Depends(get_db)) -> Optional[Report]:
    # 最新日报提供一份可直接阅读的每日简报。
    return db.scalar(select(Report).order_by(Report.id.desc()).limit(1))


@router.get("/dashboard", response_model=DashboardRead)
def dashboard(db: Session = Depends(get_db)) -> DashboardRead:
    # 单个仪表盘接口返回运行、信号、日报和主题，前端不需要理解内部调度细节。
    runs = list(db.scalars(select(AgentRun).order_by(AgentRun.id.desc()).limit(10)).all())
    latest_completed_run = db.scalar(
        select(AgentRun)
        .where(AgentRun.status == "completed")
        .order_by(AgentRun.id.desc())
        .limit(1)
    )
    latest_run_id = latest_completed_run.id if latest_completed_run is not None else None
    news_signals = list(
        db.scalars(
            _visible_signals_query(source_type="news", run_id=latest_run_id)
            .order_by(Signal.id.desc())
            .limit(80)
        ).all()
    )
    github_signals = list(
        db.scalars(
            _visible_signals_query(source_type="github", run_id=latest_run_id)
            .order_by(Signal.id.desc())
            .limit(80)
        ).all()
    )
    report = db.scalar(select(Report).order_by(Report.id.desc()).limit(1))
    topics = list(db.scalars(select(Topic).order_by(Topic.id.asc())).all())
    return DashboardRead(
        runs=runs,
        news_signals=news_signals,
        github_signals=github_signals,
        latest_report=report,
        topics=topics,
    )


@router.get("/logs/recent", response_model=list[str])
def recent_logs(settings: Settings = Depends(get_settings)) -> list[str]:
    # 最近日志用于定位Agent和模型供应商调用问题。
    if not settings.log_file.exists():
        return []
    lines = settings.log_file.read_text(encoding="utf-8", errors="replace").splitlines()
    return lines[-settings.log_tail_lines :]


@router.post("/topics", response_model=TopicRead)
def create_topic(payload: TopicCreate, db: Session = Depends(get_db)) -> Topic:
    # 主题让Agent可以在不改代码的情况下扩展关注范围。
    topic = Topic(name=payload.name, description=payload.description, enabled=payload.enabled)
    db.add(topic)
    db.commit()
    db.refresh(topic)
    return topic


@router.get("/topics", response_model=list[TopicRead])
def list_topics(db: Session = Depends(get_db)) -> list[Topic]:
    # 主题列表让操作者确认当前情报目标。
    return list(db.scalars(select(Topic).order_by(Topic.id.asc())).all())


def _visible_signals_query(
    source_type: Optional[Literal["news", "github"]] = None,
    run_id: Optional[int] = None,
):
    # 运行中的信号和完成后的信号走同一张表，查询函数保留统一排序入口。
    query = select(Signal).join(AgentRun, Signal.run_id == AgentRun.id)
    if source_type is not None:
        query = query.where(Signal.source_type == source_type)
    if run_id is not None:
        query = query.where(Signal.run_id == run_id)
    return query