# -*- coding: utf-8 -*-
"""开发规范 hard policy 守护（索引：development/POLICY_MATRIX.md）。"""
from __future__ import annotations

import os
import re
import unittest

from _support import get_skill_root

SKIP_DIR_NAMES = frozenset(
    {
        ".git",
        "__pycache__",
        ".pytest_cache",
        "rpa-artifacts",
        "videos",
    }
)

SKIP_REL_PREFIXES = ("tests/artifacts", "tests/diagnostics")

POLICY_IDS = (
    "POLICY-STRUCTURE-001",
    "POLICY-RUNTIME-001",
    "POLICY-RUNTIME-002",
    "POLICY-INSTALL-001",
    "POLICY-CONFIG-001",
    "POLICY-CONFIG-002",
    "POLICY-CONFIG-003",
    "POLICY-TESTING-001",
    "POLICY-TESTING-002",
    "POLICY-RPA-001",
    "POLICY-RPA-002",
    "POLICY-PACKAGING-001",
    "POLICY-PACKAGING-002",
    "POLICY-DOCS-001",
)

STRUCTURE_PATHS = (
    "scripts/main.py",
    "scripts/cli",
    "scripts/service",
    "scripts/db",
    "scripts/util",
)

CONFIG_ENV_BOUNDARY_FILES = frozenset(
    {
        "scripts/main.py",
        "scripts/util/config_bootstrap.py",
        "scripts/util/runtime_paths.py",
        "scripts/util/logging_config.py",
        "scripts/util/constants.py",
    }
)

INSTALL_FORBIDDEN_PATTERNS = (
    re.compile(r"\bpip\s+install\b", re.IGNORECASE),
    re.compile(r"\bpython\s+-m\s+pip\s+install\b", re.IGNORECASE),
    re.compile(r"\buv\s+pip\s+install\b", re.IGNORECASE),
    re.compile(r"\bplaywright\s+install\b", re.IGNORECASE),
)

CONFIG_ENV_FORBIDDEN_PATTERNS = (
    re.compile(r"\bos\.environ\b"),
    re.compile(r"\bos\.getenv\b"),
    re.compile(r"\bos\.putenv\b"),
    re.compile(r"\benviron\.get\b"),
)

TESTING_REAL_TARGET_ALLOWED_ROOT_TESTS = frozenset({"test_adapter_profile_policy.py"})

TESTING_001_FORBIDDEN_SUBSTRINGS = (
    "real_" + "api",
    "real_" + "rpa",
    "OPENCLAW_" + "TEST_TARGET",
)

TESTING_001_SKIP_ROOT_TESTS = frozenset({"test_development_policy_guard.py"})

RPA_001_FORBIDDEN_PATTERNS = (
    re.compile(r"\brpa_helpers\b"),
    re.compile(r"\binject_account_manager_scripts_path\b"),
    re.compile(r"\bget_account_credential\b"),
)

RPA_002_FFMPEG_PATTERNS = (
    re.compile(
        r"\bsubprocess\.(run|call|Popen|check_output|check_call)\s*\([^)]*ffmpeg",
        re.IGNORECASE | re.DOTALL,
    ),
    re.compile(
        r"\bsubprocess\.(run|call|Popen|check_output|check_call)\(\s*['\"][^'\"]*ffmpeg",
        re.IGNORECASE,
    ),
    re.compile(r"\bos\.system\s*\([^)]*ffmpeg", re.IGNORECASE | re.DOTALL),
    re.compile(r"\bos\.popen\s*\([^)]*ffmpeg", re.IGNORECASE | re.DOTALL),
    re.compile(
        r"\bsubprocess\.(run|call|Popen|check_output|check_call)\([^)]*shell\s*=\s*True[^)]*ffmpeg",
        re.IGNORECASE | re.DOTALL,
    ),
)

RPA_002_EXEMPT_REL = "scripts/service/task_run_support.py"


def _policy_msg(policy_id: str, source: str, detail: str) -> str:
    return f"{policy_id} [{source}]: {detail}"


