Files
deep_research/scripts/dr.py
T

323 lines
11 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
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from scripts.lib.model_config import (
ModelConfigError,
list_model_profiles,
parse_model_overrides,
resolve_model_profile,
)
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)
try:
resolved = resolve_model_profile(
profile=args.model_profile,
overrides=parse_model_overrides(args.model_override),
)
except ModelConfigError as exc:
raise SystemExit(f"model profile resolution failed: {exc}") from exc
roles = resolved["roles"]
cmd = [
sys.executable,
str(REPO_ROOT / "scripts" / "phase4_pipeline.py"),
str(project_root),
"--translate-workers",
str(args.translate_workers),
"--glossary-workers",
str(args.glossary_workers),
"--polish-workers",
str(args.polish_workers),
"--translate-model",
roles.get("translate", "anthropic/claude-sonnet-4.6"),
"--glossary-model",
roles.get("glossary", "anthropic/claude-haiku-4.5"),
"--polish-model",
roles.get("polish", "anthropic/claude-sonnet-4.6"),
"--glossary-mode",
args.glossary_mode,
]
if args.dry_run:
cmd.append("--dry-run")
return run_cmd(cmd, dry_run=False)
def cmd_models(args: argparse.Namespace) -> int:
if args.list:
for name in list_model_profiles():
print(name)
return 0
try:
resolved = resolve_model_profile(
profile=args.profile,
overrides=parse_model_overrides(args.model_override),
)
except ModelConfigError as exc:
raise SystemExit(f"model profile resolution failed: {exc}") from exc
if args.json:
print(json.dumps(resolved, ensure_ascii=False, indent=2))
return 0
print(f"Profile: {resolved['profile']}")
if resolved["description"]:
print(f"Description: {resolved['description']}")
print("Roles:")
for role in sorted(resolved["roles"]):
print(f" {role}: {resolved['roles'][role]}")
return 0
def cmd_apply_models(args: argparse.Namespace) -> int:
cmd = [
sys.executable,
str(REPO_ROOT / "scripts" / "apply_model_profile.py"),
"--profile",
args.profile,
"--target",
args.target,
]
for item in args.model_override:
cmd += ["--model-override", item]
if args.dry_run:
cmd.append("--dry-run")
return run_cmd(cmd, dry_run=False)
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=0)
finalize.add_argument("--glossary-workers", type=int, default=4)
finalize.add_argument("--polish-workers", type=int, default=0)
finalize.add_argument(
"--glossary-mode",
choices=["off", "low-confidence", "full"],
default="low-confidence",
)
finalize.add_argument("--model-profile", help="Model profile name from configs/models.yaml")
finalize.add_argument(
"--model-override",
action="append",
default=[],
metavar="ROLE=MODEL",
help="Override one role model, repeatable",
)
finalize.add_argument("--dry-run", action="store_true")
finalize.set_defaults(func=cmd_finalize)
models = sub.add_parser("models", help="Resolve and print model profile")
models.add_argument("--profile", help="Profile name from configs/models.yaml")
models.add_argument("--list", action="store_true", help="List available profiles")
models.add_argument(
"--model-override",
action="append",
default=[],
metavar="ROLE=MODEL",
help="Override one role model, repeatable",
)
models.add_argument("--json", action="store_true", help="Emit JSON")
models.set_defaults(func=cmd_models)
apply_models = sub.add_parser("apply-models", help="Apply profile to agent files")
apply_models.add_argument("--profile", required=True, help="Profile name from configs/models.yaml")
apply_models.add_argument("--target", choices=["opencode", "codex", "both"], default="both")
apply_models.add_argument(
"--model-override",
action="append",
default=[],
metavar="ROLE=MODEL",
help="Override one role model, repeatable",
)
apply_models.add_argument("--dry-run", action="store_true")
apply_models.set_defaults(func=cmd_apply_models)
return parser
def main() -> int:
parser = build_parser()
args = parser.parse_args()
return args.func(args)
if __name__ == "__main__":
raise SystemExit(main())