from __future__ import annotations

import json
import subprocess
import sys
import tempfile
import unittest
from copy import deepcopy
from pathlib import Path


SKILL_ROOT = Path(__file__).resolve().parents[1]
SCRIPTS_DIR = SKILL_ROOT / "scripts"
sys.path.insert(0, str(SCRIPTS_DIR))

import common  # noqa: E402


class PrivacyProtectionTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.config = common.load_config()

    def test_sanitize_user_text_redacts_common_sensitive_values(self) -> None:
        text = (
            "Authorization: Bearer sk-1234567890abcdefghijklmn "
            "email alice@example.com "
            "phone +1 415-555-1234 "
            "card 4111 1111 1111 1111 "
            "id 123-45-6789 "
            "path /Users/alice/private/project"
        )

        sanitized = common.sanitize_user_text(text, self.config)

        for forbidden in (
            "sk-1234567890abcdefghijklmn",
            "alice@example.com",
            "+1 415-555-1234",
            "4111 1111 1111 1111",
            "123-45-6789",
            "/Users/alice/private/project",
        ):
            self.assertNotIn(forbidden, sanitized)

        for marker in (
            "[redacted:secret]",
            "[redacted:email]",
            "[redacted:phone]",
            "[redacted:credit-card]",
            "[redacted:identity]",
            "[redacted:private-path]",
        ):
            self.assertIn(marker, sanitized)

    def test_sanitized_public_context_redacts_legacy_private_fields(self) -> None:
        context = {
            "bug_hash": "hash",
            "skill": {
                "name": "bensz-collect-bugs",
                "author": "Bensz Conan",
                "source_path": "/Users/alice/dev/skill",
                "source_repo": "https://github.com/example/repo",
            },
            "reporter": {
                "display_name": "Alice",
                "github_username": "octocat",
                "local_username": "alice",
            },
            "bug": {
                "summary": "password=super-secret",
                "severity": "important",
                "expected_behavior": "do x",
                "actual_behavior": "call me at +1 415-555-1234",
                "reproduction_steps": ["open /Users/alice/dev/skill"],
                "evidence": ["email alice@example.com"],
                "workaround": None,
                "impact": "important",
                "additional_notes": None,
                "is_skill_design_defect": True,
            },
            "environment": {
                "device": {"type": "laptop", "hostname": "alice-mbp"},
                "runtime": {
                    "agent_runtime": "codex-cli",
                    "shell": "/bin/zsh",
                    "cwd": "/Users/alice/dev/skill",
                    "hostname": "alice-mbp",
                    "local_username": "alice",
                },
                "software_versions": {"custom": "token=ghp_1234567890abcdefghijklmnop"},
                "os": {"family": "Darwin", "release": "24.0.0", "machine": "arm64"},
            },
            "tracking": {
                "collected_at": "2026-03-27T00:00:00Z",
                "first_seen_at": "2026-03-27T00:00:00Z",
                "last_seen_at": "2026-03-27T00:00:00Z",
                "occurrence_count": 1,
                "public_reported": False,
                "public_repo": None,
                "public_path": None,
                "reported_at": None,
                "local_path": "/Users/alice/.bensz-skills/bugs/x",
            },
            "deduplication": {
                "fingerprint_payload": {
                    "skill": {"name": "bensz-collect-bugs", "author": "Bensz Conan"},
                    "bug": {
                        "summary": "password=super-secret",
                        "expected_behavior": "do x",
                        "actual_behavior": "phone +1 415-555-1234",
                    },
                    "environment": {
                        "agent_runtime": "codex-cli",
                        "software_versions": {"custom": "alice@example.com"},
                    },
                }
            },
        }

        public_context = common.sanitized_public_context(context, self.config, placeholder="redacted")
        rendered = json.dumps(public_context, ensure_ascii=False)

        for forbidden in (
            "alice@example.com",
            "+1 415-555-1234",
            "super-secret",
            "/Users/alice",
            "alice-mbp",
        ):
            self.assertNotIn(forbidden, rendered)

        self.assertEqual(public_context["reporter"]["display_name"], "octocat")
        self.assertEqual(public_context["reporter"]["local_username"], "redacted")
        self.assertEqual(public_context["skill"]["source_path"], "redacted")

    def test_deduplication_payload_respects_stable_fields_config(self) -> None:
        config = deepcopy(self.config)
        config["hashing"]["stable_fields"] = [
            "skill.name",
            "environment.runtime.agent_runtime",
        ]
        payload = common.deduplication_payload(
            config=config,
            skill_name="demo-skill",
            skill_author="Bensz Conan",
            summary="summary should be ignored",
            expected_behavior="expected should be ignored",
            actual_behavior="actual should be ignored",
            environment={
                "os": {"family": "Darwin"},
                "runtime": {"agent_runtime": "codex-cli"},
                "software_versions": {"python3": "Python 3.12.0"},
            },
        )

        self.assertEqual(
            payload,
            {
                "skill": {"name": "demo-skill"},
                "environment": {"runtime": {"agent_runtime": "codex-cli"}},
            },
        )

    def test_build_bug_directory_respects_path_pattern(self) -> None:
        config = deepcopy(self.config)
        config["storage"]["path_pattern"] = "{reporter}/issues/{skill_name}/{bug_hash}"
        root = Path("/tmp/bugs")

        bug_dir = common.build_bug_directory(
            config,
            root,
            "Demo Skill",
            "octocat",
            "abc123",
        )

        self.assertEqual(bug_dir, Path("/tmp/bugs/octocat/issues/demo-skill/abc123"))

    def test_bug_report_markdown_uses_configured_privacy_notice(self) -> None:
        config = deepcopy(self.config)
        config["reporting"]["privacy_notice"] = "自定义隐私提示"
        template = "Privacy: {privacy_notice}\nSkill: {skill_name}\n"
        context = {
            "bug_hash": "hash",
            "skill": {"name": "demo-skill", "author": "Bensz Conan", "source_path": None, "source_repo": None},
            "reporter": {"github_username": "octocat"},
            "bug": {
                "severity": "important",
                "summary": "x",
                "expected_behavior": "y",
                "actual_behavior": "z",
                "reproduction_steps": [],
                "evidence": [],
                "impact": "i",
                "workaround": None,
                "additional_notes": None,
            },
            "environment": {"device": {}, "runtime": {}, "software_versions": {}, "os": {}},
            "tracking": {"occurrence_count": 1, "collected_at": "2026-03-27T00:00:00Z"},
        }

        rendered = common.bug_report_markdown(template, context, config)

        self.assertIn("自定义隐私提示", rendered)

    def test_collect_bug_script_writes_sanitized_local_records(self) -> None:
        with tempfile.TemporaryDirectory() as tmpdir:
            command = [
                sys.executable,
                str(SCRIPTS_DIR / "collect_bug.py"),
                "--skill-name",
                "demo-skill",
                "--skill-author",
                "Bensz Conan",
                "--bug-summary",
                "Authorization: Bearer sk-1234567890abcdefghijklmn",
                "--expected-behavior",
                "Should not leak alice@example.com",
                "--actual-behavior",
                "Logs show +1 415-555-1234",
                "--reproduction-step",
                "Open /Users/alice/private/project",
                "--evidence",
                "Card 4111 1111 1111 1111 exposed",
                "--skill-source-path",
                "/Users/alice/private/project",
                "--reporter-github",
                "octocat",
                "--bug-root",
                tmpdir,
                "--print-json",
            ]
            result = subprocess.run(command, check=True, capture_output=True, text=True)
            payload = json.loads(result.stdout)

            context_path = Path(payload["context_path"])
            report_path = Path(payload["report_path"])
            context = json.loads(context_path.read_text(encoding="utf-8"))
            report = report_path.read_text(encoding="utf-8")
            rendered_context = json.dumps(context, ensure_ascii=False)

            for forbidden in (
                "sk-1234567890abcdefghijklmn",
                "alice@example.com",
                "+1 415-555-1234",
                "4111 1111 1111 1111",
                "/Users/alice/private/project",
            ):
                self.assertNotIn(forbidden, rendered_context)
                self.assertNotIn(forbidden, report)

            self.assertIsNone(context["reporter"]["local_username"])
            self.assertIsNone(context["environment"]["runtime"]["cwd"])
            self.assertIsNone(context["environment"]["runtime"]["hostname"])
            self.assertIsNone(context["tracking"]["local_path"])
            self.assertIn("[redacted:secret]", report)
            self.assertIn("[redacted:private-path]", report)


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

