#!/usr/bin/env python3
"""Project wrapper around the official TianYanCha `tyc` CLI.

The wrapper keeps this repository's tender-enrichment workflow stable while
delegating all business data access to the official tyc-cli package.
"""

from __future__ import annotations

import argparse
import csv
import json
import os
import re
import shlex
import shutil
import stat
import subprocess
import sys
import time
from pathlib import Path
from typing import Iterable


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_OUTPUT_DIR = ROOT / "data" / "processed" / "tyc-cli"
KNOWN_COLUMNS = (
    "中标单位名称",
    "中标单位",
    "投标单位",
    "公司名称",
    "企业名称",
    "供应商名称",
    "company",
    "name",
)
TASKS: dict[str, tuple[str, ...]] = {
    "search": ("company", "companies"),
    "registration": ("company", "registration-info"),
    "bidding": ("operation", "bidding-info"),
    "qualifications": ("operation", "qualifications"),
    "risk-overview": ("risk", "overview"),
    "software-copyright": ("intellectual_property", "software-copyright-info"),
}


def clean(value: object) -> str:
    return str(value or "").strip()


def slugify(value: str, fallback: str = "company") -> str:
    text = re.sub(r"[\\/:*?\"<>|\s]+", "-", clean(value))
    text = re.sub(r"-+", "-", text).strip("-")
    return text[:80] or fallback


def print_json(data: object) -> None:
    print(json.dumps(data, ensure_ascii=False, indent=2))


def run_command(command: list[str], *, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]:
    return subprocess.run(command, text=True, capture_output=True, env=env, check=False)


def tyc_path() -> str | None:
    return shutil.which("tyc")


def config_status() -> dict[str, object]:
    path = Path.home() / ".tyc" / "config.json"
    info: dict[str, object] = {
        "path": str(path),
        "exists": path.exists(),
        "mode": None,
        "url": None,
        "has_authorization_header": False,
        "has_oauth_refresh": False,
    }
    if not path.exists():
        return info

    try:
        info["mode"] = stat.filemode(path.stat().st_mode)
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:  # noqa: BLE001 - diagnostic path only.
        info["error"] = str(exc)
        return info

    headers = data.get("headers") if isinstance(data, dict) else {}
    oauth = data.get("oauth") if isinstance(data, dict) else {}
    info["url"] = data.get("url") if isinstance(data, dict) else None
    info["has_authorization_header"] = bool(isinstance(headers, dict) and headers.get("Authorization"))
    info["has_oauth_refresh"] = bool(isinstance(oauth, dict) and oauth.get("refreshToken"))
    return info


def doctor(args: argparse.Namespace) -> int:
    result: dict[str, object] = {
        "node": None,
        "npm": None,
        "tyc_path": tyc_path(),
        "tyc_version": None,
        "config": config_status(),
    }

    for key, command in (
        ("node", ["node", "--version"]),
        ("npm", ["npm", "--version"]),
        ("tyc_version", ["tyc", "--version"]),
    ):
        completed = run_command(command)
        result[key] = clean(completed.stdout) if completed.returncode == 0 else None

    if args.auth_probe:
        probe = run_tyc(
            build_tyc_args("search", args.auth_probe, page_size=3),
            output_path=None,
            head=args.head,
            dry_run=False,
        )
        result["auth_probe"] = probe

    if args.json:
        print_json(result)
    else:
        print(f"node: {result['node'] or 'not found'}")
        print(f"npm: {result['npm'] or 'not found'}")
        print(f"tyc: {result['tyc_path'] or 'not found'}")
        print(f"tyc version: {result['tyc_version'] or 'not available'}")
        config = result["config"]
        assert isinstance(config, dict)
        print(f"config: {config['path']} ({'exists' if config['exists'] else 'missing'})")
        print(f"endpoint: {config.get('url') or 'not configured'}")
        print(f"authorization header: {'yes' if config.get('has_authorization_header') else 'no'}")
        print(f"oauth refresh: {'yes' if config.get('has_oauth_refresh') else 'no'}")
        if args.auth_probe:
            print_json(result["auth_probe"])

    return 0 if result["tyc_path"] else 1


def auth(args: argparse.Namespace) -> int:
    if args.login:
        command = ["tyc", "login"]
        if args.url:
            command += ["--url", args.url]
        if args.no_block:
            command.append("--no-block")
        display = shlex.join(command)
        if args.dry_run:
            print(display)
            return 0
        return subprocess.call(command)

    if args.resume:
        command = ["tyc", "login", "--resume"]
        if args.dry_run:
            print(shlex.join(command))
            return 0
        return subprocess.call(command)

    if args.dry_run:
        redacted = ["tyc", "init", "--authorization", "***"]
        if args.url:
            redacted += ["--url", args.url]
        if args.no_verify:
            redacted.append("--no-verify")
        print(shlex.join(redacted))
        return 0
    token = os.environ.get(args.token_env or "")
    if not token:
        print(f"missing token env: {args.token_env}", file=sys.stderr)
        return 2
    command = ["tyc", "init", "--authorization", token]
    if args.url:
        command += ["--url", args.url]
    if args.no_verify:
        command.append("--no-verify")
    return subprocess.call(command)


