import re
from pathlib import Path


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


def _env_example_keys() -> set[str]:
    keys = set()
    for raw_line in (ROOT / ".env.example").read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _value = line.split("=", 1)
        keys.add(key.strip())
    return keys


def _runtime_env_keys_from_code() -> set[str]:
    pattern = re.compile(r"""os\.(?:getenv|environ\.get)\(\s*['"]([A-Z][A-Z0-9_]+)['"]""")
    scan_paths = [
        ROOT / "app.py",
        ROOT / "config.py",
        ROOT / "scheduler.py",
        ROOT / "run_scheduler.py",
        ROOT / "routes",
        ROOT / "services",
        ROOT / "utils",
    ]

    keys = set()
    for scan_path in scan_paths:
        paths = scan_path.rglob("*.py") if scan_path.is_dir() else [scan_path]
        for path in paths:
            if not path.exists() or "__pycache__" in path.parts:
                continue
            content = path.read_text(encoding="utf-8", errors="ignore")
            keys.update(match.group(1) for match in pattern.finditer(content))
    return keys


def _compose_env_keys() -> set[str]:
    pattern = re.compile(r"""\$\{([A-Z][A-Z0-9_]+)(?::-[^}]*)?\}""")
    keys = set()
    for path in ROOT.glob("docker-compose*.yml"):
        keys.update(pattern.findall(path.read_text(encoding="utf-8", errors="ignore")))
    return keys


def test_phase3f_orphan_ai_services_stay_removed():
    orphan_services = [
        "services/elephant_alpha_decision_router.py",
        "services/telegram_ai_integration.py",
        "services/watcher_agent.py",
    ]

    assert [path for path in orphan_services if (ROOT / path).exists()] == []


def test_app_py_stays_blueprint_only_for_routes():
    app_source = (ROOT / "app.py").read_text(encoding="utf-8")

    assert "@app.route" not in app_source


def test_app_py_does_not_start_with_stale_restart_todo_banner():
    app_source = (ROOT / "app.py").read_text(encoding="utf-8")

    assert not app_source.startswith("# ================= TODO LIST")
    assert "重開機後請依序執行" not in app_source


def test_active_guides_do_not_point_to_removed_ai_services():
    active_guides = [
        ROOT / "docs" / "ELEPHANT_ALPHA_SETUP.md",
    ]
    removed_modules = [
        "services/telegram_ai_integration.py",
        "services/watcher_agent.py",
    ]

    for guide in active_guides:
        content = guide.read_text(encoding="utf-8")
        assert "Modify routing behavior in `services/elephant_alpha_decision_router.py`" not in content
        for module_path in removed_modules:
            assert module_path not in content


