#!/usr/bin/env python3
"""Build a repo-only Telegram notification egress inventory.

This scanner identifies Telegram Bot API sendMessage paths that can bypass
TelegramGateway's final-exit formatter. It does not read secrets, call
Telegram, modify workflows, or send notifications.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any


TAIPEI = timezone(timedelta(hours=8))

SCAN_ROOTS = (
    Path(".gitea/workflows"),
    Path("scripts/ops"),
    Path("scripts/ci"),
    Path("apps/api/src"),
)
SCAN_SUFFIXES = {".py", ".sh", ".js", ".yml", ".yaml"}

DIRECT_BOT_API_RE = re.compile(
    r"api\.telegram\.org/bot.*sendMessage|sendMessage.*api\.telegram\.org/bot"
)
GATEWAY_CALLSITE_RE = re.compile(
    r"(?:send_alert_notification\(|\b(?:tg|gw|gateway|telegram)\.send_text\(|_send_request\(\s*[\"']sendMessage[\"'])"
)
SECRET_INTERPOLATION_RE = re.compile(r"\$\{\{\s*secrets\.[^}]+\}\}")
BOT_TOKEN_URL_RE = re.compile(r"api\.telegram\.org/bot.*?/sendMessage")

REQUIRED_OWNER_FIELDS = [
    "egress_surface_id",
    "owner_role_or_team",
    "routing_purpose",
    "current_sender",
    "target_chat_route",
    "message_shape_contract",
    "redaction_contract",
    "formatter_convergence_plan",
    "delivery_receipt_ref",
    "dedup_or_fingerprint_plan",
    "fallback_or_degraded_mode",
    "migration_or_exception_reason",
    "maintenance_window",
    "rollback_owner",
    "postcheck_evidence_ref",
    "no_secret_value_attestation",
    "no_raw_payload_attestation",
    "no_false_green_attestation",
]

REVIEWER_CHECKS = [
    "direct_bot_api_surface_identified",
    "owner_role_present",
    "target_route_is_sre_owned",
    "message_shape_is_ai_automation_card_or_documented_exception",
    "redaction_contract_present",
    "formatter_convergence_path_present",
    "delivery_receipt_metadata_only",
    "dedup_or_fingerprint_present",
    "fallback_mode_does_not_leak_raw_payload",
    "secret_name_only_no_value",
    "workflow_or_script_change_requires_separate_approval",
    "telegram_send_not_executed_by_inventory",
    "no_false_green_claim",
    "runtime_gate_stays_zero",
]

OUTCOME_LANES = [
    "waiting_owner_response",
    "request_owner_route_supplement",
    "request_formatter_convergence_plan",
    "request_redaction_contract",
    "request_delivery_receipt_metadata",
    "quarantine_secret_or_raw_payload",
    "reject_false_green_claim",
    "ready_for_notification_egress_review",
    "waiting_runtime_gate",
]

BLOCKED_ACTIONS = [
    "telegram_send",
    "bot_api_call",
    "workflow_modification",
    "script_modification_without_owner",
    "secret_value_collection",
    "secret_hash_collection",
    "partial_token_collection",
    "chat_id_collection_without_owner",
    "store_raw_message_payload",
    "store_unredacted_workflow_log",
    "change_chat_route",
    "change_bot_token",
    "rotate_secret",
    "workflow_dispatch",
    "production_deploy",
    "accept_route_200_as_delivery_receipt",
    "accept_cd_success_as_notification_acceptance",
    "accept_ui_visible_as_notification_acceptance",
    "skip_formatter_convergence",
    "skip_redaction_review",
    "open_runtime_gate",
    "add_action_button",
]


def git_short_sha(root: Path) -> str:
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            cwd=root,
            check=True,
            capture_output=True,
            text=True,
        )
        return result.stdout.strip()
    except Exception:
        return "unknown"


def iter_scannable_files(root: Path) -> list[Path]:
    files: list[Path] = []
    for scan_root in SCAN_ROOTS:
        absolute_root = root / scan_root
        if not absolute_root.exists():
            continue
        for path in absolute_root.rglob("*"):
            if path.is_file() and path.suffix in SCAN_SUFFIXES:
                files.append(path)
    return sorted(files)


def sanitize_excerpt(line: str) -> str:
    excerpt = line.strip()
    excerpt = SECRET_INTERPOLATION_RE.sub("${{ secrets.<redacted> }}", excerpt)
    excerpt = BOT_TOKEN_URL_RE.sub("api.telegram.org/bot<redacted>/sendMessage", excerpt)
    return excerpt[:180]


def surface_kind(relative_path: str) -> str:
    if relative_path.startswith(".gitea/workflows/"):
        return "gitea_workflow_direct_bot_api"
    if relative_path.startswith("scripts/ops/"):
        return "ops_script_direct_bot_api"
    if relative_path.startswith("scripts/ci/"):
        return "ci_script_direct_bot_api"
    if relative_path.startswith("apps/api/src/"):
        return "api_direct_bot_api"
    return "other_direct_bot_api"


def line_hash(relative_path: str, line_number: int, line: str) -> str:
    payload = f"{relative_path}:{line_number}:{line.strip()}".encode("utf-8")
    return hashlib.sha256(payload).hexdigest()[:16]


def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
    generated = generated_at or datetime.now(TAIPEI).isoformat(timespec="seconds")
    files = iter_scannable_files(root)
    direct_calls: list[dict[str, Any]] = []
    gateway_calls: list[dict[str, Any]] = []

    for path in files:
        relative_path = path.relative_to(root).as_posix()
        text = path.read_text(encoding="utf-8", errors="replace")
        for line_number, line in enumerate(text.splitlines(), start=1):
            if DIRECT_BOT_API_RE.search(line):
                kind = surface_kind(relative_path)
                direct_calls.append(
                    {
                        "egress_surface_id": f"telegram_egress:{kind}:{relative_path}:{line_number}",
                        "surface_kind": kind,
                        "path": relative_path,
                        "line": line_number,
                        "line_hash": line_hash(relative_path, line_number, line),
                        "sanitized_excerpt": sanitize_excerpt(line),
                        "required_owner_fields": REQUIRED_OWNER_FIELDS,
                        "reviewer_checks": REVIEWER_CHECKS,
                        "outcome_lanes": OUTCOME_LANES,
                        "blocked_actions": BLOCKED_ACTIONS,
                        "owner_response_received": False,
                        "owner_response_accepted": False,
                        "formatter_convergence_accepted": False,
                        "redaction_contract_accepted": False,
                        "delivery_receipt_accepted": False,
                        "direct_bot_api_migration_authorized": False,
                        "telegram_send_authorized": False,
                        "bot_api_call_authorized": False,
                        "workflow_modification_authorized": False,
                        "script_modification_authorized": False,
                        "secret_value_collection_allowed": False,
                        "raw_payload_storage_allowed": False,
                        "production_write_authorized": False,
                        "runtime_gate": False,
                        "action_buttons_allowed": False,
                        "not_authorization": True,
                    }
                )

            if GATEWAY_CALLSITE_RE.search(line):
                gateway_calls.append(
                    {
                        "path": relative_path,
                        "line": line_number,
                        "line_hash": line_hash(relative_path, line_number, line),
                    }
                )

    direct_files = sorted({item["path"] for item in direct_calls})
    workflow_direct_calls = [item for item in direct_calls if item["surface_kind"] == "gitea_workflow_direct_bot_api"]
    ops_direct_calls = [item for item in direct_calls if item["surface_kind"] == "ops_script_direct_bot_api"]
    ci_direct_calls = [item for item in direct_calls if item["surface_kind"] == "ci_script_direct_bot_api"]
    api_direct_calls = [item for item in direct_calls if item["surface_kind"] == "api_direct_bot_api"]
    telegram_gateway_path = root / "apps/api/src/services/telegram_gateway.py"
    telegram_gateway_text = telegram_gateway_path.read_text(encoding="utf-8", errors="replace")
    gateway_formatter_present = "normalize_telegram_send_message_payload" in telegram_gateway_text

    return {
        "schema_version": "telegram_notification_egress_inventory_v1",
        "generated_at": generated,
        "git_commit": git_short_sha(root),
        "status": "inventory_ready_no_runtime_action",
        "mode": "repo_only_scan_no_secret_value_no_telegram_send",
        "scan_roots": [path.as_posix() for path in SCAN_ROOTS],
        "summary": {
            "scanned_file_count": len(files),
            "direct_bot_api_file_count": len(direct_files),
            "direct_bot_api_call_count": len(direct_calls),
            "workflow_direct_bot_api_call_count": len(workflow_direct_calls),
            "ops_script_direct_bot_api_call_count": len(ops_direct_calls),
            "ci_script_direct_bot_api_call_count": len(ci_direct_calls),
            "api_direct_bot_api_call_count": len(api_direct_calls),
            "gateway_normalized_callsite_count": len(gateway_calls),
            "gateway_final_exit_formatter_present_count": 1 if gateway_formatter_present else 0,
            "required_owner_field_count": len(REQUIRED_OWNER_FIELDS),
            "reviewer_check_count": len(REVIEWER_CHECKS),
            "outcome_lane_count": len(OUTCOME_LANES),
            "blocked_action_count": len(BLOCKED_ACTIONS),
            "owner_response_received_count": 0,
            "owner_response_accepted_count": 0,
            "formatter_convergence_accepted_count": 0,
            "redaction_contract_accepted_count": 0,
            "delivery_receipt_accepted_count": 0,
            "direct_bot_api_migration_authorized_count": 0,
            "telegram_send_authorized_count": 0,
            "bot_api_call_authorized_count": 0,
            "workflow_modification_authorized_count": 0,
            "script_modification_authorized_count": 0,
            "secret_value_collection_allowed_count": 0,
            "raw_payload_storage_allowed_count": 0,
            "production_write_authorized_count": 0,
            "runtime_gate_count": 0,
            "action_button_count": 0,
        },
        "execution_boundaries": {
            "runtime_execution_authorized": False,
            "telegram_send_authorized": False,
            "bot_api_call_authorized": False,
            "workflow_modification_authorized": False,
            "script_modification_authorized": False,
            "secret_value_collection_allowed": False,
            "secret_hash_collection_allowed": False,
            "partial_token_collection_allowed": False,
            "raw_payload_storage_allowed": False,
            "chat_route_change_authorized": False,
            "bot_token_change_authorized": False,
            "workflow_dispatch_authorized": False,
            "production_deploy_authorized": False,
            "action_buttons_allowed": False,
            "not_authorization": True,
        },
        "direct_bot_api_calls": direct_calls,
        "gateway_normalized_callsite_refs": gateway_calls,
        "operator_interpretation": [
            "direct_bot_api_call_count 大於 0 代表仍有 workflow / ops / API 旁路可能繞過 TelegramGateway formatter。",
            "本清冊只建立 metadata-only egress surface，不送 Telegram、不修改 workflow / script、不讀 secret value。",
            "後續要收斂 direct Bot API 必須另走 owner response、formatter convergence、redaction contract、delivery receipt 與維護窗口。",
        ],
    }


def validate(root: Path) -> None:
    report = build_report(root)
    if report["summary"]["gateway_final_exit_formatter_present_count"] != 1:
        raise SystemExit("BLOCKED telegram egress inventory: gateway formatter not found")


def main() -> None:
    parser = argparse.ArgumentParser(description="Build Telegram notification egress inventory")
    parser.add_argument("--root", default=".", help="repository root")
    parser.add_argument("--output", help="write JSON snapshot")
    parser.add_argument("--generated-at", help="fixed generated_at timestamp")
    args = parser.parse_args()

    root = Path(args.root).resolve()
    report = build_report(root, args.generated_at)
    payload = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
    if args.output:
        Path(args.output).write_text(payload, encoding="utf-8")
    else:
        sys.stdout.write(payload)

    print(
        "TELEGRAM_NOTIFICATION_EGRESS_INVENTORY_OK "
        f"direct_calls={report['summary']['direct_bot_api_call_count']} "
        f"files={report['summary']['direct_bot_api_file_count']} "
        f"workflow={report['summary']['workflow_direct_bot_api_call_count']} "
        f"ops={report['summary']['ops_script_direct_bot_api_call_count']} "
        f"api={report['summary']['api_direct_bot_api_call_count']} "
        f"runtime_gate={report['summary']['runtime_gate_count']}",
        file=sys.stderr,
    )


if __name__ == "__main__":
    main()