def build_tyc_args(task: str, company: str, *, page_size: int = 10) -> list[str]:
    if task not in TASKS:
        raise KeyError(f"unknown task: {task}")
    command = [*TASKS[task], company]
    if task == "search":
        command += ["--pageSize", str(page_size)]
    return command


def run_tyc(
    tyc_args: list[str],
    *,
    output_path: Path | None,
    head: int,
    dry_run: bool,
) -> dict[str, object]:
    command = ["tyc", *tyc_args, "--head", str(head)]
    if output_path is not None:
        command += ["--output-file", str(output_path)]

    if dry_run:
        return {"ok": True, "dry_run": True, "command": shlex.join(command), "output": str(output_path) if output_path else None}

    started = time.time()
    completed = run_command(command)
    elapsed = round(time.time() - started, 3)
    if output_path is not None and completed.returncode == 0 and not output_path.exists():
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_text(completed.stdout, encoding="utf-8")

    return {
        "ok": completed.returncode == 0,
        "returncode": completed.returncode,
        "elapsed_seconds": elapsed,
        "command": shlex.join(["tyc", *tyc_args, "--head", str(head), "--output-file", str(output_path or "")]).replace(" --output-file ''", ""),
        "output": str(output_path) if output_path else None,
        "stdout_preview": completed.stdout[:1200],
        "stderr_preview": completed.stderr[:1200],
    }


def read_text_companies(path: Path) -> list[str]:
    companies = []
    for line in path.read_text(encoding="utf-8-sig").splitlines():
        value = clean(line)
        if value and not value.startswith("#"):
            companies.append(value)
    return companies


def read_csv_companies(path: Path, column: str | None) -> list[str]:
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        if not reader.fieldnames:
            return []
        selected = column or next((name for name in KNOWN_COLUMNS if name in reader.fieldnames), reader.fieldnames[0])
        if selected not in reader.fieldnames:
            raise ValueError(f"column not found in CSV: {selected}")
        return [clean(row.get(selected)) for row in reader if clean(row.get(selected))]


def read_xlsx_companies(path: Path, column: str | None, header_row: int | None) -> list[str]:
    try:
        from openpyxl import load_workbook
    except ImportError as exc:
        raise RuntimeError("openpyxl is required to read .xlsx files") from exc

    wb = load_workbook(path, read_only=True, data_only=True)
    ws = wb.active
    rows = list(ws.iter_rows(values_only=True))
    if not rows:
        return []

    if column and column.isdigit():
        col_index = int(column) - 1
        start = (header_row or 1)
        return [clean(row[col_index] if col_index < len(row) else "") for row in rows[start:] if clean(row[col_index] if col_index < len(row) else "")]

    scan_limit = min(10, len(rows))
    candidates = [column] if column else list(KNOWN_COLUMNS)
    for row_index in range(scan_limit):
        labels = [clean(value) for value in rows[row_index]]
        for candidate in candidates:
            if candidate in labels:
                col_index = labels.index(candidate)
                return [
                    clean(row[col_index] if col_index < len(row) else "")
                    for row in rows[row_index + 1 :]
                    if clean(row[col_index] if col_index < len(row) else "")
                ]

    if column:
        raise ValueError(f"column not found in first {scan_limit} rows: {column}")
    return [clean(row[0]) for row in rows[(header_row or 1) :] if row and clean(row[0])]


def dedupe(values: Iterable[str]) -> list[str]:
    seen: set[str] = set()
    result = []
    for value in values:
        key = clean(value)
        if key and key not in seen:
            seen.add(key)
            result.append(key)
    return result


def load_companies(path: Path, column: str | None, header_row: int | None) -> list[str]:
    suffix = path.suffix.lower()
    if suffix in (".txt", ".list"):
        return dedupe(read_text_companies(path))
    if suffix == ".csv":
        return dedupe(read_csv_companies(path, column))
    if suffix == ".xlsx":
        return dedupe(read_xlsx_companies(path, column, header_row))
    raise ValueError(f"unsupported input type: {suffix}. Use .txt, .csv, or .xlsx.")


def run_single(args: argparse.Namespace) -> int:
    output_dir = args.output_dir.resolve()
    record_dir = output_dir / time.strftime("%Y%m%d-%H%M%S") / slugify(args.company)
    if not args.dry_run:
        record_dir.mkdir(parents=True, exist_ok=True)

    results = []
    for task in args.task:
        output_file = record_dir / f"{task}.json"
        tyc_args = build_tyc_args(task, args.company, page_size=args.page_size)
        result = run_tyc(tyc_args, output_path=output_file, head=args.head, dry_run=args.dry_run)
        result.update({"company": args.company, "task": task})
        results.append(result)

    print_json({"company": args.company, "record_dir": str(record_dir), "results": results})
    return 0 if all(item["ok"] for item in results) else 1