from __future__ import annotations

import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path


SKILL_ROOT = Path(__file__).resolve().parents[1]
SCRIPT = SKILL_ROOT / "scripts" / "resolve_bug.py"


class ResolutionProtocolTests(unittest.TestCase):
    def make_bug_dir(self, root: Path) -> Path:
        bug_dir = root / "demo-skill" / "octocat" / "abc123"
        bug_dir.mkdir(parents=True)
        (bug_dir / "bug-context.json").write_text("{}\n", encoding="utf-8")
        (bug_dir / "BUG_REPORT.md").write_text("# Bug\n", encoding="utf-8")
        return bug_dir

    def command(self, bug_dir: Path, *extra: str) -> list[str]:
        return [
            sys.executable,
            str(SCRIPT),
            "--bug-dir",
            str(bug_dir),
            "--status",
            "fixed",
            "--canonical-root-cause",
            "BRCC-2026-001",
            "--fixed-version-or-commit",
            "v1.2.3",
            "--verification",
            "pytest tests/test_regression.py: 4 passed",
            "--print-json",
            *extra,
        ]

    def test_dry_run_create_and_idempotent_repeat(self) -> None:
        with tempfile.TemporaryDirectory() as tmpdir:
            bug_dir = self.make_bug_dir(Path(tmpdir))
            dry_run = subprocess.run(self.command(bug_dir, "--dry-run"), check=True, capture_output=True, text=True)
            self.assertEqual(json.loads(dry_run.stdout)["action"], "create")
            self.assertFalse((bug_dir / "RESOLUTION.md").exists())

            created = subprocess.run(self.command(bug_dir), check=True, capture_output=True, text=True)
            self.assertEqual(json.loads(created.stdout)["action"], "create")
            original = (bug_dir / "RESOLUTION.md").read_text(encoding="utf-8")

            repeated = subprocess.run(self.command(bug_dir), check=True, capture_output=True, text=True)
            self.assertEqual(json.loads(repeated.stdout)["action"], "unchanged")
            self.assertEqual((bug_dir / "RESOLUTION.md").read_text(encoding="utf-8"), original)

    def test_conflicting_resolution_is_not_overwritten(self) -> None:
        with tempfile.TemporaryDirectory() as tmpdir:
            bug_dir = self.make_bug_dir(Path(tmpdir))
            subprocess.run(self.command(bug_dir), check=True, capture_output=True, text=True)
            result = subprocess.run(
                self.command(bug_dir, "--verification", "different evidence"),
                check=False,
                capture_output=True,
                text=True,
            )
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("拒绝覆盖", result.stderr)

    def test_fixed_requires_verification(self) -> None:
        with tempfile.TemporaryDirectory() as tmpdir:
            bug_dir = self.make_bug_dir(Path(tmpdir))
            command = self.command(bug_dir)
            verification_index = command.index("--verification")
            del command[verification_index : verification_index + 2]
            result = subprocess.run(command, check=False, capture_output=True, text=True)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("--verification", result.stderr)

    def test_duplicate_requires_duplicate_of(self) -> None:
        with tempfile.TemporaryDirectory() as tmpdir:
            bug_dir = self.make_bug_dir(Path(tmpdir))
            command = self.command(bug_dir)
            command[command.index("fixed")] = "duplicate"
            result = subprocess.run(command, check=False, capture_output=True, text=True)
            self.assertNotEqual(result.returncode, 0)
            self.assertIn("--duplicate-of", result.stderr)


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

# Data Model

`bug-context.json` 的目标是同时满足三件事：

1. 让 AI 能稳定重建 bug 背景
2. 让同一问题可以被稳定去重
3. 让后续公开上报时不必重新补环境信息

`RESOLUTION.md` 是独立的追加式闭环层。它不能改写或替代原始记录，只在修复与验证完成后创建。

## 顶层字段

- `schema_version`
- `bug_hash`
- `skill`
- `reporter`
- `bug`
- `environment`
- `tracking`
- `deduplication`

## 设计要点

- `skill` 保存 skill 名称、作者、来源路径、来源仓库
- `reporter` 保存展示名、GitHub 用户名；默认不再保留本地用户名
- `bug` 保存摘要、预期行为、实际行为、复现步骤、证据、临时 workaround
- `environment` 保存设备、OS、运行时、常见软件版本
- `tracking` 保存首次发现、最近发现、出现次数、是否已公开
- `deduplication.fingerprint_payload` 保存参与哈希的规范化字段，便于审计
- 所有自由文本字段在落盘前都要经过隐私清洗，避免把密钥、密码、身份信息、电话、邮箱、银行卡号与私密路径写入本地或公开副本

## 本地副本 vs 公开副本

当前版本默认采用“最小必要信息”策略，本地 `bug-context.json` 也不会再主动采集下列高风险字段：

- `reporter.local_username`
- `environment.device.hostname`
- `environment.runtime.cwd`
- `environment.runtime.local_username`
- `tracking.local_path`

若旧版本地记录仍含有上述字段，公开上报前必须生成脱敏副本，把它们替换为 `redacted` 或移除，避免公开仓库泄露本地隐私。

## 哈希策略

`bug_hash` 使用 `sha256`，基于规范化后的 bug 内容与环境指纹生成。

参与指纹的字段集合由 `config.yaml:hashing.stable_fields` 决定，脚本会按点路径从规范化 payload 中挑选字段后再计算哈希。

当前默认不把 `reporter` 放入哈希，原因是：

- 同一个人可能先在未登录 `gh` 的情况下收集 bug，后续再补 GitHub 身份
- 公开路径已经包含 `github_username`
- 这样可以减少“同一个 bug 仅因报告者身份补齐而重新生成新哈希”的重复

当前默认也不把 `reproduction_steps` 放入哈希，原因是：

- 同一 bug 在后续排查中常常会补充更多复现细节
- 如果把步骤文本直接纳入哈希，补写一条步骤就会错误地产生一个新 bug 目录
- 复现步骤仍然完整保留在 `bug-context.json` 与 `BUG_REPORT.md` 中，只是不参与去重指纹

## Resolution 模型

`RESOLUTION.md` 的 YAML frontmatter 固定包含：

- `status`：`fixed` 或 `duplicate`
- `canonical_root_cause`：稳定的 canonical 根因 ID
- `fixed_version_or_commit`：已验证的 skill 版本或源码 commit
- `resolved_at`：ISO 8601 时间
- `duplicate_of`：重复记录指向的 canonical bug；canonical 记录为 `null`
- `resolution_fingerprint`：忽略时间后的内容指纹，用于幂等判断

