from __future__ import annotations

import base64
import json
import os
import shutil
import signal
import sqlite3
import subprocess
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from urllib import error, request

import yaml

from sync_state_machine.common.persistence import PersistenceBackend

PROJECT_ROOT = Path(__file__).resolve().parents[1]


def replace_placeholders(value: Any) -> Any:
    if isinstance(value, str):
        return value.replace("${PROJECT_ROOT}", str(PROJECT_ROOT))
    if isinstance(value, list):
        return [replace_placeholders(item) for item in value]
    if isinstance(value, dict):
        return {key: replace_placeholders(item) for key, item in value.items()}
    return value


def load_yaml_config(path: str | Path) -> dict[str, Any]:
    raw_path = Path(path)
    if not raw_path.is_absolute():
        raw_path = PROJECT_ROOT / raw_path
    payload = yaml.safe_load(raw_path.read_text(encoding="utf-8")) or {}
    return replace_placeholders(payload)


def ensure_parent(path: str | Path) -> Path:
    resolved = Path(path)
    resolved.parent.mkdir(parents=True, exist_ok=True)
    return resolved


def remove_path(path: str | Path) -> None:
    target = Path(path)
    if not target.exists():
        return
    if target.is_dir():
        shutil.rmtree(target, ignore_errors=True)
    else:
        try:
            target.unlink()
        except FileNotFoundError:
            pass


def recreate_dir(path: str | Path) -> Path:
    target = Path(path)
    remove_path(target)
    target.mkdir(parents=True, exist_ok=True)
    return target


def copy_tree_contents(src: str | Path, dst: str | Path) -> None:
    src_path = Path(src)
    dst_path = recreate_dir(dst)
    for child in src_path.iterdir():
        target = dst_path / child.name
        if child.is_dir():
            shutil.copytree(child, target)
        else:
            shutil.copy2(child, target)


def copy_selected_files(src: str | Path, dst: str | Path, file_names: list[str]) -> None:
    src_path = Path(src)
    dst_path = recreate_dir(dst)
    for file_name in file_names:
        source = src_path / file_name
        if not source.exists():
            raise FileNotFoundError(f"基础数据文件不存在: {source}")
        shutil.copy2(source, dst_path / file_name)


def _jsonl_record_id(item: dict[str, Any]) -> str:
    record_id = item.get("_id") or item.get("id")
    if not record_id:
        raise ValueError(f"JSONL 记录缺少 _id/id: {item}")
    return str(record_id)


def merge_jsonl_file(src_file: str | Path, dst_file: str | Path) -> None:
    src_path = Path(src_file)
    dst_path = Path(dst_file)

    merged: dict[str, dict[str, Any]] = {}
    ordered_ids: list[str] = []

    if dst_path.exists():
        for line in dst_path.read_text(encoding="utf-8").splitlines():
            raw = line.strip()
            if not raw:
                continue
            item = json.loads(raw)
            record_id = _jsonl_record_id(item)
            merged[record_id] = item
            ordered_ids.append(record_id)

    for line in src_path.read_text(encoding="utf-8").splitlines():
        raw = line.strip()
        if not raw:
            continue
        item = json.loads(raw)
        record_id = _jsonl_record_id(item)
        if record_id in merged:
            merged[record_id] = {**merged[record_id], **item}
        else:
            merged[record_id] = item
            ordered_ids.append(record_id)

    lines = [json.dumps(merged[record_id], ensure_ascii=False) for record_id in ordered_ids]
    dst_path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def merge_tree_contents(src: str | Path, dst: str | Path) -> None:
    src_path = Path(src)
    dst_path = Path(dst)
    dst_path.mkdir(parents=True, exist_ok=True)
    for child in src_path.iterdir():
        target = dst_path / child.name
        if child.is_dir():
            if target.exists():
                shutil.rmtree(target)
            shutil.copytree(child, target)
        else:
            if child.suffix == ".jsonl" and target.exists():
                merge_jsonl_file(child, target)
            else:
                shutil.copy2(child, target)