def run_batch(args: argparse.Namespace) -> int:
    companies = load_companies(args.input.resolve(), args.column, args.header_row)
    if args.limit:
        companies = companies[: args.limit]

    batch_id = time.strftime("%Y%m%d-%H%M%S")
    output_dir = args.output_dir.resolve() / batch_id
    if not args.dry_run:
        output_dir.mkdir(parents=True, exist_ok=True)
    manifest_path = output_dir / "manifest.jsonl"
    failures = 0

    manifest_handle = manifest_path.open("w", encoding="utf-8") if not args.dry_run else None
    try:
        for index, company in enumerate(companies, start=1):
            company_dir = output_dir / f"{index:04d}-{slugify(company)}"
            if not args.dry_run:
                company_dir.mkdir(parents=True, exist_ok=True)
            for task in args.task:
                output_file = company_dir / f"{task}.json"
                tyc_args = build_tyc_args(task, company, page_size=args.page_size)
                result = run_tyc(tyc_args, output_path=output_file, head=args.head, dry_run=args.dry_run)
                result.update({"index": index, "company": company, "task": task})
                failures += 0 if result["ok"] else 1
                if manifest_handle is not None:
                    manifest_handle.write(json.dumps(result, ensure_ascii=False) + "\n")
                    manifest_handle.flush()
    finally:
        if manifest_handle is not None:
            manifest_handle.close()

    print_json(
        {
            "input": str(args.input),
            "companies": len(companies),
            "tasks": args.task,
            "output_dir": str(output_dir),
            "manifest": str(manifest_path),
            "failures": failures,
            "dry_run": args.dry_run,
        }
    )
    return 0 if failures == 0 else 1


def add_common_run_options(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--task", action="append", choices=sorted(TASKS), default=None, help="TianYanCha query task. Repeat to run multiple tasks.")
    parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR, help="Directory for raw tyc outputs. Defaults to ignored data/processed/tyc-cli.")
    parser.add_argument("--head", type=int, default=80, help="Preview line count for tyc stdout while full data is written to file.")
    parser.add_argument("--page-size", type=int, default=10, help="Page size for the L0 company search task.")
    parser.add_argument("--dry-run", action="store_true", help="Print the tyc commands and output paths without calling TianYanCha.")


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Tender-project wrapper for the official TianYanCha tyc CLI.")
    subparsers = parser.add_subparsers(dest="command", required=True)

    doctor_parser = subparsers.add_parser("doctor", help="Check local Node/npm/tyc installation and config without exposing secrets.")
    doctor_parser.add_argument("--json", action="store_true", help="Print machine-readable diagnostics.")
    doctor_parser.add_argument("--auth-probe", help="Optional company name for a real authenticated tyc search probe.")
    doctor_parser.add_argument("--head", type=int, default=20, help="Preview lines for the optional auth probe.")
    doctor_parser.set_defaults(func=doctor)

    auth_parser = subparsers.add_parser("auth", help="Run tyc OAuth login or initialize auth from an environment variable.")
    mode = auth_parser.add_mutually_exclusive_group(required=True)
    mode.add_argument("--login", action="store_true", help="Run tyc login.")
    mode.add_argument("--resume", action="store_true", help="Run tyc login --resume.")
    mode.add_argument("--token-env", default=None, help="Read an API token from this environment variable and run tyc init.")
    auth_parser.add_argument("--url", help="Optional MCP endpoint override.")
    auth_parser.add_argument("--no-block", action="store_true", help="Use tyc login --no-block with --login.")
    auth_parser.add_argument("--no-verify", action="store_true", help="Use tyc init --no-verify with --token-env.")
    auth_parser.add_argument("--dry-run", action="store_true", help="Show the command with secrets redacted.")
    auth_parser.set_defaults(func=auth)

    one_parser = subparsers.add_parser("company", help="Run one or more tyc tasks for a single company.")
    one_parser.add_argument("company", help="Company name or search keyword.")
    add_common_run_options(one_parser)
    one_parser.set_defaults(func=run_single)

    batch_parser = subparsers.add_parser("batch", help="Run one or more tyc tasks for companies in .txt, .csv, or .xlsx.")
    batch_parser.add_argument("input", type=Path, help="Input file containing company names.")
    batch_parser.add_argument("--column", help="Column name or 1-based Excel column number. Auto-detects common Chinese headers by default.")
    batch_parser.add_argument("--header-row", type=int, help="1-based header row hint for Excel files.")
    batch_parser.add_argument("--limit", type=int, help="Limit companies for a smoke run.")
    add_common_run_options(batch_parser)
    batch_parser.set_defaults(func=run_batch)

    args = parser.parse_args(argv)
    if hasattr(args, "task") and args.task is None:
        args.task = ["search"]
    return args


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv or sys.argv[1:])
    return int(args.func(args))


if __name__ == "__main__":
    raise SystemExit(main())