def _rel(skill_root: str, path: str) -> str:
    return os.path.relpath(path, skill_root).replace("\\", "/")


def _should_skip_dirname(dirname: str) -> bool:
    return dirname in SKIP_DIR_NAMES


def _should_skip_rel_path(skill_root: str, abs_path: str) -> bool:
    rel = _rel(skill_root, abs_path).replace("\\", "/")
    for prefix in SKIP_REL_PREFIXES:
        normalized = prefix.rstrip("/")
        if rel == normalized or rel.startswith(normalized + "/"):
            return True
    return False


def _walk_files(skill_root: str, rel_root: str, *, suffix: str | None = None) -> list[str]:
    base = os.path.join(skill_root, rel_root)
    if not os.path.exists(base):
        return []
    found: list[str] = []
    for root, dirnames, filenames in os.walk(base):
        kept: list[str] = []
        for d in dirnames:
            if _should_skip_dirname(d):
                continue
            sub = os.path.join(root, d)
            if _should_skip_rel_path(skill_root, sub):
                continue
            kept.append(d)
        dirnames[:] = kept
        for name in filenames:
            full = os.path.join(root, name)
            if _should_skip_rel_path(skill_root, full):
                continue
            if suffix and not name.endswith(suffix):
                continue
            found.append(full)
    return sorted(found)


def _read_text(path: str) -> str:
    with open(path, encoding="utf-8") as f:
        return f.read()


def _requirement_dependency_lines(req_text: str) -> list[str]:
    return [
        line.strip()
        for line in req_text.splitlines()
        if line.strip() and not line.strip().startswith("#")
    ]


def _scan_install_delivery_files(skill_root: str) -> list[str]:
    offenders: list[str] = []
    scan_specs: list[tuple[str, str | None]] = [
        ("scripts", ".py"),
        (".", ".ps1"),
        (".", ".sh"),
        (".", None),
        (".github/workflows", ".yml"),
        (".github/workflows", ".yaml"),
    ]
    seen: set[str] = set()
    for rel_root, suffix in scan_specs:
        if rel_root == "." and suffix is None:
            req = os.path.join(skill_root, "requirements.txt")
            if os.path.isfile(req):
                seen.add(req)
                text = _read_text(req)
                for pattern in INSTALL_FORBIDDEN_PATTERNS:
                    if pattern.search(text):
                        offenders.append(
                            _policy_msg(
                                "POLICY-INSTALL-001",
                                "development/RUNTIME.md; development/RPA.md §1.4; development/DEVELOPMENT.md §3.1",
                                f"{_rel(skill_root, req)} matches {pattern.pattern}",
                            )
                        )
            continue
        for path in _walk_files(skill_root, rel_root, suffix=suffix):
            if path in seen:
                continue
            seen.add(path)
            if suffix == ".ps1" and not path.endswith(".ps1"):
                continue
            if suffix == ".sh" and not path.endswith(".sh"):
                continue
            text = _read_text(path)
            for pattern in INSTALL_FORBIDDEN_PATTERNS:
                if pattern.search(text):
                    offenders.append(
                        _policy_msg(
                            "POLICY-INSTALL-001",
                            "development/RUNTIME.md; development/RPA.md §1.4; development/DEVELOPMENT.md §3.1",
                            f"{_rel(skill_root, path)} matches {pattern.pattern}",
                        )
                    )
    return offenders


def _gitignore_has_pattern(lines: list[str], pattern: str) -> bool:
    normalized = [ln.strip() for ln in lines if ln.strip() and not ln.strip().startswith("#")]
    return pattern in normalized


class TestPolicyStructure001(unittest.TestCase):
    def test_standard_scripts_layout_exists(self) -> None:
        skill_root = get_skill_root()
        missing: list[str] = []
        for rel in STRUCTURE_PATHS:
            path = os.path.join(skill_root, *rel.split("/"))
            if not os.path.exists(path):
                missing.append(rel)
        self.assertEqual(
            missing,
            [],
            msg=_policy_msg(
                "POLICY-STRUCTURE-001",
                "development/DEVELOPMENT.md §2; development/RUNTIME.md §目录结构",
                "missing: " + ", ".join(missing),
            ),
        )


