#!/usr/bin/env python3
"""Shared protected-data and leak-detection rules for online-skill helpers."""

from __future__ import annotations

import re
from pathlib import Path, PurePosixPath
from typing import Any

PUBLIC_PADDING_PATH = PurePosixPath("assets/online-padding/padding.bin")
PUBLIC_PADDING_SIZE_BYTES = 2_000_000
ONLINE_POLICY_MARKER = "<!-- online-auth-upgrade -->"
ONLINE_POLICY_START = "<!-- online-runtime-policy:start -->"
ONLINE_POLICY_END = "<!-- online-runtime-policy:end -->"
ONLINE_RUNTIME_POLICY_VERSION = 2
FULL_TEXT_MODES = {
    "evidence_required",
    "evidence_optional",
    "capability_probe_only",
    "unavailable",
}
DEFAULT_FULL_TEXT_MODE = "evidence_optional"
FULL_TEXT_EVIDENCE_MODES = {
    "evidence_required",
    "evidence_optional",
}
ONLINE_RUNTIME_POLICY_REQUIRED_CHECKS = {
    "cloud_runtime_policy",
    "creative_retrieval_gate",
}
ONLINE_RUNTIME_POLICY_EXPECTED = {
    "policy_version": ONLINE_RUNTIME_POLICY_VERSION,
    "protected_data_mode": "cloud_only",
    "local_protected_fallback_allowed": False,
    "creative_search_minimum": 1,
    "project_search_minimum": 2,
    "retrieval_receipt_required": True,
    "creative_evidence_source": "protected_search",
    "full_text_mode": DEFAULT_FULL_TEXT_MODE,
    "full_text_probe_satisfies_creative_search": False,
}
ONLINE_POLICY_REQUIRED_FRAGMENTS = (
    "## Mandatory Creative Retrieval Gate",
    "auth_client.py search",
    "at least two complementary searches",
    "retrieval intake",
    "云端机制库：检索 N 次",
    "Online runtime precedence:",
    "auth_client.py read-ref",
    "full-text-search",
    "full_text_mode",
    "capability_probe_only",
    "Protected data is not present in the public package.",
)


PROTECTED_DIR_NAMES = {
    "cloud-payload",
    "library",
    "libraries",
    "full-text",
    "full_text",
    "full-scripts",
    "full_scripts",
    "corpus",
    "raw",
    "search",
    "index",
    "indexes",
    "indices",
    "embedding",
    "embeddings",
    "vector",
    "vectors",
    "faiss",
    "ann",
    "database",
    "databases",
    "private",
    "protected",
    "source-map",
    "source_maps",
}

PROTECTED_SUFFIXES = {
    ".ann",
    ".arrow",
    ".bin",
    ".db",
    ".duckdb",
    ".enc",
    ".faiss",
    ".feather",
    ".idx",
    ".index",
    ".joblib",
    ".jsonl",
    ".ndjson",
    ".npy",
    ".npz",
    ".parquet",
    ".pickle",
    ".pkl",
    ".sqlite",
    ".sqlite3",
    ".sqlite-shm",
    ".sqlite-wal",
}

PROTECTED_FILENAMES = {
    "admin_source_map.json",
    "anonymous_source_map.json",
    "case_mechanisms.jsonl",
    "catalog.json",
    "catalog.jsonl",
    "cloud-payload-version.json",
    "manifest.csv",
    "manifest.json",
    "manifest.jsonl",
    "patterns.jsonl",
    "protected_references.enc.json",
    "script_manifest.jsonl",
    "source_manifest.jsonl",
    "source_map.json",
    "sources.json",
    "sources.jsonl",
    "title_inventory.json",
    "title_inventory.jsonl",
}

IDENTITY_KEY_NAMES = {
    "source_path",
    "rel_path",
    "sample_paths",
    "cache_file",
    "file_name",
    "filename",
    "path",
    "local_path",
    "source_url",
    "source_page",
    "referer",
    "manifest",
    "catalog",
    "raw_manifest",
}

TITLE_KEY_NAMES = {
    "title",
    "script_name",
    "book_name",
    "article_title",
    "work_title",
    "source_title",
    "original_title",
    "source",
    "source_name",
    "original_source",
    "source_id",
    "work",
    "book",
    "article",
    "script",
}