正文 `Verification` 至少保留一条可复核证据。已有 resolution 的指纹相同则视为幂等重跑，指纹不同则拒绝覆盖。

# Reporting Protocol

## 本地阶段

- 只写入 `~/.bensz-skills/bugs/`
- 不修改用户本地已安装 skills 的源代码
- 落盘前先对自由文本做敏感信息清洗，严禁把密钥、密码、身份信息、电话、邮箱、银行卡号和私密路径写进记录
- 每次记录都更新 `tracking.last_seen_at`
- 同一 `bug_hash` 再次出现时，只增加 `tracking.occurrence_count`
- 默认不记录本地用户名、主机名、工作目录、local_path 等高风险个人字段

## 公开阶段

- 必须使用用户本机的 `gh`
- 如使用 GitHub Enterprise，调用时需遵循 `config.yaml:github.api_host`
- 先检查 `gh auth status`
- 若未登录，先引导执行 `gh auth login`
- 先验证远端仓库可访问，再开始遍历本地 bug
- 不拉取整个 `bensz-bugs` 仓库
- 公开前先生成脱敏副本，确保本地用户名、主机名、工作目录、绝对路径，以及 bug 文本中的敏感个人信息不进入公开仓库
- 直接通过 GitHub Contents API 创建：
  - `{skill_name}/{github_username}/{bug_hash}/bug-context.json`
  - `{skill_name}/{github_username}/{bug_hash}/BUG_REPORT.md`

## 去重规则

- 若远端目标路径已存在，则视为该 bug 已公开
- 已公开 bug 不重复上传，不做远端覆盖
- 本地只把公开状态标记为完成
- `--dry-run` 只输出“预计上传项”，不修改本地文件，也不创建远端文件

## Resolution 阶段

- 只有源码修复、专项回归和版本核对全部通过后，才创建 `RESOLUTION.md`
- canonical 记录使用稳定的 `canonical_root_cause`；重复记录额外使用 `duplicate_of` 指向 canonical bug
- resolution 必须包含修复版本或 commit 与至少一条验证证据
- `resolve_bug.py --dry-run` 不创建文件；相同参数重复执行幂等
- `RESOLUTION.md` 一旦存在，不允许覆盖、截断或静默更新；内容变化应先人工审计原记录
- 原始 `BUG_REPORT.md` 与 `bug-context.json` 永远保持不变

from __future__ import annotations

import argparse

from common import (
    add_common_collection_arguments,
    bug_report_markdown,
    build_bug_directory,
    compute_bug_hash,
    current_github_username,
    deduplication_payload,
    detect_device,
    detect_os_details,
    detect_runtime,
    detect_software_versions,
    load_config,
    merge_unique,
    now_iso,
    parse_software_pairs,
    read_json,
    read_text,
    reporter_folder_name,
    sanitize_user_list,
    sanitize_user_text,
    sanitized_local_context,
    skill_root,
    skill_storage_root,
    validate_storage_root,
    write_json,
    write_text,
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="记录 Bensz skills 的设计缺陷 bug。")
    add_common_collection_arguments(parser)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    config = load_config()
    root = skill_storage_root(config, args.bug_root)
    root = validate_storage_root(root)
    github_username = current_github_username(args.reporter_github)
    reporter_display = github_username or config["defaults"]["anonymous_reporter"]
    reporter_folder = reporter_folder_name(config, github_username)

    skill_name = sanitize_user_text(args.skill_name, config)
    skill_author = sanitize_user_text(args.skill_author, config)
    bug_summary = sanitize_user_text(args.bug_summary, config)
    expected_behavior = sanitize_user_text(args.expected_behavior, config)
    actual_behavior = sanitize_user_text(args.actual_behavior, config)
    reproduction_steps = sanitize_user_list(args.reproduction_step, config)
    evidence = sanitize_user_list(args.evidence, config)
    workaround = sanitize_user_text(args.workaround, config) or None
    additional_note = sanitize_user_text(args.additional_note, config) or None
    severity = sanitize_user_text(args.severity, config) or config["defaults"]["severity"]
    skill_source_path = sanitize_user_text(args.skill_source_path, config) or None
    skill_source_repo = sanitize_user_text(args.skill_source_repo, config) or None

    software_overrides = parse_software_pairs(args.software)
    environment = {
        "device": detect_device(config, args.device_type or config["defaults"]["device_type"]),
        "os": detect_os_details(),
        "runtime": detect_runtime(config, args.agent_runtime or config["defaults"]["agent_runtime"]),
        "software_versions": detect_software_versions(config, software_overrides),
    }

    fingerprint_payload = deduplication_payload(
        config=config,
        skill_name=skill_name,
        skill_author=skill_author,
        summary=bug_summary,
        expected_behavior=expected_behavior,
        actual_behavior=actual_behavior,
        environment=environment,
    )
    bug_hash = compute_bug_hash(fingerprint_payload, config["hashing"]["algorithm"])
    bug_dir = build_bug_directory(config, root, skill_name, reporter_folder, bug_hash)
    context_path = bug_dir / config["storage"]["context_filename"]
    report_path = bug_dir / config["storage"]["report_filename"]

    now = now_iso()
    if context_path.exists():
        context = sanitized_local_context(read_json(context_path), config)
        tracking = context.setdefault("tracking", {})
        tracking["occurrence_count"] = int(tracking.get("occurrence_count", 1)) + 1
        tracking["last_seen_at"] = now
        context["bug"]["evidence"] = merge_unique(context["bug"].get("evidence", []), evidence)
        context["bug"]["reproduction_steps"] = merge_unique(
            context["bug"].get("reproduction_steps", []),
            reproduction_steps,
        )
        if workaround:
            context["bug"]["workaround"] = workaround
        if additional_note:
            context["bug"]["additional_notes"] = additional_note
        if github_username:
            context["reporter"]["github_username"] = github_username
    else:
        context = {
            "schema_version": config["defaults"]["schema_version"],
            "bug_hash": bug_hash,
            "skill": {
                "name": skill_name,
                "author": skill_author,
                "source_path": skill_source_path,
                "source_repo": skill_source_repo,
            },
            "reporter": {
                "display_name": reporter_display,
                "github_username": github_username,
                "local_username": None,
            },
            "bug": {
                "summary": bug_summary,
                "severity": severity,
                "expected_behavior": expected_behavior,
                "actual_behavior": actual_behavior,
                "reproduction_steps": reproduction_steps,
                "evidence": evidence,
                "workaround": workaround,
                "impact": config["defaults"]["impact_text"],
                "additional_notes": additional_note,
                "is_skill_design_defect": True,
            },
            "environment": environment,
            "tracking": {
                "collected_at": now,
                "first_seen_at": now,
                "last_seen_at": now,
                "occurrence_count": 1,
                "public_reported": False,
                "public_repo": None,
                "public_path": None,
                "reported_at": None,
                "local_path": None,
            },
            "deduplication": {
                "hash_algorithm": config["hashing"]["algorithm"],
                "hash_version": config["hashing"]["version"],
                "fingerprint_payload": fingerprint_payload,
            },
        }

    context = sanitized_local_context(context, config)
    template_path = skill_root() / config["templates"]["bug_report"]
    markdown = bug_report_markdown(read_text(template_path), context, config)
    write_json(context_path, context)
    write_text(report_path, markdown)

    result = {
        "bug_hash": bug_hash,
        "bug_dir": str(bug_dir),
        "context_path": str(context_path),
        "report_path": str(report_path),
        "occurrence_count": context["tracking"]["occurrence_count"],
        "public_reported": context["tracking"]["public_reported"],
    }
    if args.print_json:
        import json

        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(f"已记录 bug: {bug_hash}")
        print(f"目录: {bug_dir}")