class TestPolicyRuntime001(unittest.TestCase):
    def test_no_vendored_jiangchang_skill_core(self) -> None:
        skill_root = get_skill_root()
        vendored = os.path.join(skill_root, "scripts", "jiangchang_skill_core")
        self.assertFalse(
            os.path.isdir(vendored),
            msg=_policy_msg(
                "POLICY-RUNTIME-001",
                "development/RUNTIME.md §共享 Python Runtime",
                _rel(skill_root, vendored),
            ),
        )


class TestPolicyRuntime002(unittest.TestCase):
    def test_requirements_txt_excludes_shared_runtime_deps(self) -> None:
        skill_root = get_skill_root()
        req_path = os.path.join(skill_root, "requirements.txt")
        self.assertTrue(os.path.isfile(req_path), msg="requirements.txt missing")
        text = _read_text(req_path)
        dep_lines = _requirement_dependency_lines(text)
        offenders: list[str] = []
        for line in dep_lines:
            lower = line.lower()
            if "jiangchang-platform-kit" in lower:
                offenders.append(line)
            if re.search(r"(?<![\w.-])playwright(?![\w.-])", lower):
                offenders.append(line)
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-RUNTIME-002",
                "development/RUNTIME.md; development/DEVELOPMENT.md §3.1",
                "forbidden dependency lines: " + "; ".join(offenders),
            ),
        )


class TestPolicyInstall001(unittest.TestCase):
    def test_delivery_files_have_no_auto_install_commands(self) -> None:
        offenders = _scan_install_delivery_files(get_skill_root())
        self.assertEqual(offenders, [], msg="\n".join(offenders))


class TestPolicyConfig001(unittest.TestCase):
    def test_env_example_exists(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, ".env.example")
        self.assertTrue(
            os.path.isfile(path),
            msg=_policy_msg(
                "POLICY-CONFIG-001",
                "development/CONFIG.md",
                ".env.example missing at repo root",
            ),
        )


class TestPolicyConfig002(unittest.TestCase):
    def test_gitignore_ignores_env_files(self) -> None:
        skill_root = get_skill_root()
        path = os.path.join(skill_root, ".gitignore")
        self.assertTrue(os.path.isfile(path), msg=".gitignore missing")
        lines = _read_text(path).splitlines()
        missing: list[str] = []
        if not _gitignore_has_pattern(lines, ".env"):
            missing.append(".env")
        if not _gitignore_has_pattern(lines, "*.env.local"):
            missing.append("*.env.local")
        self.assertEqual(
            missing,
            [],
            msg=_policy_msg(
                "POLICY-CONFIG-002",
                "development/CONFIG.md §红线：敏感信息不进 .env",
                "missing gitignore entries: " + ", ".join(missing),
            ),
        )


class TestPolicyConfig003(unittest.TestCase):
    def test_business_scripts_do_not_read_os_environ_directly(self) -> None:
        skill_root = get_skill_root()
        offenders: list[str] = []
        for path in _walk_files(skill_root, "scripts", suffix=".py"):
            rel = _rel(skill_root, path)
            if rel.replace("\\", "/") in CONFIG_ENV_BOUNDARY_FILES:
                continue
            text = _read_text(path)
            for lineno, line in enumerate(text.splitlines(), 1):
                if line.strip().startswith("#"):
                    continue
                for pattern in CONFIG_ENV_FORBIDDEN_PATTERNS:
                    if pattern.search(line):
                        offenders.append(
                            f"{rel}:{lineno}: {line.strip()} — use jiangchang_skill_core.config.get*"
                        )
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-CONFIG-003",
                "development/CONFIG.md; development/ADAPTER.md",
                "\n".join(offenders),
            ),
        )


