"""Path-first unified CLI: one file or one directory, routed by suffix.""" from __future__ import annotations import argparse import sys from pathlib import Path from .commands import run_input, run_verify from .logging_utils import default_log_path, setup_run_logger from .runtime import PipelineProvider, RuntimeConfig PROJECT_ROOT = Path(__file__).resolve().parent.parent LEGACY_COMMANDS = {"image", "pdf", "batch"} def _add_device_options(parser: argparse.ArgumentParser, default: str | None) -> None: parser.add_argument("--device", choices=("cpu", "gpu"), default=default or "cpu", help="运行设备") parser.add_argument("--device-id", type=int, default=0, help="GPU 编号") parser.add_argument("--threads", type=int, default=None, help="CPU 线程数") parser.add_argument("--log-file", type=Path, default=None, help="日志文件路径") parser.add_argument("--verbose", action="store_true", help="输出详细日志") def build_input_parser(device_override: str | None = None) -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="ocr.py", description="自动按文件后缀处理图片/PDF;输入目录时批量使用同一路由逻辑", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("input", type=Path, help="图片、PDF 或目录") parser.add_argument("--recursive", action="store_true", help="目录模式递归扫描子目录") parser.add_argument("--output", type=Path, default=PROJECT_ROOT / "outputs", help="PDF 输出根目录") parser.add_argument("--fail-fast", action="store_true", help="单文件失败后立即停止目录任务") # Image options parser.add_argument("--warmup", type=int, default=0, help="首张图片预热轮数") parser.add_argument("--rounds", type=int, default=1, help="每张图片推理轮数") parser.add_argument("--benchmark-json", type=Path, default=None, help="单图片 Benchmark JSON 路径") parser.add_argument("--no-result", action="store_true", help="不记录图片 OCR 文本块") # PDF options parser.add_argument( "--pdf-mode", "--mode", dest="pdf_mode", choices=("hybrid", "text", "ocr"), default="hybrid", help="PDF 处理模式", ) parser.add_argument("--pages", help="PDF 页码范围,例如 1-5,8,10-") parser.add_argument("--dpi", type=int, default=144, help="PDF OCR 页面渲染 DPI") parser.add_argument("--password", help="PDF 密码") parser.add_argument("--resume", action="store_true", help="PDF 断点续传") parser.add_argument("--overwrite", action="store_true", help="覆盖已有 PDF 输出") parser.add_argument("--keep-rendered", action="store_true", help="保留 OCR 页面 PNG") parser.add_argument("--max-new-tokens", type=int, default=None, help="VLM 最大生成 token") parser.add_argument("--min-pixels", type=int, default=None, help="VLM 最小输入像素") parser.add_argument("--max-pixels", type=int, default=None, help="VLM 最大输入像素") parser.add_argument("--text-min-chars", type=int, default=50, help="有效文本最小字符数") parser.add_argument("--text-min-printable-ratio", type=float, default=0.85, help="可打印字符比例阈值") parser.add_argument("--text-min-content-ratio", type=float, default=0.60, help="字母/数字/CJK 比例阈值") parser.add_argument("--text-max-replacement-ratio", type=float, default=0.02, help="替换字符最大比例") parser.add_argument("--text-min-density", type=float, default=25.0, help="文本密度阈值") _add_device_options(parser, device_override) return parser def build_verify_parser(device_override: str | None = None) -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="ocr.py verify", description="验证 CPU/GPU Paddle 环境", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) _add_device_options(parser, device_override) return parser def normalize_argv(argv: list[str]) -> tuple[str, list[str]]: """Keep old image/pdf/batch prefixes as compatibility aliases.""" if argv and argv[0] == "verify": return "verify", argv[1:] if argv and argv[0] in LEGACY_COMMANDS: return "input", argv[1:] return "input", argv def main(device: str | None = None, argv: list[str] | None = None) -> int: raw_argv = list(sys.argv[1:] if argv is None else argv) command, normalized = normalize_argv(raw_argv) parser = build_verify_parser(device) if command == "verify" else build_input_parser(device) args = parser.parse_args(normalized) if device is not None: args.device = device if command == "verify": stem = "verify" category = "verify" else: stem = args.input.stem or args.input.name or "input" category = "input" log_file = args.log_file or default_log_path(PROJECT_ROOT, category, stem, device=args.device) logger = setup_run_logger(f"ocr.{category}.{args.device}", log_file, verbose=args.verbose) logger.info( "COMMAND_STARTED command=%s input=%s device=%s log=%s", command, getattr(args, "input", None), args.device, log_file, ) provider = PipelineProvider( RuntimeConfig(device=args.device, threads=args.threads, device_id=args.device_id), logger, ) try: if command == "verify": return run_verify(args, provider, logger, PROJECT_ROOT) if args.warmup < 0 or args.rounds < 1: raise ValueError("--warmup 必须 >= 0,--rounds 必须 >= 1") return run_input(args, provider, logger, PROJECT_ROOT) except Exception as exc: logger.exception("COMMAND_FAILED command=%s error=%s", command, exc) return 1