if __name__ == "__main__":
    main()

from __future__ import annotations

import argparse
import getpass
import hashlib
import json
import os
import platform
import re
import shutil
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

try:
    import yaml
except ModuleNotFoundError as exc:  # pragma: no cover
    raise SystemExit(
        "缺少 PyYAML 依赖，无法读取 config.yaml。请先执行 `python3 -m pip install pyyaml`。"
    ) from exc


SENSITIVE_TEXT_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
    (
        "secret",
        re.compile(
            r"(?i)\b(?:api[_ -]?key|access[_ -]?token|refresh[_ -]?token|client[_ -]?secret|"
            r"authorization|password|passwd|pwd|secret)\b\s*[:=]\s*(?:bearer\s+)?[^\s,;]+"
        ),
    ),
    ("secret", re.compile(r"(?i)\bbearer\s+[A-Za-z0-9._-]{8,}\b")),
    (
        "private-key",
        re.compile(
            r"-----BEGIN [A-Z0-9 ]*PRIVATE KEY-----[\s\S]*?-----END [A-Z0-9 ]*PRIVATE KEY-----"
        ),
    ),
    (
        "token",
        re.compile(
            r"\b(?:sk-[A-Za-z0-9-]{16,}|gh[pousr]_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,}|"
            r"xox[baprs]-[A-Za-z0-9-]{10,}|AIza[0-9A-Za-z\-_]{20,})\b"
        ),
    ),
    ("email", re.compile(r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", re.IGNORECASE)),
    ("identity", re.compile(r"\b\d{3}-\d{2}-\d{4}\b")),
    ("identity", re.compile(r"\b\d{17}[\dXx]\b")),
)

PRIVATE_PATH_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("private-path", re.compile(r"(?<!\w)(?:/Users|/home)/[^/\s]+(?:/[^\s]*)*")),
    ("private-path", re.compile(r"(?i)\b[A-Z]:\\Users\\[^\\\s]+(?:\\[^\s]*)*")),
)

PHONE_CANDIDATE_PATTERN = re.compile(r"(?<!\w)(?:\+?\d[\d().\-\s]{8,}\d)")
CARD_CANDIDATE_PATTERN = re.compile(r"(?<!\w)(?:\d[ -]?){13,19}(?!\w)")


def skill_root() -> Path:
    return Path(__file__).resolve().parents[1]


def load_config() -> dict[str, Any]:
    config_path = skill_root() / "config.yaml"
    with config_path.open("r", encoding="utf-8") as handle:
        data = yaml.safe_load(handle)
    if not isinstance(data, dict):
        raise SystemExit(f"配置文件格式错误：{config_path}")
    return data


def now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def expand_path(path_text: str) -> Path:
    return Path(path_text).expanduser()


def ensure_directory(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True)


def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def write_text(path: Path, content: str) -> None:
    ensure_directory(path.parent)
    path.write_text(content, encoding="utf-8")


def write_json(path: Path, payload: dict[str, Any]) -> None:
    ensure_directory(path.parent)
    with path.open("w", encoding="utf-8") as handle:
        json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
        handle.write("\n")


def read_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as handle:
        data = json.load(handle)
    if not isinstance(data, dict):
        raise SystemExit(f"JSON 结构错误：{path}")
    return data


def normalize_text(value: str | None) -> str:
    if not value:
        return ""
    return " ".join(value.strip().split())


def normalize_list(values: list[str]) -> list[str]:
    normalized = [normalize_text(item) for item in values if normalize_text(item)]
    return normalized


def privacy_settings(config: dict[str, Any]) -> dict[str, Any]:
    settings = config.get("privacy", {})
    return settings if isinstance(settings, dict) else {}


def reporting_settings(config: dict[str, Any]) -> dict[str, Any]:
    settings = config.get("reporting", {})
    return settings if isinstance(settings, dict) else {}


def redaction_token(config: dict[str, Any], kind: str) -> str:
    template = privacy_settings(config).get("redaction_placeholder", "[redacted:{kind}]")
    return str(template).format(kind=kind)


def luhn_checksum_valid(number: str) -> bool:
    digits = [int(ch) for ch in number]
    checksum = 0
    parity = len(digits) % 2
    for index, digit in enumerate(digits):
        if index % 2 == parity:
            digit *= 2
            if digit > 9:
                digit -= 9
        checksum += digit
    return checksum % 10 == 0


def redact_phone_candidates(text: str, config: dict[str, Any]) -> str:
    def replacer(match: re.Match[str]) -> str:
        digits = re.sub(r"\D", "", match.group(0))
        if 10 <= len(digits) <= 15:
            return redaction_token(config, "phone")
        return match.group(0)

    return PHONE_CANDIDATE_PATTERN.sub(replacer, text)


def redact_card_candidates(text: str, config: dict[str, Any]) -> str:
    def replacer(match: re.Match[str]) -> str:
        digits = re.sub(r"\D", "", match.group(0))
        if 13 <= len(digits) <= 19 and luhn_checksum_valid(digits):
            return redaction_token(config, "credit-card")
        return match.group(0)

    return CARD_CANDIDATE_PATTERN.sub(replacer, text)


def sanitize_user_text(value: str | None, config: dict[str, Any]) -> str:
    text = normalize_text(value)
    if not text:
        return ""
    if not privacy_settings(config).get("auto_redact_sensitive_text", True):
        return text

    for kind, pattern in SENSITIVE_TEXT_PATTERNS + PRIVATE_PATH_PATTERNS:
        text = pattern.sub(redaction_token(config, kind), text)
    text = redact_phone_candidates(text, config)
    text = redact_card_candidates(text, config)
    return text


def sanitize_user_list(values: list[str], config: dict[str, Any]) -> list[str]:
    sanitized = [sanitize_user_text(item, config) for item in values]
    return [item for item in sanitized if item]


def merge_unique(existing: list[str], incoming: list[str]) -> list[str]:
    merged: list[str] = []
    seen: set[str] = set()
    for item in normalize_list(existing) + normalize_list(incoming):
        if item in seen:
            continue
        merged.append(item)
        seen.add(item)
    return merged


def slugify(value: str, fallback: str = "unknown") -> str:
    text = normalize_text(value).lower()
    text = re.sub(r"[^a-z0-9._-]+", "-", text)
    text = text.strip("-._")
    return text or fallback


def command_exists(program: str) -> bool:
    return shutil.which(program) is not None


def run_command(command: list[str], timeout: int = 8) -> str | None:
    if not command:
        return None
    if not command_exists(command[0]):
        return None
    try:
        result = subprocess.run(
            command,
            check=False,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=timeout,
        )
    except (OSError, subprocess.TimeoutExpired):
        return None
    output = (result.stdout or result.stderr or "").strip()
    if not output:
        return None
    return output.splitlines()[0].strip()


def current_local_username() -> str | None:
    try:
        return getpass.getuser()
    except Exception:  # pragma: no cover
        return None


def current_github_username(explicit: str | None = None) -> str | None:
    if explicit:
        return normalize_text(explicit)
    output = run_command(["gh", "api", "user", "-q", ".login"])
    return normalize_text(output) or None


