#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Read-only scanner for AI automation debt and legacy human-gate residue."""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any


POLICY = "read_only_ai_automation_debt_scan"

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

SCAN_TARGETS = (
    "templates",
    "routes",
    "services",
    "scripts",
    "docs/AI_INTELLIGENCE_MODULE_SOT.md",
    "TODO_NEXT_STEPS.txt",
)

SCAN_SUFFIXES = {".py", ".html", ".js", ".md", ".txt"}

EXCLUDED_PARTS = {
    ".git",
    ".pytest_cache",
    "__pycache__",
    "node_modules",
    "data",
    "tests",
    "migrations",
    "docs/memory",
}

PRODUCT_SURFACE_PREFIXES = (
    "templates/dashboard_v2.html",
    "templates/daily_sales.html",
    "templates/growth_analysis.html",
    "routes/dashboard_routes.py",
    "routes/ai_routes.py",
    "routes/openclaw_bot_routes.py",
    "services/competitor_intel_repository.py",
    "services/competitor_match_review_service.py",
    "services/competitor_price_feeder.py",
    "services/openclaw_strategist_service.py",
    "services/pchome_mapping_backlog_service.py",
    "services/pchome_revenue_growth_service.py",
    "services/ppt_generator.py",
    "services/telegram_templates.py",
    "services/webcrumbs_host_data_service.py",
    "web/static/js/page-dashboard-v2.js",
)

MANUAL_MARKERS = (
    "需人工",
    "人工覆核",
    "人工閉環",
    "人工已",
    "人工標記",
    "人工要求",
    "人工確認",
    "人工採用",
    "人工否決",
    "人工單位價",
    "重算待人工",
    "HITL",
    "requires_hitl",
    "human_review_required",
    "manual_review_required",
    "manual_required",
    "needs_human",
    "ready_for_manual",
    "manual_operator_approval",
    "manual_approval_required",
    "manual_sample",
    "manual_fetch",
)

HARD_GATE_MARKERS = (
    "secret",
    "token",
    "private key",
    "cookie",
    "raw session",
    "authorization header",
    "DROP ",
    "TRUNCATE ",
    "destructive migration",
    "reboot",
    "force push",
    "paid provider",
)

VISIBLE_HUMAN_TEXT = (
    "需人工",
    "人工覆核",
    "人工閉環",
    "人工已",
    "人工標記",
    "人工要求",
    "人工確認",
    "人工採用",
    "人工否決",
    "人工單位價",
    "重算待人工",
    "HITL",
)


@dataclass(frozen=True)
class Finding:
    file: str
    line: int
    marker: str
    snippet: str
    category: str
    priority: str
    controlled_apply_allowed: bool
    recommended_next_action: str

    def as_dict(self) -> dict[str, Any]:
        return {
            "file": self.file,
            "line": self.line,
            "marker": self.marker,
            "snippet": self.snippet,
            "category": self.category,
            "priority": self.priority,
            "controlled_apply_allowed": self.controlled_apply_allowed,
            "recommended_next_action": self.recommended_next_action,
        }


def _relative(path: Path, root: Path) -> str:
    return path.relative_to(root).as_posix()


def _is_excluded(relative_path: str) -> bool:
    if relative_path in {
        "services/ai_automation_debt_service.py",
        "services/ai_surface_html_readback_service.py",
    }:
        return True
    parts = set(relative_path.split("/"))
    if parts & {".git", ".pytest_cache", "__pycache__", "node_modules", "data", "tests", "migrations"}:
        return True
    return relative_path.startswith("docs/memory/")


def _iter_scan_files(root: Path) -> list[Path]:
    files: list[Path] = []
    for target in SCAN_TARGETS:
        path = root / target
        if not path.exists():
            continue
        if path.is_file():
            if path.suffix in SCAN_SUFFIXES and not _is_excluded(_relative(path, root)):
                files.append(path)
            continue
        for candidate in path.rglob("*"):
            if not candidate.is_file() or candidate.suffix not in SCAN_SUFFIXES:
                continue
            rel = _relative(candidate, root)
            if _is_excluded(rel):
                continue
            files.append(candidate)
    return sorted(set(files))


def _first_marker(line: str) -> str | None:
    return next((marker for marker in MANUAL_MARKERS if marker in line), None)


def _has_hard_gate_context(line: str) -> bool:
    lower = line.lower()
    return any(marker.lower() in lower for marker in HARD_GATE_MARKERS)