def test_env_example_documents_runtime_and_ai_automation_variables():
    expected_keys = {
        "AI_CALL_LOGGING_ENABLED",
        "AUTO_FIX_ENABLED",
        "AWAITING_REVIEW_PUSH_BATCH",
        "CODE_REVIEW_AUTO_FIX_ENABLED",
        "CODE_REVIEW_OLLAMA_MODEL",
        "CODE_REVIEW_OLLAMA_TIMEOUT",
        "COST_THROTTLE_ENABLED",
        "COST_THROTTLE_PROJECT_RATIO",
        "COST_UNTHROTTLE_PROJECT_RATIO",
        "DEEPSEEK_API_KEY",
        "DEEPSEEK_BASE_URL",
        "DEEPSEEK_DIRECT_ENABLED",
        "DEEPSEEK_MODEL",
        "DEEPSEEK_TIMEOUT",
        "ELEPHANT_ALPHA_ALLOWED_SSH_HOSTS",
        "ELEPHANT_ALPHA_URL",
        "EMBEDDING_HOST",
        "EMBEDDING_TIMEOUT",
        "GUNICORN_TIMEOUT",
        "GUNICORN_THREADS",
        "GUNICORN_WORKER_CLASS",
        "LINE_ENABLED",
        "MOMO_AI_AUTOMATION_SMOKE_HISTORY",
        "MOMO_AI_AUTOMATION_SMOKE_HISTORY_LIMIT",
        "MOMO_ALLOW_INSECURE_INTERNAL_WEBHOOK_FOR_DEV",
        "MOMO_EVENT_ROUTER_DEFAULT_DEDUP_SEC",
        "MOMO_EVENT_ROUTER_QUEUE",
        "MOMO_EVENT_ROUTER_REPLAY_LIMIT",
        "MOMO_EVENT_ROUTER_REPLAY_ON_SUCCESS",
        "MCP_FILESYSTEM_URL",
        "MCP_FIRECRAWL_URL",
        "MCP_CACHE_TTL_SEC",
        "MCP_MAX_RESULT_BYTES",
        "MCP_OMNISEARCH_URL",
        "MCP_POSTGRES_URL",
        "MCP_ROUTER_ENABLED",
        "MCP_TIMEOUT_SEC",
        "MODEL_ROUTER_ENABLED",
        "N8N_HOST",
        "N8N_PASSWORD",
        "N8N_PROTOCOL",
        "N8N_USER",
        "N8N_WEBHOOK_BASE_URL",
        "NEMOTRON_OLLAMA_FIRST",
        "NEMOTRON_OLLAMA_MODEL",
        "NEMOTRON_OLLAMA_TIMEOUT",
        "OLLAMA_EMBED_TIMEOUT",
        "OPENCLAW_ADMIN_USER_IDS",
        "OPENCLAW_AGENT_DISPATCH",
        "OPENCLAW_AGENT_DISPATCH_CMDS",
        "OPENCLAW_ALLOW_PRIVATE_WITHOUT_WHITELIST",
        "OPENCLAW_DAILY_HERMES_TEMPLATE",
        "OPENCLAW_OLLAMA_MODEL",
        "OPENCLAW_PPT_CACHE_TTL_HOURS",
        "OPENCLAW_IMAGE_GEMINI_MODEL",
        "OPENCLAW_IMAGE_OLLAMA_TIMEOUT",
        "OPENCLAW_IMAGE_VISION_MODEL",
        "OPENCLAW_QA_OLLAMA_FIRST",
        "OPENCLAW_QA_OLLAMA_MODEL",
        "OPENCLAW_QA_OLLAMA_TIMEOUT",
        "PPT_VISION_ENABLED",
        "PPT_VISION_MODEL",
        "PPT_VISION_TIMEOUT",
        "PROMOTION_PENDING_BATCH_SIZE",
        "RAG_ENABLED",
        "RAG_DEFAULT_THRESHOLD",
        "RAG_DEFAULT_TOP_K",
        "RAG_EMBED_DIM",
        "RAG_EMBED_MODEL",
        "RAG_EMBED_NORMALIZE",
        "TELEGRAM_ADMIN_CHAT_ID",
        "TELEGRAM_BOT_USERNAME",
        "TELEGRAM_CHAT_ID",
        "WEB_CONCURRENCY",
    }

    assert expected_keys <= _env_example_keys()


def test_env_example_documents_runtime_os_env_keys():
    internal_runtime_keys = {
        "MOMO_ALLOW_INSECURE_CONFIG_FOR_TESTS",
        "PYTEST_CURRENT_TEST",
    }

    assert _runtime_env_keys_from_code() - internal_runtime_keys <= _env_example_keys()


def test_env_example_documents_docker_compose_variables():
    assert _compose_env_keys() <= _env_example_keys()


def test_scheduler_does_not_silently_swallow_exceptions():
    scheduler_source = (ROOT / "scheduler.py").read_text(encoding="utf-8")

    assert "except:" not in scheduler_source
    assert not re.search(r"except(?: Exception)?[^\n]*:\n\s+pass(?:\s|#|$)", scheduler_source)


def test_v2_flagged_silent_exception_sites_stay_logged():
    openclaw_source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8")
    notification_source = (ROOT / "services" / "notification_manager.py").read_text(encoding="utf-8")

    assert "trend chart temp cleanup failed" in openclaw_source
    assert "trend agg chart temp cleanup failed" in openclaw_source
    assert "[Notification] 讀取 public_url 設定失敗" in notification_source


