#!/usr/bin/env python3

from __future__ import annotations

import argparse
import csv
import json
import shutil
import sqlite3
import sys
import tempfile
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple


SCRIPT_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = SCRIPT_DIR.parent
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from config import MONITORING_DB_PATH, TASK_CSV_PATH  # noqa: E402
from analytics import AnalyticsRepository  # noqa: E402
from result_codes import (  # noqa: E402
    APP_ERROR_DESC,
    BUSINESS_ERROR_DESC,
    DOWNLOAD_ERROR_DESC,
    INFRA_ERROR_DESC,
    AppError,
    BusinessError,
    DownloadError,
    InfraError,
)


def _parse_downloads(value: Any) -> Optional[int]:
    text = str(value or "").strip().upper()
    if not text:
        return None
    normalized = text.replace(",", "").replace(" ", "")
    if normalized.endswith("+"):
        normalized = normalized[:-1]
    multiplier = 1
    if normalized.endswith("K"):
        multiplier = 1_000
        normalized = normalized[:-1]
    elif normalized.endswith("M"):
        multiplier = 1_000_000
        normalized = normalized[:-1]
    elif normalized.endswith("B"):
        multiplier = 1_000_000_000
        normalized = normalized[:-1]
    try:
        return int(float(normalized) * multiplier)
    except (TypeError, ValueError):
        return None


def _humanize_error_type(error_type: str) -> str:
    normalized = str(error_type or "").strip()
    if not normalized:
        return "未标记错误"
    category_name, _, code_text = normalized.partition("/")
    try:
        code = int(code_text)
    except (TypeError, ValueError):
        return normalized
    try:
        if category_name == "INFRA_ERROR":
            return INFRA_ERROR_DESC.get(InfraError(code), normalized)
        if category_name == "APP_ERROR":
            return APP_ERROR_DESC.get(AppError(code), normalized)
        if category_name == "BUSINESS_ERROR":
            return BUSINESS_ERROR_DESC.get(BusinessError(code), normalized)
        if category_name == "DOWNLOAD_ERROR":
            return DOWNLOAD_ERROR_DESC.get(DownloadError(code), normalized)
    except ValueError:
        return normalized
    return normalized


def _load_downloads_by_package(csv_path: str) -> Dict[str, Optional[int]]:
    path = Path(csv_path)
    if not path.exists():
        return {}
    result: Dict[str, Optional[int]] = {}
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        for row in csv.DictReader(handle):
            package_name = str(row.get("package_name") or row.get("包名") or "").strip()
            if not package_name or package_name in result:
                continue
            result[package_name] = _parse_downloads(
                row.get("downloads")
                or row.get("download")
                or row.get("下载量")
            )
    return result


def _copy_sqlite_snapshot(db_path: str) -> Tuple[str, tempfile.TemporaryDirectory[str]]:
    source = Path(db_path)
    temp_dir: tempfile.TemporaryDirectory[str] = tempfile.TemporaryDirectory(prefix="dpi-db-snapshot-")
    snapshot_path = Path(temp_dir.name) / source.name
    shutil.copy2(source, snapshot_path)
    for suffix in ("-wal", "-shm"):
        sidecar = Path(f"{db_path}{suffix}")
        if sidecar.exists():
            shutil.copy2(sidecar, Path(f"{snapshot_path}{suffix}"))
    return str(snapshot_path), temp_dir


def _open_readable_connection(db_path: str) -> Tuple[sqlite3.Connection, Optional[tempfile.TemporaryDirectory[str]], bool]:
    try:
        connection = sqlite3.connect(db_path)
        connection.row_factory = sqlite3.Row
        connection.execute("SELECT name FROM sqlite_master WHERE type = 'table' LIMIT 1").fetchone()
        return connection, None, False
    except sqlite3.Error:
        snapshot_path, temp_dir = _copy_sqlite_snapshot(db_path)
        connection = sqlite3.connect(snapshot_path)
        connection.row_factory = sqlite3.Row
        connection.execute("SELECT name FROM sqlite_master WHERE type = 'table' LIMIT 1").fetchone()
        return connection, temp_dir, True


def _extract_downloads_from_payload(task_payload_json: Optional[str]) -> Optional[int]:
    if not task_payload_json:
        return None
    try:
        payload = json.loads(task_payload_json)
    except (TypeError, ValueError):
        return None
    if not isinstance(payload, dict):
        return None
    original_row = payload.get("original_row")
    if isinstance(original_row, dict):
        for key in ("downloads", "download", "下载量"):
            if key in original_row:
                return _parse_downloads(original_row.get(key))
    for key in ("downloads", "download", "下载量"):
        if key in payload:
            return _parse_downloads(payload.get(key))
    return None


