diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4f3a733..7a4c08e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,6 +26,10 @@ jobs: run: | python -m py_compile $(git ls-files '*.py') + - name: Unit tests + run: | + python -m unittest discover -s tests -p 'test_*.py' + - name: Import and routing smoke check run: | python scripts/smoke_imports.py @@ -34,15 +38,27 @@ jobs: run: | python scripts/verify_cta_resources.py + - name: Pipeline preflight dry-run + run: | + python scripts/preflight.py --input examples/example_article.md --skip-command-checks --json + + - name: Orchestrator governance dry-run + run: | + python scripts/orchestrator.py --input examples/example_article.md --output-dir output/ci-dry-run --log /tmp/md2video-ci-pipeline.jsonl --dry-run --skip-command-checks --allow-dirty-output + - name: JSON validation run: | python -c "import json; json.load(open('harness/video-rules.json'))" python -c "import json; json.load(open('rules/segment_types.json'))" python -c "import json; json.load(open('rules/storyboard_rules.json'))" - - name: Self report dry-run + - name: Self report no-write dry-run + run: | + python harness/self_report.py --no-write --json + + - name: Privacy gate run: | - python harness/self_report.py + bash scripts/privacy-check.sh --full - name: Check LESSONS_LEARNED frontmatter run: | diff --git a/README.md b/README.md index 8686fff..14120d4 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,14 @@ storyboard_from_article(article, output_dir='output') # 运行自检 python harness/self_report.py + +# 治理 dry-run（不调用外部素材/TTS 服务） +python scripts/orchestrator.py \ + --input examples/example_article.md \ + --output-dir output/dry-run \ + --dry-run \ + --skip-command-checks \ + --allow-dirty-output ``` 零外部 API 依赖即可运行语法检查和自检。即梦 CLI 仅在有 AI 素材需求时需要。 @@ -119,6 +127,31 @@ LESSONS_LEARNED.md 更新 + video-rules.json 规则演化 ✅ ``` +### Step -1：治理预检和运行证明 + +正式生成前先跑治理 dry-run，确认本仓入口、规则、CTA 资源和自检链路是自洽的： + +```bash +python scripts/preflight.py \ + --input examples/example_article.md \ + --skip-command-checks \ + --json + +python scripts/orchestrator.py \ + --input examples/example_article.md \ + --output-dir output/dry-run \ + --dry-run \ + --skip-command-checks \ + --allow-dirty-output +``` + +`scripts/orchestrator.py` 当前是治理外壳，不会调用付费或远程素材生成服务。它会写入： + +- `.md2video-pipeline.jsonl`：每个治理步骤的结构化日志 +- `output/dry-run/run-manifest.json`：输入 hash、仓库状态、环境版本、关键产物指纹和步骤结果 + +CI 也会跑这一套 dry-run，防止入口契约、QR registry、导入路由或 self-report no-write 行为漂移。 + ### Step 0：分镜拆解 **规则驱动，无需改代码。** 将文章输入 `storyboard_ai.py`，自动输出： diff --git a/SKILL.md b/SKILL.md index f9b116a..de35597 100644 --- a/SKILL.md +++ b/SKILL.md @@ -129,6 +129,31 @@ harness.run("output/final_with_cta.mp4") 完整 pipeline 示例见 `examples/example_pipeline.py`。 +### 3b. 治理 dry-run（推荐先跑） + +在调用真实 TTS、即梦或完整视频拼接前，先运行治理入口，确认规则、CTA、 +导入路由和 self-report no-write 行为没有漂移： + +```bash +python scripts/orchestrator.py \ + --input examples/example_article.md \ + --output-dir output/dry-run \ + --dry-run \ + --skip-command-checks \ + --allow-dirty-output +``` + +这个入口不调用付费或远程素材服务。它会生成： + +- `.md2video-pipeline.jsonl`：结构化步骤日志 +- `output/dry-run/run-manifest.json`：输入 hash、仓库状态、环境版本、关键产物指纹和步骤结果 + +只需要预检时可直接运行： + +```bash +python scripts/preflight.py --input examples/example_article.md --skip-command-checks --json +``` + ## Pipeline 详解 ### 数据流 diff --git a/docs/tasks/2026-06-08-pipeline-orchestrator-governance.md b/docs/tasks/2026-06-08-pipeline-orchestrator-governance.md new file mode 100644 index 0000000..a3267cb --- /dev/null +++ b/docs/tasks/2026-06-08-pipeline-orchestrator-governance.md @@ -0,0 +1,83 @@ +# Task Card: Pipeline Orchestrator and Preflight Governance + +Status: implemented +Created: 2026-06-08 +Implemented: 2026-06-08 + +## Objective + +Adopt the useful governance pattern from `md2wechat`: every pipeline run should +have a stable preflight envelope, structured JSONL execution log, run manifest, +no-write self-report validation, CI dry-run coverage, and a privacy gate. + +## Boundary + +In scope: + +- Add `scripts/preflight.py` for machine-readable pipeline readiness checks. +- Add `scripts/orchestrator.py` as the governed dry-run entrypoint. +- Write `.md2video-pipeline.jsonl` and `output/run-manifest.json` style proof. +- Add `SelfReport.run(no_write=True)` and CLI `--no-write --json`. +- Add unit tests for the governance contracts. +- Add privacy scanning and CI coverage. + +Out of scope: + +- Paid or external video/material generation. +- Replacing `examples/example_pipeline.py` with a full production runner. +- Live E2E rendering with TTS or remote services. +- Automatic self-evolution rule generation beyond the existing `SelfReport` + behavior. + +## Implementation Notes + +- `scripts/preflight.py` checks: + - required command availability (`ffmpeg`, `ffprobe`) unless skipped + - governance JSON parseability + - storyboard animation routing + - CTA registry consistency + - TTS-sensitive input characters + - stale output artifacts +- `scripts/orchestrator.py` runs the governance chain: + - preflight + - CTA registry verification + - import/routing smoke + - self-report no-write validation + - run manifest write +- `harness/self_report.py --no-write --json` is safe for CI and local contract + checks because it does not write `LESSONS_LEARNED.md`, `video-rules.json`, or + `output/self_report.json`. +- `scripts/privacy-check.sh --full` is wired into CI to block common secret + patterns and local user paths in tracked text. + +## Validation + +Run: + +```bash +python -m unittest discover -s tests -p 'test_*.py' +python scripts/preflight.py --input examples/example_article.md --skip-command-checks --json +python scripts/orchestrator.py --input examples/example_article.md --output-dir /tmp/md2video-dry-run-output --log /tmp/md2video-pipeline.jsonl --dry-run --skip-command-checks --allow-dirty-output +python harness/self_report.py --no-write --json +bash scripts/privacy-check.sh --full +python -m py_compile $(git ls-files '*.py') +python scripts/smoke_imports.py +``` + +Expected: + +- unit tests pass +- preflight JSON reports `ok: true` +- orchestrator exits `0` and writes a JSONL log plus `run-manifest.json` +- self-report no-write exits `0` without mutating governance files +- privacy gate has no blocking findings +- syntax and import smoke checks pass + +## Residual Risks + +- The orchestrator is currently a governance dry-run wrapper, not a full + article-to-video runner. +- Full paid-service E2E remains outside this task and should stay explicit. +- Future self-evolution hardening should add observation-layer generated checks, + companion tests, audit records, and rollback snapshots before any generated + rule can be promoted. diff --git a/harness/self_report.py b/harness/self_report.py index fe2499b..8fa5a7f 100644 --- a/harness/self_report.py +++ b/harness/self_report.py @@ -151,7 +151,7 @@ def _generate_rule_id(self, category: str) -> str: safe_cat = category.lower().replace(" ", "_").replace("/", "_").replace("-", "_") return mapping.get(category, f"auto_{safe_cat}") - def auto_encode(self): + def auto_encode(self, write: bool = True): """ 自动将未编码的摩擦点加入 video-rules.json @@ -203,7 +203,8 @@ def auto_encode(self): "last_evolution": datetime.now().isoformat(), "evolution_count": self.rules.get("autopoiesis", {}).get("evolution_count", 0) + new_rules_count, } - self._save_json(self.rules_path, self.rules) + if write: + self._save_json(self.rules_path, self.rules) return new_rules_count @@ -405,21 +406,23 @@ def print_report(self): print("\n" + "=" * 60) - def run(self) -> Tuple[Path, Dict]: + def run(self, no_write: bool = False, print_human: bool = True) -> Tuple[Optional[Path], Dict]: """ 完整自检流程： 1. 加载系统状态 2. 自动编码摩擦点 - 3. 写入活记忆 - 4. 生成并保存报告 - 5. 打印报告 + 3. 写入活记忆（no_write=False 时） + 4. 生成并保存报告（no_write=False 时） + 5. 打印报告（print_human=True 时） """ self.load_system_state() - self.auto_encode() - self.write_lessons() + self.auto_encode(write=not no_write) + if not no_write: + self.write_lessons() self.generate_report() - report_path = self.save_report() - self.print_report() + report_path = None if no_write else self.save_report() + if print_human: + self.print_report() return report_path, self.report @@ -431,6 +434,9 @@ def main(): parser.add_argument("--project-dir", default=".", help="项目根目录") parser.add_argument("--capture", nargs=3, metavar=("CATEGORY", "DESC", "RESOLUTION"), help="捕获一个摩擦点：--capture '素材遗漏' 's22缺失' '补充生成'") + parser.add_argument("--no-write", action="store_true", + help="只生成内存报告，不写 LESSONS_LEARNED.md、video-rules.json 或 output/self_report.json") + parser.add_argument("--json", action="store_true", help="输出 machine-readable JSON") args = parser.parse_args() report = SelfReport(project_dir=args.project_dir) @@ -438,7 +444,9 @@ def main(): if args.capture: report.capture_friction(args.capture[0], args.capture[1], args.capture[2]) - report.run() + _, data = report.run(no_write=args.no_write, print_human=not args.json) + if args.json: + print(json.dumps(data, ensure_ascii=False, indent=2)) if __name__ == "__main__": diff --git a/scripts/orchestrator.py b/scripts/orchestrator.py new file mode 100755 index 0000000..46af3f5 --- /dev/null +++ b/scripts/orchestrator.py @@ -0,0 +1,238 @@ +#!/usr/bin/env python3 +"""Governed md2video pipeline orchestrator. + +The current orchestrator provides the stable governance envelope: preflight, +smoke validation, no-write self-reporting, JSONL logs, and a run manifest. It +does not call paid or external generation services unless a future explicit +pipeline command is added. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import subprocess +import sys +import time +from datetime import datetime +from pathlib import Path +from typing import Optional + + +REPO_ROOT = Path(__file__).resolve().parent.parent +VERSION = "0.1.0" + + +def sha256_file(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + + +def command_first_line(command: list[str]) -> Optional[str]: + try: + completed = subprocess.run( + command, + cwd=REPO_ROOT, + capture_output=True, + text=True, + check=False, + timeout=10, + ) + except Exception: + return None + output = (completed.stdout or completed.stderr or "").strip() + return output.splitlines()[0] if output else None + + +def git_value(args: list[str]) -> Optional[str]: + try: + completed = subprocess.run( + ["git", *args], + cwd=REPO_ROOT, + capture_output=True, + text=True, + check=False, + ) + except Exception: + return None + if completed.returncode != 0: + return None + return completed.stdout.strip() or None + + +def artifact_entry(path: Path) -> dict: + exists = path.exists() + entry = { + "path": str(path), + "exists": exists, + } + if exists and path.is_file(): + entry["size_bytes"] = path.stat().st_size + entry["sha256"] = sha256_file(path) + return entry + + +class PipelineLogger: + def __init__(self, log_path: Path): + self.log_path = log_path + self.entries: list[dict] = [] + self.log_path.parent.mkdir(parents=True, exist_ok=True) + self.log_path.write_text("", encoding="utf-8") + + def record(self, step: str, status: str, **meta) -> dict: + entry = { + "t": datetime.now().isoformat(), + "step": step, + "status": status, + **meta, + } + self.entries.append(entry) + with open(self.log_path, "a", encoding="utf-8") as f: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + return entry + + +def run_command(step: str, command: list[str], logger: PipelineLogger) -> int: + started = time.time() + completed = subprocess.run( + command, + cwd=REPO_ROOT, + capture_output=True, + text=True, + check=False, + ) + duration_ms = int((time.time() - started) * 1000) + logger.record( + step, + "success" if completed.returncode == 0 else "failed", + command=command, + exit_code=completed.returncode, + duration_ms=duration_ms, + stdout_preview=(completed.stdout or "")[:1000], + stderr_preview=(completed.stderr or "")[:1000], + ) + return completed.returncode + + +def build_run_manifest( + input_path: Optional[Path], + input_display_path: Optional[Path], + output_dir: Path, + log_path: Path, + mode: str, + steps: list[dict], +) -> dict: + artifacts = { + "segments": artifact_entry(output_dir / "segments.json"), + "prompts": artifact_entry(REPO_ROOT / "prompts.json"), + "timeline": artifact_entry(output_dir / "timeline.json"), + "final": artifact_entry(output_dir / "final.mp4"), + "final_with_cta": artifact_entry(output_dir / "final_with_cta.mp4"), + "frame_report": artifact_entry(output_dir / "frame_checks" / "frame_check_report.json"), + "compliance_report": artifact_entry(output_dir / "compliance_report.json"), + "cta_registry": artifact_entry(REPO_ROOT / "cta_resources.json"), + "pipeline_log": artifact_entry(log_path), + } + + return { + "generator": "md2video.orchestrator", + "version": VERSION, + "created_at": datetime.now().isoformat(), + "mode": mode, + "repo": { + "branch": git_value(["branch", "--show-current"]), + "commit": git_value(["rev-parse", "HEAD"]), + "status_short": git_value(["status", "--short"]), + }, + "environment": { + "python": sys.version.split()[0], + "ffmpeg": command_first_line(["ffmpeg", "-version"]), + "ffprobe": command_first_line(["ffprobe", "-version"]), + }, + "input": { + "path": str(input_display_path) if input_display_path else None, + "resolved_path": str(input_path) if input_path else None, + "exists": bool(input_path and input_path.exists()), + "sha256": sha256_file(input_path) if input_path and input_path.exists() else None, + }, + "output_dir": str(output_dir), + "artifacts": artifacts, + "steps": steps, + } + + +def write_manifest(manifest: dict, output_dir: Path) -> Path: + output_dir.mkdir(parents=True, exist_ok=True) + path = output_dir / "run-manifest.json" + path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") + return path + + +def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="md2video governed pipeline orchestrator") + parser.add_argument("--input", help="Source Markdown/text input") + parser.add_argument("--output-dir", default="output", help="Pipeline output directory") + parser.add_argument("--log", default=".md2video-pipeline.jsonl", help="JSONL pipeline log path") + parser.add_argument("--dry-run", action="store_true", help="Run governance checks without video generation") + parser.add_argument("--skip-command-checks", action="store_true", help="Skip ffmpeg/ffprobe preflight command checks") + parser.add_argument("--allow-dirty-output", action="store_true", help="Do not warn about existing output artifacts") + return parser.parse_args(argv) + + +def main(argv: Optional[list[str]] = None) -> int: + args = parse_args(argv) + output_dir = Path(args.output_dir).resolve() + log_path = Path(args.log).resolve() + input_display_path = Path(args.input) if args.input else None + input_path = input_display_path.resolve() if input_display_path else None + + logger = PipelineLogger(log_path) + mode = "dry-run" if args.dry_run else "governance" + + preflight_cmd = [ + sys.executable, + str(REPO_ROOT / "scripts" / "preflight.py"), + "--output-dir", + str(output_dir), + "--json", + ] + if input_path: + preflight_cmd.extend(["--input", str(input_path)]) + if args.skip_command_checks: + preflight_cmd.append("--skip-command-checks") + if args.allow_dirty_output: + preflight_cmd.append("--allow-dirty-output") + + commands = [ + ("preflight", preflight_cmd), + ("cta_resources", [sys.executable, str(REPO_ROOT / "scripts" / "verify_cta_resources.py")]), + ("smoke_imports", [sys.executable, str(REPO_ROOT / "scripts" / "smoke_imports.py")]), + ("self_report_no_write", [ + sys.executable, + str(REPO_ROOT / "harness" / "self_report.py"), + "--project-dir", + str(REPO_ROOT), + "--no-write", + "--json", + ]), + ] + + exit_code = 0 + for step, command in commands: + code = run_command(step, command, logger) + if code != 0 and exit_code == 0: + exit_code = code + + manifest = build_run_manifest(input_path, input_display_path, output_dir, log_path, mode, logger.entries) + manifest_path = write_manifest(manifest, output_dir) + logger.record("run_manifest", "success", path=str(manifest_path)) + + return exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/preflight.py b/scripts/preflight.py new file mode 100755 index 0000000..dd86795 --- /dev/null +++ b/scripts/preflight.py @@ -0,0 +1,328 @@ +#!/usr/bin/env python3 +"""Preflight checks for md2video pipeline runs.""" + +from __future__ import annotations + +import argparse +import importlib +import json +import shutil +import sys +from pathlib import Path +from typing import Iterable, Optional + + +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_JSON_FILES = [ + "harness/video-rules.json", + "rules/segment_types.json", + "rules/storyboard_rules.json", + "cta_resources.json", +] + + +def result( + check_id: str, + name: str, + level: str, + passed: bool, + detail: str, + block_on_fail: bool, + **extra, +) -> dict: + data = { + "id": check_id, + "name": name, + "level": level, + "passed": passed, + "detail": detail, + "block_on_fail": block_on_fail, + } + data.update(extra) + return data + + +def check_required_commands(commands: Iterable[str]) -> dict: + missing = [cmd for cmd in commands if shutil.which(cmd) is None] + if missing: + return result( + "required_commands", + "Required command availability", + "L1", + False, + "Missing required command(s): " + ", ".join(missing), + True, + missing=missing, + ) + return result( + "required_commands", + "Required command availability", + "L1", + True, + "All required commands are available", + True, + missing=[], + ) + + +def check_json_files(paths: Iterable[str]) -> dict: + errors = [] + for rel_path in paths: + path = REPO_ROOT / rel_path + if not path.exists(): + errors.append(f"{rel_path}: missing") + continue + try: + with open(path, "r", encoding="utf-8") as f: + json.load(f) + except Exception as exc: + errors.append(f"{rel_path}: {exc}") + + if errors: + return result( + "json_validity", + "Governance JSON validity", + "L1", + False, + "; ".join(errors), + True, + errors=errors, + ) + return result( + "json_validity", + "Governance JSON validity", + "L1", + True, + "All governance JSON files parse successfully", + True, + errors=[], + ) + + +def check_animation_routing() -> dict: + sys.path.insert(0, str(REPO_ROOT)) + from extensions.animations.animation_templates import available_animation_types + + rules_path = REPO_ROOT / "rules" / "storyboard_rules.json" + with open(rules_path, "r", encoding="utf-8") as f: + rules = json.load(f) + + supported = set(available_animation_types()) + missing = [] + for segment_type, mapping in rules.get("segment_type_mapping", {}).items(): + if mapping.get("visual_type") != "animation": + continue + animation_type = mapping.get("animation_type") + if animation_type not in supported: + missing.append(f"{segment_type}:{animation_type}") + + if missing: + return result( + "animation_routing", + "Storyboard animation routing", + "L1", + False, + "Unsupported animation type(s): " + ", ".join(missing), + True, + missing=missing, + ) + return result( + "animation_routing", + "Storyboard animation routing", + "L1", + True, + "All storyboard animation types resolve", + True, + missing=[], + ) + + +def check_cta_resources() -> dict: + sys.path.insert(0, str(REPO_ROOT)) + verify = importlib.import_module("scripts.verify_cta_resources") + errors = verify.validate_registry() + if errors: + return result( + "cta_registry", + "CTA resource registry", + "L1", + False, + "; ".join(errors), + True, + errors=errors, + ) + return result( + "cta_registry", + "CTA resource registry", + "L1", + True, + "CTA resource registry is consistent", + True, + errors=[], + ) + + +def check_input_text(input_path: Optional[Path]) -> dict: + if not input_path: + return result( + "tts_text_sanitization", + "TTS text sanitization", + "L2", + True, + "Skipped: no input file supplied", + False, + findings=[], + ) + if not input_path.exists(): + return result( + "tts_text_sanitization", + "TTS text sanitization", + "L2", + False, + f"Input file does not exist: {input_path}", + False, + findings=["missing_input"], + ) + + text = input_path.read_text(encoding="utf-8") + findings = [] + if "---" in text: + findings.append("contains '---', which should be normalized before edge-tts") + if "~" in text: + findings.append("contains '~', which should be normalized before edge-tts") + + return result( + "tts_text_sanitization", + "TTS text sanitization", + "L2", + len(findings) == 0, + "No TTS-sensitive characters found" if not findings else "; ".join(findings), + False, + findings=findings, + ) + + +def check_output_dir(output_dir: Path, allow_dirty_output: bool = False) -> dict: + known_artifacts = [ + "segments.json", + "timeline.json", + "final.mp4", + "final_with_cta.mp4", + "compliance_report.json", + "frame_checks", + "run-manifest.json", + ] + if allow_dirty_output or not output_dir.exists(): + return result( + "output_dir_state", + "Output directory state", + "L2", + True, + "Output directory is clean or explicitly allowed", + False, + stale_artifacts=[], + ) + + stale = [name for name in known_artifacts if (output_dir / name).exists()] + return result( + "output_dir_state", + "Output directory state", + "L2", + len(stale) == 0, + "No known stale artifacts found" if not stale else "Known output artifacts already exist: " + ", ".join(stale), + False, + stale_artifacts=stale, + ) + + +def summarize(checks: list[dict]) -> dict: + l1 = [c for c in checks if c["level"] == "L1"] + l2 = [c for c in checks if c["level"] == "L2"] + return { + "l1": { + "total": len(l1), + "passed": sum(1 for c in l1 if c["passed"]), + "failed": sum(1 for c in l1 if not c["passed"]), + }, + "l2": { + "total": len(l2), + "passed": sum(1 for c in l2 if c["passed"]), + "failed": sum(1 for c in l2 if not c["passed"]), + }, + } + + +def run_preflight( + input_path: Optional[Path] = None, + output_dir: Path = Path("output"), + skip_command_checks: bool = False, + allow_dirty_output: bool = False, +) -> dict: + checks = [] + if skip_command_checks: + checks.append(result( + "required_commands", + "Required command availability", + "L1", + True, + "Skipped by --skip-command-checks", + True, + skipped=True, + missing=[], + )) + else: + checks.append(check_required_commands(["ffmpeg", "ffprobe"])) + + checks.extend([ + check_json_files(DEFAULT_JSON_FILES), + check_animation_routing(), + check_cta_resources(), + check_input_text(input_path), + check_output_dir(output_dir, allow_dirty_output=allow_dirty_output), + ]) + + summary = summarize(checks) + return { + "preflight": "md2video.pipeline_preflight", + "ok": summary["l1"]["failed"] == 0, + "summary": summary, + "checks": checks, + } + + +def format_human(report: dict) -> str: + lines = ["md2video Preflight"] + for check in report["checks"]: + status = "PASS" if check["passed"] else ("FAIL" if check["level"] == "L1" else "WARN") + lines.append(f"[{check['level']}] {status} {check['id']}: {check['detail']}") + lines.append(f"Result: {'PASS' if report['ok'] else 'FAIL'}") + return "\n".join(lines) + + +def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="md2video pipeline preflight checks") + parser.add_argument("--input", help="Source Markdown/text input") + parser.add_argument("--output-dir", default="output", help="Pipeline output directory") + parser.add_argument("--skip-command-checks", action="store_true", help="Skip ffmpeg/ffprobe availability checks") + parser.add_argument("--allow-dirty-output", action="store_true", help="Do not warn about existing output artifacts") + parser.add_argument("--json", action="store_true", help="Print machine-readable JSON") + return parser.parse_args(argv) + + +def main(argv: Optional[list[str]] = None) -> int: + args = parse_args(argv) + report = run_preflight( + input_path=Path(args.input).resolve() if args.input else None, + output_dir=Path(args.output_dir).resolve(), + skip_command_checks=args.skip_command_checks, + allow_dirty_output=args.allow_dirty_output, + ) + if args.json: + print(json.dumps(report, ensure_ascii=False, indent=2)) + else: + print(format_human(report)) + return 0 if report["ok"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/privacy-check.sh b/scripts/privacy-check.sh new file mode 100755 index 0000000..670fc17 --- /dev/null +++ b/scripts/privacy-check.sh @@ -0,0 +1,95 @@ +#!/usr/bin/env bash +# Privacy gate for md2video. +# +# Usage: +# bash scripts/privacy-check.sh [--full] +# +# Without --full, scans staged files when available. With --full, scans tracked +# repository content. The script blocks known token/private-key/credential +# patterns and warns on likely personal identifiers. + +set -euo pipefail + +MODE="${1:-}" + +blocked=0 +warned=0 + +EXCLUDE=( + ':!scripts/privacy-check.sh' + ':!.git/**' + ':!.venv/**' + ':!venv/**' + ':!output/**' + ':!scenes/**' + ':!rebuild_animations/**' + ':!animations/**' +) + +run_grep() { + local pattern="$1" + if [ "$MODE" = "--full" ]; then + git grep -n --color=never -P "$pattern" -- "${EXCLUDE[@]}" 2>/dev/null || true + else + local files + files=$(git diff --cached --name-only --diff-filter=ACMR 2>/dev/null || true) + if [ -n "$files" ]; then + echo "$files" | xargs -r git grep -n --color=never -P "$pattern" -- 2>/dev/null || true + fi + fi +} + +block() { + local desc="$1" + local pattern="$2" + local matches + matches=$(run_grep "$pattern") + if [ -n "$matches" ]; then + echo "[BLOCKED] $desc" + echo "$matches" + echo "" + blocked=$((blocked + $(echo "$matches" | wc -l | tr -d ' '))) + fi +} + +warn() { + local desc="$1" + local pattern="$2" + local matches + matches=$(run_grep "$pattern") + if [ -n "$matches" ]; then + echo "[WARN] $desc" + echo "$matches" + echo "" + warned=$((warned + $(echo "$matches" | wc -l | tr -d ' '))) + fi +} + +echo "md2video Privacy Gate" +echo "" + +block "OpenAI API key" 'sk-(proj-)?[A-Za-z0-9]{20,}' +block "GitHub token" 'gh[pousr]_[A-Za-z0-9_]{30,}' +block "Slack token" 'xox[baprs]-[A-Za-z0-9-]{10,}' +block "AWS access key" 'AKIA[0-9A-Z]{16}' +block "Private key" '-----BEGIN (RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----' +block "JWT token" 'eyJ[A-Za-z0-9_-]+\.eyJ[A-Za-z0-9_-]{10,}' +block "Credential assignment" '(SECRET|TOKEN|PASSWORD|API_KEY|APP_SECRET|ACCESS_KEY)\s*=\s*["'\''"]?[A-Za-z0-9!@#$%^&*()_+\-]{8,}["'\''"]?' +block "Internal IP" '(192\.168\.\d{1,3}\.\d{1,3}|10\.\d{1,3}\.\d{1,3}\.\d{1,3}|172\.(1[6-9]|2\d|3[01])\.\d{1,3}\.\d{1,3})' +block "Local absolute user path in tracked text" '/Users/[A-Za-z0-9._-]+/' + +warn "Personal email" '[A-Za-z0-9._%+-]+@(?!example\.com|test\.com|localhost)[A-Za-z0-9.-]+\.[A-Za-z]{2,}' +warn "CN mobile phone" '1[3-9][0-9]{9}' +warn "WeChat payload-like URL" '(weixin://|wecom://|https?://[^[:space:]"'\''<>]*(mp\.weixin\.qq\.com|work\.weixin\.qq\.com|u\.wechat\.com)[^[:space:]"'\''<>]*)' + +echo "" +if [ "$blocked" -gt 0 ]; then + echo "Privacy gate failed: $blocked blocking finding(s)" + exit 1 +fi + +if [ "$warned" -gt 0 ]; then + echo "Privacy gate passed with $warned warning finding(s)" +else + echo "Privacy gate passed" +fi diff --git a/scripts/smoke_imports.py b/scripts/smoke_imports.py index 3a6bfe2..3c97a8c 100644 --- a/scripts/smoke_imports.py +++ b/scripts/smoke_imports.py @@ -22,6 +22,8 @@ "harness.harness", "harness.memory_loader", "harness.self_report", + "scripts.preflight", + "scripts.orchestrator", ] diff --git a/tests/test_pipeline_governance.py b/tests/test_pipeline_governance.py new file mode 100644 index 0000000..d3894ad --- /dev/null +++ b/tests/test_pipeline_governance.py @@ -0,0 +1,106 @@ +import importlib.util +import json +import os +import tempfile +import unittest +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def load_module(name: str, path: Path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class PipelineGovernanceTests(unittest.TestCase): + def test_preflight_reports_missing_required_command(self): + preflight = load_module("preflight_script", REPO_ROOT / "scripts" / "preflight.py") + + result = preflight.check_required_commands(["definitely-not-md2video-command"]) + + self.assertEqual(result["id"], "required_commands") + self.assertFalse(result["passed"]) + self.assertIn("definitely-not-md2video-command", result["missing"]) + self.assertEqual(result["level"], "L1") + + def test_orchestrator_writes_jsonl_log_and_run_manifest_in_dry_run(self): + orchestrator = load_module("orchestrator_script", REPO_ROOT / "scripts" / "orchestrator.py") + + with tempfile.TemporaryDirectory(prefix="md2video-orchestrator-test-") as tmp: + output_dir = Path(tmp) / "output" + log_path = Path(tmp) / ".md2video-pipeline.jsonl" + article_path = Path(tmp) / "article.md" + article_path.write_text("# Test\n\nA governed dry run.", encoding="utf-8") + + code = orchestrator.main([ + "--input", str(article_path), + "--output-dir", str(output_dir), + "--log", str(log_path), + "--dry-run", + "--skip-command-checks", + ]) + + self.assertEqual(code, 0) + self.assertTrue(log_path.exists()) + log_entries = [ + json.loads(line) + for line in log_path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + self.assertGreaterEqual(len(log_entries), 2) + self.assertTrue(all("step" in entry and "status" in entry for entry in log_entries)) + + manifest_path = output_dir / "run-manifest.json" + self.assertTrue(manifest_path.exists()) + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + self.assertEqual(manifest["mode"], "dry-run") + self.assertEqual(manifest["input"]["path"], str(article_path)) + self.assertEqual(manifest["input"]["sha256"], orchestrator.sha256_file(article_path)) + self.assertIn("steps", manifest) + + def test_self_report_no_write_preserves_governance_files(self): + from harness.self_report import SelfReport + + with tempfile.TemporaryDirectory(prefix="md2video-self-report-test-") as tmp: + project = Path(tmp) + (project / "harness").mkdir() + (project / "docs").mkdir() + (project / "output").mkdir() + rules_path = project / "harness" / "video-rules.json" + lessons_path = project / "docs" / "LESSONS_LEARNED.md" + + rules_before = { + "version": "test", + "l3_render_checks": {}, + "autopoiesis": {"self_report_enabled": True, "evolution_count": 0}, + } + lessons_before = """--- +autopoiesis: true +memory_type: "living" +last_updated: "2026-06-08" +evolution_count: 0 +friction_points: +--- + +# LESSONS +""" + rules_path.write_text(json.dumps(rules_before, ensure_ascii=False, indent=2), encoding="utf-8") + lessons_path.write_text(lessons_before, encoding="utf-8") + + report = SelfReport(project_dir=str(project)) + report.capture_friction("测试", "no-write should not persist", "keep files unchanged") + report_path, data = report.run(no_write=True, print_human=False) + + self.assertIsNone(report_path) + self.assertEqual(json.loads(rules_path.read_text(encoding="utf-8")), rules_before) + self.assertEqual(lessons_path.read_text(encoding="utf-8"), lessons_before) + self.assertFalse((project / "output" / "self_report.json").exists()) + self.assertEqual(data["friction_summary"]["total"], 1) + + +if __name__ == "__main__": + unittest.main()