COMMON_POSIX_ROOT_RE = (
    r"/(?:Users|Volumes|private|tmp|opt|var|home|mnt|srv|root|Library|Applications)"
    r"(?:/|\b)"
)
HOME_RELATIVE_PATH_RE = r"(?:^|(?<=[\s`\"'(]))~/[^\s`\"'<>)]+"
WINDOWS_DRIVE_PATH_RE = r"\b[A-Za-z]:[\\/][^\s`\"'<>)]+"
WINDOWS_UNC_PATH_RE = r"(?<!:)(?:\\\\|//)[^\\/\s`\"']+[\\/][^\\/\s`\"']+"
LOCAL_MACHINE_PATH_RE = re.compile(
    rf"(?:{COMMON_POSIX_ROOT_RE}|{HOME_RELATIVE_PATH_RE}|"
    rf"{WINDOWS_DRIVE_PATH_RE}|{WINDOWS_UNC_PATH_RE})",
    re.IGNORECASE,
)
PATH_OR_URL_RE = re.compile(
    rf"(?:{COMMON_POSIX_ROOT_RE}|{HOME_RELATIVE_PATH_RE}|"
    rf"{WINDOWS_DRIVE_PATH_RE}|{WINDOWS_UNC_PATH_RE}|"
    r"https?://|s3://|cos://|\.jsonl\b|"
    r"\.sqlite(?:3|-wal|-shm)?\b|\.db\b|\.parquet\b|\.faiss\b|\.idx\b)",
    re.IGNORECASE,
)
ANON_ID_RE = re.compile(r"^(case|source|mechanism|scene|ref)_[a-z]{0,4}_?\d+$", re.IGNORECASE)
ONLINE_LEGACY_INSTRUCTION_PATTERNS = (
    (
        "E_LOCAL_FIRST",
        re.compile(r"\b(?:LOCAL[-_ ]?FIRST|OFFLINE[-_ ]?FIRST)\b", re.IGNORECASE),
        "public instructions still declare a local-first runtime",
    ),
    (
        "E_LOCAL_FIRST",
        re.compile(
            r"(?:本地(?:资料|数据库|数据|知识|机制|案例|素材)?库?\s*优先|"
            r"优先(?:使用|读取|查询|检索|搜索)?\s*本地(?:资料|数据库|数据|知识|机制|案例|素材)?库?)"
        ),
        "public instructions still declare a local-first runtime",
    ),
    (
        "E_LOCAL_SEARCH_COMMAND",
        re.compile(
            r"(?:(?:python3?|uv\s+run\s+python|bash)\s+)?(?:\./)?scripts/"
            r"(?!auth_client\.py|online_version_guard\.py|woo_online_client\.py)"
            r"[^\s`\"']*(?:search|query|retrieve|inspect|lookup|fts|full[_-]?text)"
            r"[^\s`\"']*",
            re.IGNORECASE,
        ),
        "public instructions still invoke a local retrieval script",
    ),
    (
        "E_LOCAL_FALLBACK",
        re.compile(
            r"(?:if|when)[^\n.。]{0,100}(?:cloud|remote|api)"
            r"[^\n.。]{0,60}(?:fail(?:s|ed)?|unavailable|offline|timeout|error)"
            r"[^\n.。]{0,100}(?:use|read|query|search|continue\s+with|fall\s+back\s+to)"
            r"[^\n.。]{0,50}(?:local|bundled|on-device)"
            r"[^\n.。]{0,30}(?:database|library|data|references?|corpus)",
            re.IGNORECASE,
        ),
        "public instructions permit protected local-data fallback after a cloud failure",
    ),
    (
        "E_LOCAL_FALLBACK",
        re.compile(
            r"(?:如果|若|当)[^\n。]{0,100}(?:云端|远程|API)"
            r"[^\n。]{0,60}(?:失败|不可用|离线|中断|超时|报错)"
            r"[^\n。]{0,100}(?:使用|改用|读取|查询|检索|搜索|继续用|回退到)"
            r"[^\n。]{0,30}本地(?:数据库|资料库|知识库|机制库|案例库|数据|资料|语料)",
            re.IGNORECASE,
        ),
        "public instructions permit protected local-data fallback after a cloud failure",
    ),
    (
        "E_DIRECT_LOCAL_READ",
        re.compile(
            r"(?:read|open|load|search|query|inspect|读取|打开|加载|检索|搜索|查询)"
            r"[^\n。]{0,100}(?:references/|assets/(?:library|libraries|search|index|indexes|"
            r"corpus|database|databases|full[-_]?text)/|"
            r"\.(?:sqlite3?|db|jsonl|ndjson|parquet|faiss)\b)",
            re.IGNORECASE,
        ),
        "public instructions still require a direct protected local-data read",
    ),
)
ABSOLUTE_LOCAL_PATH_RE = LOCAL_MACHINE_PATH_RE