def build_reason_popularity_report(
    db_path: str = MONITORING_DB_PATH,
    csv_path: str = TASK_CSV_PATH,
    *,
    threshold: int = 100_000,
) -> Dict[str, Any]:
    connection, temp_dir, used_snapshot = _open_readable_connection(db_path)
    try:
        repo = AnalyticsRepository.__new__(AnalyticsRepository)
        summaries = repo._list_catalog_summaries(connection)
        rows = [
            item for item in summaries
            if item.get("restriction_status") == "severe_restricted"
            and item.get("retryability") == "non_retryable"
        ]
        rows.sort(key=lambda item: (str(item.get("latest_failure_type") or ""), str(item.get("package_name") or "")))
    finally:
        connection.close()
        if temp_dir is not None:
            temp_dir.cleanup()

    downloads_by_package = _load_downloads_by_package(csv_path)

    reason_buckets: Dict[str, Dict[str, Any]] = {}
    missing_download_packages: List[str] = []
    matched_downloads_apps = 0
    for row in rows:
        package_name = str(row.get("package_name") or "").strip()
        downloads = downloads_by_package.get(package_name)
        if downloads is None:
            downloads = _extract_downloads_from_payload(row.get("task_payload_json"))
        if downloads is None and row.get("downloads") is not None:
            downloads = int(row.get("downloads") or 0)
        if downloads is None:
            missing_download_packages.append(package_name)
            continue
        matched_downloads_apps += 1
        error_type = str(row.get("latest_failure_type") or "").strip()
        bucket = reason_buckets.setdefault(
            error_type,
            {
                "error_type": error_type,
                "reason_label": _humanize_error_type(error_type),
                "hot_count": 0,
                "non_hot_count": 0,
                "total_count": 0,
            },
        )
        if downloads > threshold:
            bucket["hot_count"] += 1
        else:
            bucket["non_hot_count"] += 1
        bucket["total_count"] += 1

    report_rows = sorted(
        reason_buckets.values(),
        key=lambda item: (-int(item["total_count"]), -int(item["hot_count"]), str(item["error_type"])),
    )
    return {
        "db_path": db_path,
        "csv_path": csv_path,
        "threshold": int(threshold),
        "used_snapshot": used_snapshot,
        "total_non_retryable_apps": len(rows),
        "matched_downloads_apps": matched_downloads_apps,
        "missing_downloads_apps": len(missing_download_packages),
        "missing_download_packages": missing_download_packages,
        "rows": report_rows,
    }


def _iter_table_rows(report_rows: Iterable[Dict[str, Any]]) -> Iterable[List[str]]:
    for item in report_rows:
        yield [
            item["reason_label"],
            item["error_type"] or "-",
            str(item["hot_count"]),
            str(item["non_hot_count"]),
            str(item["total_count"]),
        ]


def _format_table(report_rows: List[Dict[str, Any]]) -> str:
    headers = ["失败原因", "错误码", "热门", "非热门", "总数"]
    rows = list(_iter_table_rows(report_rows))
    if not rows:
        return "没有符合条件的应用。"
    widths = [len(header) for header in headers]
    for row in rows:
        for index, cell in enumerate(row):
            widths[index] = max(widths[index], len(cell))

    def _render(row: List[str]) -> str:
        return " | ".join(cell.ljust(widths[index]) for index, cell in enumerate(row))

    divider = "-+-".join("-" * width for width in widths)
    lines = [_render(headers), divider]
    lines.extend(_render(row) for row in rows)
    return "\n".join(lines)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="统计严重受限不可重试应用中，各失败原因对应的热门/非热门数量。"
    )
    parser.add_argument("--db-path", default=MONITORING_DB_PATH, help="monitoring sqlite 路径。")
    parser.add_argument("--csv-path", default=TASK_CSV_PATH, help="包含 downloads 列的应用清单 CSV。")
    parser.add_argument(
        "--threshold",
        type=int,
        default=100_000,
        help="热门阈值。downloads > threshold 视为热门，默认 100000。",
    )
    parser.add_argument(
        "--format",
        choices=("table", "json"),
        default="table",
        help="输出格式，默认 table。",
    )
    parser.add_argument(
        "--show-missing-packages",
        action="store_true",
        help="额外打印缺失下载量的包名。",
    )
    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()
    report = build_reason_popularity_report(
        db_path=args.db_path,
        csv_path=args.csv_path,
        threshold=args.threshold,
    )

    if args.format == "json":
        print(json.dumps(report, ensure_ascii=False, indent=2))
        return 0

    print(f"数据库: {report['db_path']}")
    print(f"应用清单: {report['csv_path']}")
    print(f"热门阈值: downloads > {report['threshold']}")
    print("统计范围: restriction_status=severe_restricted AND retryability=non_retryable")
    print(f"严重受限不可重试应用总数: {report['total_non_retryable_apps']}")
    print(f"已匹配下载量应用数: {report['matched_downloads_apps']}")
    print(f"下载量缺失应用数: {report['missing_downloads_apps']}")
    if report["used_snapshot"]:
        print("读取方式: 已自动复制 live sqlite 快照后再统计")
    print()
    print(_format_table(report["rows"]))

    if args.show_missing_packages and report["missing_download_packages"]:
        print()
        print("下载量缺失包名:")
        for package_name in report["missing_download_packages"]:
            print(package_name)
    return 0


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