def test_v2_flagged_imports_are_removed_or_active():
    app_source = (ROOT / "app.py").read_text(encoding="utf-8")
    scheduler_source = (ROOT / "scheduler.py").read_text(encoding="utf-8")

    for module_name in ["math", "hashlib", "zipfile", "io", "traceback"]:
        assert not re.search(rf"^\s*import\s+{module_name}\b", app_source, re.MULTILINE)
    assert "EXCEL_EXPORT_DIR" not in app_source
    assert "DATABASE_TYPE" not in app_source

    assert re.search(r"^\s*import\s+schedule\b", app_source, re.MULTILINE)
    assert "schedule.run_pending()" in app_source
    assert "schedule.every(" in app_source
    assert not re.search(r"^\s*import\s+schedule\b", scheduler_source, re.MULTILINE)


def test_tracked_backup_artifacts_stay_removed():
    forbidden_artifacts = [
        "app.py.backup_login_required",
        "vendor_stockout_list.html.backup",
        "database/edm_dashboard.html",
        "templates/list.html",
        "tests/main_test.py",
    ]

    assert [path for path in forbidden_artifacts if (ROOT / path).exists()] == []


def test_active_code_no_longer_references_legacy_5888_port():
    active_paths = [
        ROOT / "app.py",
        ROOT / "tests",
        ROOT / "AUTO_IMPORT_README.md",
        ROOT / "GOOGLE_DRIVE_SETUP.md",
        ROOT / "start_momo.command",
        ROOT / "scripts" / "archive" / "check_email_status.py",
    ]

    offenders = []
    for active_path in active_paths:
        paths = active_path.rglob("*") if active_path.is_dir() else [active_path]
        for path in paths:
            if (
                not path.is_file()
                or path == Path(__file__).resolve()
                or "__pycache__" in path.parts
                or path.suffix == ".pyc"
            ):
                continue
            content = path.read_text(encoding="utf-8", errors="ignore")
            if "5888" in content:
                offenders.append(str(path.relative_to(ROOT)))

    assert offenders == []


def test_executable_scripts_do_not_use_remove_orphans():
    script_paths = [
        ROOT / "scripts",
        ROOT / "docker",
        ROOT / ".gitea" / "workflows",
    ]

    offenders = []
    for script_root in script_paths:
        if not script_root.exists():
            continue
        for path in script_root.rglob("*"):
            if not path.is_file():
                continue
            content = path.read_text(encoding="utf-8", errors="ignore")
            if "--remove-orphans" in content or "docker compose down" in content:
                offenders.append(str(path.relative_to(ROOT)))

    assert offenders == []


def test_gitea_cd_uses_ewoooc_dedicated_runner_label():
    workflow = (ROOT / ".gitea" / "workflows" / "cd.yaml").read_text(encoding="utf-8")

    assert "runs-on: ewoooc-host" in workflow
    assert "runs-on: ubuntu-latest" not in workflow
    assert "runs-on: ubuntu-22.04" not in workflow
    assert "runs-on: ubuntu-24.04" not in workflow


def test_devops_handbook_uses_current_docker_runtime_commands():
    handbook = (ROOT / "docs" / "guides" / "devops_handbook.md").read_text(encoding="utf-8")

    assert "kubectl" not in handbook
    assert "docker restart momo-pro-system" not in handbook
    assert "docker compose build --no-cache momo-pro-system" not in handbook
    assert "docker kill -s HUP momo-pro-system" in handbook
    assert "docker compose build --no-cache momo-app" in handbook
    assert "docker compose up -d --no-deps --force-recreate momo-app" in handbook
    assert "--remove-orphans" in handbook
    assert "禁止" in handbook


def test_deployment_docs_cover_jump_host_known_hosts_repair():
    sop = (ROOT / "docs" / "guides" / "deployment_sop.md").read_text(encoding="utf-8")
    handbook = (ROOT / "docs" / "guides" / "devops_handbook.md").read_text(encoding="utf-8")

    for content in (sop, handbook):
        assert "Host key verification failed" in content
        assert "ssh-keygen -R 192.168.0.188" in content
        assert "ssh-keyscan -H 192.168.0.188" in content
        assert "StrictHostKeyChecking=no" in content
        assert "不要" in content
    assert "momo_scp_smoke.txt" in sop