#!/usr/bin/env python3
"""带 no-cache 头的看板静态服务器（替代 python -m http.server 8080）。

用法：
  kill 62754 && python3 serve_dashboard.py
（或：pkill -f "http.server 8080" && python3 serve_dashboard.py）

对所有响应发送 Cache-Control: no-store，浏览器每次都拉最新文件，
彻底避免"改了代码但浏览器一直显示旧版"的问题。

同时提供埋点接收 API：
  POST /api/track       — 接收前端埋点事件
  GET  /api/analytics   — 获取汇总分析数据（供后台管理页面使用）
  GET  /api/events      — 获取原始事件列表（分页）

用户系统 API：
  POST /api/user/login  — 登录/注册（{username} → 返回用户数据）
  GET  /api/user/data    — 获取用户数据（?username=xxx）
  POST /api/user/sync    — 同步用户数据（{username, watchlist, paperAccount, paperPortfolios, layout}）
"""
import http.server
import socketserver
import os
import json
import time
import gzip
import io
import threading
import re
import socket
import ssl
import subprocess
import urllib.request
import urllib.parse
from urllib.parse import urlparse, parse_qs
from datetime import datetime

# 跳过 SSL 证书验证（东财 API 证书链问题）
_SSL_CTX = ssl.create_default_context()
_SSL_CTX.check_hostname = False
_SSL_CTX.verify_mode = ssl.CERT_NONE

PORT = int(os.environ.get("PORT", "8080"))
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, "data")
EVENTS_FILE = os.path.join(DATA_DIR, "analytics_events.jsonl")

os.chdir(BASE_DIR)
os.makedirs(DATA_DIR, exist_ok=True)


# ===== 智能助理 LLM 代理 =====
# 配置来源：llm_config.json；环境变量 LLM_BASE_URL / LLM_API_KEY / LLM_MODEL 优先覆盖（推荐用环境变量注入密钥）
_LLM_CACHE = None  # 每次请求实时读取，改配置无需重启


def _load_llm_config():
    global _LLM_CACHE
    cfg = {}
    cfg_file = os.path.join(BASE_DIR, "llm_config.json")
    if os.path.exists(cfg_file):
        try:
            with open(cfg_file, "r", encoding="utf-8") as f:
                cfg = json.load(f)
        except Exception:
            cfg = {}
    # 环境变量优先覆盖（推荐用环境变量注入密钥，避免在项目文件留明文）
    for k, ek in (("base_url", "LLM_BASE_URL"), ("api_key", "LLM_API_KEY"), ("model", "LLM_MODEL")):
        v = os.environ.get(ek)
        if v:
            cfg[k] = v.strip()
    _LLM_CACHE = cfg
    return cfg


def _strip_html(s):
    return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", s or "")).strip()


def _fetch_em_flashes(limit=20):
    """东财实时快讯：当日市场新闻流（带时间戳），无需 API key，mac mini 直连可达。
    返回 [(标题, 时间, '东财快讯'), ...]"""
    try:
        url = "https://newsapi.eastmoney.com/kuaixun/v1/getlist_101_ajaxResult_50_1_.html"
        req = urllib.request.Request(url, headers={
            "User-Agent": ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
                           "(KHTML, like Gecko) Chrome/120 Safari/537.36"),
            "Referer": "https://kuaixun.eastmoney.com/",
            "Accept-Encoding": "identity",
        })
        with urllib.request.urlopen(req, timeout=6, context=_SSL_CTX) as r:
            body = r.read(300000).decode("utf-8", "ignore")
        m = re.search(r"var ajaxResult=(\{.*\})\s*;?\s*$", body, re.S)
        if not m:
            return []
        data = json.loads(m.group(1))
        out = []
        for it in (data.get("LivesList") or [])[:limit]:
            title = _strip_html(str(it.get("title") or ""))
            tm = _strip_html(str(it.get("showtime") or ""))
            if title:
                out.append((title, tm, "东财快讯"))
        return out
    except Exception:
        return []


def _bing_search(query, limit=7):
    """Bing 搜索页解析（国内可达、无需 key）。返回 [(标题, 摘要, 链接), ...]"""
    q = urllib.parse.quote((query or "").strip()[:40])
    ua = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
          "(KHTML, like Gecko) Chrome/120 Safari/537.36")
    sources = [
        "https://cn.bing.com/search?q=%s&setlang=zh-hans&MKT=zh-CN&count=%d" % (q, limit + 5),
        "https://www.bing.com/search?q=%s&setlang=zh-hans&MKT=zh-CN&count=%d" % (q, limit + 5),
    ]
    items = []
    for url in sources:
        try:
            req = urllib.request.Request(url, headers={
                "User-Agent": ua,
                "Accept": "*/*",
                "Accept-Encoding": "identity",
                "Accept-Language": "zh-CN,zh;q=0.9",
            })
            with urllib.request.urlopen(req, timeout=8, context=_SSL_CTX) as r:
                html = r.read(1200000).decode("utf-8", "ignore")
            for m in re.finditer(r'<li class="b_algo"(.*?)</li>', html, re.S):
                blk = m.group(1)
                h = re.search(r'<h2[^>]*>\s*<a[^>]*href=["\'](https?://[^"\']+)["\'][^>]*>(.*?)</a>', blk, re.S)
                p = re.search(r'<p(?: [^>]*)?>(.*?)</p>', blk, re.S)
                if h:
                    title = _strip_html(h.group(2))
                    link = h.group(1)
                    snippet = _strip_html(p.group(1)) if p else ""
                    if title and title not in [x[0] for x in items]:
                        items.append((title, snippet, link))
                if len(items) >= limit:
                    break
            if items:
                break
        except Exception:
            continue
    return items


def _fetch_web(query, limit=7):
    """联网检索编排：金融类查询优先东财实时快讯（按关键词过滤），Bing 搜索补充；
    通用查询仅 Bing。全部失败返回空列表，调用方优雅降级。返回 [(标题, 摘要, 来源), ...]"""
    query = (query or "").strip()
    items = []
    if _FIN_HINT.search(query):
        broad = ("股票", "最新", "A股", "大盘", "行情", "盘面", "资讯", "新闻", "大新闻",
                 "消息", "热点", "事件", "动态", "涨了", "跌了", "出了")
        toks = [t for t in re.split(r"\s+", query) if len(t) >= 2 and t not in broad]
        flashes = _fetch_em_flashes(30)
        if flashes:
            hits = [f for f in flashes if any(t in f[0] for t in toks)] if toks else flashes
            if hits or not toks:
                items.extend(hits[:limit])
    if len(items) < limit:
        for t, d, l in _bing_search(query, limit):
            if t not in [x[0] for x in items]:
                items.append((t, d, l))
            if len(items) >= limit:
                break
    return items


# 需要联网检索的时效性/行情类关键词（通用知识问答不搜，避免注入无关噪音干扰模型）
_WEB_TRIGGER = re.compile(
    r"最新|新闻|消息|今天|今日|现在|目前|当前|最近|刚刚|昨日|昨天|本周|上周|本月|今年|"
    r"股价|涨了|跌了|涨停|跌停|大盘|盘面|行情|板块|概念|龙头|政策|公告|央行|美联储|证监会|"
    r"发布会|利率|降息|加息|关税|财报|季报|年报|IPO|回购|增持|减持|解禁|热点|出了|发生"
)


def _needs_web(q):
    return bool(_WEB_TRIGGER.search(q or ""))


# 口语填充词/疑问词/时间诱饵词（"今天"会命中黄历站）——按长度降序排列避免前缀截断
_Q_FILLER = re.compile(
    r"怎么样|是怎样|怎么样|怎样|怎么|如何|为什么|为啥|有什么|有没有|是不是|哪些|哪个|什么|多少|"
    r"请问|帮我|帮忙|给我|讲讲|看看|说说|聊聊|告诉我|问下|一下|请|"
    r"今天|今日|现在|目前|当前|最近|刚刚|昨天|昨日|吗|呢|啊|吧|啦|哦|呀"
)
_TIME_HINT = re.compile(r"今天|今日|现在|目前|当前|最近|刚刚|昨天|昨日|本周|上周|本月|今年|最新|新闻|消息")
_FIN_HINT = re.compile(r"股|大盘|盘面|行情|板块|涨停|跌停|指数|基金|债券|期货|牛市|熊市")


def _web_query(q):
    """把口语化问题清洗成搜索关键词：去填充词/疑问词/时间诱饵，时间词转'最新'"""
    s = re.sub(r"[?？!！。,，、;；:\s]+", " ", (q or ""))
    cleaned = re.sub(r"\s+", " ", _Q_FILLER.sub(" ", s)).strip()
    if not cleaned:
        cleaned = s.strip()
    if not cleaned:
        return ""
    if _TIME_HINT.search(s):
        cleaned += " 最新"
    return cleaned[:40]


def _chat_completion(messages, cfg):
    """OpenAI 兼容接口调用。返回 (answer, error)"""
    base_url = (cfg.get("base_url") or "").strip().rstrip("/")
    api_key = (cfg.get("api_key") or "").strip()
    model = (cfg.get("model") or "").strip()
    if not base_url or not api_key:
        return None, ("未配置大模型。请在 llm_config.json 填写 {base_url, api_key, model}（OpenAI 兼容接口，"
                      "如 DeepSeek / 通义 / Kimi 等），或设置环境变量 LLM_BASE_URL / LLM_API_KEY / LLM_MODEL。")
    if not model:
        model = "gpt-4o-mini"
    url = base_url if base_url.endswith("/chat/completions") else base_url + "/chat/completions"
    payload = {
        "model": model,
        "messages": messages,
        "temperature": 0.6,
        "max_tokens": 2500,
    }
    body = json.dumps(payload).encode("utf-8")
    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + api_key,
    }
    try:
        req = urllib.request.Request(url, data=body, headers=headers, method="POST")
        with urllib.request.urlopen(req, timeout=120, context=_SSL_CTX) as resp:
            data = json.loads(resp.read().decode("utf-8", errors="replace"))
        try:
            return data["choices"][0]["message"]["content"], None
        except Exception:
            return None, "模型返回异常: " + json.dumps(data, ensure_ascii=False)[:500]
    except urllib.error.HTTPError as e:
        try:
            detail = json.loads(e.read().decode("utf-8", errors="replace"))
            msg = json.dumps(detail, ensure_ascii=False)[:500]
        except Exception:
            msg = str(e)
        return None, "模型接口返回错误 %s: %s" % (e.code, msg)
    except Exception as e:
        return None, "请求模型失败: %s" % e


# ===== 公告解读引擎 =====
_ANN_CACHE = {}  # code → {ts, data}


def _fetch_url(url, timeout=10):
    """发起 HTTP 请求"""
    try:
        req = urllib.request.Request(url, headers={
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
            "Accept": "application/json, text/plain, */*",
            "Referer": "https://data.eastmoney.com/",
        })
        with urllib.request.urlopen(req, timeout=timeout, context=_SSL_CTX) as resp:
            return resp.read().decode("utf-8", errors="replace")
    except Exception:
        return None


# ===== 财务数据 API =====
_FIN_CACHE = {}  # code → {ts, data}


def _fetch_financial_data(code):
    """从东财财务数据 API 拉取完整资产负债表+利润表"""
    cache = _FIN_CACHE.get(code)
    if cache and time.time() - cache["ts"] < 300:
        return cache["data"]

    # 确定市场前缀
    if code.startswith("6"):
        mkt = "SH"
    else:
        mkt = "SZ"

    result = {"balance_sheet": {}, "income_statement": {}}

    # 资产负债表
    bs_url = (
        "https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/zcfzbAjaxNew"
        "?companyType=4&reportDateType=0&reportType=1&dates=2026-06-30,2025-12-31,2025-06-30&code=" + mkt + code
    )
    bs_raw = _fetch_url(bs_url, timeout=10)
    if bs_raw:
        try:
            bs_data = json.loads(bs_raw)
            rows = bs_data.get("data", [])
            if rows:
                # 找到最新报告期
                latest = None
                prev_year = None
                for r in rows:
                    rd = r.get("REPORT_DATE", "")
                    if "2026-06-30" in rd:
                        latest = r
                    elif "2025-06-30" in rd:
                        prev_year = r
                if latest:
                    result["balance_sheet"] = {
                        "current": latest,
                        "prev_year": prev_year or {},
                    }
        except Exception:
            pass

    # 利润表
    is_url = (
        "https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/lrbAjaxNew"
        "?companyType=4&reportDateType=0&reportType=1&dates=2026-06-30,2025-06-30&code=" + mkt + code
    )
    is_raw = _fetch_url(is_url, timeout=10)
    if is_raw:
        try:
            is_data = json.loads(is_raw)
            rows = is_data.get("data", [])
            if rows:
                latest = None
                prev_year = None
                for r in rows:
                    rd = r.get("REPORT_DATE", "")
                    if "2026-06-30" in rd:
                        latest = r
                    elif "2025-06-30" in rd:
                        prev_year = r
                if latest:
                    result["income_statement"] = {
                        "current": latest,
                        "prev_year": prev_year or {},
                    }
        except Exception:
            pass

    _FIN_CACHE[code] = {"ts": time.time(), "data": result}
    return result