def _is_product_surface(relative_path: str) -> bool:
    return any(relative_path == prefix or relative_path.startswith(prefix) for prefix in PRODUCT_SURFACE_PREFIXES)


def _is_legacy_compatibility_line(line: str) -> bool:
    stripped = line.strip()
    if any(text in stripped for text in VISIBLE_HUMAN_TEXT):
        return False
    if "requires_hitl" in stripped and "True" not in stripped and "true" not in stripped:
        return True
    if 'get("human_review_required")' in stripped or "get('human_review_required')" in stripped:
        return True
    compatibility_tokens = (
        "requires_hitl",
        "manual_",
        "human_review_required",
        "manual_review_required",
        "legacy_human_review_required",
        "hitl_count",
    )
    if not any(token in stripped for token in compatibility_tokens):
        return False
    false_or_count_zero = (
        "False" in stripped
        or "false" in stripped
        or "_count" in stripped
        or "legacy_" in stripped
        or "manual_" in stripped
    )
    return false_or_count_zero


def _classify(relative_path: str, line: str, marker: str) -> tuple[str, str, bool, str]:
    legacy_compatibility_line = _is_legacy_compatibility_line(line)
    if _has_hard_gate_context(line) and not legacy_compatibility_line:
        return (
            "incident_hard_gate",
            "P0",
            False,
            "Keep as hard gate; require break-glass path, replay/shadow/canary, and explicit external approval.",
        )

    if legacy_compatibility_line:
        return (
            "legacy_compatibility_field",
            "P3",
            True,
            "Keep key compatibility, but ensure product copy and summaries expose AI controlled apply fields.",
        )

    if relative_path == "routes/openclaw_bot_routes.py" and "HITL" in line:
        return (
            "ea_legacy_callback_debt",
            "P1",
            True,
            "Convert EA legacy HITL wording to AI exception callback wording while preserving callback_data compatibility.",
        )

    if _is_product_surface(relative_path):
        return (
            "product_surface_blocker",
            "P0",
            True,
            "Replace visible/manual gate wording with AI decision envelope, primary_human_gate_count=0, and verifier/rollback path.",
        )

    if relative_path.startswith("services/market_intel/") or relative_path.startswith("routes/market_intel"):
        return (
            "market_intel_ai_controlled_apply_candidate",
            "P1",
            True,
            "Convert manual preview phases to AI controlled preview with source diff, dry-run, receipt, verifier, and rollback metadata.",
        )

    if relative_path.startswith("docs/") or relative_path == "TODO_NEXT_STEPS.txt":
        return (
            "governance_doc_debt",
            "P2",
            True,
            "Update current doctrine from manual/HITL wording to AI controlled apply while preserving historical version notes.",
        )

    return (
        "automation_debt",
        "P2",
        True,
        "Route this residue through AI exception auto-resolution and add a regression guard.",
    )


def _scan_file(path: Path, root: Path, per_file_limit: int) -> list[Finding]:
    relative_path = _relative(path, root)
    findings: list[Finding] = []
    try:
        lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()
    except OSError:
        return findings

    for index, line in enumerate(lines, start=1):
        marker = _first_marker(line)
        if not marker:
            continue
        category, priority, allowed, action = _classify(relative_path, line, marker)
        findings.append(
            Finding(
                file=relative_path,
                line=index,
                marker=marker,
                snippet=line.strip()[:220],
                category=category,
                priority=priority,
                controlled_apply_allowed=allowed,
                recommended_next_action=action,
            )
        )
        if len(findings) >= per_file_limit:
            break
    return findings


def _priority_key(finding: Finding) -> tuple[int, str, int]:
    rank = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}.get(finding.priority, 9)
    return rank, finding.file, finding.line


def _market_intel_ai_alias_count() -> int:
    try:
        from services.market_intel.ai_controlled_route_aliases import (
            AI_CONTROLLED_ROUTE_ALIASES,
        )
    except Exception:
        return 0
    return len(AI_CONTROLLED_ROUTE_ALIASES)