class TestPolicyTesting001(unittest.TestCase):
    def test_root_tests_do_not_hardcode_real_targets(self) -> None:
        skill_root = get_skill_root()
        tests_dir = os.path.join(skill_root, "tests")
        offenders: list[str] = []
        for name in sorted(os.listdir(tests_dir)):
            if not (name.startswith("test_") and name.endswith(".py")):
                continue
            if name in TESTING_001_SKIP_ROOT_TESTS:
                continue
            if name in TESTING_REAL_TARGET_ALLOWED_ROOT_TESTS:
                continue
            path = os.path.join(tests_dir, name)
            text = _read_text(path)
            for lineno, line in enumerate(text.splitlines(), 1):
                stripped = line.strip()
                if stripped.startswith("#"):
                    continue
                for token in TESTING_001_FORBIDDEN_SUBSTRINGS:
                    if token in line:
                        offenders.append(f"{name}:{lineno}: {stripped}")
                        break
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-TESTING-001",
                "development/TESTING.md §默认必跑测试要做什么",
                "\n".join(offenders),
            ),
        )


class TestPolicyTesting002(unittest.TestCase):
    def test_integration_python_tests_are_sample_only(self) -> None:
        skill_root = get_skill_root()
        integration_dir = os.path.join(skill_root, "tests", "integration")
        if not os.path.isdir(integration_dir):
            return
        offenders: list[str] = []
        for name in sorted(os.listdir(integration_dir)):
            if not name.endswith(".py"):
                continue
            if name.endswith(".sample"):
                continue
            offenders.append(f"tests/integration/{name}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-TESTING-002",
                "development/TESTING.md §7 真实联调测试的安全约束",
                "non-sample integration tests: " + ", ".join(offenders),
            ),
        )


class TestPolicyRpa001(unittest.TestCase):
    def test_scripts_have_no_account_manager_internal_imports(self) -> None:
        skill_root = get_skill_root()
        offenders: list[str] = []
        for path in _walk_files(skill_root, "scripts", suffix=".py"):
            rel = _rel(skill_root, path)
            text = _read_text(path)
            for pattern in RPA_001_FORBIDDEN_PATTERNS:
                if pattern.search(text):
                    offenders.append(f"{rel}: {pattern.pattern}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-RPA-001",
                "development/ADAPTER.md §兄弟依赖; development/RPA.md §1.7",
                "\n".join(offenders),
            ),
        )


class TestPolicyRpa002(unittest.TestCase):
    def test_scripts_do_not_invoke_ffmpeg_directly(self) -> None:
        skill_root = get_skill_root()
        offenders: list[str] = []
        for path in _walk_files(skill_root, "scripts", suffix=".py"):
            rel = _rel(skill_root, path)
            if rel.replace("\\", "/") == RPA_002_EXEMPT_REL:
                continue
            text = _read_text(path)
            for pattern in RPA_002_FFMPEG_PATTERNS:
                if pattern.search(text):
                    offenders.append(f"{rel}: {pattern.pattern}")
        self.assertEqual(
            offenders,
            [],
            msg=_policy_msg(
                "POLICY-RPA-002",
                "development/RPA.md §5.3 录屏成片标准",
                "\n".join(offenders),
            ),
        )


class TestPolicyDocs001(unittest.TestCase):
    def test_policy_matrix_exists_and_lists_policy_ids(self) -> None:
        skill_root = get_skill_root()
        matrix_path = os.path.join(skill_root, "development", "POLICY_MATRIX.md")
        self.assertTrue(
            os.path.isfile(matrix_path),
            msg=_policy_msg(
                "POLICY-DOCS-001",
                "development/POLICY_MATRIX.md",
                "file missing",
            ),
        )
        text = _read_text(matrix_path)
        missing = [pid for pid in POLICY_IDS if pid not in text]
        self.assertEqual(
            missing,
            [],
            msg=_policy_msg(
                "POLICY-DOCS-001",
                "development/POLICY_MATRIX.md",
                "missing policy_id entries: " + ", ".join(missing),
            ),
        )


if __name__ == "__main__":
    unittest.main()