def _deep_analyze_financials(code, ann_text=""):
    """深度财务分析：从东财 API 获取完整财务数据，提取亮点和风险点"""
    fin = _fetch_financial_data(code)
    bs = fin.get("balance_sheet", {})
    inc = fin.get("income_statement", {})

    if not bs.get("current") and not inc.get("current"):
        return None

    bs_curr = bs.get("current", {})
    bs_prev = bs.get("prev_year", {})
    inc_curr = inc.get("current", {})
    inc_prev = inc.get("prev_year", {})

    highlights = []
    risks = []

    def _to_yi(val):
        if val is None or val == "":
            return None
        try:
            return round(float(val) / 1e8, 4)
        except Exception:
            return None

    def _fmt(val):
        """格式化亿元数值为可读字符串"""
        if val is None:
            return "--"
        if val >= 1:
            return f"{val:.2f}亿元"
        elif val > 0:
            return f"{val*10000:.0f}万元"
        return "0"

    def _yoy_str(yoy):
        if yoy is None:
            return ""
        if yoy > 0:
            return f"同比+{yoy:.1f}%"
        else:
            return f"同比{yoy:.1f}%"

    # ===== 利润表分析 =====
    if inc_curr:
        revenue = _to_yi(inc_curr.get("OPERATE_INCOME"))
        revenue_yoy = inc_curr.get("OPERATE_INCOME_YOY")
        cost = _to_yi(inc_curr.get("OPERATE_COST"))
        cost_yoy = inc_curr.get("OPERATE_COST_YOY")
        net_profit = _to_yi(inc_curr.get("PARENT_NETPROFIT"))
        profit_yoy = inc_curr.get("PARENT_NETPROFIT_YOY")
        deduct_profit = _to_yi(inc_curr.get("DEDUCT_PARENT_NETPROFIT"))
        deduct_yoy = inc_curr.get("DEDUCT_PARENT_NETPROFIT_YOY")
        op_profit = _to_yi(inc_curr.get("OPERATE_PROFIT"))
        op_profit_yoy = inc_curr.get("OPERATE_PROFIT_YOY")
        rd_expense = _to_yi(inc_curr.get("RESEARCH_EXPENSE"))
        rd_yoy = inc_curr.get("RESEARCH_EXPENSE_YOY")
        sale_expense = _to_yi(inc_curr.get("SALE_EXPENSE"))
        sale_yoy = inc_curr.get("SALE_EXPENSE_YOY")
        manage_expense = _to_yi(inc_curr.get("MANAGE_EXPENSE"))
        manage_yoy = inc_curr.get("MANAGE_EXPENSE_YOY")
        finance_expense = _to_yi(inc_curr.get("FINANCE_EXPENSE"))
        finance_yoy = inc_curr.get("FINANCE_EXPENSE_YOY")

        # 1. 营收增长
        if revenue and revenue_yoy is not None:
            if revenue_yoy > 20:
                highlights.append({
                    "category": "营收高增长",
                    "description": f"营业收入{_fmt(revenue)}，{_yoy_str(revenue_yoy)}",
                    "insight": f"营收增速{revenue_yoy:.1f}%表现亮眼，反映市场需求旺盛或公司份额提升，关注增长驱动因素是否可持续",
                })
            elif revenue_yoy < -10:
                risks.append({
                    "category": "营收下滑",
                    "description": f"营业收入{_fmt(revenue)}，{_yoy_str(revenue_yoy)}",
                    "insight": f"营收同比下降{abs(revenue_yoy):.1f}%，需关注是行业周期下行还是公司竞争力减弱",
                })

        # 2. 成本增速超过营收
        if revenue_yoy and cost_yoy and cost_yoy > revenue_yoy + 5:
            risks.append({
                "category": "成本压力",
                "description": f"营业成本增速({cost_yoy:.1f}%)超过营收增速({revenue_yoy:.1f}%)",
                "insight": "成本增速快于营收，毛利率承压，需关注原材料价格、产能利用率等因素",
            })

        # 3. 净利润大幅下滑
        if profit_yoy is not None and profit_yoy < -30:
            risks.append({
                "category": "利润大幅下滑",
                "description": f"归母净利润{_fmt(net_profit)}，{_yoy_str(profit_yoy)}",
                "insight": f"净利润同比{profit_yoy:.1f}%大幅下滑，需深挖原因：是成本上升、费用增加、还是毛利率下降",
            })

        # 4. 扣非净利润 vs 净利润差异大
        if net_profit and deduct_profit and net_profit > 0:
            non_recurring = net_profit - deduct_profit
            ratio = non_recurring / net_profit * 100 if net_profit else 0
            if ratio > 30:
                risks.append({
                    "category": "利润质量",
                    "description": f"非经常性损益占净利润{ratio:.0f}%（扣非净利{_fmt(deduct_profit)} vs 净利润{_fmt(net_profit)}）",
                    "insight": "净利润中非经常性损益占比较高，核心盈利能力弱于表面数字，需关注扣非后的真实盈利水平",
                })

        # 5. 研发投入大幅增长
        if rd_expense and rd_yoy and rd_yoy > 20:
            highlights.append({
                "category": "研发加码",
                "description": f"研发费用{_fmt(rd_expense)}，{_yoy_str(rd_yoy)}",
                "insight": f"研发投入大幅增长{rd_yoy:.1f}%，占营收比{rd_expense/revenue*100:.1f}%{'，高研发投入有望转化为技术壁垒和产品竞争力' if revenue else ''}",
            })

        # 6. 管理费用大增
        if manage_expense and manage_yoy and manage_yoy > 50:
            risks.append({
                "category": "管理费用激增",
                "description": f"管理费用{_fmt(manage_expense)}，{_yoy_str(manage_yoy)}",
                "insight": f"管理费用同比激增{manage_yoy:.1f}%，侵蚀利润，需关注是股权激励、人员扩张还是一次性费用",
            })

        # 7. 财务费用大增
        if finance_expense and finance_yoy and finance_yoy > 100:
            risks.append({
                "category": "财务费用飙升",
                "description": f"财务费用{_fmt(finance_expense)}，{_yoy_str(finance_yoy)}",
                "insight": f"财务费用同比飙升{finance_yoy:.1f}%，反映借款增加或利率上升，需关注负债结构和偿债压力",
            })

        # 8. 营收增长但利润下滑（增收不增利）
        if revenue_yoy and profit_yoy and revenue_yoy > 5 and profit_yoy < 0:
            risks.append({
                "category": "增收不增利",
                "description": f"营收增长{revenue_yoy:.1f}%但净利润下降{abs(profit_yoy):.1f}%",
                "insight": "增收不增利，利润被成本或费用侵蚀，需关注毛利率变化和费用率趋势",
            })

        # 9. 经营利润大幅下滑
        if op_profit_yoy and op_profit_yoy < -50:
            risks.append({
                "category": "经营利润萎缩",
                "description": f"营业利润{_fmt(op_profit)}，{_yoy_str(op_profit_yoy)}",
                "insight": f"营业利润同比{op_profit_yoy:.1f}%大幅萎缩，主营盈利能力恶化",
            })

    # ===== 资产负债表分析 =====
    if bs_curr:
        inventory = _to_yi(bs_curr.get("INVENTORY"))
        inventory_yoy = bs_curr.get("INVENTORY_YOY")
        cip = _to_yi(bs_curr.get("CONSTRUCT_IN_PROGRESS"))
        cip_yoy = bs_curr.get("CONSTRUCT_IN_PROGRESS_YOY")
        accounts_rece = _to_yi(bs_curr.get("ACCOUNTS_RECE"))
        accounts_rece_yoy = bs_curr.get("ACCOUNTS_RECE_YOY")
        contract_liab = _to_yi(bs_curr.get("CONTRACT_LIAB"))
        contract_liab_yoy = bs_curr.get("CONTRACT_LIAB_YOY")
        monetary = _to_yi(bs_curr.get("MONETARYFUNDS"))
        monetary_yoy = bs_curr.get("MONETARYFUNDS_YOY")
        short_loan = _to_yi(bs_curr.get("SHORT_LOAN"))
        long_loan = _to_yi(bs_curr.get("LONG_LOAN"))
        prepayment = _to_yi(bs_curr.get("PREPAYMENT"))
        prepayment_yoy = bs_curr.get("PREPAYMENT_YOY")
        accounts_payable = _to_yi(bs_curr.get("ACCOUNTS_PAYABLE"))
        accounts_payable_yoy = bs_curr.get("ACCOUNTS_PAYABLE_YOY")
        fixed_asset = _to_yi(bs_curr.get("FIXED_ASSET"))
        fixed_asset_yoy = bs_curr.get("FIXED_ASSET_YOY")
        total_assets = _to_yi(bs_curr.get("TOTAL_ASSETS"))
        total_assets_yoy = bs_curr.get("TOTAL_ASSETS_YOY")
        total_liab = _to_yi(bs_curr.get("TOTAL_LIABILITIES"))
        total_liab_yoy = bs_curr.get("TOTAL_LIABILITIES_YOY")
        total_equity = _to_yi(bs_curr.get("TOTAL_EQUITY"))
        total_equity_yoy = bs_curr.get("TOTAL_EQUITY_YOY")

        # 10. 存货大幅增长
        if inventory_yoy and abs(inventory_yoy) >= 20:
            if inventory_yoy > 0:
                # 检查是否有"以销定产"等关键词
                has_keyword = any(k in ann_text for k in ["以销定产", "备货", "订单充足", "需求旺盛"])
                if has_keyword:
                    highlights.append({
                        "category": "存货增长+以销定产",
                        "description": f"存货{_fmt(inventory)}，同比+{inventory_yoy:.1f}%，公司提及「以销定产/备货」策略",
                        "insight": f"存货大增{inventory_yoy:.1f}%但公司强调以销定产，说明是因客户需求旺盛而主动备货，非库存积压。结合在建工程情况，产能扩张+备货双重信号指向下游需求强劲",
                    })
                else:
                    risks.append({
                        "category": "存货激增",
                        "description": f"存货{_fmt(inventory)}，同比+{inventory_yoy:.1f}%",
                        "insight": f"存货大幅增长{inventory_yoy:.1f}%，若营收未同步增长则可能存在库存积压风险，需结合存货周转率和营收增速综合判断",
                    })
            else:
                highlights.append({
                    "category": "存货优化",
                    "description": f"存货{_fmt(inventory)}，同比{inventory_yoy:.1f}%",
                    "insight": "存货下降，库存管理优化或销售加速，有利于资金周转",
                })

        # 11. 在建工程增长（扩产）
        if cip_yoy and cip_yoy > 20:
            highlights.append({
                "category": "产能扩张",
                "description": f"在建工程{_fmt(cip)}，同比+{cip_yoy:.1f}%",
                "insight": f"在建工程大增{cip_yoy:.1f}%，公司正在扩大产能。结合存货增长情况，如果存货同步增加说明下游有需求才扩产；新产能投放后有望带来业绩增量",
            })
        elif cip_yoy and cip_yoy < -30 and cip:
            highlights.append({
                "category": "产能转固",
                "description": f"在建工程{_fmt(cip)}，同比{cip_yoy:.1f}%",
                "insight": "在建工程减少，可能项目已完工转入固定资产，即将进入产能释放期",
            })

        # 12. 合同负债增长（订单预收）
        if contract_liab_yoy and abs(contract_liab_yoy) >= 20:
            if contract_liab_yoy > 0:
                highlights.append({
                    "category": "预收款大增",
                    "description": f"合同负债{_fmt(contract_liab)}，同比+{contract_liab_yoy:.1f}%",
                    "insight": f"合同负债大增{contract_liab_yoy:.1f}%，预收款反映客户订单充足，未来收入有保障，是业绩前瞻的积极信号",
                })
            else:
                risks.append({
                    "category": "预收款下降",
                    "description": f"合同负债{_fmt(contract_liab)}，同比{contract_liab_yoy:.1f}%",
                    "insight": "合同负债下降，预收款减少可能反映订单需求走弱",
                })

        # 13. 应收账款增长
        if accounts_rece_yoy and accounts_rece_yoy > 30:
            risks.append({
                "category": "应收账款增长",
                "description": f"应收账款{_fmt(accounts_rece)}，同比+{accounts_rece_yoy:.1f}%",
                "insight": "应收账款增速较快，需关注回款质量和客户信用风险，若增速超过营收增速则更需警惕",
            })

        # 14. 预付款项大增
        if prepayment_yoy and prepayment_yoy > 50:
            highlights.append({
                "category": "预付款项",
                "description": f"预付款项{_fmt(prepayment)}，同比+{prepayment_yoy:.1f}%",
                "insight": "预付款项大增，可能是为锁定上游原材料或产能，反映公司对后续订单的预期",
            })

        # 15. 应付账款大增
        if accounts_payable_yoy and accounts_payable_yoy > 50:
            highlights.append({
                "category": "议价能力",
                "description": f"应付账款{_fmt(accounts_payable)}，同比+{accounts_payable_yoy:.1f}%",
                "insight": "应付账款大增，公司对供应商的议价能力强，占用上游资金支持自身运营",
            })

        # 16. 固定资产大增
        if fixed_asset_yoy and fixed_asset_yoy > 30:
            highlights.append({
                "category": "固定资产扩张",
                "description": f"固定资产{_fmt(fixed_asset)}，同比+{fixed_asset_yoy:.1f}%",
                "insight": "固定资产大增，可能在建工程转固或新购设备，产能扩张进入收获期",
            })

        # 17. 货币资金充裕
        if monetary and monetary_yoy and monetary_yoy > 30:
            highlights.append({
                "category": "资金充裕",
                "description": f"货币资金{_fmt(monetary)}，同比+{monetary_yoy:.1f}%",
                "insight": "货币资金充裕，公司抗风险能力强，有充足资金支持扩张或并购",
            })

        # 18. 短期借款
        if short_loan and short_loan > 0:
            risks.append({
                "category": "短期借款",
                "description": f"短期借款{_fmt(short_loan)}",
                "insight": "存在短期借款，需关注偿债压力和财务费用对利润的影响",
            })

        # 19. 负债率
        if total_assets and total_liab:
            debt_ratio = total_liab / total_assets * 100 if total_assets else 0
            if debt_ratio > 60:
                risks.append({
                    "category": "负债率偏高",
                    "description": f"资产负债率{debt_ratio:.1f}%（总负债{_fmt(total_liab)} / 总资产{_fmt(total_assets)}）",
                    "insight": "资产负债率偏高，财务杠杆较大，需关注偿债能力和现金流覆盖",
                })

        # 20. 总资产大增
        if total_assets_yoy and total_assets_yoy > 30:
            highlights.append({
                "category": "规模扩张",
                "description": f"总资产{_fmt(total_assets)}，同比+{total_assets_yoy:.1f}%",
                "insight": "总资产大幅增长，公司处于快速扩张期，规模效应有望逐步显现",
            })

    # 检查公告正文中的关键词
    if ann_text:
        if "以销定产" in ann_text:
            highlights.append({
                "category": "经营策略",
                "description": "公司强调「以销定产」策略",
                "insight": "以销定产说明公司根据客户需求安排生产，存货增长有实际订单支撑，非盲目备货",
            })
        if "产能" in ann_text and ("扩张" in ann_text or "扩大" in ann_text or "新增" in ann_text):
            highlights.append({
                "category": "产能规划",
                "description": "公司提及产能扩张/扩大",
                "insight": "公司主动扩大产能，说明管理层对市场需求有信心，未来业绩有产能支撑",
            })
        if "客户集中" in ann_text or "前五名客户" in ann_text:
            risks.append({
                "category": "客户集中度",
                "description": "公司存在客户集中度较高的风险",
                "insight": "大客户依赖度高，若核心客户流失或需求下降，对公司业绩影响较大",
            })
        if "供应商集中" in ann_text or "前五名供应商" in ann_text:
            risks.append({
                "category": "供应商集中度",
                "description": "公司存在供应商集中度较高的风险",
                "insight": "上游供应商集中，原材料供应稳定性和价格谈判能力受限",
            })

    return {
        "highlights": highlights,
        "risks": risks,
        "balance_sheet": _build_bs_table(bs_curr) if bs_curr else None,
        "income_statement": _build_is_table(inc_curr) if inc_curr else None,
    }