def build_ai_automation_debt_report(
    *,
    root: Path | str | None = None,
    max_findings: int = 120,
    per_file_limit: int = 8,
) -> dict[str, Any]:
    """Build a read-only, machine-actionable AI automation debt inventory."""
    scan_root = Path(root) if root is not None else ROOT
    max_findings = max(10, min(int(max_findings or 120), 500))
    per_file_limit = max(1, min(int(per_file_limit or 8), 40))

    files = _iter_scan_files(scan_root)
    findings: list[Finding] = []
    for path in files:
        findings.extend(_scan_file(path, scan_root, per_file_limit=per_file_limit))

    findings.sort(key=_priority_key)
    all_finding_dicts = [finding.as_dict() for finding in findings]
    visible_findings = all_finding_dicts[:max_findings]

    category_counts: dict[str, int] = {}
    priority_counts: dict[str, int] = {}
    for finding in findings:
        category_counts[finding.category] = category_counts.get(finding.category, 0) + 1
        priority_counts[finding.priority] = priority_counts.get(finding.priority, 0) + 1

    product_surface_blocker_count = category_counts.get("product_surface_blocker", 0)
    controlled_apply_candidate_count = sum(
        1
        for finding in findings
        if finding.controlled_apply_allowed and finding.category != "legacy_compatibility_field"
    )
    hard_gate_count = category_counts.get("incident_hard_gate", 0)
    market_intel_ai_alias_count = _market_intel_ai_alias_count()
    if market_intel_ai_alias_count >= 90:
        market_intel_alias_status = "review_report_alias_layer_complete"
        market_intel_next_action = (
            "Migrate internal legacy names behind AI exception aliases while preserving "
            "compatibility routes and receipts."
        )
    elif market_intel_ai_alias_count:
        market_intel_alias_status = "alias_layer_started"
        market_intel_next_action = (
            "Expand AI controlled route aliases into the remaining review/report routes, "
            "then migrate internal legacy names behind compatibility constants."
        )
    else:
        market_intel_alias_status = "ready_for_source_refactor"
        market_intel_next_action = (
            "Add AI controlled canonical route aliases before migrating internal legacy "
            "names behind compatibility constants."
        )

    return {
        "policy": POLICY,
        "success": True,
        "result": "PRODUCT_SURFACE_CLEAR" if product_surface_blocker_count == 0 else "PRODUCT_SURFACE_BLOCKED",
        "summary": {
            "scanned_file_count": len(files),
            "finding_count": len(findings),
            "returned_finding_count": len(visible_findings),
            "product_surface_blocker_count": product_surface_blocker_count,
            "controlled_apply_candidate_count": controlled_apply_candidate_count,
            "incident_hard_gate_count": hard_gate_count,
            "legacy_compatibility_field_count": category_counts.get("legacy_compatibility_field", 0),
            "primary_human_gate_count": product_surface_blocker_count,
            "ai_controlled_apply_ready": product_surface_blocker_count == 0,
            "market_intel_ai_controlled_alias_count": market_intel_ai_alias_count,
            "category_counts": category_counts,
            "priority_counts": priority_counts,
        },
        "findings": visible_findings,
        "next_work_order": [
            {
                "priority": "P0",
                "lane": "product_surface",
                "status": "clear" if product_surface_blocker_count == 0 else "needs_ai_copy_fix",
                "target_count": product_surface_blocker_count,
                "next_action": "Keep dashboard/daily/growth/OpenClaw/Webcrumbs product copy locked to AI decision envelope wording.",
            },
            {
                "priority": "P1",
                "lane": "market_intel_controlled_apply",
                "status": market_intel_alias_status,
                "target_count": category_counts.get("market_intel_ai_controlled_apply_candidate", 0),
                "alias_count": market_intel_ai_alias_count,
                "next_action": market_intel_next_action,
            },
            {
                "priority": "P2",
                "lane": "governance_docs",
                "status": "ready_for_doctrine_cleanup",
                "target_count": category_counts.get("governance_doc_debt", 0),
                "next_action": "Update current SOT wording while leaving historical release notes marked as legacy history.",
            },
            {
                "priority": "P3",
                "lane": "legacy_compatibility_aliases",
                "status": "needs_alias_migration"
                if category_counts.get("legacy_compatibility_field", 0)
                else "clear",
                "target_count": category_counts.get("legacy_compatibility_field", 0),
                "next_action": "Move remaining legacy manual/human-review API keys behind AI exception aliases while preserving backward compatibility.",
            },
        ],
        "safety": {
            "read_only": True,
            "writes_database": False,
            "executes_network": False,
            "uses_llm": False,
            "scans_raw_sessions": False,
            "github_used": False,
        },
    }
