stock/serve/backend/app/api/financials.py

"""财务数据 API — 独立路由, Cap.FINANCIAL 门控。"""
from __future__ import annotations

import logging

import polars as pl
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from app.services.financial_sync import get_financial_df
from app.services.financial_analyzer import analyze_financials_stream
from app.services import ai_reports
from app.tickflow.capabilities import Cap

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/financials", tags=["financials"])


@router.get("/status")
def financial_status(request: Request):
    """返回各财务表的同步状态。"""
    data_dir = request.app.state.repo.store.data_dir
    tables = {}

    for table in ("metrics", "income", "balance_sheet", "cash_flow"):
        path = data_dir / "financials" / table / "part.parquet"
        if path.exists():
            try:
                df = pl.read_parquet(path, columns=["symbol"])
                tables[table] = {
                    "rows": len(df),
                    "symbols": df["symbol"].n_unique() if not df.is_empty() else 0,
                }
            except Exception:
                tables[table] = {"rows": 0, "symbols": 0}
        else:
            tables[table] = {"rows": 0, "symbols": 0}

    capset = request.app.state.capabilities
    has_cap = capset.has(Cap.FINANCIAL)
    fs = getattr(request.app.state, "financial_scheduler", None)
    last_sync = fs.last_sync if fs else {}
    has_data = any(t.get("rows", 0) > 0 for t in tables.values())

    return {
        "available": has_cap or has_data,
        "tables": tables,
        "last_sync": last_sync,
        "syncing": bool(fs and fs.is_syncing),
    }


@router.get("/metrics")
def get_metrics(request: Request, symbol: str | None = None):
    """查询核心财务指标。"""
    df = get_financial_df(request.app.state.repo.store.data_dir, "metrics")
    if df.is_empty():
        return {"data": []}
    if symbol:
        df = df.filter(pl.col("symbol") == symbol)
    return {"data": df.to_dicts()}


@router.get("/income")
def get_income(request: Request, symbol: str | None = None):
    """查询利润表。"""
    df = get_financial_df(request.app.state.repo.store.data_dir, "income")
    if df.is_empty():
        return {"data": []}
    if symbol:
        df = df.filter(pl.col("symbol") == symbol)
    return {"data": df.to_dicts()}


@router.get("/balance-sheet")
def get_balance_sheet(request: Request, symbol: str | None = None):
    """查询资产负债表。"""
    df = get_financial_df(request.app.state.repo.store.data_dir, "balance_sheet")
    if df.is_empty():
        return {"data": []}
    if symbol:
        df = df.filter(pl.col("symbol") == symbol)
    return {"data": df.to_dicts()}


@router.get("/cash-flow")
def get_cash_flow(request: Request, symbol: str | None = None):
    """查询现金流量表。"""
    df = get_financial_df(request.app.state.repo.store.data_dir, "cash_flow")
    if df.is_empty():
        return {"data": []}
    if symbol:
        df = df.filter(pl.col("symbol") == symbol)
    return {"data": df.to_dicts()}


@router.post("/sync/{table}")
def sync_table(request: Request, table: str):
    """手动触发同步 — serve 端为只读模式，已禁用。"""
    raise HTTPException(403, "serve 端为只读模式，数据仅通过 local→serve 同步")


class AnalyzeRequest(BaseModel):
    """AI 财务分析请求。"""
    symbol: str
    focus: str = ""  # 可选:用户追加的分析关注点


@router.post("/analyze")
async def analyze_financials(request: Request, req: AnalyzeRequest):
    """AI 财务分析 — SSE 流式返回。

    后端读取该标的 4 张财务表 → 注入 CFA 分析师级提示词 → 流式调用 LLM →
    逐 chunk 以 SSE 形式推给前端(JSON per line, 非 text/event-stream,
    以便前端用 ReadableStream 逐行解析,更简单可靠)。
    """
    capset = request.app.state.capabilities
    capset.require(Cap.FINANCIAL)

    if not req.symbol:
        raise HTTPException(400, "symbol 不能为空")

    data_dir = request.app.state.repo.store.data_dir

    async def stream_gen():
        async for chunk in analyze_financials_stream(data_dir, req.symbol, req.focus, username=request.state.username):
            yield chunk + "\n"

    return StreamingResponse(
        stream_gen(),
        media_type="application/x-ndjson",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


# ================================================================
# AI 报告 CRUD(历史报告持久化)
# ================================================================

class SaveReportRequest(BaseModel):
    """保存一条 AI 财务分析报告。"""
    symbol: str
    name: str = ""
    focus: str = ""
    content: str
    periods: int | None = None
    summary: str = ""


@router.get("/reports")
def list_reports(request: Request):
    """获取当前用户的历史报告(按时间降序,后端已裁剪到上限)。"""
    capset = request.app.state.capabilities
    if not capset.has(Cap.FINANCIAL):
        return {"reports": []}
    username = getattr(request.state, "username", None)
    return {"reports": ai_reports.list_reports(username)}


@router.post("/reports")
def save_report(request: Request, req: SaveReportRequest):
    """保存一条报告。"""
    capset = request.app.state.capabilities
    capset.require(Cap.FINANCIAL)
    username = getattr(request.state, "username", None)
    report = ai_reports.save_report({
        "symbol": req.symbol,
        "name": req.name,
        "focus": req.focus,
        "content": req.content,
        "periods": req.periods,
        "summary": req.summary,
    }, username)
    return {"ok": True, "report": report}


@router.delete("/reports/{report_id}")
def delete_report(request: Request, report_id: str):
    """删除一条报告。"""
    capset = request.app.state.capabilities
    capset.require(Cap.FINANCIAL)
    username = getattr(request.state, "username", None)
    ok = ai_reports.delete_report(report_id, username)
    return {"ok": ok}