def _build_bs_table(bs):
    """构建资产负债表对比"""
    items = [
        ("货币资金", "MONETARYFUNDS"),
        ("应收账款", "ACCOUNTS_RECE"),
        ("预付款项", "PREPAYMENT"),
        ("存货", "INVENTORY"),
        ("合同资产", "CONTRACT_ASSET"),
        ("固定资产", "FIXED_ASSET"),
        ("在建工程", "CONSTRUCT_IN_PROGRESS"),
        ("无形资产", "INTANGIBLE_ASSET"),
        ("商誉", "GOODWILL"),
        ("短期借款", "SHORT_LOAN"),
        ("应付账款", "ACCOUNTS_PAYABLE"),
        ("合同负债", "CONTRACT_LIAB"),
        ("长期借款", "LONG_LOAN"),
        ("总资产", "TOTAL_ASSETS"),
        ("总负债", "TOTAL_LIABILITIES"),
        ("股东权益", "TOTAL_EQUITY"),
    ]
    result = []
    for name, key in items:
        val = bs.get(key)
        yoy = bs.get(key + "_YOY")
        if val is not None and val != 0 and val != "":
            try:
                yi = round(float(val) / 1e8, 4)
                result.append({
                    "name": name,
                    "current": yi,
                    "yoy": yoy,
                })
            except Exception:
                pass
    return result if result else None


def _build_is_table(inc):
    """构建利润表对比"""
    items = [
        ("营业收入", "OPERATE_INCOME"),
        ("营业成本", "OPERATE_COST"),
        ("研发费用", "RESEARCH_EXPENSE"),
        ("销售费用", "SALE_EXPENSE"),
        ("管理费用", "MANAGE_EXPENSE"),
        ("财务费用", "FINANCE_EXPENSE"),
        ("营业利润", "OPERATE_PROFIT"),
        ("利润总额", "TOTAL_PROFIT"),
        ("净利润", "NETPROFIT"),
        ("归母净利润", "PARENT_NETPROFIT"),
        ("扣非净利润", "DEDUCT_PARENT_NETPROFIT"),
        ("基本每股收益", "BASIC_EPS"),
    ]
    result = []
    for name, key in items:
        val = inc.get(key)
        yoy = inc.get(key + "_YOY")
        if val is not None and val != 0 and val != "":
            try:
                if key == "BASIC_EPS":
                    result.append({"name": name, "current": float(val), "yoy": yoy, "unit": "元"})
                else:
                    yi = round(float(val) / 1e8, 4)
                    result.append({"name": name, "current": yi, "yoy": yoy, "unit": "亿"})
            except Exception:
                pass
    return result if result else None


def _fetch_announcements(code):
    """从东财拉取个股公告列表"""
    cache = _ANN_CACHE.get(code)
    if cache and time.time() - cache["ts"] < 300:
        return cache["data"]
    url = (
        "https://np-anotice-stock.eastmoney.com/api/security/ann"
        "?sr=-1&page_size=20&page_index=1&ann_type=A&client_source=web"
        "&stock_list=" + code + "&f_node=0&s_node=0"
    )
    raw = _fetch_url(url, timeout=10)
    if not raw:
        _ANN_CACHE[code] = {"ts": time.time(), "data": []}
        return []
    try:
        data = json.loads(raw)
    except Exception:
        # 有时返回带 callback 包裹
        m = re.search(r"\{.*\}", raw)
        if m:
            try:
                data = json.loads(m.group(0))
            except Exception:
                _ANN_CACHE[code] = {"ts": time.time(), "data": []}
                return []
        else:
            _ANN_CACHE[code] = {"ts": time.time(), "data": []}
            return []
    items = (data.get("data") or {}).get("list") or []
    if not items:
        _ANN_CACHE[code] = {"ts": time.time(), "data": []}
        return []
    result = []
    for idx, item in enumerate(items[:15]):
        title = item.get("title", "")
        ann_id = item.get("art_code", "")
        pub_date = (item.get("notice_date") or "")[:10]
        content_url = item.get("art_code", "")
        # 前15条拉取正文（确保覆盖完整财报）
        summary = ""
        if content_url and idx < 15:
            c_url = (
                "https://np-cnotice-stock.eastmoney.com/api/content/ann"
                "?art_code=" + content_url + "&client_source=web"
            )
            c_raw = _fetch_url(c_url, timeout=8)
            if c_raw:
                try:
                    c_data = json.loads(c_raw)
                    content_data = c_data.get("data", {})
                    attach_data = content_data.get("notice_content", "") or content_data.get("attach_data", "")
                    if attach_data:
                        text = re.sub(r"<[^>]+>", " ", attach_data)
                        text = re.sub(r"&nbsp;", " ", text)
                        text = re.sub(r"&amp;", "&", text)
                        text = re.sub(r"&#\d+;", "", text)
                        text = re.sub(r"\s+", " ", text).strip()
                        summary = text[:4000]
                except Exception:
                    pass
        result.append({
            "title": title,
            "date": pub_date,
            "type": _classify_announcement(title),
            "summary": summary,
            "art_code": ann_id,
        })
    _ANN_CACHE[code] = {"ts": time.time(), "data": result}
    return result


def _classify_announcement(title):
    """根据标题分类公告"""
    t = title or ""
    if any(k in t for k in ["年报", "季报", "半年报", "半年度", "一季报", "三季报", "业绩预告", "业绩快报", "业绩说明", "营收", "净利润", "归母", "年度报告", "季度报告"]):
        return "财报"
    if any(k in t for k in ["研报", "评级", "上调", "下调", "买入", "增持", "调高", "调低"]):
        return "研报"
    if any(k in t for k in ["重组", "并购", "收购", "合并", "分立", "借壳", "重大资产"]):
        return "重大事项"
    if any(k in t for k in ["增发", "减持", "增持", "回购", "股份变动", "股权激励", "限售", "解禁"]):
        return "股权变动"
    if any(k in t for k in ["分红", "派息", "送转", "转增", "利润分配"]):
        return "分红送转"
    if any(k in t for k in ["合同", "中标", "订单", "协议", "战略合作", "框架协议"]):
        return "经营合同"
    if any(k in t for k in ["担保", "质押", "冻结", "诉讼", "仲裁", "违规", "处罚", "监管", "警示"]):
        return "风险警示"
    if any(k in t for k in ["募资", "募投", "募集资金", "超募", "资金使用"]):
        return "募资动态"
    if any(k in t for k in ["董事会", "股东大会", "监事会", "决议", "人事", "辞职", "聘任", "离职"]):
        return "公司治理"
    return "其他"


