from __future__ import annotations

import argparse
import getpass
import hashlib
import json
import os
import re
from datetime import datetime
from typing import Iterable

import pymysql


ARCHIVE_BATCH_ID = "DL_FULL_ARCHIVE_20260624_V1"
SOURCE_ID = "SOURCE_LEGACY_EXTERNAL_INFORMATION_CASEBOOK"
SOURCE_PATH = "legacy_source/external_information_casebook.md"


def sha256_text(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def get_password(args: argparse.Namespace) -> str:
    if args.password:
        return args.password
    if os.environ.get("DARKLINE_MYSQL_PASSWORD"):
        return os.environ["DARKLINE_MYSQL_PASSWORD"]
    if os.environ.get("TIANXIA_MYSQL_PASSWORD"):
        return os.environ["TIANXIA_MYSQL_PASSWORD"]
    return getpass.getpass("MySQL password: ")


def connect(args: argparse.Namespace):
    return pymysql.connect(
        host=args.host,
        port=args.port,
        user=args.user,
        password=get_password(args),
        database=args.database,
        charset="utf8mb4",
        autocommit=True,
        cursorclass=pymysql.cursors.DictCursor,
    )


def normalize_text(text: str) -> str:
    return re.sub(r"\s+", " ", text.strip())


def compact(text: str, limit: int = 480) -> str:
    text = normalize_text(text)
    return text if len(text) <= limit else text[: limit - 3] + "..."


def split_reasoning_units(body: str) -> list[dict[str, str]]:
    units: list[dict[str, str]] = []
    current_heading = ""
    for raw in body.splitlines():
        line = raw.strip()
        if not line:
            continue
        if line.startswith("#"):
            current_heading = line.strip("# ").strip()
            units.append({"heading": current_heading, "text": line, "is_heading": "1"})
            continue
        # Preserve bullets, numbered lines and dense paragraphs. They usually carry
        # the reasoning chain in our darkline casebook.
        if len(line) >= 8:
            units.append({"heading": current_heading, "text": line, "is_heading": "0"})
    return units


def classify_step_type(text: str) -> str:
    lower = text.lower()
    if re.search(r"为什么|為什麼|why", text, flags=re.I):
        return "WHY_NODE"
    if re.search(r"暗线|意图|目标|人心|组织意志|反推|操纵|主力", text):
        return "DARKLINE_INTENTION"
    if re.search(r"新闻|公告|事件|看到|叶子|b/c|初始", lower):
        return "INITIAL_LEAF"
    if re.search(r"之前|前置|铺垫|早已|提前|m节点|M：|M:", text):
        return "PRIOR_EXPECTED_LINE"
    if re.search(r"后续|接下来|将来|下一步|应该发生|n/l|N：|L：|N:|L:", text):
        return "FOLLOWUP_EXPECTED_LINE"
    if re.search(r"K线|股价|涨停|跌停|上涨|下跌|放量|缩量|换手|市场|显影|输出", text, flags=re.I):
        return "MARKET_MANIFESTATION"
    if re.search(r"替代解释|也可能|可能是|不是|风险|缺口|不确定|held|review", lower):
        return "ALTERNATIVE_OR_GAP"
    if re.search(r"总结|结论|启发|当前读法|收口", text):
        return "SUMMARY_READOUT"
    if text.startswith("#"):
        return "SECTION_HEADING"
    return "BODY_REASONING"


def evidence_type_for_step(step_type: str) -> str:
    mapping = {
        "INITIAL_LEAF": "INITIAL_INFORMATION_LEAF",
        "DARKLINE_INTENTION": "INTENTION_REASONING",
        "WHY_NODE": "WHY_REASONING",
        "PRIOR_EXPECTED_LINE": "PRIOR_NODE_REASONING",
        "FOLLOWUP_EXPECTED_LINE": "FOLLOWUP_NODE_REASONING",
        "MARKET_MANIFESTATION": "MARKET_MANIFESTATION_TEXT",
        "ALTERNATIVE_OR_GAP": "ALTERNATIVE_EXPLANATION_TEXT",
        "SUMMARY_READOUT": "SUMMARY_TEXT",
    }
    return mapping.get(step_type, "CASEBOOK_TEXT")


def extract_symbols(text: str) -> list[str]:
    symbols = sorted(set(re.findall(r"\b(?:[036]\d{5}|688\d{3}|8\d{5})\.(?:SZ|SH|BJ)\b", text)))
    return symbols


def has_event_signal(text: str) -> bool:
    return bool(
        re.search(
            r"\d{4}[-/年]\d{1,2}[-/月]\d{0,2}|公告|发布|披露|立案|减持|回购|增持|诉讼|仲裁|并购|重组|订单|涨停|跌停",
            text,
        )
    )


def has_market_signal(text: str) -> bool:
    return bool(re.search(r"K线|股价|涨停|跌停|上涨|下跌|放量|缩量|换手|D[0-9]+|收益|显影|拉升|砸盘", text, flags=re.I))


def contains_expected_line(step_type: str, text: str) -> bool:
    if step_type in {"PRIOR_EXPECTED_LINE", "FOLLOWUP_EXPECTED_LINE", "WHY_NODE"}:
        return True
    return bool(re.search(r"如果.*应该|理论上.*会|后续.*会|之前.*应该|预期|应有之线", text))


def create_tables(cur) -> None:
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS dl_source_document (
            source_id VARCHAR(128) PRIMARY KEY,
            source_title VARCHAR(512) NOT NULL,
            source_url_or_path VARCHAR(512) NOT NULL,
            source_type VARCHAR(64) NOT NULL,
            publish_time VARCHAR(64) NULL,
            available_time VARCHAR(64) NULL,
            raw_text_hash CHAR(64) NOT NULL,
            source_reliability VARCHAR(64) NOT NULL,
            archive_status VARCHAR(64) NOT NULL
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
        """
    )
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS dl_case_reasoning_step (
            reasoning_step_id VARCHAR(128) PRIMARY KEY,
            case_id VARCHAR(96) NOT NULL,
            darkline_hypothesis_id VARCHAR(96) NOT NULL,
            step_order INT NOT NULL,
            step_type VARCHAR(64) NOT NULL,
            section_heading VARCHAR(512) NULL,
            step_text LONGTEXT NOT NULL,
            extraction_rule VARCHAR(128) NOT NULL,
            archive_batch_id VARCHAR(96) NOT NULL,
            KEY idx_reason_case (case_id),
            KEY idx_reason_type (step_type)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
        """
    )
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS dl_expected_line (
            expected_line_id VARCHAR(128) PRIMARY KEY,
            darkline_hypothesis_id VARCHAR(96) NOT NULL,
            case_id VARCHAR(96) NOT NULL,
            expected_line_type VARCHAR(64) NOT NULL,
            expected_event LONGTEXT NOT NULL,
            expected_timing VARCHAR(128) NULL,
            expected_direction VARCHAR(64) NOT NULL,
            observed_flag TINYINT NOT NULL,
            observed_event_node_id VARCHAR(128) NULL,
            validation_status VARCHAR(64) NOT NULL,
            source_reasoning_step_id VARCHAR(128) NULL,
            archive_batch_id VARCHAR(96) NOT NULL,
            KEY idx_expected_case (case_id),
            KEY idx_expected_hypothesis (darkline_hypothesis_id)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
        """
    )
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS dl_event_node (
            event_node_id VARCHAR(128) PRIMARY KEY,
            darkline_hypothesis_id VARCHAR(96) NOT NULL,
            case_id VARCHAR(96) NOT NULL,
            event_title VARCHAR(512) NOT NULL,
            event_date VARCHAR(64) NULL,
            available_time VARCHAR(64) NULL,
            actor_list TEXT NULL,
            target_list TEXT NULL,
            event_type VARCHAR(96) NOT NULL,
            event_level VARCHAR(32) NOT NULL,
            source_id VARCHAR(128) NOT NULL,
            node_status VARCHAR(64) NOT NULL,
            source_reasoning_step_id VARCHAR(128) NULL,
            archive_batch_id VARCHAR(96) NOT NULL,
            KEY idx_event_case (case_id),
            KEY idx_event_hypothesis (darkline_hypothesis_id)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
        """
    )
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS dl_evidence (
            evidence_id VARCHAR(128) PRIMARY KEY,
            source_id VARCHAR(128) NOT NULL,
            case_id VARCHAR(96) NOT NULL,
            raw_excerpt LONGTEXT NOT NULL,
            evidence_type VARCHAR(96) NOT NULL,
            evidence_strength VARCHAR(64) NOT NULL,
            extracted_time DATETIME NOT NULL,
            evidence_hash CHAR(64) NOT NULL,
            source_reasoning_step_id VARCHAR(128) NULL,
            archive_batch_id VARCHAR(96) NOT NULL,
            KEY idx_evidence_case (case_id),
            KEY idx_evidence_type (evidence_type)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
        """
    )
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS dl_evidence_node_link (
            evidence_node_link_id VARCHAR(160) PRIMARY KEY,
            evidence_id VARCHAR(128) NOT NULL,
            event_node_id VARCHAR(128) NULL,
            darkline_hypothesis_id VARCHAR(96) NOT NULL,
            expected_line_id VARCHAR(128) NULL,
            link_role VARCHAR(64) NOT NULL,
            archive_batch_id VARCHAR(96) NOT NULL,
            KEY idx_link_evidence (evidence_id),
            KEY idx_link_event (event_node_id),
            KEY idx_link_hypothesis (darkline_hypothesis_id)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
        """
    )
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS dl_impact_target (
            impact_target_id VARCHAR(128) PRIMARY KEY,
            darkline_hypothesis_id VARCHAR(96) NOT NULL,
            case_id VARCHAR(96) NOT NULL,
            target_type VARCHAR(64) NOT NULL,
            target_id VARCHAR(96) NOT NULL,
            target_name VARCHAR(256) NULL,
            expected_impact_direction VARCHAR(64) NOT NULL,
            expected_window VARCHAR(128) NULL,
            archive_batch_id VARCHAR(96) NOT NULL,
            KEY idx_target_case (case_id),
            KEY idx_target_hypothesis (darkline_hypothesis_id)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
        """
    )
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS dl_manifestation_bridge (
            bridge_id VARCHAR(128) PRIMARY KEY,
            darkline_hypothesis_id VARCHAR(96) NOT NULL,
            case_id VARCHAR(96) NOT NULL,
            expected_line_id VARCHAR(128) NULL,
            impact_target_id VARCHAR(128) NULL,
            output_layer VARCHAR(96) NOT NULL,
            manifestation_time VARCHAR(64) NULL,
            manifestation_value LONGTEXT NOT NULL,
            control_group_id VARCHAR(128) NULL,
            support_status VARCHAR(64) NOT NULL,
            source_reasoning_step_id VARCHAR(128) NULL,
            archive_batch_id VARCHAR(96) NOT NULL,
            KEY idx_bridge_case (case_id),
            KEY idx_bridge_hypothesis (darkline_hypothesis_id)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
        """
    )
    cur.execute(
        """
        CREATE TABLE IF NOT EXISTS dl_alternative_explanation (
            alternative_id VARCHAR(128) PRIMARY KEY,
            darkline_hypothesis_id VARCHAR(96) NOT NULL,
            case_id VARCHAR(96) NOT NULL,
            bridge_id VARCHAR(128) NULL,
            alternative_type VARCHAR(96) NOT NULL,
            explanation LONGTEXT NOT NULL,
            strength VARCHAR(64) NOT NULL,
            current_status VARCHAR(64) NOT NULL,
            source_reasoning_step_id VARCHAR(128) NULL,
            archive_batch_id VARCHAR(96) NOT NULL,
            KEY idx_alt_case (case_id),
            KEY idx_alt_hypothesis (darkline_hypothesis_id)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
        """
    )


def clear_archive_rows(cur) -> None:
    for table in [
        "dl_case_reasoning_step",
        "dl_expected_line",
        "dl_event_node",
        "dl_evidence",
        "dl_evidence_node_link",
        "dl_impact_target",
        "dl_manifestation_bridge",
        "dl_alternative_explanation",
    ]:
        cur.execute(f"DELETE FROM `{table}` WHERE archive_batch_id=%s", (ARCHIVE_BATCH_ID,))
    cur.execute("DELETE FROM dl_source_document WHERE source_id=%s", (SOURCE_ID,))


def first_event_date(text: str) -> str | None:
    m = re.search(r"(\d{4}[-/年]\d{1,2}(?:[-/月]\d{1,2})?)", text)
    return m.group(1) if m else None


def insert_many(cur, sql: str, rows: Iterable[tuple]) -> int:
    rows = list(rows)
    if rows:
        cur.executemany(sql, rows)
    return len(rows)


def main() -> None:
    parser = argparse.ArgumentParser(description="Build full darkline archive tables from imported case records.")
    parser.add_argument("--host", default=os.environ.get("DARKLINE_MYSQL_HOST", "127.0.0.1"))
    parser.add_argument("--port", type=int, default=int(os.environ.get("DARKLINE_MYSQL_PORT", "3306")))
    parser.add_argument("--user", default=os.environ.get("DARKLINE_MYSQL_USER", "root"))
    parser.add_argument("--password", default=None)
    parser.add_argument("--database", default=os.environ.get("DARKLINE_MYSQL_DATABASE", "tianxia"))
    args = parser.parse_args()

    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    conn = connect(args)
    counts: dict[str, int] = {}

    with conn.cursor() as cur:
        create_tables(cur)
        clear_archive_rows(cur)
        cur.execute("SELECT COALESCE(MAX(source_hash), '') AS source_hash FROM dl_case_import_batch")
        source_hash = cur.fetchone()["source_hash"] or "UNKNOWN_SOURCE_HASH"
        cur.execute(
            """
            INSERT INTO dl_source_document (
                source_id, source_title, source_url_or_path, source_type, publish_time,
                available_time, raw_text_hash, source_reliability, archive_status
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """,
            (
                SOURCE_ID,
                "Legacy external information casebook",
                SOURCE_PATH,
                "MARKDOWN_CASEBOOK",
                None,
                None,
                source_hash,
                "HISTORICAL_PROJECT_SOURCE",
                "READY",
            ),
        )
        counts["dl_source_document"] = 1

        cur.execute(
            """
            SELECT case_id, case_title, case_level, case_family, case_body_md, case_body_hash
            FROM dl_case_record
            ORDER BY case_order
            """
        )
        cases = cur.fetchall()

        reasoning_rows = []
        expected_rows = []
        event_rows = []
        evidence_rows = []
        link_rows = []
        target_rows = []
        bridge_rows = []
        alt_rows = []

        for case in cases:
            case_id = case["case_id"]
            hypothesis_id = case_id
            body = case["case_body_md"] or ""
            units = split_reasoning_units(body)
            symbols = extract_symbols(case["case_title"] + "\n" + body)
            if not symbols:
                symbols = ["CASE_SCOPE_UNKNOWN"]

            for target_order, symbol in enumerate(symbols, start=1):
                target_rows.append(
                    (
                        f"{case_id}_TARGET_{target_order:03d}",
                        hypothesis_id,
                        case_id,
                        "SYMBOL" if symbol != "CASE_SCOPE_UNKNOWN" else "CASE_SCOPE",
                        symbol,
                        symbol if symbol != "CASE_SCOPE_UNKNOWN" else None,
                        "UNKNOWN",
                        "CASE_DEFINED_WINDOW",
                        ARCHIVE_BATCH_ID,
                    )
                )

            expected_id_for_case: str | None = None
            event_id_for_step: dict[str, str] = {}

            for order, unit in enumerate(units, start=1):
                text = unit["text"]
                step_type = classify_step_type(text)
                step_id = f"{case_id}_STEP_{order:04d}"
                reasoning_rows.append(
                    (
                        step_id,
                        case_id,
                        hypothesis_id,
                        order,
                        step_type,
                        compact(unit.get("heading", ""), 480) or None,
                        text,
                        "MARKDOWN_LINE_HEURISTIC_V1",
                        ARCHIVE_BATCH_ID,
                    )
                )

                evidence_id = f"{case_id}_EVID_{order:04d}"
                evidence_rows.append(
                    (
                        evidence_id,
                        SOURCE_ID,
                        case_id,
                        text,
                        evidence_type_for_step(step_type),
                        "REVIEW",
                        now,
                        sha256_text(f"{case_id}|{order}|{text}"),
                        step_id,
                        ARCHIVE_BATCH_ID,
                    )
                )

                expected_line_id = None
                if contains_expected_line(step_type, text):
                    expected_line_id = f"{case_id}_EXPECTED_{len([r for r in expected_rows if r[2] == case_id]) + 1:03d}"
                    expected_id_for_case = expected_id_for_case or expected_line_id
                    expected_rows.append(
                        (
                            expected_line_id,
                            hypothesis_id,
                            case_id,
                            step_type,
                            compact(text, 1000),
                            "TEXT_INFERRED_WINDOW",
                            "UNKNOWN",
                            0,
                            None,
                            "REVIEW_EXTRACTED",
                            step_id,
                            ARCHIVE_BATCH_ID,
                        )
                    )

                event_node_id = None
                if has_event_signal(text):
                    event_node_id = f"{case_id}_EVENT_{order:04d}"
                    event_id_for_step[step_id] = event_node_id
                    event_rows.append(
                        (
                            event_node_id,
                            hypothesis_id,
                            case_id,
                            compact(text, 480),
                            first_event_date(text),
                            None,
                            None,
                            json.dumps(symbols, ensure_ascii=False),
                            step_type,
                            case["case_level"] or "UNKNOWN",
                            SOURCE_ID,
                            "REVIEW_EXTRACTED",
                            step_id,
                            ARCHIVE_BATCH_ID,
                        )
                    )

                link_rows.append(
                    (
                        f"{case_id}_LINK_{order:04d}",
                        evidence_id,
                        event_node_id,
                        hypothesis_id,
                        expected_line_id or expected_id_for_case,
                        "SUPPORT" if step_type != "ALTERNATIVE_OR_GAP" else "REVIEW",
                        ARCHIVE_BATCH_ID,
                    )
                )

                if has_market_signal(text):
                    bridge_rows.append(
                        (
                            f"{case_id}_BRIDGE_{len([r for r in bridge_rows if r[2] == case_id]) + 1:03d}",
                            hypothesis_id,
                            case_id,
                            expected_line_id or expected_id_for_case,
                            f"{case_id}_TARGET_001",
                            "KLINE_OR_MARKET_TEXT",
                            first_event_date(text),
                            text,
                            None,
                            "REVIEW_EXTRACTED",
                            step_id,
                            ARCHIVE_BATCH_ID,
                        )
                    )

                if step_type == "ALTERNATIVE_OR_GAP":
                    alt_rows.append(
                        (
                            f"{case_id}_ALT_{len([r for r in alt_rows if r[2] == case_id]) + 1:03d}",
                            hypothesis_id,
                            case_id,
                            None,
                            "TEXT_ALTERNATIVE_OR_GAP",
                            text,
                            "REVIEW",
                            "OPEN",
                            step_id,
                            ARCHIVE_BATCH_ID,
                        )
                    )

            if not expected_id_for_case:
                expected_rows.append(
                    (
                        f"{case_id}_EXPECTED_001",
                        hypothesis_id,
                        case_id,
                        "CASE_IMPLIED_EXPECTED_LINE",
                        compact(case["case_title"], 1000),
                        "CASE_REVIEW_WINDOW",
                        "UNKNOWN",
                        0,
                        None,
                        "REVIEW_EXTRACTED",
                        None,
                        ARCHIVE_BATCH_ID,
                    )
                )

        counts["dl_case_reasoning_step"] = insert_many(
            cur,
            """
            INSERT INTO dl_case_reasoning_step (
                reasoning_step_id, case_id, darkline_hypothesis_id, step_order, step_type,
                section_heading, step_text, extraction_rule, archive_batch_id
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """,
            reasoning_rows,
        )
        counts["dl_expected_line"] = insert_many(
            cur,
            """
            INSERT INTO dl_expected_line (
                expected_line_id, darkline_hypothesis_id, case_id, expected_line_type,
                expected_event, expected_timing, expected_direction, observed_flag,
                observed_event_node_id, validation_status, source_reasoning_step_id, archive_batch_id
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """,
            expected_rows,
        )
        counts["dl_event_node"] = insert_many(
            cur,
            """
            INSERT INTO dl_event_node (
                event_node_id, darkline_hypothesis_id, case_id, event_title, event_date,
                available_time, actor_list, target_list, event_type, event_level, source_id,
                node_status, source_reasoning_step_id, archive_batch_id
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """,
            event_rows,
        )
        counts["dl_evidence"] = insert_many(
            cur,
            """
            INSERT INTO dl_evidence (
                evidence_id, source_id, case_id, raw_excerpt, evidence_type, evidence_strength,
                extracted_time, evidence_hash, source_reasoning_step_id, archive_batch_id
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """,
            evidence_rows,
        )
        counts["dl_evidence_node_link"] = insert_many(
            cur,
            """
            INSERT INTO dl_evidence_node_link (
                evidence_node_link_id, evidence_id, event_node_id, darkline_hypothesis_id,
                expected_line_id, link_role, archive_batch_id
            ) VALUES (%s,%s,%s,%s,%s,%s,%s)
            """,
            link_rows,
        )
        counts["dl_impact_target"] = insert_many(
            cur,
            """
            INSERT INTO dl_impact_target (
                impact_target_id, darkline_hypothesis_id, case_id, target_type, target_id,
                target_name, expected_impact_direction, expected_window, archive_batch_id
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """,
            target_rows,
        )
        counts["dl_manifestation_bridge"] = insert_many(
            cur,
            """
            INSERT INTO dl_manifestation_bridge (
                bridge_id, darkline_hypothesis_id, case_id, expected_line_id, impact_target_id,
                output_layer, manifestation_time, manifestation_value, control_group_id,
                support_status, source_reasoning_step_id, archive_batch_id
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """,
            bridge_rows,
        )
        counts["dl_alternative_explanation"] = insert_many(
            cur,
            """
            INSERT INTO dl_alternative_explanation (
                alternative_id, darkline_hypothesis_id, case_id, bridge_id, alternative_type,
                explanation, strength, current_status, source_reasoning_step_id, archive_batch_id
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """,
            alt_rows,
        )

    conn.close()
    print(
        json.dumps(
            {
                "archive_batch_id": ARCHIVE_BATCH_ID,
                "database": args.database,
                "counts": counts,
                "status": "PASS",
            },
            ensure_ascii=False,
            indent=2,
        )
    )


if __name__ == "__main__":
    main()