def gh_auth_ok() -> bool:
    if not command_exists("gh"):
        return False
    try:
        result = subprocess.run(
            ["gh", "auth", "status"],
            check=False,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=8,
        )
    except (OSError, subprocess.TimeoutExpired):
        return False
    return result.returncode == 0


def parse_software_pairs(values: list[str]) -> dict[str, str]:
    parsed: dict[str, str] = {}
    for item in values:
        if "=" not in item:
            raise SystemExit(f"`--software` 参数格式错误，应为 key=value：{item}")
        key, value = item.split("=", 1)
        key = slugify(key, fallback="")
        if not key:
            raise SystemExit(f"`--software` 参数键为空：{item}")
        parsed[key] = normalize_text(value)
    return parsed


def detect_os_details() -> dict[str, Any]:
    uname = platform.uname()
    details: dict[str, Any] = {
        "family": uname.system or platform.system(),
        "release": uname.release or platform.release(),
        "version": uname.version or platform.version(),
        "machine": uname.machine or platform.machine(),
        "processor": uname.processor or platform.processor(),
    }

    if details["family"] == "Darwin":
        sw_vers = {
            "product_name": run_command(["sw_vers", "-productName"]),
            "product_version": run_command(["sw_vers", "-productVersion"]),
            "build_version": run_command(["sw_vers", "-buildVersion"]),
        }
        details["macos"] = sw_vers
    elif details["family"] == "Linux":
        os_release = Path("/etc/os-release")
        if os_release.exists():
            parsed: dict[str, str] = {}
            for line in os_release.read_text(encoding="utf-8").splitlines():
                if "=" not in line or line.startswith("#"):
                    continue
                key, value = line.split("=", 1)
                parsed[key] = value.strip().strip('"')
            details["linux_distribution"] = parsed

    return details


def detect_runtime(config: dict[str, Any], agent_runtime: str | None) -> dict[str, Any]:
    privacy = privacy_settings(config)
    return {
        "agent_runtime": sanitize_user_text(agent_runtime, config) or "unknown",
        "shell": os.environ.get("SHELL") or None,
        "cwd": str(Path.cwd()) if privacy.get("collect_working_directory", False) else None,
        "hostname": platform.node() or None if privacy.get("collect_hostnames", False) else None,
        "local_username": (
            current_local_username() if privacy.get("collect_local_username", False) else None
        ),
        "python_runtime": platform.python_version(),
    }


def detect_device(config: dict[str, Any], device_type: str | None) -> dict[str, Any]:
    privacy = privacy_settings(config)
    return {
        "type": sanitize_user_text(device_type, config) or "unknown",
        "architecture": platform.machine() or None,
        "hostname": platform.node() or None if privacy.get("collect_hostnames", False) else None,
    }


def detect_software_versions(config: dict[str, Any], extra_versions: dict[str, str]) -> dict[str, Any]:
    detected: dict[str, Any] = {}
    commands = config.get("environment", {}).get("version_commands", [])
    for entry in commands:
        key = entry.get("key")
        command = entry.get("command")
        if not key or not isinstance(command, list):
            continue
        detected[key] = sanitize_user_text(run_command([str(item) for item in command]), config) or None
    for key, value in extra_versions.items():
        detected[key] = sanitize_user_text(value, config) or None
    return detected


def canonical_json(payload: dict[str, Any]) -> str:
    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def get_nested_value(payload: dict[str, Any], path: str) -> Any:
    current: Any = payload
    for part in path.split("."):
        if not isinstance(current, dict):
            return None
        current = current.get(part)
    return current


def set_nested_value(payload: dict[str, Any], path: str, value: Any) -> None:
    current = payload
    parts = path.split(".")
    for part in parts[:-1]:
        nested = current.get(part)
        if not isinstance(nested, dict):
            nested = {}
            current[part] = nested
        current = nested
    current[parts[-1]] = json.loads(json.dumps(value, ensure_ascii=False))


def compute_bug_hash(payload: dict[str, Any], algorithm: str) -> str:
    try:
        hasher = hashlib.new(algorithm)
    except ValueError as exc:  # pragma: no cover
        raise SystemExit(f"不支持的哈希算法：{algorithm}") from exc
    hasher.update(canonical_json(payload).encode("utf-8"))
    return hasher.hexdigest()


def deduplication_payload(
    config: dict[str, Any],
    skill_name: str,
    skill_author: str,
    summary: str,
    expected_behavior: str,
    actual_behavior: str,
    environment: dict[str, Any],
) -> dict[str, Any]:
    software_versions = environment.get("software_versions", {})
    canonical_payload = {
        "skill": {
            "name": normalize_text(skill_name),
            "author": normalize_text(skill_author),
        },
        "bug": {
            "summary": normalize_text(summary),
            "expected_behavior": normalize_text(expected_behavior),
            "actual_behavior": normalize_text(actual_behavior),
        },
        "environment": {
            "os": environment.get("os", {}),
            "runtime": {
                "agent_runtime": environment.get("runtime", {}).get("agent_runtime"),
            },
            "software_versions": software_versions,
        },
    }
    stable_fields = config.get("hashing", {}).get("stable_fields", [])
    normalized_fields = [normalize_text(str(field)) for field in stable_fields if normalize_text(str(field))]
    if not normalized_fields:
        return canonical_payload

    selected_payload: dict[str, Any] = {}
    for field in normalized_fields:
        value = get_nested_value(canonical_payload, field)
        if value is None:
            continue
        set_nested_value(selected_payload, field, value)
    return selected_payload or canonical_payload


def skill_storage_root(config: dict[str, Any], override: str | None = None) -> Path:
    root_text = override or config["storage"]["local_root"]
    return expand_path(root_text)


def reporter_folder_name(config: dict[str, Any], github_username: str | None) -> str:
    fallback = config.get("defaults", {}).get("reporter_fallback", "unknown-reporter")
    return slugify(github_username or fallback, fallback=fallback)


def build_bug_directory(
    config: dict[str, Any],
    root: Path,
    skill_name: str,
    reporter_folder: str,
    bug_hash: str,
) -> Path:
    path_pattern = config.get("storage", {}).get("path_pattern") or "{skill_name}/{reporter}/{bug_hash}"
    try:
        relative_path = path_pattern.format(
            skill_name=slugify(skill_name),
            reporter=slugify(reporter_folder),
            bug_hash=normalize_text(bug_hash),
        )
    except KeyError as exc:
        raise SystemExit(f"storage.path_pattern 包含不支持的占位符：{exc}") from exc

    path = Path(relative_path)
    if path.is_absolute() or ".." in path.parts:
        raise SystemExit(f"storage.path_pattern 解析后越界：{relative_path}")
    return root / path


def format_list_block(values: list[str], empty_text: str = "- None") -> str:
    normalized = normalize_list(values)
    if not normalized:
        return empty_text
    return "\n".join(f"- {item}" for item in normalized)


def format_software_block(versions: dict[str, Any]) -> str:
    lines = []
    for key in sorted(versions):
        value = versions[key]
        lines.append(f"  - {key}: {value or 'unavailable'}")
    return "\n".join(lines) if lines else "  - unavailable"


def os_summary(environment: dict[str, Any]) -> str:
    os_info = environment.get("os", {})
    family = os_info.get("family") or "unknown"
    release = os_info.get("release") or ""
    machine = os_info.get("machine") or ""
    parts = [part for part in [family, release, machine] if part]
    return " / ".join(parts) if parts else "unknown"


