#!/usr/bin/env python3
"""Mid-year point-in-time refresh for every active stock valuation.

This is a valuation-analyst task utility, not a second production service.  It
uses the accepted MySQL ledger as the universe, Eastmoney's registered
structured financial mirror for report facts, and dated institution details
plus business identity from 10jqka.  Raw responses remain in the valuation
temporary area; formal outputs are written by the later revalue/publish phase.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import re
import subprocess
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
from typing import Any

import requests
from bs4 import BeautifulSoup


ROOT = Path(__file__).resolve().parents[2]
BATCH_ID = "BATCH-STOCK-VALUATION-MIDYEAR-FULL-REFRESH-20260826-001"
TMP_ROOT = ROOT / "ana-data/tmp/股票估值" / BATCH_ID
RAW_ROOT = TMP_ROOT / "raw"
CASE_ROOT = ROOT / "ana-data/cases/股票估值" / BATCH_ID
AS_OF = "2026-08-26"
PRICE_DATE = "2026-08-25"
FINANCE_URL = "https://datacenter-web.eastmoney.com/api/data/v1/get"
THS_BASE = "https://basic.10jqka.com.cn"
FINANCE_REPORTS = {
    "main": "RPT_F10_FINANCE_MAINFINADATA",
    "income": "RPT_DMSK_FN_INCOME",
    "balance": "RPT_F10_FINANCE_GBALANCE",
    "cashflow": "RPT_DMSK_FN_CASHFLOW",
}
THREAD_STATE = threading.local()


def mysql_rows(sql: str) -> list[dict[str, str]]:
    proc = subprocess.run(
        [
            "mysql",
            "--login-path=ana_semi_admin_preflight",
            "--default-character-set=utf8mb4",
            "--batch",
            "--raw",
            "-e",
            sql,
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="strict",
        check=True,
    )
    lines = proc.stdout.splitlines()
    if not lines:
        return []
    reader = csv.DictReader(lines, delimiter="\t")
    return list(reader)


def universe() -> list[dict[str, Any]]:
    sql = f"""
SELECT s.ticker,s.company,s.market,s.currency,
       v.valuation_id,v.valuation_date,v.method,v.base_low,v.base_high,
       v.optimistic_low,v.optimistic_high,v.normalized_profit,
       v.normalized_pe,v.pb,v.ps,v.consensus_year,v.consensus_profit,
       v.consensus_count,v.report_path,v.snapshot_path,v.source_hash,
       p.close,p.trade_date,
       JSON_UNQUOTE(JSON_EXTRACT(i.raw_json,'$.TotalVolume')) AS static_total_shares
FROM stock_valuation.security s
JOIN stock_valuation.valuation_version v
  ON v.ticker=s.ticker AND v.active_to IS NULL
LEFT JOIN stock_valuation.daily_price p
  ON p.ticker=s.ticker AND p.trade_date='{PRICE_DATE}'
LEFT JOIN trading_xuntou.cn_stock_instrument_static i
  ON BINARY i.symbol=BINARY s.ticker
