as-event/backend/app/importers.py

"""录入与导入：人工 CSV 导入 + 自动化接入接口（预留）。

- parse_events_csv / parse_etf_csv：解析导入模板（见 ddl/*_template.csv）
- AutoEventSource / AutoEtfSource：自动化抓取接口（未来挂新闻/公告/资金流），
  现在留空注册表；实现后在 _AUTO_* 注册即可被定时任务调用。
"""
from __future__ import annotations

import csv
import io
import logging
from datetime import date, datetime
from typing import Dict, List, Protocol

log = logging.getLogger("as-event.importers")

_VALID_DIRECTION = {"bullish", "bearish", "neutral"}
_VALID_CYCLE = {"top", "bottom", "none"}


def _parse_date(s: str) -> date:
    s = (s or "").strip()
    for fmt in ("%Y%m%d", "%Y-%m-%d", "%Y/%m/%d"):
        try:
            return datetime.strptime(s, fmt).date()
        except ValueError:
            continue
    raise ValueError(f"日期格式错误: {s!r}")


def _clean(s) -> str:
    return (s or "").strip()


def parse_events_csv(content: str) -> List[dict]:
    """列：event_date,title,category,impact_direction,severity,related_indices,cycle_tag,description,source_url"""
    reader = csv.DictReader(io.StringIO(content))
    out: List[dict] = []
    for i, row in enumerate(reader, start=2):  # 含表头，数据从第2行
        title = _clean(row.get("title"))
        if not title:
            continue
        try:
            ev = dict(
                event_date=_parse_date(row.get("event_date", "")),
                title=title,
                category=_clean(row.get("category")) or "其他",
                impact_direction=(_clean(row.get("impact_direction")).lower() or "neutral"),
                severity=int(_clean(row.get("severity")) or 3),
                related_indices=_clean(row.get("related_indices")) or None,
                cycle_tag=(_clean(row.get("cycle_tag")).lower() or "none"),
                description=_clean(row.get("description")) or None,
                source_url=_clean(row.get("source_url")) or None,
                source="import",
            )
        except ValueError as e:
            raise ValueError(f"第 {i} 行解析失败: {e}") from e
        if ev["impact_direction"] not in _VALID_DIRECTION:
            ev["impact_direction"] = "neutral"
        if ev["cycle_tag"] not in _VALID_CYCLE:
            ev["cycle_tag"] = "none"
        ev["severity"] = min(5, max(1, ev["severity"]))
        out.append(ev)
    return out


def parse_etf_csv(content: str) -> List[dict]:
    """列：trade_date,etf_code,etf_name,category,related_index,net_inflow,shares_change,note"""
    reader = csv.DictReader(io.StringIO(content))
    out: List[dict] = []
    for i, row in enumerate(reader, start=2):
        d = _clean(row.get("trade_date"))
        if not d:
            continue
        try:
            rec = dict(
                trade_date=_parse_date(d),
                etf_code=_clean(row.get("etf_code")),
                etf_name=_clean(row.get("etf_name")) or None,
                category=_clean(row.get("category")) or None,
                related_index=_clean(row.get("related_index")) or None,
                net_inflow=float(_clean(row.get("net_inflow")) or 0) if _clean(row.get("net_inflow")) else None,
                shares_change=float(_clean(row.get("shares_change"))) if _clean(row.get("shares_change")) else None,
                note=_clean(row.get("note")) or None,
                source="import",
            )
        except ValueError as e:
            raise ValueError(f"第 {i} 行解析失败: {e}") from e
        out.append(rec)
    return out


# ============================ 自动化接入接口（预留）============================
class AutoEventSource(Protocol):
    """未来自动抓取事件源的接口。实现后注册到 _AUTO_EVENT_SOURCES。"""

    name: str

    def pull(self, start: date, end: date) -> List[dict]:  # 返回同 parse_events_csv 的 dict 列表
        ...


class AutoEtfSource(Protocol):
    name: str

    def pull(self, start: date, end: date) -> List[dict]:
        ...


_AUTO_EVENT_SOURCES: Dict[str, AutoEventSource] = {}
_AUTO_ETF_SOURCES: Dict[str, AutoEtfSource] = {}


def register_auto_event_source(src: AutoEventSource) -> None:
    _AUTO_EVENT_SOURCES[src.name] = src
    log.info("注册自动事件源: %s", src.name)


def register_auto_etf_source(src: AutoEtfSource) -> None:
    _AUTO_ETF_SOURCES[src.name] = src
    log.info("注册自动ETF源: %s", src.name)


def list_auto_sources() -> Dict[str, List[str]]:
    return {
        "event": list(_AUTO_EVENT_SOURCES.keys()),
        "etf": list(_AUTO_ETF_SOURCES.keys()),
    }