"""
新闻公告数据源。

支持三种数据来源：
1. AkShare stock_news_em（东方财富个股新闻）
2. MariaDB xwlb_daily_ext（新闻联播分割数据）
3. MCP trendradar-news（外部新闻聚合服务）

统一返回格式：DataFrame (date, title, content, source, url)
"""

import json
import os
import time
from datetime import datetime, timedelta

import akshare as ak
import pandas as pd
import requests

# 确保 .env 已加载
import config.settings  # noqa: F401


class NewsSource:
    """
    新闻数据源聚合器。

    参数:
        use_akshare: 启用 AkShare 新闻接口
        use_xwlb: 启用新闻联播数据库
        use_mcp: 启用 MCP 新闻服务
        mcp_url: MCP 服务地址
        request_delay: 请求间隔（秒），避免被限频
    """

    def __init__(
        self,
        use_akshare: bool = True,
        use_xwlb: bool = True,
        use_mcp: bool = True,
        mcp_url: str | None = None,
        request_delay: float = 1.0,
    ):
        self.use_akshare = use_akshare
        self.use_xwlb = use_xwlb
        self.use_mcp = use_mcp
        self.mcp_url = mcp_url or os.getenv("NEWS_MCP_URL", "http://192.168.1.160:3333/mcp")
        self.request_delay = request_delay
        self._mcp_session_id: str | None = None

    # ── 统一入口 ──────────────────────────────────────────

    def fetch(
        self,
        ts_code: str,
        start: str | None = None,
        end: str | None = None,
        max_news: int | None = None,
    ) -> pd.DataFrame:
        """
        获取个股相关新闻（聚合多源）。

        参数:
            ts_code: 如 '000001.SZ'
            start: 起始日期 'YYYYMMDD'，默认 30 天前
            end: 结束日期 'YYYYMMDD'，默认今天
            max_news: 最多返回条数，默认从 .env SENTIMENT_MAX_NEWS_PER_STOCK 读取，fallback 30

        返回:
            DataFrame (date, title, content, source, url)
        """
        if max_news is None:
            max_news = int(os.getenv("SENTIMENT_MAX_NEWS_PER_STOCK", "30"))

        if end is None:
            end = time.strftime("%Y%m%d")
        if start is None:
            start = (datetime.strptime(end, "%Y%m%d") - timedelta(days=30)).strftime("%Y%m%d")

        frames = []

        if self.use_akshare:
            try:
                df = self._fetch_akshare(ts_code, start, end)
                if not df.empty:
                    frames.append(df)
                    time.sleep(self.request_delay)
            except Exception as e:
                print("  [WARN] AkShare 新闻获取失败 ({}): {}".format(ts_code, e))

        if self.use_xwlb:
            try:
                # xwlb: news_date 是播出日，+1day = 交易日
                # 所以查询时 start/end 各减 1 天
                xwlb_start = (datetime.strptime(start, "%Y%m%d") - timedelta(days=1)).strftime("%Y%m%d")
                xwlb_end = (datetime.strptime(end, "%Y%m%d") - timedelta(days=1)).strftime("%Y%m%d")
                df = self._fetch_xwlb(xwlb_start, xwlb_end)
                if not df.empty:
                    frames.append(df)
            except Exception as e:
                print("  [WARN] xwlb 新闻获取失败: {}".format(e))

        if self.use_mcp:
            try:
                df = self._fetch_mcp(ts_code, start, end)
                if not df.empty:
                    frames.append(df)
            except Exception as e:
                print("  [WARN] MCP 新闻获取失败: {}".format(e))

        if not frames:
            return pd.DataFrame(columns=["date", "title", "content", "source", "url"])

        result = pd.concat(frames, ignore_index=True)
        result = result.drop_duplicates(subset=["title", "date"])
        result = result.sort_values("date", ascending=False)

        if len(result) > max_news:
            result = result.head(max_news)

        return result.reset_index(drop=True)

    # ── AkShare 数据源 ────────────────────────────────────

    def _fetch_akshare(self, ts_code: str, start: str = "", end: str = "") -> pd.DataFrame:
        """
        东方财富个股新闻。

        stock_news_em 返回最新约 10 条（无日期筛选），
        如需更多应使用 InfoMoney 或自建爬虫。
        """
        symbol = ts_code.replace(".SZ", "").replace(".SH", "").replace(".BJ", "")
        try:
            df = ak.stock_news_em(symbol=symbol.zfill(6))
        except Exception:
            return pd.DataFrame()

        if df is None or df.empty:
            return pd.DataFrame()

        df = df.rename(columns={
            "新闻标题": "title",
            "新闻内容": "content",
            "发布时间": "date",
            "文章来源": "source",
            "新闻链接": "url",
        })
        df["date"] = pd.to_datetime(df["date"]).dt.strftime("%Y%m%d")
        df["source"] = "akshare_" + df["source"].fillna("unknown")

        # 日期筛选
        if start:
            df = df[df["date"] >= start]
        if end:
            df = df[df["date"] <= end]

        return df[["date", "title", "content", "source", "url"]]

    # ── 新闻联播数据源 ────────────────────────────────────

    def _fetch_xwlb(self, start: str, end: str) -> pd.DataFrame:
        """
        从 MariaDB xwlb_daily_ext 表获取新闻联播分割数据。

        news_date 是播出日期，+1day 后成为对市场产生影响的交易日。
        调用方应传入 (target_start-1, target_end-1)。

        连接失败时 database/connection.py 自动重建 SSH 隧道。
        """
        from database.connection import get_engine

        start_fmt = "{}-{}-{}".format(start[:4], start[4:6], start[6:8])
        end_fmt = "{}-{}-{}".format(end[:4], end[4:6], end[6:8])

        day_span = (datetime.strptime(end, "%Y%m%d") - datetime.strptime(start, "%Y%m%d")).days
        row_limit = max(100, (day_span + 1) * 40)

        sql = """
            SELECT news_date, news_title, news_content
            FROM xwlb_daily_ext
            WHERE news_date BETWEEN %(start)s AND %(end)s
            ORDER BY news_date DESC
            LIMIT {}
        """.format(row_limit)

        try:
            df = pd.read_sql(
                sql, get_engine(),
                params={"start": start_fmt, "end": end_fmt},
            )
            if df.empty:
                return pd.DataFrame()

            df = df.rename(columns={
                "news_date": "date",
                "news_title": "title",
                "news_content": "content",
            })
            df["date"] = (
                pd.to_datetime(df["date"]) + pd.Timedelta(days=1)
            ).dt.strftime("%Y%m%d")
            df["source"] = "xwlb"
            df["url"] = ""
            return df[["date", "title", "content", "source", "url"]]

        except Exception:
            return pd.DataFrame()

    # ── MCP 数据源 ────────────────────────────────────────

    def _fetch_mcp(self, ts_code: str, start: str, end: str) -> pd.DataFrame:
        """通过 MCP streamable HTTP 调用 trendradar-news 服务。"""
        session_id = self._mcp_initialize()
        if not session_id:
            return pd.DataFrame()

        symbol = ts_code.replace(".SZ", "").replace(".SH", "").replace(".BJ", "")
        try:
            result = self._mcp_call_tool(
                session_id,
                "search_news",
                {"keyword": symbol, "start_date": start, "end_date": end, "limit": 30},
            )
            if not result:
                return pd.DataFrame()

            records = self._parse_mcp_result(result)
            if not records:
                return pd.DataFrame()

            df = pd.DataFrame(records)
            df["source"] = "mcp_trendradar"
            return df[["date", "title", "content", "source", "url"]]
        except Exception:
            return pd.DataFrame()

    def _mcp_initialize(self) -> str | None:
        """MCP streamable HTTP 初始化，获取 session ID。"""
        if self._mcp_session_id:
            return self._mcp_session_id
        try:
            resp = requests.post(
                self.mcp_url,
                json={
                    "jsonrpc": "2.0",
                    "method": "initialize",
                    "params": {
                        "protocolVersion": "2024-11-05",
                        "capabilities": {},
                        "clientInfo": {"name": "cc-cursor", "version": "1.0"},
                    },
                    "id": 1,
                },
                headers={
                    "Content-Type": "application/json",
                    "Accept": "application/json, text/event-stream",
                },
                timeout=10,
            )
            # headers 是大小写不敏感的，尝试多种格式
            session_id = (
                resp.headers.get("mcp-session-id")
                or resp.headers.get("Mcp-Session-Id")
                or resp.headers.get("MCP-Session-Id")
            )
            if session_id:
                self._mcp_session_id = session_id
            else:
                print("  [WARN] MCP initialize 未返回 session-id")
            return session_id
        except Exception as e:
            print("  [WARN] MCP 连接失败: {}".format(e))
            return None

    def _mcp_call_tool(
        self, session_id: str, tool_name: str, arguments: dict
    ) -> dict | None:
        """调用 MCP 工具。"""
        try:
            resp = requests.post(
                self.mcp_url,
                json={
                    "jsonrpc": "2.0",
                    "method": "tools/call",
                    "params": {"name": tool_name, "arguments": arguments},
                    "id": 2,
                },
                headers={
                    "Content-Type": "application/json",
                    "Accept": "application/json, text/event-stream",
                    "mcp-session-id": session_id,
                },
                timeout=15,
            )
            # 解析 SSE 响应
            for line in resp.text.split("\n"):
                if line.startswith("data:"):
                    data = json.loads(line[5:].strip())
                    return data.get("result", {})
            return None
        except Exception:
            return None

    @staticmethod
    def _parse_mcp_result(result: dict) -> list[dict]:
        """解析 MCP 返回内容为统一格式。"""
        records = []
        content = result.get("content", [])
        for item in content:
            text = item.get("text", "") if isinstance(item, dict) else str(item)
            if text:
                try:
                    parsed = json.loads(text)
                    if isinstance(parsed, list):
                        for p in parsed:
                            records.append({
                                "date": str(p.get("date", p.get("publish_date", ""))),
                                "title": str(p.get("title", "")),
                                "content": str(p.get("content", p.get("summary", ""))),
                                "url": str(p.get("url", "")),
                            })
                    elif isinstance(parsed, dict):
                        records.append({
                            "date": str(parsed.get("date", parsed.get("publish_date", ""))),
                            "title": str(parsed.get("title", "")),
                            "content": str(parsed.get("content", parsed.get("summary", ""))),
                            "url": str(parsed.get("url", "")),
                        })
                except (json.JSONDecodeError, TypeError):
                    records.append({
                        "date": time.strftime("%Y%m%d"),
                        "title": text[:100],
                        "content": text,
                        "url": "",
                    })
        return records


# ── 日期对齐工具 ──────────────────────────────────────────

def align_news_to_trading_days(
    news_df: pd.DataFrame,
    trading_calendar: pd.DatetimeIndex,
) -> pd.DataFrame:
    """
    将新闻日期对齐到最近交易日。

    非交易日的新闻归入最近的下一个交易日。
    例：周六新闻 → 下周一交易日的因子值中体现。

    参数:
        news_df: 新闻 DataFrame，含 'date' 列 (YYYYMMDD)
        trading_calendar: 交易日 DatetimeIndex

    返回:
        news_df，'date' 列已替换为对齐后的交易日
    """
    calendar = trading_calendar.sort_values()

    news_dates = pd.to_datetime(news_df["date"], format="%Y%m%d")
    aligned = []

    for nd in news_dates:
        future_dates = calendar[calendar >= nd]
        if len(future_dates) > 0:
            aligned.append(future_dates[0])
        else:
            aligned.append(calendar[-1])

    news_df = news_df.copy()
    news_df["date"] = pd.DatetimeIndex(aligned).strftime("%Y%m%d")
    return news_df