def _analyze_announcement(ann, stock_code):
    """规则引擎：解读单条公告，提取关键点+分析判断"""
    title = ann.get("title", "")
    text = ann.get("summary", "")
    full = title + " " + text
    ann_type = ann.get("type", "其他")

    key_points = []
    sentiment = "neutral"  # bullish / bearish / neutral
    sentiment_score = 0
    analysis = ""
    impact_level = "low"  # high / medium / low

    # === 财报类 ===
    if ann_type == "财报":
        impact_level = "high"
        def _to_yi(num_str):
            """将数字（元）转为亿元"""
            n = float(num_str.replace(",", ""))
            return round(n / 1e8, 4)

        def _find_3col_analysis(name_pattern, txt):
            pattern = name_pattern + r"\s*(?:（[^）]*）)?\s*([\d,]+\.?\d*)\s+([\d,]+\.?\d*)\s+(-?[\d.]+)%?"
            m = re.search(pattern, txt)
            if m:
                return _to_yi(m.group(1)), float(m.group(3))
            return None, None

        # 营业收入
        rev, rev_chg = _find_3col_analysis(r"营业(?:总收入|收入)", text)
        # 净利润
        profit, profit_chg = _find_3col_analysis(r"归属于[^净]*净利润", text)
        if profit is None:
            profit, profit_chg = _find_3col_analysis(r"净利润", text)
        # 每股收益
        eps_m = re.search(r"(?:基本)?每股收益[^0-9]*([\d.]+)", text)
        eps = eps_m.group(1) if eps_m else None

        def _fmt_yi(val):
            """格式化亿元数值"""
            if val >= 1:
                return f"{val:.2f}亿元"
            else:
                return f"{val*10000:.2f}万元"

        if rev is not None:
            key_points.append(f"营业收入：{_fmt_yi(rev)}")
        if profit is not None:
            key_points.append(f"净利润：{_fmt_yi(profit)}")
        if rev_chg is not None:
            key_points.append(f"营收同比：{rev_chg:+.1f}%")
            sentiment_score += 1 if rev_chg > 0 else -1
        if profit_chg is not None:
            key_points.append(f"净利润同比：{profit_chg:+.1f}%")
            sentiment_score += 2 if profit_chg > 10 else (1 if profit_chg > 0 else (-2 if profit_chg < -10 else -1))
        if eps:
            key_points.append(f"每股收益：{eps}元")

        if "亏损" in full or "减亏" in full:
            key_points.append("本期存在亏损")
            sentiment_score -= 2
        if "扭亏" in full:
            key_points.append("本期实现扭亏为盈")
            sentiment_score += 3

        if sentiment_score >= 2:
            sentiment = "bullish"
            analysis = "财报业绩表现亮眼，营收和利润均实现增长，反映公司经营状况向好。关注后续业绩能否持续，以及市场对财报数据的反应。"
        elif sentiment_score <= -2:
            sentiment = "bearish"
            analysis = "财报业绩不及预期，营收或利润出现下滑，反映公司经营面临压力。需关注业绩下滑的原因是周期性还是结构性，以及对后续季度的传导影响。"
        else:
            analysis = "财报表现平稳，未出现显著超预期或低于预期的情况。关注细分业务数据和后续业绩指引。"

    # === 研报类 ===
    elif ann_type == "研报":
        impact_level = "medium"
        rating = ""
        if "买入" in full and "增持" not in full:
            rating = "买入"
            sentiment_score += 2
        elif "增持" in full or "调高" in full or "上调" in full:
            rating = "增持"
            sentiment_score += 1
        elif "减持" in full or "调低" in full or "下调" in full or "卖出" in full:
            rating = "减持/下调"
            sentiment_score -= 1
        if rating:
            key_points.append(f"评级：{rating}")
        tp = re.findall(r"目标价[^0-9]*([\d.]+)\s*元", text)
        if tp:
            key_points.append(f"目标价：{tp[0]}元")
            sentiment_score += 1
        if sentiment_score > 0:
            sentiment = "bullish"
            analysis = "机构发布正面研报，对公司未来发展持乐观态度。机构评级和目标价可作为投资参考，但需注意研报发布时点和股价位置。"
        elif sentiment_score < 0:
            sentiment = "bearish"
            analysis = "机构下调评级或目标价，对公司短期表现持谨慎态度。需关注下调原因，以及是否有其他机构跟进调整。"
        else:
            analysis = "机构发布研报，评级中性，提供公司基本面参考信息。"

    # === 重大事项 ===
    elif ann_type == "重大事项":
        impact_level = "high"
        if "重组" in full or "并购" in full or "收购" in full:
            key_points.append("涉及重大资产重组/并购")
            target = re.findall(r"(?:收购|受让|控股|收购)\s*([^，。、]{2,20})", text)
            if target:
                key_points.append(f"标的：{target[0]}")
            amt = re.findall(r"(?:交易|收购|转让)\s*(?:金额|对价)[^0-9]*([\d.]+)\s*亿", text)
            if amt:
                key_points.append(f"交易金额：{amt[0]}亿元")
            if "通过" in full or "获批" in full or "同意" in full:
                key_points.append("重组方案已获批")
                sentiment_score += 2
                analysis = "重大资产重组方案获批，有望改善公司业务结构和盈利能力。关注重组后标的资产的并表时间和业绩贡献。"
            elif "终止" in full or "中止" in full or "停止" in full:
                key_points.append("重组方案已终止")
                sentiment_score -= 2
                analysis = "重组方案终止，市场预期落空，短期可能面临情绪冲击。需关注公司后续是否有替代方案或新的战略方向。"
            else:
                analysis = "公司筹划重大资产重组事项，存在不确定性。重组若成功可能显著改变公司基本面，需持续跟踪方案进展。"
                sentiment_score += 1
        elif "借壳" in full:
            key_points.append("涉及借壳上市")
            sentiment_score += 2
            analysis = "借壳上市事项，若成功则公司基本面将发生根本性变化。关注借壳方资产质量和审批进展。"

    # === 股权变动 ===
    elif ann_type == "股权变动":
        impact_level = "medium"
        if "回购" in full:
            amt = re.findall(r"回购[^0-9]*([\d.]+)\s*亿", text)
            pct = re.findall(r"回购[^0-9]*([\d.]+)%", text)
            if amt:
                key_points.append(f"回购金额：{amt[0]}亿元")
            if pct:
                key_points.append(f"回购比例：{pct[0]}%")
            sentiment_score += 2
            analysis = "公司实施股份回购，通常意味着管理层认为股价被低估，回购将减少流通股本，对股价形成支撑。"
        elif "增持" in full and "减持" not in full:
            pct = re.findall(r"增持[^0-9]*([\d.]+)\s*万股", text)
            if pct:
                key_points.append(f"增持股数：{pct[0]}万股")
            sentiment_score += 1
            analysis = "重要股东或管理层增持股份，显示对公司未来发展的信心，通常被视为正面信号。"
        elif "减持" in full:
            pct = re.findall(r"减持[^0-9]*([\d.]+)\s*万股", text)
            if pct:
                key_points.append(f"减持股数：{pct[0]}万股")
            sentiment_score -= 1
            analysis = "重要股东减持股份，可能反映股东资金需求或对公司短期前景的谨慎判断，需关注减持规模和节奏。"
        elif "解禁" in full:
            amt = re.findall(r"解禁[^0-9]*([\d.]+)\s*亿股", text)
            if amt:
                key_points.append(f"解禁数量：{amt[0]}亿股")
            sentiment_score -= 1
            analysis = "限售股解禁，短期内可能面临抛售压力，需关注解禁股东类型和减持意愿。"
        elif "股权激励" in full:
            pct = re.findall(r"激励[^0-9]*([\d.]+)\s*万股", text)
            if pct:
                key_points.append(f"激励数量：{pct[0]}万股")
            sentiment_score += 1
            analysis = "公司推出股权激励计划，有利于绑定核心团队利益，设定业绩考核目标也隐含了管理层对业绩增长的预期。"

    # === 分红送转 ===
    elif ann_type == "分红送转":
        impact_level = "medium"
        cash = re.findall(r"每股派[^0-9]*([\d.]+)\s*元", text)
        if not cash:
            cash = re.findall(r"分红[^0-9]*([\d.]+)\s*元", text)
        send = re.findall(r"每\d0?股送转?[^0-9]*([\d.]+)\s*股", text)
        if cash:
            key_points.append(f"每股分红：{cash[0]}元")
            sentiment_score += 1
        if send:
            key_points.append(f"送转比例：每10股送转{send[0]}股")
            sentiment_score += 1
        if sentiment_score > 0:
            sentiment = "bullish"
            analysis = "公司实施分红送转方案，体现盈利能力和回报股东意愿。高送转可能引发填权行情，分红则直接增加股东收益。"

    # === 经营合同 ===
    elif ann_type == "经营合同":
        impact_level = "medium"
        amt = re.findall(r"(?:合同|订单|中标)\s*金额[^0-9]*([\d.]+)\s*亿", text)
        if not amt:
            amt = re.findall(r"(?:合同|订单|中标)\s*金额[^0-9]*([\d.]+)\s*万", text)
        if amt:
            key_points.append(f"合同金额：{amt[0]}亿/万元")
        counterparty = re.findall(r"(?:与|同|跟)\s*([^，。、]{2,20})\s*(?:签订|签署|达成)", text)
        if counterparty:
            key_points.append(f"合作方：{counterparty[0]}")
        if "中标" in full:
            key_points.append("类型：中标")
            sentiment_score += 1
        if "战略合作" in full:
            key_points.append("类型：战略合作")
            sentiment_score += 1
        if sentiment_score > 0:
            sentiment = "bullish"
            analysis = "公司签订重大经营合同或中标项目，有望增厚未来业绩。关注合同金额占公司营收的比重，以及项目执行的确定性。"
        else:
            analysis = "公司披露经营合同信息，关注合同执行进度和对业绩的实际贡献。"

    # === 风险警示 ===
    elif ann_type == "风险警示":
        impact_level = "high"
        if "诉讼" in full or "仲裁" in full:
            key_points.append("涉及诉讼/仲裁")
            amt = re.findall(r"(?:诉讼|仲裁)\s*(?:金额|标的)[^0-9]*([\d.]+)\s*亿", text)
            if amt:
                key_points.append(f"涉诉金额：{amt[0]}亿元")
            sentiment_score -= 2
            analysis = "公司涉及诉讼或仲裁事项，存在不确定的赔付风险。需关注案件进展和可能对公司业绩的影响。"
        elif "违规" in full or "处罚" in full or "监管" in full:
            key_points.append("涉及监管处罚/违规")
            sentiment_score -= 2
            analysis = "公司收到监管处罚或违规警示，可能影响公司声誉和经营，需关注处罚措施和整改情况。"
        elif "质押" in full:
            pct = re.findall(r"质押[^0-9]*([\d.]+)\s*万股", text)
            if pct:
                key_points.append(f"质押数量：{pct[0]}万股")
            ratio = re.findall(r"质押比例[^0-9]*([\d.]+)%", text)
            if ratio:
                key_points.append(f"质押比例：{ratio[0]}%")
            sentiment_score -= 1
            analysis = "股东股份质押，若质押比例较高可能面临平仓风险，需关注股价波动和质押率变化。"
        elif "担保" in full:
            amt = re.findall(r"担保[^0-9]*([\d.]+)\s*亿", text)
            if amt:
                key_points.append(f"担保金额：{amt[0]}亿元")
            sentiment_score -= 1
            analysis = "公司对外担保事项，需关注被担保方资信状况和担保风险敞口。"

    # === 募资动态 ===
    elif ann_type == "募资动态":
        impact_level = "medium"
        amt = re.findall(r"募集资金[^0-9]*([\d.]+)\s*亿", text)
        if amt:
            key_points.append(f"募集资金：{amt[0]}亿元")
        if "变更" in full:
            key_points.append("募投项目变更")
            sentiment_score -= 1
            analysis = "募集资金用途变更，需关注变更原因和新项目前景。"
        else:
            analysis = "披露募集资金使用情况，关注资金投入进度和项目效益。"

    # === 公司治理 ===
    elif ann_type == "公司治理":
        impact_level = "low"
        if "辞职" in full or "离职" in full:
            key_points.append("涉及高管变动")
            sentiment_score -= 1
            analysis = "高管辞职/离职，需关注对公司经营稳定性的影响，以及继任人选的安排。"
        elif "聘任" in full:
            key_points.append("聘任新高管")
            sentiment_score += 1
            analysis = "公司聘任新高管，关注新高管的背景和对公司战略的影响。"
        else:
            analysis = "公司治理相关事项，关注股东大会决议和董事会决策内容。"

    else:
        # 其他类型
        impact_level = "low"
        analysis = "公司发布一般性公告，关注公告具体内容是否涉及影响股价的重大信息。"

    # 确定最终情绪
    if sentiment_score >= 2:
        sentiment = "bullish"
    elif sentiment_score <= -2:
        sentiment = "bearish"
    else:
        sentiment = "neutral"

    # 如果没有提取到关键点，从标题生成
    if not key_points:
        key_points.append(title[:60])

    return {
        "type": ann_type,
        "key_points": key_points,
        "sentiment": sentiment,
        "impact": impact_level,
        "analysis": analysis,
        "title": title,
        "date": ann.get("date", ""),
        "highlights": _extract_financial_highlights(text, ann_type),
        "financial_comparison": _extract_financial_comparison(text) if ann_type == "财报" else None,
    }


def _extract_financial_comparison(text):
    """提取财务结构对比（本期 vs 上年同期）"""
    if not text:
        return None

    def _to_yi(num_str):
        n = float(num_str.replace(",", ""))
        return round(n / 1e8, 4)

    def _fmt_yi(val):
        if val >= 1:
            return f"{val:.2f}"
        elif val > 0:
            return f"{val:.4f}"
        return "0"

    items = []

    def _find_metric(name_pattern, txt):
        pattern = name_pattern + r"\s*(?:（[^）]*）)?\s*([\d,]+\.?\d*)\s+([\d,]+\.?\d*)\s+(-?[\d.]+)%?"
        m = re.search(pattern, txt)
        if m:
            try:
                curr = _to_yi(m.group(1))
                prev = _to_yi(m.group(2))
                chg = float(m.group(3))
                return {"current": _fmt_yi(curr), "previous": _fmt_yi(prev), "change_pct": chg}
            except Exception:
                pass
        return None

    metrics = [
        ("营业收入", r"营业(?:总收入|收入)"),
        ("归属于上市公司股东的净利润", r"归属于上市公司股东的净利润"),
        ("扣非净利润", r"归属于上市公司股东的扣除非经常性损益的净利润"),
        ("经营活动产生的现金流量净额", r"经营活动产生的现金流量净额"),
        ("总资产", r"(?:总资产|资产总计)"),
        ("归属于上市公司股东的净资产", r"归属于上市公司股东的净资产"),
        ("存货", r"存货"),
        ("在建工程", r"在建工程"),
        ("研发费用", r"研发(?:费用|投入)"),
        ("合同负债", r"合同负债"),
        ("应收账款", r"应收账款"),
        ("预付款项", r"预付款项"),
        ("短期借款", r"短期借款"),
        ("货币资金", r"货币资金"),
    ]

    for name, pattern in metrics:
        result = _find_metric(pattern, text)
        if result:
            result["name"] = name
            items.append(result)

    return items if items else None