def run_command(
    command: list[str],
    cwd: str | Path,
    *,
    capture_output: bool = True,
    extra_env: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    if extra_env:
        env.update({key: str(value) for key, value in extra_env.items()})
    return subprocess.run(
        command,
        cwd=str(cwd),
        text=True,
        capture_output=capture_output,
        check=False,
        env=env,
    )


def start_background_process(
    command: list[str],
    cwd: str | Path,
    log_file: str | Path,
    *,
    extra_env: dict[str, str] | None = None,
) -> subprocess.Popen[str]:
    log_path = ensure_parent(log_file)
    handle = open(log_path, "a", encoding="utf-8")
    env = os.environ.copy()
    if extra_env:
        env.update({key: str(value) for key, value in extra_env.items()})
    return subprocess.Popen(
        command,
        cwd=str(cwd),
        stdout=handle,
        stderr=subprocess.STDOUT,
        text=True,
        start_new_session=True,
        env=env,
    )


def stop_process_from_pid_file(pid_file: str | Path) -> bool:
    path = Path(pid_file)
    if not path.exists():
        return False
    try:
        pid = int(path.read_text(encoding="utf-8").strip())
    except ValueError:
        path.unlink(missing_ok=True)
        return False

    try:
        os.killpg(pid, signal.SIGTERM)
        time.sleep(1)
    except PermissionError:
        try:
            os.kill(pid, signal.SIGTERM)
            time.sleep(1)
        except (PermissionError, ProcessLookupError):
            pass
    except ProcessLookupError:
        pass
    finally:
        path.unlink(missing_ok=True)
    return True


def stop_backend_processes(backend_root_dir: str | Path) -> list[int]:
    backend_root = str(Path(backend_root_dir).resolve())
    result = run_command(
        ["ps", "-ax", "-o", "pid=,pgid=,command="],
        cwd=PROJECT_ROOT,
    )
    if result.returncode != 0:
        return []

    killed_pgids: list[int] = []
    for line in result.stdout.splitlines():
        raw = line.strip()
        if not raw or backend_root not in raw or " run.py" not in raw:
            continue
        parts = raw.split(None, 2)
        if len(parts) < 3:
            continue
        try:
            pgid = int(parts[1])
        except ValueError:
            continue
        if pgid in killed_pgids:
            continue
        try:
            os.killpg(pgid, signal.SIGTERM)
            time.sleep(1)
        except ProcessLookupError:
            pass
        killed_pgids.append(pgid)
    return killed_pgids


def wait_http_ready(url: str, timeout_seconds: int, interval_seconds: int) -> None:
    deadline = time.time() + timeout_seconds
    last_error = ""
    while time.time() < deadline:
        try:
            with request.urlopen(url, timeout=5) as response:
                if 200 <= response.status < 500:
                    return
        except Exception as exc:  # noqa: BLE001
            last_error = str(exc)
        time.sleep(interval_seconds)
    raise TimeoutError(f"Backend not ready: {url}; last_error={last_error}")


def _auth_header(username: str, password: str) -> dict[str, str]:
    if not username:
        return {}
    token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
    return {"Authorization": f"Basic {token}"}


def delete_couchdb_database(base_url: str, database: str, username: str = "", password: str = "") -> int:
    url = f"{base_url.rstrip('/')}/{database}"
    req = request.Request(url, method="DELETE", headers=_auth_header(username, password))
    try:
        with request.urlopen(req, timeout=10) as response:
            return response.status
    except error.HTTPError as exc:
        if exc.code == 404:
            return 404
        raise


def create_couchdb_database(base_url: str, database: str, username: str = "", password: str = "") -> int:
    url = f"{base_url.rstrip('/')}/{database}"
    req = request.Request(url, method="PUT", headers=_auth_header(username, password))
    with request.urlopen(req, timeout=10) as response:
        return response.status


def latest_log_file(log_dir: str | Path) -> str | None:
    path = Path(log_dir)
    if not path.exists():
        return None
    files = [item for item in path.iterdir() if item.is_file() and item.suffix == ".log"]
    if not files:
        return None
    files.sort(key=lambda item: item.stat().st_mtime, reverse=True)
    return str(files[0])


def tail_text_file(file_path: str | Path, max_chars: int = 4000) -> str:
    path = Path(file_path)
    if not path.exists() or not path.is_file():
        return ""
    content = path.read_text(encoding="utf-8", errors="replace")
    if len(content) <= max_chars:
        return content
    return content[-max_chars:]


def sqlite_rows(db_path: str | Path, sql: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
    path = Path(db_path)
    if not path.exists():
        return []
    with sqlite3.connect(path) as conn:
        conn.row_factory = sqlite3.Row
        rows = conn.execute(sql, params).fetchall()
        return [{key: row[key] for key in row.keys()} for row in rows]


def persistence_rows(
    *,
    backend: str,
    sql: str,
    db_path: str | Path | None = None,
    url: str | None = None,
    limit: int = 200,
) -> list[dict[str, Any]]:
    _, rows = PersistenceBackend.query_records(
        backend=backend,
        db_path=str(db_path) if db_path is not None else None,
        url=url,
        sql=sql,
        limit=limit,
    )
    return rows


def persistence_summary(
    db_path: str | Path | None,
    node_type: str | None = None,
    *,
    backend: str = "sqlite",
    url: str | None = None,
) -> dict[str, Any]:
    records = PersistenceBackend.records_summary(
        backend=backend,
        db_path=str(db_path) if db_path is not None else None,
        url=url,
    )
    summary: dict[str, Any] = {
        "backend": backend,
        "db_path": str(db_path),
        "url": url,
        "db_exists": records.get("db_exists", False),
        "bindings": [],
        "node_status": [],
    }
    if not summary["db_exists"]:
        return summary

    target_db_path = str(db_path) if db_path is not None else None
    if node_type:
        escaped_node_type = node_type.replace("'", "''")
        summary["bindings"] = persistence_rows(
            backend=backend,
            db_path=target_db_path,
            url=url,
            sql=(
                "SELECT node_type, local_id, remote_id, last_updated "
                f"FROM bindings WHERE node_type = '{escaped_node_type}' ORDER BY last_updated DESC LIMIT 20"
            ),
        )
        summary["node_status"] = persistence_rows(
            backend=backend,
            db_path=target_db_path,
            url=url,
            sql=(
                "SELECT collection_id, node_type, action, status, binding_status, COUNT(1) AS count "
                f"FROM nodes WHERE node_type = '{escaped_node_type}' "
                "GROUP BY collection_id, node_type, action, status, binding_status "
                "ORDER BY collection_id, action, status"
            ),
        )
    return summary


def read_jsonl_records(file_path: str | Path) -> dict[str, dict[str, Any]]:
    records: dict[str, dict[str, Any]] = {}
    path = Path(file_path)
    if not path.exists():
        return records
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        item = json.loads(line)
        records[str(item.get("_id") or item.get("id"))] = item
    return records


def diff_record_fields(before: dict[str, Any], after: dict[str, Any]) -> list[str]:
    keys = set(before) | set(after)
    ignored = {"_rev", "updated_at", "updated_by", "updated_by_name"}
    changed = [key for key in sorted(keys) if key not in ignored and before.get(key) != after.get(key)]
    return changed


def now_slug() -> str:
    return datetime.now().strftime("%Y%m%d-%H%M%S")


def enrich_supplier_display_fields(data_dir: str | Path, region_file: str | Path) -> None:
    data_dir_path = Path(data_dir)
    supplier_file = data_dir_path / "supplier_1.jsonl"
    region_path = Path(region_file)
    if not supplier_file.exists() or not region_path.exists():
        return

    country_name_map: dict[str, str] = {}
    province_name_map: dict[str, str] = {}
    city_name_map: dict[str, str] = {}
    for line in region_path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        item = json.loads(line)
        region_name = str(item.get("name") or "")
        region_code = str(item.get("code") or "")
        region_type = str(item.get("region_type") or "")
        if not region_code:
            continue
        if region_type == "country":
            country_name_map[region_code] = region_name
        elif region_type == "province":
            province_name_map[region_code] = region_name
        elif region_type == "city":
            city_name_map[region_code] = region_name

    normalized_lines: list[str] = []
    changed = False
    for line in supplier_file.read_text(encoding="utf-8").splitlines():
        raw = line.strip()
        if not raw:
            continue
        item = json.loads(raw)
        country = item.get("country")
        province = item.get("province")
        city = item.get("city")
        expected_country_name = country_name_map.get(str(country), "") if country else None
        expected_province_name = province_name_map.get(str(province), "") if province else None
        expected_city_name = city_name_map.get(str(city), "") if city else None

        if item.get("country_name") != expected_country_name:
            item["country_name"] = expected_country_name
            changed = True
        if item.get("province_name") != expected_province_name:
            item["province_name"] = expected_province_name
            changed = True
        if item.get("city_name") != expected_city_name:
            item["city_name"] = expected_city_name
            changed = True

        normalized_lines.append(json.dumps(item, ensure_ascii=False))

    if changed:
        supplier_file.write_text("\n".join(normalized_lines) + "\n", encoding="utf-8")


def normalize_construction_task_plan_nodes(data_dir: str | Path) -> None:
    task_file = Path(data_dir) / "construction_task_1.jsonl"
    if not task_file.exists():
        return

    normalized_lines: list[str] = []
    changed = False
    for line in task_file.read_text(encoding="utf-8").splitlines():
        raw = line.strip()
        if not raw:
            continue
        item = json.loads(raw)
        plan_node = item.get("plan_node")
        if isinstance(plan_node, list) and len(plan_node) >= 2:
            trimmed = list(plan_node)
            if trimmed and float(trimmed[0].get("quantity") or 0) == 0:
                trimmed = trimmed[1:]
            if trimmed and item.get("contract_quantity") is not None:
                end_quantity = float(trimmed[-1].get("quantity") or 0)
                contract_quantity = float(item.get("contract_quantity") or 0)
                if end_quantity == contract_quantity:
                    trimmed = trimmed[:-1]
            if trimmed != plan_node:
                item["plan_node"] = trimmed
                changed = True
            if not trimmed:
                for field_name in ["plan_start_date", "plan_complete_date", "contract_quantity"]:
                    if item.get(field_name) is not None:
                        item[field_name] = None
                        changed = True
        normalized_lines.append(json.dumps(item, ensure_ascii=False))

    if changed:
        task_file.write_text("\n".join(normalized_lines) + "\n", encoding="utf-8")


def normalize_contract_settlement_records(data_dir: str | Path) -> None:
    settlement_file = Path(data_dir) / "contract_settlement_1.jsonl"
    if not settlement_file.exists():
        return

    excluded_ids = {
        "544430cd-1301-43e0-8f7e-a471d9438983",
    }

    normalized_lines: list[str] = []
    changed = False
    for line in settlement_file.read_text(encoding="utf-8").splitlines():
        raw = line.strip()
        if not raw:
            continue
        item = json.loads(raw)

        if str(item.get("_id") or item.get("id")) in excluded_ids:
            changed = True
            continue

        if item.get("contract_id") == "1351be9d-3ca3-48f1-be29-fb39b2c3ff15" and item.get("show_name") == "施工总承包合同":
            item["show_name"] = "施工总承包合同(不含设计、采购)"
            changed = True

        plan_node = item.get("plan_node")
        if isinstance(plan_node, list) and plan_node:
            normalized_plan_node: list[dict[str, Any]] = []
            plan_changed = False
            for node in plan_node:
                if isinstance(node, dict) and node.get("is_audit") is False:
                    normalized_plan_node.append({**node, "is_audit": True})
                    plan_changed = True
                else:
                    normalized_plan_node.append(node)
            if plan_changed:
                item["plan_node"] = normalized_plan_node
                changed = True

        normalized_lines.append(json.dumps(item, ensure_ascii=False))

    if changed:
        settlement_file.write_text("\n".join(normalized_lines) + "\n", encoding="utf-8")