#!/usr/bin/env python3
"""Fetch V2 evidence packages for uncovered industry-research securities.

The target list is derived at runtime from the four governed research pools and
the MySQL valuation ledger.  Fetches run concurrently, while each V2 process
retains its own 90-second monotonic deadline and writes only to the analyst's
temporary workspace.
"""

from __future__ import annotations

import csv
import json
import os
import re
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import pymysql


ROOT = Path(__file__).resolve().parents[2]
AS_OF = "2026-08-13"
OUT = ROOT / "ai-valuation-analyst" / "tmp" / "valuation_industry_research_gaps_20260814"
CACHE = ROOT / "ai-valuation-analyst" / "tmp" / "v2_cache"


def mysql_connection(database: str) -> pymysql.Connection:
    return pymysql.connect(
        host=os.environ.get("MYSQL_HOST", "127.0.0.1"),
        port=int(os.environ.get("MYSQL_PORT", "3306")),
        user=os.environ.get("MYSQL_USER", "root"),
        password=os.environ["MYSQL_PASSWORD"],
        database=database,
        charset="utf8mb4",
        autocommit=True,
        cursorclass=pymysql.cursors.DictCursor,
    )


def semiconductor_tickers() -> set[str]:
    path = ROOT / "ana-data/cases/半导体案例/ANA-SEMI-20260722-001/outputs/核心文档/国内企业信息表.csv"
    market_map = {"SSE": "SH", "SSE STAR": "SH", "SZSE": "SZ", "SZSE CHINEXT": "SZ", "BSE": "BJ"}
    rows = csv.DictReader(path.open(encoding="utf-8-sig", newline=""))
    result = set()
    for row in rows:
        code = row["ticker"].strip()
        suffix = market_map.get(row["listed_market"].strip())
        if suffix and re.fullmatch(r"\d{6}", code):
            result.add(f"{code}.{suffix}")
    # The North Exchange completed its 920-series code migration after the
    # research snapshot.  The local formal market contract uses the new code.
    result.discard("835179.BJ")
    result.add("920179.BJ")
    return result


def robot_tickers() -> set[str]:
    path = ROOT / "ana-data/cases/机器人案例/ANA-ROBOT-INDUSTRY-001/evidence/next_robot_036_market_database_company_universe_authority_20260726.csv"
    return {row["symbol"].strip() for row in csv.DictReader(path.open(encoding="utf-8-sig", newline=""))}


def newenergy_tickers() -> set[str]:
    path = ROOT / "ana-data/cases/新能源案例/核心文档/全量公司横向总表.md"
    result = set()
    for code in re.findall(r"\| (\d{6}) \| \[", path.read_text(encoding="utf-8")):
        suffix = "SH" if code.startswith("6") else ("BJ" if code.startswith(("8", "9")) else "SZ")
        result.add(f"{code}.{suffix}")
    return result


def missing_tickers() -> list[str]:
    universe = semiconductor_tickers() | robot_tickers() | newenergy_tickers()
    with mysql_connection("stock_valuation") as connection, connection.cursor() as cursor:
        cursor.execute("SELECT ticker FROM security")
        covered = {row["ticker"] for row in cursor.fetchall()}
    return sorted(universe - covered)


def fetch(ticker: str) -> dict[str, object]:
    target = OUT / f"v2_{ticker.replace('.', '_')}"
    env = os.environ.copy()
    env["PYTHONPATH"] = str(ROOT / "dev/project-dev")
    proc = subprocess.run(
        [
            sys.executable,
            "-m",
            "stock_valuation_pipeline_v2",
            "--ticker",
            ticker,
            "--as-of",
            AS_OF,
            "--cache-dir",
            str(CACHE),
            "--output-dir",
            str(target),
        ],
        cwd=ROOT,
        env=env,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=120,
    )
    return {
        "ticker": ticker,
        "returncode": proc.returncode,
        "stdout": proc.stdout.strip().splitlines()[-1] if proc.stdout.strip() else "",
        "stderr": proc.stderr.strip()[-500:],
    }


def latest_package_exists(ticker: str) -> bool:
    return any(OUT.glob(f"v2_{ticker.replace('.', '_')}.failed-*")) or (OUT / f"v2_{ticker.replace('.', '_')}").exists()


def main() -> None:
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    tickers = missing_tickers()
    if "--missing-packages-only" in sys.argv:
        tickers = [ticker for ticker in tickers if not latest_package_exists(ticker)]
    OUT.mkdir(parents=True, exist_ok=True)
    rows: list[dict[str, object]] = []
    with ThreadPoolExecutor(max_workers=12) as pool:
        pending = {pool.submit(fetch, ticker): ticker for ticker in tickers}
        for future in as_completed(pending):
            ticker = pending[future]
            try:
                row = future.result()
            except Exception as exc:  # retain every batch failure
                row = {"ticker": ticker, "returncode": 99, "error": repr(exc)}
            rows.append(row)
            print(f"[{len(rows):03d}/{len(tickers):03d}] {ticker} rc={row.get('returncode')}", flush=True)
    rows.sort(key=lambda item: str(item["ticker"]))
    manifest = {"as_of": AS_OF, "count": len(rows), "items": rows}
    (OUT / "fetch_manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps({"count": len(rows), "returncodes": {str(code): sum(row.get("returncode") == code for row in rows) for code in sorted({int(row.get("returncode", 99)) for row in rows})}}, ensure_ascii=False))


if __name__ == "__main__":
    main()