def _extract_financial_highlights(text, ann_type):
    """从公告正文中提取亮点"""
    if not text:
        return []

    highlights = []

    def _try_num(pattern, txt):
        m = re.search(pattern, txt)
        if m:
            return m.group(1).replace(",", "")
        return None

    def _to_yi_str(num_str):
        n = float(num_str.replace(",", ""))
        yi = n / 1e8
        if yi >= 1:
            return f"{yi:.2f}亿元"
        else:
            return f"{yi*10000:.2f}万元"

    if ann_type != "财报":
        return highlights

    # 通用正则：支持带%和不带%的百分比
    def _find_3col(name_pattern, txt):
        pattern = name_pattern + r"\s*(?:（[^）]*）)?\s*([\d,]+\.?\d*)\s+([\d,]+\.?\d*)\s+(-?[\d.]+)%?"
        m = re.search(pattern, txt)
        if m:
            return m.group(1), m.group(2), float(m.group(3))
        return None, None, None

    # 1. 存货大幅增长
    inv_curr, inv_prev, inv_chg = _find_3col(r"存货", text)
    if inv_curr:
        try:
            chg = inv_chg
            curr = _to_yi_str(inv_curr)
            if abs(chg) >= 20:
                direction = "增长" if chg > 0 else "下降"
                desc = f"存货{direction}{abs(chg):.1f}%至{curr}"
                inv_context = ""
                for keyword in ["以销定产", "备货", "订单", "备货需求", "生产备货"]:
                    if keyword in text:
                        inv_context = f"，公司提及「{keyword}」"
                        break
                if not inv_context and chg > 50:
                    inv_context = "，需关注存货增长原因"
                highlights.append({
                    "category": "存货变动",
                    "description": desc + inv_context,
                    "impact": "bullish" if chg > 30 else ("bearish" if chg < -30 else "neutral"),
                    "insight": f"存货{direction}{abs(chg):.1f}%{'，结合以销定产策略说明下游需求旺盛' if '以销定产' in text and chg > 0 else '，反映公司产能扩张或备货需求' if chg > 0 else '，可能反映销售放缓或库存优化'}",
                })
        except Exception:
            pass

    # 2. 在建工程增长（扩产信号）
    cip_curr, cip_prev, cip_chg = _find_3col(r"在建工程", text)
    if cip_curr:
        try:
            chg = cip_chg
            curr = _to_yi_str(cip_curr)
            if abs(chg) >= 20:
                direction = "增长" if chg > 0 else "减少"
                highlights.append({
                    "category": "产能扩张",
                    "description": f"在建工程{direction}{abs(chg):.1f}%至{curr}",
                    "impact": "bullish" if chg > 0 else "neutral",
                    "insight": f"在建工程大幅{'增长' if chg > 0 else '减少'}{'，反映公司正在扩大产能，未来产能释放将带动业绩增长' if chg > 0 else '，可能反映项目完工转入固定资产'}",
                })
        except Exception:
            pass

    # 3. 研发投入变化
    rd_curr, rd_prev, rd_chg = _find_3col(r"研发(?:费用|投入)", text)
    if rd_curr:
        try:
            chg = rd_chg
            curr = _to_yi_str(rd_curr)
            if abs(chg) >= 10:
                direction = "增长" if chg > 0 else "下降"
                highlights.append({
                    "category": "研发投入",
                    "description": f"研发费用{direction}{abs(chg):.1f}%至{curr}",
                    "impact": "bullish" if chg > 0 else "neutral",
                    "insight": f"研发投入{'增加' if chg > 0 else '减少'}{'，显示公司持续加大技术创新投入，有利于长期竞争力' if chg > 0 else '，需关注是否影响长期竞争力'}",
                })
        except Exception:
            pass

    # 4. 合同负债变化（订单预收款）
    cl_curr, cl_prev, cl_chg = _find_3col(r"合同负债", text)
    if cl_curr:
        try:
            chg = cl_chg
            curr = _to_yi_str(cl_curr)
            if abs(chg) >= 20:
                direction = "增长" if chg > 0 else "下降"
                highlights.append({
                    "category": "订单预收",
                    "description": f"合同负债{direction}{abs(chg):.1f}%至{curr}",
                    "impact": "bullish" if chg > 0 else "bearish",
                    "insight": f"合同负债{'大幅增长' if chg > 0 else '下降'}{'，预收款增加说明客户订单充足，未来收入有保障' if chg > 0 else '，预收款减少可能反映订单需求走弱'}",
                })
        except Exception:
            pass

    # 5. 经营性现金流变化
    ocf_curr, ocf_prev, ocf_chg = _find_3col(r"经营活动产生的现金流量净额", text)
    if ocf_curr:
        try:
            chg = ocf_chg
            curr = _to_yi_str(ocf_curr)
            if abs(chg) >= 30:
                direction = "增长" if chg > 0 else "下降"
                highlights.append({
                    "category": "现金流",
                    "description": f"经营现金流{direction}{abs(chg):.1f}%至{curr}",
                    "impact": "bullish" if chg > 0 else "bearish",
                    "insight": f"经营现金流大幅{'增长' if chg > 0 else '下降'}{'，反映公司现金回收能力强，盈利质量高' if chg > 0 else '，需关注应收账款回收和资金周转情况'}",
                })
        except Exception:
            pass

    # 6. 营收增长但利润增速更高（经营杠杆）
    _, _, rev_chg = _find_3col(r"营业(?:总收入|收入)", text)
    _, _, profit_chg = _find_3col(r"归属于[^净]*净利润", text)
    if rev_chg is not None and profit_chg is not None:
        try:
            if rev_chg > 5 and profit_chg > rev_chg * 1.5:
                highlights.append({
                    "category": "经营杠杆",
                    "description": f"营收增长{rev_chg:.1f}%，净利润增长{profit_chg:.1f}%，利润增速远超营收",
                    "impact": "bullish",
                    "insight": "利润增速大幅超过营收增速，反映经营杠杆释放，规模效应显现或成本控制出色",
                })
            elif rev_chg > 0 and profit_chg < 0:
                highlights.append({
                    "category": "增收不增利",
                    "description": f"营收增长{rev_chg:.1f}%但净利润下降{abs(profit_chg):.1f}%",
                    "impact": "bearish",
                    "insight": "增收不增利，需关注成本上升、毛利率下滑或费用增加的原因",
                })
        except Exception:
            pass

    # 7. 应收账款变化
    ar_curr, ar_prev, ar_chg = _find_3col(r"应收账款", text)
    if ar_curr:
        try:
            chg = ar_chg
            if chg > 50:
                curr = _to_yi_str(ar_curr)
                highlights.append({
                    "category": "应收账款",
                    "description": f"应收账款增长{chg:.1f}%至{curr}",
                    "impact": "bearish",
                    "insight": "应收账款大幅增长，需关注回款风险和客户信用状况",
                })
        except Exception:
            pass

    # 8. 货币资金充裕
    cash_curr, cash_prev, cash_chg = _find_3col(r"货币资金", text)
    if cash_curr:
        try:
            chg = cash_chg
            curr = _to_yi_str(cash_curr)
            if chg > 30:
                highlights.append({
                    "category": "资金面",
                    "description": f"货币资金增长{chg:.1f}%至{curr}",
                    "impact": "bullish",
                    "insight": "货币资金充裕，公司抗风险能力强，有并购或扩张的资金基础",
                })
        except Exception:
            pass

    # 9. 短期借款变化
    loan_curr, loan_prev, loan_chg = _find_3col(r"短期借款", text)
    if loan_curr:
        try:
            chg = loan_chg
            if abs(chg) >= 30:
                curr = _to_yi_str(loan_curr)
                direction = "增加" if chg > 0 else "减少"
                highlights.append({
                    "category": "负债结构",
                    "description": f"短期借款{direction}{abs(chg):.1f}%至{curr}",
                    "impact": "bearish" if chg > 0 else "bullish",
                    "insight": f"短期借款{direction}{'，需关注偿债压力和财务费用' if chg > 0 else '，债务结构优化，财务压力减轻'}",
                })
        except Exception:
            pass

    # 10. 业绩说明会/分红/回购等特殊亮点
    if "业绩说明会" in text or "业绩说明会" in text:
        highlights.append({
            "category": "投资者关系",
            "description": "公司召开业绩说明会",
            "impact": "neutral",
            "insight": "公司主动召开业绩说明会，展示管理层对业绩的信心和与投资者沟通的意愿",
        })

    # 11. 亏损或扭亏
    if "扭亏" in text:
        highlights.append({
            "category": "业绩拐点",
            "description": "本期实现扭亏为盈",
            "impact": "bullish",
            "insight": "公司扭亏为盈，经营出现拐点，需关注盈利的可持续性",
        })

    if "亏损" in text and "扭亏" not in text:
        highlights.append({
            "category": "业绩风险",
            "description": "本期存在亏损",
            "impact": "bearish",
            "insight": "公司本期出现亏损，需关注亏损原因和后续改善措施",
        })

    return highlights


def get_announcements(code):
    """获取个股公告列表+解读（仅展示5条）"""
    anns = _fetch_announcements(code)
    if not anns:
        return {"ok": False, "code": code, "announcements": [], "msg": "暂无公告"}
    analyzed = []
    for ann in anns:
        result = _analyze_announcement(ann, code)
        analyzed.append(result)
    # 按重要性排序：high > medium > low
    impact_order = {"high": 0, "medium": 1, "low": 2}
    analyzed.sort(key=lambda x: impact_order.get(x.get("impact", "low"), 2))
    # 仅返回前5条
    analyzed = analyzed[:5]
    # 对财报类公告附加深度财务分析
    has_financial = any(a.get("type") == "财报" for a in analyzed)
    if has_financial:
        deep = _deep_analyze_financials(code, analyzed[0].get("analysis", ""))
        if deep:
            for a in analyzed:
                if a.get("type") == "财报":
                    a["deep_analysis"] = deep
                    break  # 只在第一条财报上附加
    return {"ok": True, "code": code, "announcements": analyzed}


_F10_CACHE = {}  # code → {ts, data}


def get_f10_info(code):
    """获取个股F10资料：公司基本概况 + 最新报告期主营构成（按产品/行业/地区）"""
    cache = _F10_CACHE.get(code)
    # 公司基本资料几乎不变，缓存24小时
    if cache and time.time() - cache["ts"] < 86400:
        return cache["data"]
    code = re.sub(r"\D", "", code or "")
    if not code:
        return {"ok": False, "code": code, "msg": "缺少 code 参数"}
    # 生成 SECUCODE：沪市6开头.SH，北交所4/8开头.BJ，其余深市.SZ
    if code.startswith("6"):
        secucode = code + ".SH"
    elif code.startswith(("4", "8")):
        secucode = code + ".BJ"
    else:
        secucode = code + ".SZ"

    def _dc_get(report, columns, extra=""):
        url = (
            "https://datacenter.eastmoney.com/securities/api/data/v1/get"
            "?reportName=" + report + "&columns=" + columns +
            "&filter=" + urllib.parse.quote('(SECUCODE="' + secucode + '")') +
            "&pageNumber=1&source=HSF10&client=PC" + extra
        )
        raw = _fetch_url(url, timeout=10)
        if not raw:
            return None
        try:
            d = json.loads(raw)
        except Exception:
            m = re.search(r"\{.*\}", raw, re.DOTALL)
            if not m:
                return None
            try:
                d = json.loads(m.group(0))
            except Exception:
                return None
        if not d.get("success"):
            return None
        return (d.get("result") or {}).get("data") or []

    result = {"ok": False, "code": code, "basic": None, "mainop": None}
    # --- 公司基本概况 ---
    try:
        rows = _dc_get(
            "RPT_F10_BASIC_ORGINFO",
            "SECURITY_NAME_ABBR,ORG_NAME,CHAIRMAN,PRESIDENT,SECRETARY,REG_CAPITAL,"
            "EMP_NUM,LISTING_DATE,FOUND_DATE,ADDRESS,ORG_WEB,ORG_PROFILE,BUSINESS_SCOPE,EM2016",
        )
        if rows:
            r = rows[0]
            result["basic"] = {
                "abbr": r.get("SECURITY_NAME_ABBR") or "",
                "org_name": r.get("ORG_NAME") or "",
                "chairman": r.get("CHAIRMAN") or "",
                "president": r.get("PRESIDENT") or "",
                "secretary": r.get("SECRETARY") or "",
                "reg_capital": r.get("REG_CAPITAL"),  # 万元
                "emp_num": r.get("EMP_NUM"),
                "listing_date": (r.get("LISTING_DATE") or "")[:10],
                "found_date": (r.get("FOUND_DATE") or "")[:10],
                "address": r.get("ADDRESS") or "",
                "web": r.get("ORG_WEB") or "",
                "profile": (r.get("ORG_PROFILE") or "").strip(),
                "business_scope": (r.get("BUSINESS_SCOPE") or "").strip(),
                "em2016": r.get("EM2016") or "",
            }
    except Exception:
        pass
    # --- 主营构成（最新报告期） ---
    try:
        rows = _dc_get(
            "RPT_F10_FN_MAINOP",
            "REPORT_DATE,MAINOP_TYPE,ITEM_NAME,MAIN_BUSINESS_INCOME,MBI_RATIO,GROSS_RPOFIT_RATIO",
            "&pageSize=200&sortTypes=-1,-1&sortColumns=REPORT_DATE,RANK",
        )
        if rows:
            latest = max((x.get("REPORT_DATE") or "")[:10] for x in rows)
            by_type = {"product": [], "industry": [], "region": []}
            type_map = {"2": "product", "1": "industry", "3": "region"}
            for x in rows:
                if (x.get("REPORT_DATE") or "")[:10] != latest:
                    continue
                bucket = type_map.get(str(x.get("MAINOP_TYPE", "")))
                if not bucket:
                    continue
                income = x.get("MAIN_BUSINESS_INCOME") or 0
                if income <= 0 and not (x.get("MBI_RATIO") or 0):
                    continue
                by_type[bucket].append({
                    "name": x.get("ITEM_NAME") or "",
                    "income": round(income / 1e8, 2),  # 亿元
                    "ratio": round((x.get("MBI_RATIO") or 0) * 100, 2),  # %
                    "gross": round((x.get("GROSS_RPOFIT_RATIO") or 0) * 100, 1),  # %
                })
            for k in by_type:
                by_type[k].sort(key=lambda v: -v["ratio"])
            result["mainop"] = {"report_date": latest, **by_type}
    except Exception:
        pass
    result["ok"] = bool(result["basic"] or result["mainop"])
    if not result["ok"]:
        result["msg"] = "F10资料暂不可用"
    _F10_CACHE[code] = {"ts": time.time(), "data": result}
    return result