def sanitized_local_context(context: dict[str, Any], config: dict[str, Any]) -> dict[str, Any]:
    sanitized = json.loads(json.dumps(context))
    privacy = privacy_settings(config)

    skill = sanitized.setdefault("skill", {})
    for key in ("name", "author", "source_path", "source_repo"):
        if isinstance(skill.get(key), str):
            skill[key] = sanitize_user_text(skill[key], config)

    reporter = sanitized.setdefault("reporter", {})
    github_username = reporter.get("github_username")
    reporter["display_name"] = github_username or config.get("defaults", {}).get(
        "anonymous_reporter", "anonymous-reporter"
    )
    if not privacy.get("collect_local_username", False):
        reporter["local_username"] = None

    bug = sanitized.setdefault("bug", {})
    for key in (
        "summary",
        "severity",
        "expected_behavior",
        "actual_behavior",
        "workaround",
        "impact",
        "additional_notes",
    ):
        if isinstance(bug.get(key), str):
            bug[key] = sanitize_user_text(bug[key], config)
    bug["reproduction_steps"] = sanitize_user_list(bug.get("reproduction_steps", []), config)
    bug["evidence"] = sanitize_user_list(bug.get("evidence", []), config)

    environment = sanitized.setdefault("environment", {})
    device = environment.setdefault("device", {})
    runtime = environment.setdefault("runtime", {})
    software_versions = environment.setdefault("software_versions", {})

    if isinstance(device.get("type"), str):
        device["type"] = sanitize_user_text(device["type"], config) or "unknown"
    if not privacy.get("collect_hostnames", False):
        device["hostname"] = None
    elif isinstance(device.get("hostname"), str):
        device["hostname"] = sanitize_user_text(device["hostname"], config)

    if isinstance(runtime.get("agent_runtime"), str):
        runtime["agent_runtime"] = sanitize_user_text(runtime["agent_runtime"], config) or "unknown"
    if isinstance(runtime.get("shell"), str):
        runtime["shell"] = sanitize_user_text(runtime["shell"], config)
    if not privacy.get("collect_working_directory", False):
        runtime["cwd"] = None
    elif isinstance(runtime.get("cwd"), str):
        runtime["cwd"] = sanitize_user_text(runtime["cwd"], config)
    if not privacy.get("collect_hostnames", False):
        runtime["hostname"] = None
    elif isinstance(runtime.get("hostname"), str):
        runtime["hostname"] = sanitize_user_text(runtime["hostname"], config)
    if not privacy.get("collect_local_username", False):
        runtime["local_username"] = None
    elif isinstance(runtime.get("local_username"), str):
        runtime["local_username"] = sanitize_user_text(runtime["local_username"], config)

    for key, value in list(software_versions.items()):
        if isinstance(value, str):
            software_versions[key] = sanitize_user_text(value, config)

    tracking = sanitized.setdefault("tracking", {})
    if not privacy.get("store_local_path", False):
        tracking["local_path"] = None
    elif isinstance(tracking.get("local_path"), str):
        tracking["local_path"] = sanitize_user_text(tracking["local_path"], config)

    deduplication = sanitized.get("deduplication", {})
    fingerprint_payload = deduplication.get("fingerprint_payload")
    if isinstance(fingerprint_payload, dict):
        fingerprint_skill = fingerprint_payload.get("skill", {})
        fingerprint_bug = fingerprint_payload.get("bug", {})
        fingerprint_environment = fingerprint_payload.get("environment", {})
        for key in ("name", "author"):
            if isinstance(fingerprint_skill.get(key), str):
                fingerprint_skill[key] = sanitize_user_text(fingerprint_skill[key], config)
        for key in ("summary", "expected_behavior", "actual_behavior"):
            if isinstance(fingerprint_bug.get(key), str):
                fingerprint_bug[key] = sanitize_user_text(fingerprint_bug[key], config)
        if isinstance(fingerprint_environment.get("agent_runtime"), str):
            fingerprint_environment["agent_runtime"] = sanitize_user_text(
                fingerprint_environment["agent_runtime"],
                config,
            )
        nested_runtime = fingerprint_environment.get("runtime", {})
        if isinstance(nested_runtime, dict) and isinstance(nested_runtime.get("agent_runtime"), str):
            nested_runtime["agent_runtime"] = sanitize_user_text(
                nested_runtime["agent_runtime"],
                config,
            )
        software = fingerprint_environment.get("software_versions", {})
        if isinstance(software, dict):
            for key, value in list(software.items()):
                if isinstance(value, str):
                    software[key] = sanitize_user_text(value, config)

    return sanitized


def bug_report_markdown(template_text: str, context: dict[str, Any], config: dict[str, Any]) -> str:
    bug = context["bug"]
    environment = context["environment"]
    tracking = context["tracking"]
    reporter = context["reporter"]
    replacements = {
        "skill_name": context["skill"]["name"],
        "skill_author": context["skill"]["author"],
        "reporter_github": reporter.get("github_username") or "unknown",
        "bug_hash": context["bug_hash"],
        "severity": bug.get("severity") or "important",
        "occurrence_count": tracking.get("occurrence_count") or 1,
        "first_seen_at": tracking.get("first_seen_at") or tracking.get("collected_at") or "unknown",
        "last_seen_at": tracking.get("last_seen_at") or tracking.get("collected_at") or "unknown",
        "privacy_notice": reporting_settings(config).get(
            "privacy_notice",
            "Sensitive user data is auto-redacted before storage and public reporting.",
        ),
        "summary": bug.get("summary") or "",
        "expected_behavior": bug.get("expected_behavior") or "",
        "actual_behavior": bug.get("actual_behavior") or "",
        "reproduction_steps_block": format_list_block(bug.get("reproduction_steps", [])),
        "evidence_block": format_list_block(bug.get("evidence", [])),
        "skill_source_path": context["skill"].get("source_path") or "None",
        "skill_source_repo": context["skill"].get("source_repo") or "None",
        "device_type": environment.get("device", {}).get("type") or "unknown",
        "os_summary": os_summary(environment),
        "shell": environment.get("runtime", {}).get("shell") or "unknown",
        "agent_runtime": environment.get("runtime", {}).get("agent_runtime") or "unknown",
        "software_versions_block": format_software_block(environment.get("software_versions", {})),
        "impact": bug.get("impact") or "该问题会削弱 skill 在真实环境中的稳定性与可预期性，属于需要跟踪修复的设计缺陷。",
        "workaround": bug.get("workaround") or "None",
        "additional_notes": bug.get("additional_notes") or "None",
    }
    return template_text.format(**replacements)


