Files
deep_research/scripts/dr.py

239 lines
7.8 KiB
Python

#!/usr/bin/env python3
"""Platform-neutral Deep Research CLI for Codex and other adapters.
This CLI intentionally keeps deterministic orchestration in Python while
allowing Codex/OpenCode/Gemini/Claude Code to provide the agentic layer.
"""
from __future__ import annotations
import argparse
import json
import re
import subprocess
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
PROJECTS_DIR = REPO_ROOT / "projects"
CODEX_COMMANDS_DIR = REPO_ROOT / ".codex" / "commands"
CODEX_COMMAND_TEMPLATES_DIR = REPO_ROOT / "codex_adapter_templates" / "codex" / "commands"
def resolve_project(project: str | None) -> Path:
if project:
p = Path(project)
if p.is_dir():
return p.resolve()
cand = PROJECTS_DIR / project
if cand.is_dir():
return cand.resolve()
raise SystemExit(f"project not found: {project}")
manifests = sorted(
PROJECTS_DIR.glob("*/manifest.json"),
key=lambda p: p.stat().st_mtime,
reverse=True,
)
if not manifests:
raise SystemExit("no projects found")
return manifests[0].parent.resolve()
def load_manifest(project_root: Path) -> dict:
path = project_root / "manifest.json"
if not path.exists():
raise SystemExit(f"manifest not found: {path}")
return json.loads(path.read_text(encoding="utf-8"))
def count_words(text: str) -> int:
return len(re.findall(r"[A-Za-z]+(?:[-'][A-Za-z]+)*", text))
def count_chinese_chars(text: str) -> int:
return sum(1 for c in text if "\u4e00" <= c <= "\u9fff")
def file_state(path: Path) -> str:
return "yes" if path.exists() else "no"
def run_cmd(cmd: list[str], *, dry_run: bool) -> int:
printable = " ".join(cmd)
print(f"$ {printable}")
if dry_run:
return 0
return subprocess.run(cmd, cwd=REPO_ROOT, check=False).returncode
def cmd_status(args: argparse.Namespace) -> int:
project_root = resolve_project(args.project)
manifest = load_manifest(project_root)
slug = project_root.name
drafts = sorted((project_root / "phase2" / "drafts").glob("ch*.md"))
evidence = sorted((project_root / "phase2" / "evidence").glob("ch*-evidence.md"))
sources = project_root / "phase2" / "sources.jsonl"
final_en = project_root / "phase4" / "final_en.md"
final_zh = project_root / "phase4" / "final_zh.md"
final_zh_polished = project_root / "phase4" / "final_zh_polished.md"
glossary = project_root / "phase4" / "glossary.json"
en_words = count_words(final_en.read_text(encoding="utf-8")) if final_en.exists() else 0
zh_chars = count_chinese_chars(final_zh_polished.read_text(encoding="utf-8")) if final_zh_polished.exists() else 0
source_count = 0
if sources.exists():
source_count = sum(1 for line in sources.read_text(encoding="utf-8").splitlines() if line.strip())
print(f"Project: {manifest.get('topic', slug)}")
print(f"Slug: {slug}")
print(f"Title: {manifest.get('report_title', '(unset)')}")
print(f"Type: {manifest.get('type', '(unset)')}")
print()
print("Phases:")
for phase in ("phase1", "phase2", "phase3", "phase4"):
p = manifest.get(phase, {})
print(f" {phase}: {p.get('status', 'pending')} approved={p.get('approved', False)}")
print()
print("Artifacts:")
print(f" framework: {file_state(project_root / 'phase1' / 'framework.md')}")
print(f" drafts: {len(drafts)}")
print(f" evidence files: {len(evidence)}")
print(f" sources: {source_count}")
print(f" final_en.md: {file_state(final_en)} ({en_words:,} words)")
print(f" final_zh.md: {file_state(final_zh)}")
print(f" final_zh_polished.md: {file_state(final_zh_polished)} ({zh_chars:,} Chinese chars)")
print(f" glossary.json: {file_state(glossary)}")
return 0
def cmd_prompt(args: argparse.Namespace) -> int:
name = args.command
if not name.startswith("dr-"):
name = f"dr-{name}"
path = CODEX_COMMANDS_DIR / f"{name}.md"
if not path.exists():
fallback = CODEX_COMMAND_TEMPLATES_DIR / f"{name}.md"
if fallback.exists():
path = fallback
else:
raise SystemExit(f"Codex command template not found: {path}")
text = path.read_text(encoding="utf-8")
if args.argument:
text = text.replace("$ARGUMENTS", args.argument)
else:
text = text.replace("$ARGUMENTS", "")
print(text)
return 0
def cmd_glossary(args: argparse.Namespace) -> int:
project_root = resolve_project(args.project)
cmd = [
sys.executable,
str(REPO_ROOT / "scripts" / "build_glossary.py"),
str(project_root),
"--workers",
str(args.workers),
]
if args.force:
cmd.append("--force")
if args.only:
cmd += ["--only", args.only]
if args.input:
cmd += ["--input", args.input]
if args.output:
cmd += ["--output", args.output]
return run_cmd(cmd, dry_run=args.dry_run)
def cmd_finalize(args: argparse.Namespace) -> int:
project_root = resolve_project(args.project)
steps = [
[
sys.executable,
str(REPO_ROOT / "scripts" / "translate.py"),
str(project_root),
"--workers",
str(args.translate_workers),
],
[
sys.executable,
str(REPO_ROOT / "scripts" / "build_glossary.py"),
str(project_root),
"--workers",
str(args.glossary_workers),
],
[
sys.executable,
str(REPO_ROOT / "scripts" / "apply_glossary.py"),
str(project_root),
"--input",
"phase4/final_zh.md",
],
[
sys.executable,
str(REPO_ROOT / "scripts" / "polish.py"),
str(project_root),
"--workers",
str(args.polish_workers),
],
[
sys.executable,
str(REPO_ROOT / "scripts" / "build_report.py"),
str(project_root),
],
]
for step in steps:
rc = run_cmd(step, dry_run=args.dry_run)
if rc != 0:
return rc
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Deep Research platform-neutral CLI")
sub = parser.add_subparsers(dest="cmd", required=True)
status = sub.add_parser("status", help="Show project status")
status.add_argument("project", nargs="?", help="Project slug or path")
status.set_defaults(func=cmd_status)
prompt = sub.add_parser("prompt", help="Print a Codex command prompt template")
prompt.add_argument("command", help="Command name, e.g. dr-frame or frame")
prompt.add_argument("argument", nargs="?", help="Replacement for $ARGUMENTS")
prompt.set_defaults(func=cmd_prompt)
glossary = sub.add_parser("glossary", help="Run glossary verification")
glossary.add_argument("project", help="Project slug or path")
glossary.add_argument("--workers", type=int, default=4)
glossary.add_argument("--force", action="store_true")
glossary.add_argument("--only")
glossary.add_argument("--input")
glossary.add_argument("--output")
glossary.add_argument("--dry-run", action="store_true")
glossary.set_defaults(func=cmd_glossary)
finalize = sub.add_parser("finalize", help="Run Phase 4 deterministic pipeline")
finalize.add_argument("project", help="Project slug or path")
finalize.add_argument("--translate-workers", type=int, default=4)
finalize.add_argument("--glossary-workers", type=int, default=4)
finalize.add_argument("--polish-workers", type=int, default=4)
finalize.add_argument("--dry-run", action="store_true")
finalize.set_defaults(func=cmd_finalize)
return parser
def main() -> int:
parser = build_parser()
args = parser.parse_args()
return args.func(args)
if __name__ == "__main__":
raise SystemExit(main())