def append_events(events):
    """将事件追加到 JSONL 文件"""
    with open(EVENTS_FILE, "a", encoding="utf-8") as f:
        for ev in events:
            f.write(json.dumps(ev, ensure_ascii=False) + "\n")


def load_events(limit=10000, offset=0, event_type=None, uid=None):
    """从文件加载事件"""
    if not os.path.exists(EVENTS_FILE):
        return []
    events = []
    with open(EVENTS_FILE, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                ev = json.loads(line)
                if event_type and ev.get("type") != event_type:
                    continue
                if uid and ev.get("uid") != uid:
                    continue
                events.append(ev)
            except json.JSONDecodeError:
                continue
    events.reverse()  # 最新的在前
    total = len(events)
    return events[offset:offset + limit], total


def get_analytics_summary():
    """生成汇总统计数据 v2 - 支持丰富的埋点事件"""
    events, total = load_events(limit=100000)
    if not events:
        return {"total_events": 0, "total_users": 0, "total_sessions": 0, "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}

    uids = set(); sids = set()
    page_views = 0; heartbeats = 0; idle_count = 0; error_count = 0
    daily_active = {}; hourly_active = {}
    first_seen = {}; last_seen = {}
    # v2 事件收集
    section_enters = {}; section_leaves = {}; section_scroll = {}; section_inter = {}
    clicks = {}; click_coords = []; searches = []; filter_changes = {}
    content_expands = {}; chart_interacts = {}; trade_actions = []; watchlist_changes = []
    exits = []; event_types = {}; devices = {}; idle_sections = {}
    user_sessions = {}  # uid → {sid → [events]}
    user_terminals = {}  # uid → terminal info (latest)

    SEC_NAMES = {
        'layer1': '市场全景', 'layer2': '板块轮动', 'layer0': '交易信号',
        'layer4': '量化学习', 'layer-paper': '模拟交易',
        'layer-watch': '我关注的', 'layer-tpo': '四度空间',
    }

    for ev in events:
        uid = ev.get("uid", ""); sid = ev.get("sid", "")
        uids.add(uid); sids.add(sid)
        ts = ev.get("ts", 0)
        dt = datetime.fromtimestamp(ts / 1000)
        date_str = dt.strftime("%Y-%m-%d"); hour_str = dt.strftime("%H")
        daily_active[date_str] = daily_active.get(date_str, 0) + 1
        hourly_active[hour_str] = hourly_active.get(hour_str, 0) + 1
        if uid not in first_seen or ts < first_seen[uid]: first_seen[uid] = ts
        if uid not in last_seen or ts > last_seen[uid]: last_seen[uid] = ts

        # 提取终端信息（每次更新为最新）
        term = ev.get("term")
        if term and isinstance(term, dict):
            user_terminals[uid] = term

        t = ev.get("type", ""); d = ev.get("data", {})
        event_types[t] = event_types.get(t, 0) + 1

        if t == "page_view":
            page_views += 1
            vp = d.get("viewport", ""); dpr = d.get("dpr", 1)
            if vp: devices[vp] = devices.get(vp, 0) + 1
        elif t == "section_enter":
            sec = d.get("section", ev.get("page", ""))
            section_enters[sec] = section_enters.get(sec, 0) + 1
        elif t == "section_leave":
            sec = d.get("section", "")
            dur = d.get("duration_s", 0)
            if sec and dur > 0:
                if sec not in section_leaves: section_leaves[sec] = []
                section_leaves[sec].append(dur)
            sp = d.get("scroll_pct", 0)
            if sec and sp > 0:
                if not isinstance(section_scroll.get(sec), list): section_scroll[sec] = []
                section_scroll[sec].append(sp)
            si = d.get("interactions", 0)
            if sec:
                if not isinstance(section_inter.get(sec), list): section_inter[sec] = []
                section_inter[sec].append(si)
        elif t == "scroll_depth":
            sec = d.get("section", ""); sp = d.get("scroll_pct", 0)
            if sec and sp > 0:
                if not isinstance(section_scroll.get(sec), list): section_scroll[sec] = []
                section_scroll[sec].append(sp)
        elif t == "click":
            key = d.get("label", "")[:40]
            if key: clicks[key] = clicks.get(key, 0) + 1
            x = d.get("x"); y = d.get("y")
            if x is not None and y is not None:
                click_coords.append({"x": x, "y": y, "section": d.get("section", ""), "label": key})
        elif t == "search":
            q = d.get("query", "")
            if q: searches.append(q)
        elif t == "filter_change":
            fid = d.get("id", ""); val = d.get("label", val if (val := d.get("value", "")) else "")
            key = fid + "=" + str(d.get("value", ""))
            filter_changes[key] = filter_changes.get(key, 0) + 1
        elif t == "content_expand":
            key = d.get("label", "")[:30]
            if key: content_expands[key] = content_expands.get(key, 0) + 1
        elif t == "chart_interact":
            key = d.get("type", "") + ":" + d.get("name", "")[:20]
            chart_interacts[key] = chart_interacts.get(key, 0) + 1
        elif t == "trade_action":
            trade_actions.append(d)
        elif t == "watchlist_change":
            watchlist_changes.append(d)
        elif t == "idle":
            idle_count += 1
            sec = d.get("section", "")
            idle_sections[sec] = idle_sections.get(sec, 0) + 1
        elif t == "error":
            error_count += 1
        elif t == "page_exit":
            dur = d.get("duration_s", 0)
            if dur > 0: exits.append(dur)
        elif t == "heartbeat":
            heartbeats += 1

    # 计算各模块平均停留、最大滚动深度、平均交互数
    avg_dwell = {}; avg_scroll = {}; avg_inter = {}
    for sec, durs in section_leaves.items():
        avg_dwell[sec] = round(sum(durs) / len(durs)) if durs else 0
    for sec, scs in section_scroll.items():
        avg_scroll[sec] = round(sum(scs) / len(scs)) if scs else 0
    for sec, ins in section_inter.items():
        avg_inter[sec] = round(sum(ins) / len(ins)) if ins else 0

    top_clicks = sorted(clicks.items(), key=lambda x: -x[1])[:30]
    search_counts = {}
    for q in searches: search_counts[q] = search_counts.get(q, 0) + 1
    top_searches = sorted(search_counts.items(), key=lambda x: -x[1])[:20]
    top_filters = sorted(filter_changes.items(), key=lambda x: -x[1])[:15]
    top_expands = sorted(content_expands.items(), key=lambda x: -x[1])[:15]
    top_charts = sorted(chart_interacts.items(), key=lambda x: -x[1])[:15]

    avg_session = round(sum(exits) / len(exits)) if exits else 0

    # 用户列表（带详细行为）
    users = []
    for uid in sorted(uids):
        fs = first_seen.get(uid, 0); ls = last_seen.get(uid, 0)
        u_events = [e for e in events if e.get("uid") == uid]
        u_sids = set(e.get("sid") for e in u_events)
        u_clicks = sum(1 for e in u_events if e.get("type") == "click")
        u_searches = sum(1 for e in u_events if e.get("type") == "search")
        u_trades = sum(1 for e in u_events if e.get("type") == "trade_action")
        u_exits = [e for e in u_events if e.get("type") == "page_exit"]
        u_durs = [e.get("data", {}).get("duration_s", 0) for e in u_exits]
        u_avg_dur = round(sum(u_durs) / len(u_durs)) if u_durs else 0
        u_secs = set()
        for e in u_events:
            if e.get("type") in ("section_enter", "section_leave"):
                s = e.get("data", {}).get("section", "")
                if s: u_secs.add(s)
        users.append({
            "uid": uid,
            "first_seen": datetime.fromtimestamp(fs / 1000).strftime("%Y-%m-%d %H:%M") if fs else "",
            "last_seen": datetime.fromtimestamp(ls / 1000).strftime("%Y-%m-%d %H:%M") if ls else "",
            "sessions": len(u_sids),
            "events": len(u_events),
            "clicks": u_clicks,
            "searches": u_searches,
            "trades": u_trades,
            "avg_session_s": u_avg_dur,
            "sections_visited": len(u_secs),
            "terminal": user_terminals.get(uid, {}),
        })

    # 点击热力图坐标（聚合到网格）
    heat_grid = {}
    for c in click_coords[:5000]:
        gx = c["x"] // 50 * 50; gy = c["y"] // 50 * 50
        key = str(gx) + "," + str(gy)
        heat_grid[key] = heat_grid.get(key, 0) + 1
    top_heat = sorted(heat_grid.items(), key=lambda x: -x[1])[:20]

    # 终端分布统计
    term_devices = {}; term_os = {}; term_browsers = {}; term_screens = {}
    for t in user_terminals.values():
        if t.get("device"): term_devices[t["device"]] = term_devices.get(t["device"], 0) + 1
        if t.get("os"): term_os[t["os"]] = term_os.get(t["os"], 0) + 1
        if t.get("browser"): term_browsers[t["browser"]] = term_browsers.get(t["browser"], 0) + 1
        if t.get("screen"): term_screens[t["screen"]] = term_screens.get(t["screen"], 0) + 1

    return {
        "total_events": total, "total_users": len(uids), "total_sessions": len(sids),
        "total_page_views": page_views, "total_heartbeats": heartbeats,
        "total_idle": idle_count, "total_errors": error_count,
        "total_trades": len(trade_actions), "total_watchlist_changes": len(watchlist_changes),
        "avg_session_s": avg_session,
        "event_type_dist": dict(sorted(event_types.items(), key=lambda x: -x[1])),
        "daily_active": dict(sorted(daily_active.items())),
        "hourly_active": dict(sorted(hourly_active.items())),
        "section_enters": dict(sorted(section_enters.items(), key=lambda x: -x[1])),
        "avg_dwell_s": avg_dwell, "avg_scroll_pct": avg_scroll, "avg_interactions": avg_inter,
        "top_clicks": top_clicks, "top_searches": top_searches,
        "top_filters": top_filters, "top_expands": top_expands, "top_charts": top_charts,
        "click_heatmap": top_heat, "idle_sections": dict(sorted(idle_sections.items(), key=lambda x: -x[1])),
        "devices": dict(sorted(devices.items(), key=lambda x: -x[1])[:10]),
        "term_devices": dict(sorted(term_devices.items(), key=lambda x: -x[1])),
        "term_os": dict(sorted(term_os.items(), key=lambda x: -x[1])),
        "term_browsers": dict(sorted(term_browsers.items(), key=lambda x: -x[1])),
        "term_screens": dict(sorted(term_screens.items(), key=lambda x: -x[1])),
        "users": users,
        "section_names": SEC_NAMES,
        "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    }


# ===== 用户系统 =====
USERS_DIR = os.path.join(DATA_DIR, "users")
os.makedirs(USERS_DIR, exist_ok=True)

# 简单的用户索引文件
USERS_INDEX = os.path.join(USERS_DIR, "_index.json")


def load_users_index():
    """加载用户索引"""
    if not os.path.exists(USERS_INDEX):
        return {}
    try:
        with open(USERS_INDEX, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}


def save_users_index(index):
    """保存用户索引"""
    with open(USERS_INDEX, "w", encoding="utf-8") as f:
        json.dump(index, f, ensure_ascii=False, indent=2)


def user_login(username):
    """用户登录/注册，返回用户完整数据"""
    username = username.strip()[:20]
    if not username:
        return {"ok": False, "error": "用户名不能为空"}
    # 安全过滤文件名
    safe_name = "".join(c for c in username if c.isalnum() or c in "_-")
    if not safe_name:
        return {"ok": False, "error": "用户名仅允许字母、数字、下划线、横杠"}
    user_file = os.path.join(USERS_DIR, safe_name + ".json")
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    index = load_users_index()
    if os.path.exists(user_file):
        # 已有用户，更新最后登录
        try:
            with open(user_file, "r", encoding="utf-8") as f:
                data = json.load(f)
        except Exception:
            data = {}
        data["last_login"] = now
        if safe_name not in index:
            index[safe_name] = {"created": now, "last_login": now}
        else:
            index[safe_name]["last_login"] = now
    else:
        # 新用户
        data = {
            "username": username,
            "created": now,
            "last_login": now,
            "watchlist": [],
            "paperAccount": None,
            "paperPortfolios": [],
        }
        index[safe_name] = {"created": now, "last_login": now}
    save_users_index(index)
    with open(user_file, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    return {"ok": True, "data": data}


def user_get_data(username):
    """获取用户数据"""
    safe_name = "".join(c for c in username.strip()[:20] if c.isalnum() or c in "_-")
    if not safe_name:
        return {"ok": False, "error": "invalid username"}
    user_file = os.path.join(USERS_DIR, safe_name + ".json")
    if not os.path.exists(user_file):
        return {"ok": False, "error": "user not found"}
    try:
        with open(user_file, "r", encoding="utf-8") as f:
            data = json.load(f)
        return {"ok": True, "data": data}
    except Exception as e:
        return {"ok": False, "error": str(e)}


def user_sync_data(username, watchlist=None, paper_account=None, paper_portfolios=None, layout=None):
    """同步用户数据"""
    safe_name = "".join(c for c in username.strip()[:20] if c.isalnum() or c in "_-")
    if not safe_name:
        return {"ok": False, "error": "invalid username"}
    user_file = os.path.join(USERS_DIR, safe_name + ".json")
    if not os.path.exists(user_file):
        return {"ok": False, "error": "user not found, please login first"}
    try:
        with open(user_file, "r", encoding="utf-8") as f:
            data = json.load(f)
        if watchlist is not None:
            data["watchlist"] = watchlist
        if paper_account is not None:
            data["paperAccount"] = paper_account
        if paper_portfolios is not None:
            data["paperPortfolios"] = paper_portfolios
        if layout is not None:
            data["layout"] = layout
        data["last_sync"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with open(user_file, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        return {"ok": True, "last_sync": data["last_sync"]}
    except Exception as e:
        return {"ok": False, "error": str(e)}


# ===== 经济日历实时数据 (ForexFactory API) =====
_ECON_CAL_CACHE = {"ts": 0, "data": None}
_ECON_CAL_TTL = 3600  # 1小时缓存

# ForexFactory事件标题 → 看板事件标题的映射
_FF_TITLE_MAP = {
    "Non-Farm Employment Change": "美国非农就业",
    "Unemployment Rate": "美国失业率",
    "Average Hourly Earnings m/m": "美国平均时薪月率",
    "Average Hourly Earnings y/y": "美国平均时薪年率",
    "ADP Non-Farm Employment Change": "美国ADP就业",
    "CPI m/m": "美国CPI月率",
    "CPI y/y": "美国CPI年率",
    "Core CPI m/m": "美国核心CPI月率",
    "Core CPI y/y": "美国核心CPI年率",
    "PPI m/m": "美国PPI月率",
    "PPI y/y": "美国PPI年率",
    "Core PPI m/m": "美国核心PPI月率",
    "Core PCE Price Index m/m": "美国核心PCE月率",
    "Core PCE Price Index y/y": "美国核心PCE年率",
    "PCE Price Index m/m": "美国PCE月率",
    "PCE Price Index y/y": "美国PCE年率",
    "ISM Manufacturing PMI": "美国ISM制造业PMI",
    "ISM Services PMI": "美国ISM服务业PMI",
    "FOMC Statement": "美联储FOMC声明",
    "FOMC Economic Projections": "美联储经济预测",
    "Fed Chair Press Conference": "美联储主席新闻发布会",
    "Interest Rate Decision": "美联储利率决议",
    "Retail Sales m/m": "美国零售销售月率",
    "Core Retail Sales m/m": "美国核心零售销售月率",
    "GDP q/q": "美国GDP季率",
    "GDP q/q adv": "美国GDP季率(预估)",
    "GDP q/q final": "美国GDP季率(终值)",
    "JOLTS Job Openings": "美国JOLTts职位空缺",
    "Unemployment Claims": "美国初请失业金",
    "Crude Oil Inventories": "美国原油库存变动",
}

_IMPACT_MAP = {"High": "high", "Medium": "medium", "Low": "low"}
_COUNTRY_MAP = {"USD": "美国", "CNY": "中国", "EUR": "欧盟", "JPY": "日本", "GBP": "英国", "CAD": "加拿大", "AUD": "澳大利亚"}


def _fetch_econ_calendar():
    """从ForexFactory获取本周+下周经济日历数据"""
    now_ts = time.time()
    if _ECON_CAL_CACHE["data"] and now_ts - _ECON_CAL_CACHE["ts"] < _ECON_CAL_TTL:
        return _ECON_CAL_CACHE["data"]

    # 磁盘缓存路径
    disk_cache = os.path.join(BASE_DIR, "data", "econ_cal_cache.json")

    events = []
    ff_ok = False
    for week_url in [
        "https://nfs.faireconomy.media/ff_calendar_thisweek.json",
        "https://nfs.faireconomy.media/ff_calendar_nextweek.json",
    ]:
        try:
            req = urllib.request.Request(week_url, headers={
                "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
                "Accept": "application/json",
                "Referer": "https://www.forexfactory.com/",
            })
            with urllib.request.urlopen(req, timeout=15, context=_SSL_CTX) as resp:
                raw = resp.read().decode("utf-8", errors="replace")
            items = json.loads(raw)
            for item in items:
                title_en = item.get("title", "")
                country = item.get("country", "")
                mapped = _FF_TITLE_MAP.get(title_en)
                if not mapped:
                    continue
                ff_date = item.get("date", "")
                if not ff_date:
                    continue
                # ISO时间 → 北京时间
                try:
                    from datetime import timezone, timedelta
                    dt = datetime.fromisoformat(ff_date)
                    if dt.tzinfo is None:
                        dt = dt.replace(tzinfo=timezone.utc)
                    beijing = dt.astimezone(timezone(timedelta(hours=8)))
                    date_str = beijing.strftime("%Y-%m-%d")
                    time_str = beijing.strftime("%H:%M")
                except Exception:
                    date_str = ff_date[:10]
                    time_str = "待定"

                impact_en = item.get("impact", "Low")
                impact_cn = _IMPACT_MAP.get(impact_en, "low")
                country_cn = _COUNTRY_MAP.get(country, country)

                # 根据事件类型设置category
                cat = "重磅" if impact_en == "High" else "宏观"
                if country == "CNY":
                    cat = "宏观" if impact_en != "High" else "重磅"

                # 判断利好利空（默认中性，已发布的根据实际vs预期判断）
                impact = "neutral"
                forecast = item.get("forecast", "") or "待定"
                previous = item.get("previous", "") or "待定"
                actual = item.get("actual", "") or ""

                events.append({
                    "date": date_str,
                    "time": time_str,
                    "title": mapped,
                    "cat": cat,
                    "importance": impact_cn,
                    "country": country_cn,
                    "desc": title_en,
                    "forecast": forecast,
                    "prior": previous,
                    "actual": actual,
                    "impact": impact,
                    "impact_text": "",
                    "markets": _event_markets(country),
                    "status": "actual" if actual else "pending",
                    "source": "ForexFactory",
                })
            ff_ok = True
        except Exception as e:
            pass

    # 去重：同一日期+标题只保留一条
    seen = set()
    deduped = []
    for e in events:
        key = e["date"] + e["title"]
        if key not in seen:
            seen.add(key)
            deduped.append(e)

    # 按日期+时间排序
    deduped.sort(key=lambda x: x["date"] + x["time"])

    # 过滤已过期事件
    today_str = datetime.now().strftime("%Y-%m-%d")
    deduped = [e for e in deduped if e["date"] >= today_str]

    if not deduped and not ff_ok:
        # ForexFactory失败，尝试磁盘缓存
        try:
            if os.path.exists(disk_cache):
                with open(disk_cache, "r", encoding="utf-8") as f:
                    cached = json.load(f)
                if cached.get("events"):
                    cached_events = [e for e in cached["events"] if e["date"] >= today_str]
                    if cached_events:
                        deduped = cached_events
                        result = {
                            "ok": True,
                            "fetch_time": cached.get("fetch_time", "") + " (磁盘缓存)",
                            "events": deduped,
                            "source": "ForexFactory(disk-cache)",
                            "summary": {
                                "total": len(deduped),
                                "high_impact": sum(1 for e in deduped if e["importance"] == "high"),
                                "bullish": sum(1 for e in deduped if e["impact"] == "bullish"),
                                "bearish": sum(1 for e in deduped if e["impact"] == "bearish"),
                                "neutral": sum(1 for e in deduped if e["impact"] == "neutral"),
                            },
                        }
                        return result
        except Exception:
            pass
        return {"ok": False, "fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M"),
                "events": [], "source": "ForexFactory", "summary": {}}

    result = {
        "ok": len(deduped) > 0,
        "fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M"),
        "events": deduped,
        "source": "ForexFactory",
        "summary": {
            "total": len(deduped),
            "high_impact": sum(1 for e in deduped if e["importance"] == "high"),
            "bullish": sum(1 for e in deduped if e["impact"] == "bullish"),
            "bearish": sum(1 for e in deduped if e["impact"] == "bearish"),
            "neutral": sum(1 for e in deduped if e["impact"] == "neutral"),
        },
    }
    # 写入磁盘缓存
    if ff_ok and deduped:
        try:
            os.makedirs(os.path.dirname(disk_cache), exist_ok=True)
            with open(disk_cache, "w", encoding="utf-8") as f:
                json.dump(result, f, ensure_ascii=False)
        except Exception:
            pass
    _ECON_CAL_CACHE["data"] = result
    _ECON_CAL_CACHE["ts"] = now_ts
    return result


def _event_markets(country):
    if country == "USD":
        return ["美股", "债券", "外汇"]
    elif country == "CNY":
        return ["A股", "港股"]
    elif country == "JPY":
        return ["外汇", "美股"]
    else:
        return ["外汇"]


class DashboardHandler(http.server.SimpleHTTPRequestHandler):
    # gzip 压缩缓存：(path, mtime_ns, size) -> bytes
    _GZ_CACHE = {}
    _GZ_LOCK = threading.Lock()
    GZIP_EXTS = ('.html', '.js', '.css')
    CACHEABLE_EXTS = ('.js', '.css', '.png', '.jpg', '.jpeg', '.svg', '.woff', '.woff2', '.ico')

    def __init__(self, *args, **kwargs):
        super().__init__(*args, directory=BASE_DIR, **kwargs)

    def end_headers(self):
        # 静态资源（echarts等库文件）允许缓存1小时，HTML保持no-store确保数据新鲜
        ext = os.path.splitext(urlparse(self.path).path)[1].lower()
        if ext in self.CACHEABLE_EXTS:
            self.send_header("Cache-Control", "public, max-age=3600")
        else:
            self.send_header("Cache-Control", "no-store, must-revalidate")
            self.send_header("Pragma", "no-cache")
            self.send_header("Expires", "0")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        super().end_headers()

    def _gzip_content(self, path):
        """读取文件并gzip压缩，带内存缓存避免重复压缩"""
        st = os.stat(path)
        key = (path, st.st_mtime_ns, st.st_size)
        with self._GZ_LOCK:
            cached = self._GZ_CACHE.get(key)
        if cached is not None:
            return cached
        with open(path, 'rb') as f:
            raw = f.read()
        buf = io.BytesIO()
        with gzip.GzipFile(fileobj=buf, mode='wb', compresslevel=6, mtime=0) as gz:
            gz.write(raw)
        data = buf.getvalue()
        with self._GZ_LOCK:
            if len(self._GZ_CACHE) > 32:
                self._GZ_CACHE.clear()
            self._GZ_CACHE[key] = data
        return data

    def send_head(self):
        """对文本类文件启用gzip传输（1.3MB HTML → ~300KB）"""
        try:
            path = self.translate_path(self.path)
            ext = os.path.splitext(path)[1].lower()
            if (os.path.isfile(path) and ext in self.GZIP_EXTS
                    and 'gzip' in self.headers.get('Accept-Encoding', '')):
                data = self._gzip_content(path)
                self.send_response(200)
                ctype = self.guess_type(path)
                if ctype.startswith('text/') or ctype == 'application/javascript':
                    ctype += '; charset=utf-8'
                self.send_header("Content-Type", ctype)
                self.send_header("Content-Length", str(len(data)))
                self.send_header("Content-Encoding", "gzip")
                self.send_header("Vary", "Accept-Encoding")
                self.send_header("Last-Modified", self.date_time_string(os.path.getmtime(path)))
                self.end_headers()
                return io.BytesIO(data)
        except Exception:
            pass
        return super().send_head()

    def do_OPTIONS(self):
        self.send_response(204)
        self.end_headers()

    def do_POST(self):
        parsed = urlparse(self.path)
        if parsed.path == "/api/track":
            content_len = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_len).decode("utf-8") if content_len else ""
            try:
                payload = json.loads(body)
                events = payload.get("events", [])
                if events:
                    append_events(events)
                self._json_response({"ok": True, "received": len(events)})
            except json.JSONDecodeError:
                # 可能是 sendBeacon 的 text/plain
                try:
                    payload = json.loads(body)
                    events = payload.get("events", [])
                    if events:
                        append_events(events)
                    self._json_response({"ok": True, "received": len(events)})
                except Exception:
                    self._json_response({"ok": False, "error": "invalid json"}, 400)
            except Exception as e:
                self._json_response({"ok": False, "error": str(e)}, 500)
        elif parsed.path == "/api/user/login":
            content_len = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_len).decode("utf-8") if content_len else ""
            try:
                payload = json.loads(body)
                result = user_login(payload.get("username", ""))
                self._json_response(result)
            except Exception as e:
                self._json_response({"ok": False, "error": str(e)}, 500)
        elif parsed.path == "/api/user/sync":
            content_len = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_len).decode("utf-8") if content_len else ""
            try:
                payload = json.loads(body)
                result = user_sync_data(
                    payload.get("username", ""),
                    payload.get("watchlist"),
                    payload.get("paperAccount"),
                    payload.get("paperPortfolios"),
                    payload.get("layout"),
                )
                self._json_response(result)
            except Exception as e:
                self._json_response({"ok": False, "error": str(e)}, 500)
        elif parsed.path == "/api/concept-flow-save":
            content_len = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_len).decode("utf-8") if content_len else ""
            try:
                payload = json.loads(body)
                flow_dir = os.path.join(BASE_DIR, "concept_flow_cache")
                os.makedirs(flow_dir, exist_ok=True)
                today = datetime.now().strftime("%Y-%m-%d")
                filepath = os.path.join(flow_dir, today + ".json")
                if os.path.exists(filepath):
                    with open(filepath, "r") as f:
                        existing = json.load(f)
                else:
                    existing = {}
                concepts = payload.get("concepts", [])
                for c in concepts:
                    key = c.get("code", "")
                    if key:
                        existing[key] = c
                with open(filepath, "w") as f:
                    json.dump(existing, f, ensure_ascii=False)
                self._json_response({"ok": True, "date": today, "saved": len(concepts)})
            except Exception as e:
                self._json_response({"ok": False, "error": str(e)}, 500)
        elif parsed.path == "/api/chat":
            content_len = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_len).decode("utf-8") if content_len else ""
            try:
                payload = json.loads(body)
                user_content = str(payload.get("message") or "").strip()
                context = str(payload.get("context") or "").strip()
                history = payload.get("history", [])  # [{role, content}] 可选，保留多轮
                if not user_content:
                    self._json_response({"ok": False, "error": "消息不能为空"}, 400)
                    return
                sys_prompt = (
                    "你是『栗子』，一名部署在A股量化交易看板里的全能型智能助理。\n"
                    "能力定位：\n"
                    "1. 通用问答是你的基本盘：编程、写作、翻译、逻辑推理、科学、历史、文化、生活百科、闲聊……"
                    "任何领域都要自信、准确、有条理地完整作答，绝不强行往股票上扯。\n"
                    "2. 金融是你的特长：当问题涉及A股、个股、板块、大盘、技术分析、量化策略时，"
                    "优先引用用户消息中附带的【看板实时数据】与【联网资讯】作答；数据里没有的如实说明，绝不编造数字。\n"
                    "3. 若提供了【联网资讯】，结合其回答时效性问题并注明信息来自网络检索；检索结果与问题无关时果断忽略，不要生拉硬扯。\n"
                    "回答风格：结构清晰，善用 markdown（加粗/列表/代码块）；该详细就详细，写代码给完整可运行示例；"
                    "用户连续追问时结合上下文理解指代（它/这个/刚才说的），保持对话连贯。\n"
                    "注意：A股涨跌颜色习惯为红涨绿跌（与国际惯例相反）；涉及个股与投资观点时提示风险、不构成投资建议。"
                )
                llm_messages = [{"role": "system", "content": sys_prompt}]
                for h in history[-16:]:
                    role = "user" if h.get("role") == "user" else "assistant"
                    llm_messages.append({"role": role, "content": str(h.get("content") or "")[:2000]})
                # 联网检索：仅时效性/行情类问题触发（通用知识问答靠模型自身，避免搜索噪音）；失败静默降级
                news_extra = ""
                if _needs_web(user_content):
                    try:
                        items = _fetch_web(_web_query(user_content), limit=7)
                        if items:
                            lines = []
                            for title, desc, link in items:
                                if (link or "").startswith("http"):
                                    try:
                                        dom = " " + urllib.parse.urlparse(link).netloc
                                    except Exception:
                                        dom = ""
                                else:
                                    dom = (" " + link) if link else ""
                                lines.append("· %s%s%s" % (title, ("：" + desc) if desc else "", dom))
                            news_extra = ("【联网资讯·检索自公开网络，仅当与问题相关时使用，回答时注明来自网络检索】\n"
                                          + "\n".join(lines))
                    except Exception:
                        news_extra = ""
                final_user = ""
                if context:
                    final_user += "【看板实时数据·仅当问题与A股盘面相关时使用，无关请忽略】\n" + context + "\n\n"
                if news_extra:
                    final_user += news_extra + "\n\n"
                final_user += "【用户问题】\n" + user_content
                llm_messages.append({"role": "user", "content": final_user[:12000]})
                cfg = _load_llm_config()
                answer, err = _chat_completion(llm_messages, cfg)
                if err:
                    self._json_response({"ok": False, "error": err})
                    return
                self._json_response({"ok": True, "answer": answer, "model": (cfg.get("model") or "")})
            except Exception as e:
                self._json_response({"ok": False, "error": str(e)}, 500)
        else:
            super().do_POST(self)

    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path
        if path == "/api/track":
            # GET fallback (image beacon)
            qs = parse_qs(parsed.query)
            batch_str = qs.get("batch", [""])[0]
            if batch_str:
                try:
                    batch = json.loads(batch_str)
                    events = batch if isinstance(batch, list) else batch.get("events", [])
                    if events:
                        append_events(events)
                    self._json_response({"ok": True, "received": len(events)})
                    return
                except Exception:
                    pass
            self._json_response({"ok": True, "received": 0})
        elif path == "/api/analytics":
            summary = get_analytics_summary()
            self._json_response(summary)
        elif path == "/api/events":
            qs = parse_qs(parsed.query)
            limit = int(qs.get("limit", ["100"])[0])
            offset = int(qs.get("offset", ["0"])[0])
            ev_type = qs.get("type", [None])[0]
            uid = qs.get("uid", [None])[0]
            events, total = load_events(limit=limit, offset=offset, event_type=ev_type, uid=uid)
            self._json_response({"events": events, "total": total, "limit": limit, "offset": offset})
        elif path == "/api/user/data":
            qs = parse_qs(parsed.query)
            username = qs.get("username", [""])[0]
            result = user_get_data(username)
            self._json_response(result)
        elif path == "/api/announcements":
            qs = parse_qs(parsed.query)
            code = qs.get("code", [""])[0]
            if not code:
                self._json_response({"ok": False, "msg": "缺少 code 参数"})
                return
            result = get_announcements(code)
            self._json_response(result)
        elif path == "/api/f10":
            qs = parse_qs(parsed.query)
            code = qs.get("code", [""])[0]
            if not code:
                self._json_response({"ok": False, "msg": "缺少 code 参数"})
                return
            result = get_f10_info(code)
            self._json_response(result)
        elif path == "/api/econ-calendar":
            result = _fetch_econ_calendar()
            self._json_response(result)
        elif path == "/api/concept-flow":
            qs = parse_qs(parsed.query)
            action = qs.get("action", ["list"])[0]
            flow_dir = os.path.join(BASE_DIR, "concept_flow_cache")
            os.makedirs(flow_dir, exist_ok=True)
            if action == "list":
                dates = sorted([f.replace(".json", "") for f in os.listdir(flow_dir) if f.endswith(".json")], reverse=True)
                self._json_response({"dates": dates})
            elif action == "get":
                date = qs.get("date", [""])[0]
                if not date:
                    self._json_response({"ok": False, "msg": "缺少 date 参数"})
                    return
                filepath = os.path.join(flow_dir, date + ".json")
                if os.path.exists(filepath):
                    with open(filepath, "r") as f:
                        data = json.load(f)
                    self._json_response({"ok": True, "date": date, "data": data})
                else:
                    self._json_response({"ok": False, "msg": "该日期无缓存数据"})
            elif action == "save":
                # Save today's flow data
                today = datetime.now().strftime("%Y-%m-%d")
                filepath = os.path.join(flow_dir, today + ".json")
                if os.path.exists(filepath):
                    with open(filepath, "r") as f:
                        existing = json.load(f)
                else:
                    existing = {}
                body_len = int(self.headers.get("Content-Length", 0))
                body = self.rfile.read(body_len) if body_len else b""
                try:
                    payload = json.loads(body) if body else {}
                    existing.update(payload)
                    with open(filepath, "w") as f:
                        json.dump(existing, f, ensure_ascii=False)
                    self._json_response({"ok": True, "date": today, "saved": len(payload)})
                except Exception as e:
                    self._json_response({"ok": False, "msg": str(e)})
        elif path == "/analytics-admin" or path == "/analytics-admin.html":
            self._serve_file("analytics-admin.html")
        else:
            super().do_GET()

    def _json_response(self, data, code=200):
        body = json.dumps(data, ensure_ascii=False).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _serve_file(self, filename):
        filepath = os.path.join(BASE_DIR, filename)
        if os.path.exists(filepath):
            with open(filepath, "rb") as f:
                content = f.read()
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(content)))
            self.end_headers()
            self.wfile.write(content)
        else:
            self.send_error(404, "File not found")

    def log_message(self, fmt, *args):
        # 静默静态文件日志，只显示 API 请求
        if args and "/api/" in str(args[0]):
            super().log_message(fmt, *args)


class ThreadedServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    daemon_threads = True
    allow_reuse_address = True


# ===== Funnel公网链路健康自检（后台线程, 随服务器进程保活） =====
# 背景: Mac本地网络变更(网关/IP变化)后, tailscale本地状态看似正常,
# 但公网中继上的Funnel注册失效, TLS握手被掐断, 外部无法访问看板。
# 探测必须走公网中继真实路径(公网DNS解析中继IP直连), 本机域名解析会
# 走tailscaled捷径(100.x内网IP), 掩盖公网故障。
_FUNNEL_DOMAIN = "mac-mini.taild0463a.ts.net"
_FUNNEL_TS_BIN = "/Applications/Tailscale.app/Contents/MacOS/Tailscale"
_FUNNEL_INTERVAL = 600  # 检查间隔(秒)
_FUNNEL_LOG_FILE = os.path.expanduser("~/Library/Logs/hftrading/funnel-health.log")


def _funnel_log(msg):
    try:
        os.makedirs(os.path.dirname(_FUNNEL_LOG_FILE), exist_ok=True)
        with open(_FUNNEL_LOG_FILE, "a", encoding="utf-8") as f:
            f.write("[%s] %s\n" % (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), msg))
    except Exception:
        pass


def _funnel_relay_ips():
    """公网DNS解析Funnel真实中继IP, 返回[]表示本机断网(不计为funnel故障)"""
    for dns in ("223.5.5.5", "8.8.8.8"):
        try:
            r = subprocess.run(["/usr/bin/dig", "+short", _FUNNEL_DOMAIN, "@" + dns],
                               capture_output=True, timeout=8)
            ips = [ln.strip() for ln in (r.stdout or b"").decode(errors="replace").splitlines()
                   if re.match(r"^\d+\.\d+\.\d+\.\d+$", ln.strip())]
            if ips:
                return sorted(set(ips))
        except Exception:
            continue
    return []