def add_common_collection_arguments(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--skill-name", required=True)
    parser.add_argument("--skill-author", required=True)
    parser.add_argument("--bug-summary", required=True)
    parser.add_argument("--expected-behavior", required=True)
    parser.add_argument("--actual-behavior", required=True)
    parser.add_argument("--reproduction-step", action="append", default=[])
    parser.add_argument("--evidence", action="append", default=[])
    parser.add_argument("--workaround")
    parser.add_argument("--severity")
    parser.add_argument("--device-type")
    parser.add_argument("--agent-runtime")
    parser.add_argument("--skill-source-path")
    parser.add_argument("--skill-source-repo")
    parser.add_argument("--reporter-display-name", help="已弃用；为保护隐私，此参数当前会被忽略")
    parser.add_argument("--reporter-github")
    parser.add_argument("--additional-note")
    parser.add_argument("--software", action="append", default=[])
    parser.add_argument("--bug-root")
    parser.add_argument("--print-json", action="store_true")


def find_context_files(root: Path, filename: str) -> list[Path]:
    if not root.exists():
        return []
    return sorted(root.rglob(filename))


def validate_storage_root(path: Path) -> Path:
    if path.exists() and not path.is_dir():
        raise SystemExit(f"bug 根目录不是文件夹：{path}")
    if path.is_symlink():
        raise SystemExit(f"拒绝使用符号链接作为 bug 根目录：{path}")
    return path.resolve()


def sanitized_public_context(
    context: dict[str, Any],
    config: dict[str, Any],
    placeholder: str = "redacted",
) -> dict[str, Any]:
    sanitized = sanitized_local_context(context, config)
    sanitized["reporter"]["display_name"] = sanitized["reporter"].get("github_username") or placeholder
    sanitized["reporter"]["local_username"] = placeholder

    skill_block = sanitized.get("skill", {})
    if skill_block.get("source_path"):
        skill_block["source_path"] = placeholder

    environment = sanitized.get("environment", {})
    device = environment.get("device", {})
    runtime = environment.get("runtime", {})
    device["hostname"] = placeholder if device.get("hostname") else None
    runtime["hostname"] = placeholder if runtime.get("hostname") else None
    runtime["cwd"] = placeholder if runtime.get("cwd") else None
    runtime["local_username"] = placeholder if runtime.get("local_username") else None

    tracking = sanitized.get("tracking", {})
    if tracking.get("local_path"):
        tracking["local_path"] = placeholder
    return sanitized

from __future__ import annotations

import argparse
import base64
import json
import subprocess
from typing import Any

from common import (
    bug_report_markdown,
    current_github_username,
    find_context_files,
    gh_auth_ok,
    load_config,
    now_iso,
    read_json,
    read_text,
    sanitized_local_context,
    sanitized_public_context,
    skill_root,
    skill_storage_root,
    slugify,
    validate_storage_root,
    write_json,
    write_text,
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="把本地收集的 bug 公开上报到 GitHub 仓库。")
    parser.add_argument("--bug-root")
    parser.add_argument("--skill-name")
    parser.add_argument("--repo-owner")
    parser.add_argument("--repo-name")
    parser.add_argument("--reporter-github")
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--print-json", action="store_true")
    return parser.parse_args()


def gh_api(
    endpoint: str,
    *,
    host: str,
    method: str = "GET",
    fields: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    command = ["gh", "api", endpoint]
    if host and host != "github.com":
        command.extend(["--hostname", host])
    if method != "GET":
        command.extend(["--method", method])
    for key, value in (fields or {}).items():
        command.extend(["-f", f"{key}={value}"])
    return subprocess.run(
        command,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=20,
    )


def remote_file_exists(owner: str, repo: str, path: str, host: str) -> bool:
    result = gh_api(f"repos/{owner}/{repo}/contents/{path}", host=host)
    return result.returncode == 0


def verify_remote_repo_access(owner: str, repo: str, host: str) -> None:
    result = gh_api(f"repos/{owner}/{repo}", host=host)
    if result.returncode != 0:
        raise SystemExit(
            f"无法访问远端仓库 {owner}/{repo}。\n"
            "请确认仓库存在、当前 `gh` 账号有权限，并且网络可用。"
        )


def upload_text_file(
    owner: str,
    repo: str,
    path: str,
    content: str,
    message: str,
    dry_run: bool,
    host: str,
) -> None:
    if dry_run:
        return
    payload = base64.b64encode(content.encode("utf-8")).decode("ascii")
    result = gh_api(
        f"repos/{owner}/{repo}/contents/{path}",
        host=host,
        method="PUT",
        fields={"message": message, "content": payload},
    )
    if result.returncode != 0:
        raise SystemExit(f"上传失败：{path}\n{result.stderr.strip()}")


def main() -> None:
    args = parse_args()
    config = load_config()
    if not gh_auth_ok():
        raise SystemExit("`gh` 未登录或不可用。请先执行 `gh auth login`，然后重试。")

    owner = args.repo_owner or config["github"]["owner"]
    repo = args.repo_name or config["github"]["repo"]
    host = config.get("github", {}).get("api_host") or "github.com"
    bug_root = skill_storage_root(config, args.bug_root)
    bug_root = validate_storage_root(bug_root)
    reporter_github = current_github_username(args.reporter_github)
    if not reporter_github:
        raise SystemExit("无法确定当前 GitHub 用户名。请先确认 `gh auth status` 可用。")
    verify_remote_repo_access(owner, repo, host)

    context_filename = config["storage"]["context_filename"]
    context_files = find_context_files(bug_root, context_filename)
    if args.skill_name:
        expected_skill = slugify(args.skill_name)
        context_files = [
            path
            for path in context_files
            if slugify(read_json(path)["skill"]["name"]) == expected_skill
        ]

    uploaded = 0
    would_upload = 0
    skipped = 0
    already_public = 0
    results: list[dict[str, Any]] = []
    template_path = skill_root() / config["templates"]["bug_report"]
    template_text = read_text(template_path)

    for context_path in context_files:
        context = sanitized_local_context(read_json(context_path), config)
        if not context.get("bug", {}).get("is_skill_design_defect", False):
            skipped += 1
            continue

        bug_hash = context["bug_hash"]
        skill_slug = slugify(context["skill"]["name"])
        remote_dir = f"{skill_slug}/{reporter_github}/{bug_hash}"
        remote_context_path = f"{remote_dir}/{config['storage']['context_filename']}"
        remote_report_path = f"{remote_dir}/{config['storage']['report_filename']}"

        tracking = context.setdefault("tracking", {})
        if not args.dry_run:
            tracking["public_reported"] = True
            tracking["public_repo"] = f"{owner}/{repo}"
            tracking["public_path"] = remote_dir
            tracking["reported_at"] = now_iso()
        context["reporter"]["github_username"] = reporter_github
        public_context = sanitized_public_context(
            context,
            config,
            placeholder=config["reporting"]["redacted_placeholder"],
        )
        markdown = bug_report_markdown(template_text, context, config)
        public_markdown = bug_report_markdown(template_text, public_context, config)
        report_path = context_path.parent / config["storage"]["report_filename"]
        if not args.dry_run:
            write_text(report_path, markdown)

        context_exists = remote_file_exists(owner, repo, remote_context_path, host)
        report_exists = remote_file_exists(owner, repo, remote_report_path, host)
        if context_exists and report_exists:
            already_public += 1
        else:
            would_upload += 1
            message = config["reporting"]["commit_message_template"].format(
                bug_hash=bug_hash,
                skill_name=context["skill"]["name"],
            )
            upload_text_file(
                owner,
                repo,
                remote_context_path,
                json.dumps(public_context, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
                message,
                args.dry_run,
                host,
            )
            upload_text_file(
                owner,
                repo,
                remote_report_path,
                public_markdown,
                message,
                args.dry_run,
                host,
            )
            if not args.dry_run:
                uploaded += 1

        if not args.dry_run:
            write_json(context_path, context)
        results.append(
            {
                "bug_hash": bug_hash,
                "remote_dir": remote_dir,
                "uploaded": (not args.dry_run) and not (context_exists and report_exists),
                "would_upload": args.dry_run and not (context_exists and report_exists),
            }
        )

    summary = {
        "bug_root": str(bug_root),
        "repo": f"{owner}/{repo}",
        "reporter": reporter_github,
        "dry_run": args.dry_run,
        "uploaded": uploaded,
        "would_upload": would_upload,
        "already_public": already_public,
        "skipped": skipped,
        "total_scanned": len(context_files),
        "results": results,
    }
    if args.print_json:
        print(json.dumps(summary, ensure_ascii=False, indent=2))
    else:
        if args.dry_run:
            print(
                f"扫描 {len(context_files)} 个本地 bug，预计新增公开 {would_upload} 个，"
                f"已存在 {already_public} 个，跳过 {skipped} 个。"
            )
        else:
            print(
                f"扫描 {len(context_files)} 个本地 bug，新增公开 {uploaded} 个，"
                f"已存在 {already_public} 个，跳过 {skipped} 个。"
            )
        print(f"远端仓库: https://{host}/{owner}/{repo}")


if __name__ == "__main__":
    main()

from __future__ import annotations

import argparse
import hashlib
import json
import re
from datetime import datetime
from pathlib import Path
from typing import Any

from common import load_config, now_iso, read_text, sanitize_user_list, sanitize_user_text, skill_root, write_text


CANONICAL_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,127}$")
FINGERPRINT_PATTERN = re.compile(r"^resolution_fingerprint:\s*([a-f0-9]{64})\s*$", re.MULTILINE)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="为已有 Bensz bug 追加不可覆盖的 resolution 记录。")
    parser.add_argument("--bug-dir", required=True, help="包含 bug-context.json 与 BUG_REPORT.md 的 bug 目录")
    parser.add_argument("--status", required=True, choices=("fixed", "duplicate"))
    parser.add_argument("--canonical-root-cause", required=True, help="稳定的 canonical 根因 ID")
    parser.add_argument("--fixed-version-or-commit", required=True)
    parser.add_argument("--verification", action="append", default=[], help="可重复传入的验证证据")
    parser.add_argument("--duplicate-of", help="重复记录指向的 canonical bug hash 或公开路径")
    parser.add_argument("--resolved-at", help="ISO 8601 时间；默认使用当前 UTC 时间")
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--print-json", action="store_true")
    return parser.parse_args()


def validated_bug_dir(path_text: str, config: dict[str, Any]) -> Path:
    path = Path(path_text).expanduser()
    if path.is_symlink():
        raise SystemExit(f"拒绝使用符号链接 bug 目录：{path}")
    path = path.resolve()
    if not path.is_dir():
        raise SystemExit(f"bug 目录不存在：{path}")
    required = (config["storage"]["context_filename"], config["storage"]["report_filename"])
    missing = [name for name in required if not (path / name).is_file()]
    if missing:
        raise SystemExit(f"bug 目录缺少原始证据文件：{', '.join(missing)}")
    return path


def validate_resolved_at(value: str | None) -> str:
    if not value:
        return now_iso()
    candidate = value.replace("Z", "+00:00")
    try:
        datetime.fromisoformat(candidate)
    except ValueError as exc:
        raise SystemExit("--resolved-at 必须是 ISO 8601 时间") from exc
    return value


def build_payload(args: argparse.Namespace, config: dict[str, Any]) -> dict[str, Any]:
    canonical_id = sanitize_user_text(args.canonical_root_cause, config)
    if not CANONICAL_ID_PATTERN.fullmatch(canonical_id):
        raise SystemExit("--canonical-root-cause 仅允许 3-128 位字母、数字、点、下划线、冒号或连字符")
    version = sanitize_user_text(args.fixed_version_or_commit, config)
    verification = sanitize_user_list(args.verification, config)
    duplicate_of = sanitize_user_text(args.duplicate_of, config) or None
    if not version:
        raise SystemExit("标记 resolved 必须提供 --fixed-version-or-commit")
    if not verification:
        raise SystemExit("标记 resolved 必须至少提供一条 --verification")
    if args.status == "duplicate" and not duplicate_of:
        raise SystemExit("status=duplicate 必须提供 --duplicate-of")
    return {
        "status": args.status,
        "canonical_root_cause": canonical_id,
        "fixed_version_or_commit": version,
        "verification": verification,
        "duplicate_of": duplicate_of,
    }


def payload_fingerprint(payload: dict[str, Any]) -> str:
    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def render_resolution(payload: dict[str, Any], resolved_at: str, config: dict[str, Any]) -> str:
    template = read_text(skill_root() / config["templates"]["resolution"])
    verification_block = "\n".join(f"- {item}" for item in payload["verification"])
    return template.format(
        status=payload["status"],
        canonical_root_cause=payload["canonical_root_cause"],
        fixed_version_or_commit=payload["fixed_version_or_commit"],
        resolved_at=resolved_at,
        duplicate_of=payload["duplicate_of"] or "null",
        resolution_fingerprint=payload_fingerprint(payload),
        verification_block=verification_block,
    )


def main() -> None:
    args = parse_args()
    config = load_config()
    bug_dir = validated_bug_dir(args.bug_dir, config)
    payload = build_payload(args, config)
    resolved_at = validate_resolved_at(args.resolved_at)
    fingerprint = payload_fingerprint(payload)
    resolution_path = bug_dir / config["storage"]["resolution_filename"]
    action = "create"

    if resolution_path.exists():
        existing = read_text(resolution_path)
        match = FINGERPRINT_PATTERN.search(existing)
        if match and match.group(1) == fingerprint:
            action = "unchanged"
        else:
            raise SystemExit(
                f"resolution 已存在且内容不同，追加式协议拒绝覆盖：{resolution_path}"
            )
    elif not args.dry_run:
        write_text(resolution_path, render_resolution(payload, resolved_at, config))

    result = {
        "action": action,
        "dry_run": bool(args.dry_run),
        "resolution_path": str(resolution_path),
        "status": payload["status"],
        "canonical_root_cause": payload["canonical_root_cause"],
        "duplicate_of": payload["duplicate_of"],
    }
    if args.print_json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        prefix = "预计创建" if args.dry_run and action == "create" else ("无需变更" if action == "unchanged" else "已创建")
        print(f"{prefix}: {resolution_path}")


if __name__ == "__main__":
    main()

# Bug Report

## Metadata
- Skill: {skill_name}
- Skill author: {skill_author}
- Reporter GitHub: {reporter_github}
- Bug hash: {bug_hash}
- Severity: {severity}
- Occurrence count: {occurrence_count}
- First seen at: {first_seen_at}
- Last seen at: {last_seen_at}
- Privacy: {privacy_notice}

## Summary
{summary}

## Expected Behavior
{expected_behavior}

## Actual Behavior
{actual_behavior}

## Reproduction Steps
{reproduction_steps_block}

## Evidence
{evidence_block}

## Environment Notes
- Skill source path: {skill_source_path}
- Skill source repo: {skill_source_repo}
- Device type: {device_type}
- OS: {os_summary}
- Shell: {shell}
- Agent runtime: {agent_runtime}
- Key software versions:
{software_versions_block}

## Impact
{impact}

## Workaround
{workaround}

## Additional Notes
{additional_notes}

---
status: {status}
canonical_root_cause: {canonical_root_cause}
fixed_version_or_commit: {fixed_version_or_commit}
resolved_at: {resolved_at}
duplicate_of: {duplicate_of}
resolution_fingerprint: {resolution_fingerprint}
---

# Bug Resolution

## Verification

{verification_block}

## Notes

This file is an append-only resolution record. The original `BUG_REPORT.md` and `bug-context.json` remain unchanged.