def posix_rel(path: str | Path, *, strip_root: str | None = None) -> PurePosixPath:
    text = str(path).replace("\\", "/").strip("/")
    rel = PurePosixPath(text)
    if strip_root and rel.parts and rel.parts[0] == strip_root:
        rel = PurePosixPath(*rel.parts[1:]) if len(rel.parts) > 1 else PurePosixPath("")
    return rel


def unsafe_member_path(name: str) -> bool:
    if "\\" in name or re.match(r"^[A-Za-z]:", name):
        return True
    path = PurePosixPath(name)
    if path.is_absolute():
        return True
    return any(part in ("", ".", "..") for part in path.parts)


def is_public_padding_path(path: str | Path, *, strip_root: str | None = None) -> bool:
    return posix_rel(path, strip_root=strip_root) == PUBLIC_PADDING_PATH


def is_protected_path(
    path: str | Path,
    *,
    strip_root: str | None = None,
    allow_padding: bool = True,
    protect_references_md: bool = True,
) -> bool:
    rel = posix_rel(path, strip_root=strip_root)
    if not rel.parts:
        return False
    if "online-padding" in {part.lower() for part in rel.parts}:
        return not (allow_padding and is_public_padding_path(rel))
    parts = {part.lower() for part in rel.parts}
    if parts & PROTECTED_DIR_NAMES:
        return True
    name = rel.name.lower()
    if name in PROTECTED_FILENAMES:
        return True
    suffix = "".join(PurePosixPath(name).suffixes[-2:])
    if suffix in PROTECTED_SUFFIXES or PurePosixPath(name).suffix.lower() in PROTECTED_SUFFIXES:
        return True
    if protect_references_md and len(rel.parts) >= 2 and "references" in parts:
        return True
    return False


def looks_like_source_identity(value: Any) -> bool:
    if not isinstance(value, str):
        return bool(value)
    text = value.strip()
    if not text:
        return False
    if PATH_OR_URL_RE.search(text):
        return True
    if ANON_ID_RE.match(text):
        return False
    return True


def find_protected_identity_leaks(value: Any, *, path: str = "$") -> list[str]:
    leaks: list[str] = []
    if isinstance(value, dict):
        for key, child in value.items():
            key_text = str(key)
            key_lower = key_text.lower()
            child_path = f"{path}.{key_text}"
            if key_lower in IDENTITY_KEY_NAMES and child:
                leaks.append(child_path)
                continue
            if key_lower in TITLE_KEY_NAMES and looks_like_source_identity(child):
                leaks.append(child_path)
                continue
            leaks.extend(find_protected_identity_leaks(child, path=child_path))
    elif isinstance(value, list):
        for index, child in enumerate(value):
            leaks.extend(find_protected_identity_leaks(child, path=f"{path}[{index}]"))
    elif isinstance(value, str) and PATH_OR_URL_RE.search(value):
        leaks.append(path)
    return sorted(set(leaks))


def full_text_mode_contract(mode: Any) -> dict[str, Any]:
    normalized = str(mode or "").strip().lower()
    if normalized not in FULL_TEXT_MODES:
        raise ValueError(
            "full_text_mode must be one of: " + ", ".join(sorted(FULL_TEXT_MODES))
        )
    return {
        "mode": normalized,
        "request_required": normalized in {"evidence_required", "capability_probe_only"},
        "nonempty_evidence_required": normalized == "evidence_required",
        "creative_evidence_eligible": normalized in FULL_TEXT_EVIDENCE_MODES,
        "creative_search_substitute": False,
    }