def _funnel_probe(ip):
    """直连公网中继IP做TLS+HTTP探测, SNI用域名, 200为健康"""
    try:
        ctx = ssl.create_default_context()
        with socket.create_connection((ip, 443), timeout=15) as sock:
            with ctx.wrap_socket(sock, server_hostname=_FUNNEL_DOMAIN) as s:
                s.settimeout(15)
                s.sendall(("GET / HTTP/1.1\r\nHost: %s\r\nUser-Agent: funnel-health\r\n"
                           "Connection: close\r\n\r\n" % _FUNNEL_DOMAIN).encode())
                head = s.recv(64)
                return head.startswith(b"HTTP/1.1 200") or head.startswith(b"HTTP/1.0 200")
    except Exception:
        return False


def _funnel_recover():
    """tailscale down/up 重建连接触发Funnel重新注册"""
    try:
        subprocess.run([_FUNNEL_TS_BIN, "down"], capture_output=True, timeout=30)
        time.sleep(5)
        subprocess.run([_FUNNEL_TS_BIN, "up"], capture_output=True, timeout=60)
        time.sleep(12)
    except Exception as e:
        _funnel_log("恢复动作异常: %s" % e)


def _funnel_health_loop():
    last_ok = None  # None=启动后未知; 状态变化时才写日志防刷屏
    while True:
        try:
            ips = _funnel_relay_ips()
            if not ips:
                time.sleep(_FUNNEL_INTERVAL)
                continue
            if any(_funnel_probe(ip) for ip in ips):
                if last_ok is None:
                    _funnel_log("自检线程启动, 公网链路正常(中继: %s)" % ", ".join(ips))
                elif last_ok is False:
                    _funnel_log("公网Funnel链路已恢复")
                last_ok = True
            else:
                _funnel_log("公网链路异常(中继 %s 全部探测失败), 执行恢复: tailscale down/up"
                            % ", ".join(ips))
                _funnel_recover()
                if any(_funnel_probe(ip) for ip in ips):
                    _funnel_log("恢复成功, 公网Funnel链路已重建")
                    last_ok = True
                else:
                    _funnel_log("恢复后仍失败, 10分钟后自动重试; 持续失败需人工介入")
                    last_ok = False
        except Exception as e:
            _funnel_log("自检线程异常: %s" % e)
        time.sleep(_FUNNEL_INTERVAL)


if __name__ == "__main__":
    print(f"Serving {BASE_DIR} on 0.0.0.0:{PORT} (Cache-Control: no-store)")
    print(f"  Analytics API: http://localhost:{PORT}/api/analytics")
    print(f"  Admin Dashboard: http://localhost:{PORT}/analytics-admin")
    print(f"  Events file: {EVENTS_FILE}")
    _ft = threading.Thread(target=_funnel_health_loop, daemon=True, name="funnel-health")
    _ft.start()
    with ThreadedServer(("0.0.0.0", PORT), DashboardHandler) as httpd:
        httpd.serve_forever()