from __future__ import annotations

import argparse
import csv
import getpass
import hashlib
import json
import os
import re
from datetime import datetime
from pathlib import Path

import pymysql


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_OUT_DIR = ROOT / "data" / "legacy_case_import_20260624"
IMPORT_BATCH_ID = "DL_LEGACY_CASE_IMPORT_20260624_V1"
SOURCE_NAME = "legacy_external_information_casebook"
SOURCE_KIND = "markdown_casebook"


def project_rel(path: Path) -> str:
    try:
        return str(path.resolve().relative_to(ROOT)).replace("\\", "/")
    except ValueError:
        return f"external_source/{path.name}"


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


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def infer_level(body: str) -> str:
    levels = []
    for level in ("L1", "L2", "L3"):
        if re.search(rf"(?<![A-Z0-9]){level}(?![A-Z0-9])", body):
            levels.append(level)
    return "+".join(levels) if levels else "UNKNOWN"


def infer_case_family(title: str, body: str) -> str:
    text = title + "\n" + body[:2000]
    rules = [
        ("AUTO_BATCH", r"自动批量案例|AUTOBATCH"),
        ("COUNTERINTUITIVE_KLINE", r"反常识|不跌反涨|坏标题|利空.*上涨"),
        ("STATE_OWNED_REPURCHASE", r"招商局|回购|增持"),
        ("REGULATORY_RISK", r"立案|证监会|监管|违规|实控人|关键人"),
        ("AUDIT_REMEDIATION", r"审计|非标|影响消除|保留意见"),
        ("DISTRESSED_RESTRUCTURING", r"重整|ST|退市|债务|司法拍卖|壳"),
        ("INDUSTRIAL_GROWTH", r"机器人|商业航天|AI|订单|扩产|股权激励"),
    ]
    for family, pattern in rules:
        if re.search(pattern, text, flags=re.I):
            return family
    return "GENERAL_DARKLINE_CASE"


def split_cases(source_text: str, source_rel: str, body_dir: Path) -> list[dict[str, str]]:
    heading_pattern = re.compile(r"^##\s+(.+)$", re.M)
    matches = list(heading_pattern.finditer(source_text))
    cases: list[dict[str, str]] = []
    case_no = 0

    for i, match in enumerate(matches):
        title = match.group(1).strip()
        start = match.start()
        end = matches[i + 1].start() if i + 1 < len(matches) else len(source_text)
        body = source_text[start:end].strip()

        is_case = (
            title.startswith("案例")
            or title.startswith("自动批量案例")
            or title.startswith("DARKLINE-CASE")
        )
        if not is_case or title.startswith("案例 X"):
            continue

        case_no += 1
        case_id = f"DLCASE_LEGACY_{case_no:04d}"
        body_hash = sha256_text(body)
        body_path = body_dir / f"{case_id}.md"
        body_path.write_text(body + "\n", encoding="utf-8")

        cases.append(
            {
                "case_id": case_id,
                "import_batch_id": IMPORT_BATCH_ID,
                "case_order": str(case_no),
                "case_title": title,
                "case_family": infer_case_family(title, body),
                "case_level": infer_level(body),
                "current_status": "LEGACY_IMPORTED_REVIEW",
                "case_body_hash": body_hash,
                "local_body_path": project_rel(body_path),
                "source_document_path": source_rel,
                "source_section_heading": f"## {title}",
            }
        )

    return cases


