myquant/djapi/api/report/query.py

"""
news_report / news_event 只读查询层（日报结构化入库，见 docs/db_schema_v1.1.md）。

连接配置来自环境变量（与 docs/db_schema_v1.1.md 保持一致）：
    NEWS_DB_HOST / NEWS_DB_PORT / NEWS_DB_USER / NEWS_DB_PASSWORD / NEWS_DB_NAME
NEWS_DB_PASSWORD 缺失时直接报错，禁止默认密码。

所有 SQL 均为 MariaDB 方言、参数化查询（防 SQL 注入），不依赖 ORM。
"""

import json
import os
from datetime import date, datetime, timedelta

import mysql.connector

_REPORT_FIELDS = (
    "id, report_date, report_type, file_name, generated_at, "
    "ai_summary, stats, created_at"
)
_EVENT_FIELDS = (
    "id, report_id, section, rank, importance, event_type, title, "
    "summary, sentiment, source, url"
)


def load_db_config() -> dict:
    """从环境变量读取 NEWS_DB_* 连接配置，密码缺失时抛错。"""
    password = os.getenv("NEWS_DB_PASSWORD")
    if not password:
        raise RuntimeError(
            "NEWS_DB_PASSWORD 未设置，禁止使用默认密码连接 news 库"
        )
    return {
        "host": os.getenv("NEWS_DB_HOST", "127.0.0.1"),
        "port": int(os.getenv("NEWS_DB_PORT", "3306")),
        "user": os.getenv("NEWS_DB_USER", "myquant"),
        "password": password,
        "database": os.getenv("NEWS_DB_NAME", "myquant"),
        "charset": "utf8mb4",
    }


def _connect():
    return mysql.connector.connect(**load_db_config())


def _parse_json(value):
    """把 JSON 字符串列（如 sources）解析为 dict/list；已是对象则原样返回。"""
    if value is None:
        return None
    if isinstance(value, str):
        try:
            return json.loads(value)
        except (TypeError, ValueError):
            return None
    return value


def _row_to_dict(row: dict) -> dict:
    """序列化行：stats JSON 解析、日期/时间转 ISO 字符串。"""
    d = dict(row)
    if d.get("stats") is not None:
        try:
            d["stats"] = json.loads(d["stats"])
        except (TypeError, ValueError):
            d["stats"] = None
    for k, v in d.items():
        if isinstance(v, (date, datetime)):
            d[k] = v.isoformat()
    return d


def fetch_reports(
    report_type: str | None = None,
    start_date: date | None = None,
    end_date: date | None = None,
    report_id: int | None = None,
):
    """
    日报查询。

    report_id 给定 → 返回单份详情 dict（含 events，按 section, rank 排序）；
    不存在返回 None。
    否则 → 返回列表：每天每类型取最新一份（MAX(generated_at) 子查询），
    仅主表字段（轻量，不带 events）。
    """
    conn = _connect()
    try:
        cur = conn.cursor(dictionary=True)
        if report_id is not None:
            cur.execute(
                f"SELECT {_REPORT_FIELDS} FROM news_report WHERE id = %s",
                (report_id,),
            )
            row = cur.fetchone()
            if row is None:
                return None
            report = _row_to_dict(row)
            cur.execute(
                "SELECT id, section, rank, importance, event_type, title, "
                "summary, sentiment, source, sources, url "
                "FROM news_event WHERE report_id = %s ORDER BY section, rank",
                (report_id,),
            )
            report["events"] = [
                {**dict(r), "sources": _parse_json(r["sources"])}
                for r in cur.fetchall()
            ]
            return report

        where, params = [], []
        if report_type:
            where.append("r.report_type = %s")
            params.append(report_type)
        if start_date:
            where.append("r.report_date >= %s")
            params.append(start_date.isoformat())
        if end_date:
            where.append("r.report_date <= %s")
            params.append(end_date.isoformat())
        cond = (" WHERE " + " AND ".join(where)) if where else ""

        sql = (
            "SELECT r.id, r.report_date, r.report_type, r.file_name, "
            "r.generated_at, r.ai_summary, r.stats, r.created_at "
            "FROM news_report r "
            "JOIN ("
            "    SELECT report_date, report_type, MAX(generated_at) AS g "
            "    FROM news_report GROUP BY report_date, report_type"
            ") t ON r.report_date = t.report_date "
            "     AND r.report_type = t.report_type "
            "     AND r.generated_at = t.g"
            + cond
            + " ORDER BY r.report_date DESC, r.report_type"
        )
        cur.execute(sql, tuple(params))
        return [_row_to_dict(r) for r in cur.fetchall()]
    finally:
        conn.close()


def fetch_important_events(
    days: int = 7,
    importance: int = 4,
    report_type: str | None = None,
    section: str | None = None,
    limit: int = 100,
) -> list:
    """
    跨日报重要事件聚合检索（最近 N 天，importance >= 阈值）。

    按 importance DESC, report_date DESC 排序。
    """
    conn = _connect()
    try:
        cur = conn.cursor(dictionary=True)
        since = (date.today() - timedelta(days=days)).isoformat()
        where = ["r.report_date >= %s", "e.importance >= %s"]
        params = [since, int(importance)]
        if report_type:
            where.append("r.report_type = %s")
            params.append(report_type)
        if section:
            where.append("e.section = %s")
            params.append(section)

        sql = (
            "SELECT r.report_date, r.report_type, e.id, e.section, e.rank, "
            "e.importance, e.event_type, e.title, e.summary, e.sentiment, "
            "e.source, e.sources, e.url "
            "FROM news_event e "
            "JOIN news_report r ON r.id = e.report_id "
            "WHERE " + " AND ".join(where)
            + " ORDER BY e.importance DESC, r.report_date DESC, "
              "e.section, e.rank "
            + "LIMIT %s"
        )
        params.append(int(limit))
        cur.execute(sql, tuple(params))
        rows = cur.fetchall()
        return [
            {**dict(r), "sources": _parse_json(r["sources"])}
            for r in rows
        ]
    finally:
        conn.close()