WHERE s.active=1
ORDER BY s.ticker
"""
    rows = mysql_rows(sql)
    for row in rows:
        for key in (
            "base_low", "base_high", "optimistic_low", "optimistic_high",
            "normalized_profit", "normalized_pe", "pb", "ps",
            "consensus_profit", "close", "static_total_shares",
        ):
            value = row.get(key)
            row[key] = float(value) if value not in (None, "", "NULL") else None
        row["consensus_count"] = int(row["consensus_count"]) if row.get("consensus_count") not in (None, "", "NULL") else 0
    return rows


def session() -> requests.Session:
    value = getattr(THREAD_STATE, "session", None)
    if value is None:
        value = requests.Session()
        value.headers.update(
            {
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/140 Safari/537.36",
                "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.6",
            }
        )
        THREAD_STATE.session = value
    return value


def get(url: str, *, params: dict[str, str] | None = None, referer: str | None = None) -> bytes:
    headers = {"Referer": referer} if referer else None
    last: Exception | None = None
    for attempt in range(3):
        try:
            response = session().get(url, params=params, headers=headers, timeout=(8, 25))
            response.raise_for_status()
            if not response.content:
                raise RuntimeError("empty response")
            return response.content
        except Exception as exc:
            last = exc
            time.sleep(0.5 * (attempt + 1))
    raise RuntimeError(f"GET failed: {url}: {last}")


def finance_payload(ticker: str, report_name: str) -> tuple[dict[str, Any], str]:
    params = {
        "reportName": report_name,
        "columns": "ALL",
        "filter": f'(SECUCODE="{ticker}")',
        "pageNumber": "1",
        "pageSize": "20",
        "sortTypes": "-1",
        "sortColumns": "REPORT_DATE",
    }
    body = get(FINANCE_URL, params=params)
    payload = json.loads(body.decode("utf-8-sig"))
    return payload, hashlib.sha256(body).hexdigest().upper()


def records(payload: dict[str, Any]) -> list[dict[str, Any]]:
    return list(((payload.get("result") or {}).get("data") or payload.get("data") or []))


def eligible(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    result = []
    for row in rows:
        report_date = str(row.get("REPORT_DATE") or "")[:10]
        notice_date = str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE") or "")[:10]
        if report_date and notice_date and report_date <= AS_OF and notice_date <= AS_OF:
            result.append(row)
    return result


def select_periods(income_rows: list[dict[str, Any]]) -> tuple[str, str, str]:
    periods = sorted({str(row["REPORT_DATE"])[:10] for row in eligible(income_rows)})
    annuals = [period for period in periods if period.endswith("-12-31")]
    if not annuals:
        raise RuntimeError("no annual report")
    annual = annuals[-1]
    current_candidates = [period for period in periods if period > annual and not period.endswith("-12-31")]
    if not current_candidates:
        raise RuntimeError("no current cumulative report")
    current = current_candidates[-1]
    prior = f"{int(current[:4]) - 1}{current[4:]}"
    if prior not in periods:
        raise RuntimeError(f"no comparative period {prior}")
    return annual, current, prior


def pick(rows: list[dict[str, Any]], period: str) -> dict[str, Any]:
    matches = [row for row in eligible(rows) if str(row["REPORT_DATE"])[:10] == period]
    if not matches:
        raise RuntimeError(f"missing {period}")
    return max(matches, key=lambda row: str(row.get("NOTICE_DATE") or row.get("UPDATE_DATE") or ""))


def compact(row: dict[str, Any], keys: tuple[str, ...]) -> dict[str, Any]:
    return {key: row.get(key) for key in keys if key in row}


INCOME_KEYS = (
    "REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE", "SECURITY_CODE", "SECURITY_NAME_ABBR",
    "TOTAL_OPERATE_INCOME", "TOTALOPERATEREVE", "PARENT_NETPROFIT", "PARENTNETPROFIT",
    "DEDUCT_PARENT_NETPROFIT", "KCFJCXSYJLR", "BASIC_EPS", "DILUTED_EPS",
)
CASHFLOW_KEYS = ("REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE", "NETCASH_OPERATE", "CONSTRUCT_LONG_ASSET")
BALANCE_KEYS = (
    "REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE", "TOTAL_EQUITY", "TOTAL_EQUITY_PARENT",
    "MONETARYFUNDS", "MINORITY_EQUITY", "TOTAL_ASSETS", "TOTAL_LIABILITIES",
    "SHORT_LOAN", "NONCURRENT_LIAB_1YEAR", "LONG_LOAN", "BOND_PAYABLE", "LEASE_LIAB",
    "SHORT_BOND_PAYABLE", "TRADE_FINASSET_NOTFVTPL", "TRADE_FINASSET", "FVTPL_FINASSET",
    "APPOINT_FVTPL_FINASSET", "AVAILABLE_SALE_FINASSET", "DERIVE_FINASSET", "BUY_RESALE_FINASSET",
)
MAIN_KEYS = (
    "REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE", "TOTAL_SHARE", "TOTAL_EQUITY_PARENT",
    "ROEJQ", "XSMLL", "ZCFZL",
)


def parse_money(text: str) -> float | None:
    cleaned = text.replace(",", "").strip()
    if cleaned in {"", "--", "-"}:
        return None
    match = re.search(r"(-?\d+(?:\.\d+)?)\s*([万亿]?)", cleaned)
    if not match:
        return None
    value = float(match.group(1))
    return value * ({"亿": 1e8, "万": 1e4}.get(match.group(2), 1.0))


def parse_forecast_page(body: bytes) -> tuple[list[dict[str, Any]], str]:
    text = body.decode("gb18030", errors="replace")
    soup = BeautifulSoup(text, "lxml")
    root = soup.find(id="forecastdetail")
    rows: list[dict[str, Any]] = []
    if root:
        table = root.find("table")
        if table:
            for tr in table.select("tbody tr"):
                cells = [cell.get_text(" ", strip=True) for cell in tr.find_all(["th", "td"])]
                if len(cells) < 9 or not re.fullmatch(r"20\d{2}-\d{2}-\d{2}", cells[-1]):
                    continue
                rows.append(
                    {
                        "institution": cells[0],
                        "analyst": cells[1],
                        "eps_2026": parse_money(cells[2]),
                        "eps_2027": parse_money(cells[3]),
                        "eps_2028": parse_money(cells[4]),
                        "profit_2026": parse_money(cells[5]),
                        "profit_2027": parse_money(cells[6]),
                        "profit_2028": parse_money(cells[7]),
                        "report_date": cells[8],
                    }
                )
    return rows, hashlib.sha256(body).hexdigest().upper()


def extract_after(text: str, label: str, next_labels: tuple[str, ...]) -> str:
    start = text.find(label)
    if start < 0:
        return ""
    value = text[start + len(label):]
    stops = [value.find(item) for item in next_labels if value.find(item) >= 0]
    return value[: min(stops) if stops else len(value)].strip()


def parse_business_page(body: bytes) -> tuple[dict[str, str], str]:
    text = body.decode("gb18030", errors="replace")
    soup = BeautifulSoup(text, "lxml")
    visible = re.sub(r"\s+", " ", soup.get_text(" ", strip=True))
    main = extract_after(visible, "主营业务：", ("产品类型：", "产品名称：", "经营范围："))
    product_types = extract_after(visible, "产品类型：", ("产品名称：", "经营范围："))
    product_names = extract_after(visible, "产品名称：", ("经营范围：", "主营构成分析"))
    return {
        "main_business": main,
        "product_types": product_types,
        "product_names": product_names,
    }, hashlib.sha256(body).hexdigest().upper()


def fetch_one(item: dict[str, Any]) -> dict[str, Any]:
    ticker = item["ticker"]
    code = ticker.split(".")[0]
    if ticker.endswith(".HK"):
        return {"ticker": ticker, "status": "UNSUPPORTED_HK", "errors": ["A股公开源不覆盖港股"]}
    errors: list[str] = []
    payloads: dict[str, Any] = {}
    hashes: dict[str, str] = {}
    try:
        for kind, report_name in FINANCE_REPORTS.items():
            payloads[kind], hashes[f"finance_{kind}"] = finance_payload(ticker, report_name)
        income_rows = records(payloads["income"])
        annual, current, prior = select_periods(income_rows)
        selected = {
            "annual_income": compact(pick(income_rows, annual), INCOME_KEYS),
            "current_income": compact(pick(income_rows, current), INCOME_KEYS),
            "prior_income": compact(pick(income_rows, prior), INCOME_KEYS),
            "annual_cashflow": compact(pick(records(payloads["cashflow"]), annual), CASHFLOW_KEYS),
            "current_cashflow": compact(pick(records(payloads["cashflow"]), current), CASHFLOW_KEYS),
            "prior_cashflow": compact(pick(records(payloads["cashflow"]), prior), CASHFLOW_KEYS),
            "current_balance": compact(pick(records(payloads["balance"]), current), BALANCE_KEYS),
            "current_main": compact(pick(records(payloads["main"]), current), MAIN_KEYS),
        }
    except Exception as exc:
        errors.append(f"finance:{type(exc).__name__}:{exc}")
        selected = {}
        annual = current = prior = ""
        # Banks and a few financial issuers can omit the industrial cash-flow
        # comparative row.  Preserve income/balance evidence so their PB model
        # can still be refreshed without inventing cash flow.
        try:
            income_rows = records(payloads.get("income") or {})
            annual, current, prior = select_periods(income_rows)
            selected = {
                "annual_income": compact(pick(income_rows, annual), INCOME_KEYS),
                "current_income": compact(pick(income_rows, current), INCOME_KEYS),
                "prior_income": compact(pick(income_rows, prior), INCOME_KEYS),
                "current_main": compact(pick(records(payloads.get("main") or {}), current), MAIN_KEYS),
            }
            balance_rows = records(payloads.get("balance") or {})
            if balance_rows:
                selected["current_balance"] = compact(pick(balance_rows, current), BALANCE_KEYS)
            cash_rows = records(payloads.get("cashflow") or {})
            if cash_rows:
                selected.update(
                    {
                        "annual_cashflow": compact(pick(cash_rows, annual), CASHFLOW_KEYS),
                        "current_cashflow": compact(pick(cash_rows, current), CASHFLOW_KEYS),
                        "prior_cashflow": compact(pick(cash_rows, prior), CASHFLOW_KEYS),
                    }
                )
        except Exception as partial_exc:
            errors.append(f"finance_partial:{type(partial_exc).__name__}:{partial_exc}")
    try:
        worth = get(f"{THS_BASE}/{code}/worth.html", referer=f"{THS_BASE}/{code}/")
        forecasts, hashes["institution_detail"] = parse_forecast_page(worth)
    except Exception as exc:
        forecasts = []
        errors.append(f"forecast:{type(exc).__name__}:{exc}")
    try:
        operate = get(f"{THS_BASE}/{code}/operate.html", referer=f"{THS_BASE}/{code}/")
        business, hashes["business_identity"] = parse_business_page(operate)
    except Exception as exc:
        business = {"main_business": "", "product_types": "", "product_names": ""}
        errors.append(f"business:{type(exc).__name__}:{exc}")
    result = {
        "ticker": ticker,
        "company": item["company"],
        "status": "OK" if not errors else ("PARTIAL" if selected else "FAILED"),
        "as_of": AS_OF,
        "periods": {"annual": annual, "current": current, "prior": prior},
        "selected": selected,
        "forecasts": [row for row in forecasts if row["report_date"] <= AS_OF],
        "business": business,
        "raw_sha256": hashes,
        "errors": errors,
    }
    return result


def write_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def action_inventory() -> None:
    rows = universe()
    TMP_ROOT.mkdir(parents=True, exist_ok=True)
    columns = list(rows[0])
    with (TMP_ROOT / "universe.csv").open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns)
        writer.writeheader()
        writer.writerows(rows)
    write_json(
        TMP_ROOT / "inventory.json",
        {
            "batch_id": BATCH_ID,
            "as_of": AS_OF,
            "price_date": PRICE_DATE,
            "created_at": datetime.now().astimezone().isoformat(),
            "universe_count": len(rows),
            "a_share_count": sum(not row["ticker"].endswith(".HK") for row in rows),
            "price_covered": sum(row["close"] is not None for row in rows),
        },
    )
    print(json.dumps({"universe": len(rows), "price_covered": sum(row["close"] is not None for row in rows)}, ensure_ascii=False))


def action_fetch(workers: int) -> None:
    rows = universe()
    RAW_ROOT.mkdir(parents=True, exist_ok=True)
    completed: list[dict[str, Any]] = []
    started = time.monotonic()
    with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="midyear-refresh") as pool:
        futures = {pool.submit(fetch_one, item): item for item in rows}
        for index, future in enumerate(as_completed(futures), 1):
            item = futures[future]
            try:
                result = future.result()
            except Exception as exc:
                result = {"ticker": item["ticker"], "company": item["company"], "status": "FAILED", "errors": [repr(exc)]}
            write_json(RAW_ROOT / f"{result['ticker'].replace('.', '_')}.json", result)
            completed.append(result)
            if index % 50 == 0 or index == len(rows):
                print(f"[{index}/{len(rows)}] elapsed={time.monotonic()-started:.1f}s", flush=True)
    counts: dict[str, int] = {}
    for row in completed:
        counts[row["status"]] = counts.get(row["status"], 0) + 1
    write_json(
        TMP_ROOT / "fetch_manifest.json",
        {
            "batch_id": BATCH_ID,
            "as_of": AS_OF,
            "price_date": PRICE_DATE,
            "finished_at": datetime.now().astimezone().isoformat(),
            "wall_seconds": round(time.monotonic() - started, 3),
            "counts": counts,
            "files": len(completed),
        },
    )
    print(json.dumps(counts, ensure_ascii=False))


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("action", choices=("inventory", "fetch"))
    parser.add_argument("--workers", type=int, default=16)
    args = parser.parse_args()
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    if args.action == "inventory":
        action_inventory()
    else:
        action_fetch(args.workers)


if __name__ == "__main__":
    main()