def write_csv(path: Path, rows: list[dict[str, str]], fields: list[str]) -> None:
    with path.open("w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=fields)
        writer.writeheader()
        writer.writerows(rows)


def write_manifest(manifest_csv: Path, paths: list[Path]) -> None:
    rows = []
    for path in sorted(paths):
        rows.append(
            {
                "artifact_path": project_rel(path),
                "artifact_name": path.name,
                "artifact_type": path.suffix.lstrip(".") or "directory",
                "size_bytes": path.stat().st_size,
                "sha256": sha256_file(path),
                "artifact_status": "READY",
            }
        )
    write_csv(
        manifest_csv,
        rows,
        [
            "artifact_path",
            "artifact_name",
            "artifact_type",
            "size_bytes",
            "sha256",
            "artifact_status",
        ],
    )


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_mysql(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,
    )


def import_mysql(
    args: argparse.Namespace,
    cases: list[dict[str, str]],
    source_hash: str,
    source_rel: str,
    out_dir: Path,
) -> dict[str, object]:
    imported_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    conn = connect_mysql(args)
    with conn.cursor() as cur:
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS dl_case_import_batch (
                import_batch_id VARCHAR(96) PRIMARY KEY,
                source_name VARCHAR(128) NOT NULL,
                source_kind VARCHAR(64) NOT NULL,
                source_document_path VARCHAR(512) NOT NULL,
                source_hash CHAR(64) NOT NULL,
                case_count INT NOT NULL,
                data_dir VARCHAR(512) NOT NULL,
                imported_at DATETIME NOT NULL,
                note VARCHAR(512) NULL
            ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
            """
        )
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS dl_case_record (
                case_id VARCHAR(96) PRIMARY KEY,
                import_batch_id VARCHAR(96) NOT NULL,
                case_order INT NOT NULL,
                case_title VARCHAR(512) NOT NULL,
                case_family VARCHAR(96) NOT NULL,
                case_level VARCHAR(32) NOT NULL,
                current_status VARCHAR(64) NOT NULL,
                case_body_md LONGTEXT NOT NULL,
                case_body_hash CHAR(64) NOT NULL,
                local_body_path VARCHAR(512) NOT NULL,
                source_document_path VARCHAR(512) NOT NULL,
                source_section_heading VARCHAR(512) NOT NULL,
                created_at DATETIME NOT NULL,
                updated_at DATETIME NOT NULL,
                KEY idx_dl_case_batch (import_batch_id),
                KEY idx_dl_case_family (case_family),
                KEY idx_dl_case_level (case_level)
            ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
            """
        )
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS dl_darkline_hypothesis (
                darkline_hypothesis_id VARCHAR(96) PRIMARY KEY,
                case_id VARCHAR(96) NOT NULL,
                hypothesis_title VARCHAR(512) NOT NULL,
                darkline_level VARCHAR(32) NOT NULL,
                hypothesis_family VARCHAR(96) NOT NULL,
                hypothesis_status VARCHAR(64) NOT NULL,
                source_case_body_hash CHAR(64) NOT NULL,
                created_at DATETIME NOT NULL,
                updated_at DATETIME NOT NULL,
                KEY idx_dl_hypothesis_case (case_id),
                KEY idx_dl_hypothesis_level (darkline_level)
            ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
            """
        )
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS dl_chain_state (
                state_record_id VARCHAR(128) PRIMARY KEY,
                darkline_hypothesis_id VARCHAR(96) NOT NULL,
                case_id VARCHAR(96) NOT NULL,
                chain_clarity_status VARCHAR(64) NOT NULL,
                reality_confirmation_status VARCHAR(64) NOT NULL,
                market_manifestation_status VARCHAR(64) NOT NULL,
                validation_readout_status VARCHAR(64) NOT NULL,
                current_consumption_level VARCHAR(64) NOT NULL,
                state_record_time DATETIME NOT NULL,
                note VARCHAR(512) NULL,
                KEY idx_dl_chain_case (case_id),
                KEY idx_dl_chain_hypothesis (darkline_hypothesis_id)
            ) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
            """
        )

        cur.execute(
            """
            INSERT INTO dl_case_import_batch (
                import_batch_id, source_name, source_kind, source_document_path,
                source_hash, case_count, data_dir, imported_at, note
            ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
            ON DUPLICATE KEY UPDATE
                source_hash=VALUES(source_hash),
                case_count=VALUES(case_count),
                data_dir=VALUES(data_dir),
                imported_at=VALUES(imported_at),
                note=VALUES(note)
            """,
            (
                IMPORT_BATCH_ID,
                SOURCE_NAME,
                SOURCE_KIND,
                source_rel,
                source_hash,
                len(cases),
                project_rel(out_dir),
                imported_at,
                "Legacy casebook imported as historical review data for standalone darkline.",
            ),
        )

        for case in cases:
            body = (ROOT / case["local_body_path"]).read_text(encoding="utf-8")
            cur.execute(
                """
                INSERT INTO dl_case_record (
                    case_id, import_batch_id, case_order, case_title, case_family,
                    case_level, current_status, case_body_md, case_body_hash,
                    local_body_path, source_document_path, source_section_heading,
                    created_at, updated_at
                ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                ON DUPLICATE KEY UPDATE
                    case_title=VALUES(case_title),
                    case_family=VALUES(case_family),
                    case_level=VALUES(case_level),
                    current_status=VALUES(current_status),
                    case_body_md=VALUES(case_body_md),
                    case_body_hash=VALUES(case_body_hash),
                    local_body_path=VALUES(local_body_path),
                    source_document_path=VALUES(source_document_path),
                    source_section_heading=VALUES(source_section_heading),
                    updated_at=VALUES(updated_at)
                """,
                (
                    case["case_id"],
                    case["import_batch_id"],
                    int(case["case_order"]),
                    case["case_title"],
                    case["case_family"],
                    case["case_level"],
                    case["current_status"],
                    body,
                    case["case_body_hash"],
                    case["local_body_path"],
                    case["source_document_path"],
                    case["source_section_heading"],
                    imported_at,
                    imported_at,
                ),
            )
            cur.execute(
                """
                INSERT INTO dl_darkline_hypothesis (
                    darkline_hypothesis_id, case_id, hypothesis_title, darkline_level,
                    hypothesis_family, hypothesis_status, source_case_body_hash,
                    created_at, updated_at
                ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
                ON DUPLICATE KEY UPDATE
                    hypothesis_title=VALUES(hypothesis_title),
                    darkline_level=VALUES(darkline_level),
                    hypothesis_family=VALUES(hypothesis_family),
                    hypothesis_status=VALUES(hypothesis_status),
                    source_case_body_hash=VALUES(source_case_body_hash),
                    updated_at=VALUES(updated_at)
                """,
                (
                    case["case_id"],
                    case["case_id"],
                    case["case_title"],
                    case["case_level"],
                    case["case_family"],
                    "LEGACY_IMPORTED_REVIEW",
                    case["case_body_hash"],
                    imported_at,
                    imported_at,
                ),
            )
            cur.execute(
                """
                INSERT INTO dl_chain_state (
                    state_record_id, darkline_hypothesis_id, case_id,
                    chain_clarity_status, reality_confirmation_status,
                    market_manifestation_status, validation_readout_status,
                    current_consumption_level, state_record_time, note
                ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                ON DUPLICATE KEY UPDATE
                    chain_clarity_status=VALUES(chain_clarity_status),
                    reality_confirmation_status=VALUES(reality_confirmation_status),
                    market_manifestation_status=VALUES(market_manifestation_status),
                    validation_readout_status=VALUES(validation_readout_status),
                    current_consumption_level=VALUES(current_consumption_level),
                    state_record_time=VALUES(state_record_time),
                    note=VALUES(note)
                """,
                (
                    f"{case['case_id']}_STATE_LEGACY_IMPORT_V1",
                    case["case_id"],
                    case["case_id"],
                    "LEGACY_IMPORTED_REVIEW",
                    "LEGACY_IMPORTED_REVIEW",
                    "LEGACY_IMPORTED_REVIEW",
                    "CASE_IMPORTED_FOR_REVIEW",
                    "CASEBOOK",
                    imported_at,
                    "Historical case imported. Re-validation is required before reuse.",
                ),
            )

        cur.execute("SELECT COUNT(*) FROM dl_case_record WHERE import_batch_id=%s", (IMPORT_BATCH_ID,))
        db_case_count = cur.fetchone()[0]

    conn.close()
    return {
        "mysql_import_status": "PASS",
        "db_case_count": db_case_count,
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="Import legacy markdown cases into darkline package files and MySQL.")
    parser.add_argument("--source", required=True, help="Legacy markdown casebook path.")
    parser.add_argument("--out-dir", default=str(DEFAULT_OUT_DIR))
    parser.add_argument("--skip-mysql", action="store_true")
    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()

    source = Path(args.source)
    out_dir = Path(args.out_dir)
    body_dir = out_dir / "case_body"
    index_csv = out_dir / "legacy_case_index.csv"
    summary_json = out_dir / "legacy_case_import_summary.json"
    manifest_csv = out_dir / "legacy_case_import_manifest.csv"

    out_dir.mkdir(parents=True, exist_ok=True)
    body_dir.mkdir(parents=True, exist_ok=True)

    source_text = source.read_text(encoding="utf-8")
    source_hash = sha256_text(source_text)
    source_rel = project_rel(source)
    cases = split_cases(source_text, source_rel, body_dir)

    fields = [
        "case_id",
        "import_batch_id",
        "case_order",
        "case_title",
        "case_family",
        "case_level",
        "current_status",
        "case_body_hash",
        "local_body_path",
        "source_document_path",
        "source_section_heading",
    ]
    write_csv(index_csv, cases, fields)

    mysql_result: dict[str, object] = {"mysql_import_status": "SKIPPED"}
    if not args.skip_mysql:
        mysql_result = import_mysql(args, cases, source_hash, source_rel, out_dir)

    body_files = sorted(body_dir.glob("*.md"))
    summary = {
        "run_id": "legacy_case_import_20260624",
        "import_batch_id": IMPORT_BATCH_ID,
        "source_document_path": source_rel,
        "source_hash": source_hash,
        "output_dir": project_rel(out_dir),
        "case_count": len(cases),
        "case_body_file_count": len(body_files),
        "file_export_status": "PASS" if len(cases) == len(body_files) else "FAIL",
        **mysql_result,
    }
    summary_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

    write_manifest(manifest_csv, [index_csv, summary_json, *body_files])
    summary["manifest_path"] = project_rel(manifest_csv)
    summary["manifest_sha256"] = sha256_file(manifest_csv)
    summary_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    write_manifest(manifest_csv, [index_csv, summary_json, *body_files])

    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()