def runtime_policy_for(full_text_mode: str = DEFAULT_FULL_TEXT_MODE) -> dict[str, Any]:
    policy = dict(ONLINE_RUNTIME_POLICY_EXPECTED)
    policy["full_text_mode"] = full_text_mode_contract(full_text_mode)["mode"]
    return policy


def runtime_policy_errors(manifest: Any) -> list[str]:
    if not isinstance(manifest, dict):
        return ["online-version.json must contain a JSON object"]
    policy = manifest.get("runtime_policy")
    if not isinstance(policy, dict):
        return ["online-version.json runtime_policy is missing"]
    errors: list[str] = []
    for key, expected in ONLINE_RUNTIME_POLICY_EXPECTED.items():
        if key == "full_text_mode":
            continue
        actual = policy.get(key)
        if actual != expected:
            errors.append(f"runtime_policy.{key} must be {expected!r}, got {actual!r}")
    try:
        contract = full_text_mode_contract(policy.get("full_text_mode"))
    except ValueError as exc:
        errors.append(f"runtime_policy.{exc}")
    else:
        if (
            contract["mode"] in {"capability_probe_only", "unavailable"}
            and policy.get("full_text_probe_satisfies_creative_search") is not False
        ):
            errors.append(
                "runtime_policy.full_text_probe_satisfies_creative_search must be False "
                f"when full_text_mode is {contract['mode']!r}"
            )
    update_policy = manifest.get("public_update_policy")
    if not isinstance(update_policy, dict):
        errors.append("public_update_policy is missing")
        return errors
    checks = update_policy.get("required_package_checks")
    if not isinstance(checks, list):
        errors.append("public_update_policy.required_package_checks must be a list")
        return errors
    missing = sorted(ONLINE_RUNTIME_POLICY_REQUIRED_CHECKS - {str(item) for item in checks})
    if missing:
        errors.append(f"required_package_checks missing: {', '.join(missing)}")
    return errors


def online_skill_policy_errors(text: str) -> list[str]:
    errors: list[str] = []
    if text.count(ONLINE_POLICY_MARKER) != 1:
        errors.append("SKILL.md must contain exactly one online authorization marker")
    if text.count(ONLINE_POLICY_START) != 1 or text.count(ONLINE_POLICY_END) != 1:
        errors.append("SKILL.md must contain exactly one online runtime policy block")
        return errors
    start = text.find(ONLINE_POLICY_START)
    end = text.find(ONLINE_POLICY_END)
    if start >= end:
        errors.append("SKILL.md online runtime policy markers are out of order")
        return errors
    policy = text[start : end + len(ONLINE_POLICY_END)]
    for fragment in ONLINE_POLICY_REQUIRED_FRAGMENTS:
        if fragment not in policy:
            errors.append(f"SKILL.md online runtime policy missing: {fragment}")
    return errors


def online_instruction_conflicts(
    text: str,
    *,
    exclude_policy: bool = True,
) -> list[dict[str, Any]]:
    # Keep the keyword argument for caller compatibility, but scan the complete
    # document. The generated policy block is deliberately worded so it does
    # not match forbidden local-runtime patterns; excluding the block would let
    # an appended LOCAL-FIRST rule hide inside trusted markers.
    _ = exclude_policy
    spans = [(0, len(text))]
    conflicts: list[dict[str, Any]] = []
    for span_start, span_end in spans:
        segment = text[span_start:span_end]
        for code, pattern, message in ONLINE_LEGACY_INSTRUCTION_PATTERNS:
            for match in pattern.finditer(segment):
                offset = span_start + match.start()
                conflicts.append(
                    {
                        "code": code,
                        "line": text.count("\n", 0, offset) + 1,
                        "message": message,
                        "evidence": " ".join(match.group(0).strip().split())[:180],
                    }
                )
        for match in ABSOLUTE_LOCAL_PATH_RE.finditer(segment):
            offset = span_start + match.start()
            conflicts.append(
                {
                    "code": "E_ABSOLUTE_LOCAL_PATH",
                    "line": text.count("\n", 0, offset) + 1,
                    "message": "public instructions expose a machine-local absolute path",
                    "evidence": " ".join(match.group(0).strip().split())[:180],
                }
            )
    return conflicts
