#!/usr/bin/env python3 """Dental dictionary management CLI — single entry point. Subcommands: add-terms Import terms from JSON seed files (seeds/) -> workflow_terms add-images Map terms to images from local/remote sources -> workflow_images verify Re-check that mapped images fit their terms -> workflow_verify stats Print per-category term / image-coverage report Run `manage.py --help` for options. See docs/howto.md. """ import argparse import os import sqlite3 import sys from datetime import datetime from config import DB_PATH, connect_db import workflow_images import workflow_terms import workflow_verify class _Tee: """Write to multiple streams; used to mirror stdout/stderr to a log file.""" def __init__(self, *streams): self.streams = streams def write(self, data): for s in self.streams: try: s.write(data) s.flush() except Exception: pass def flush(self): for s in self.streams: try: s.flush() except Exception: pass def isatty(self): return False def _open_log(path): """Open `path` and tee stdout+stderr to it. Returns the file handle.""" os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True) f = open(path, "w", encoding="utf-8", buffering=1) sys.stdout = _Tee(sys.__stdout__, f) sys.stderr = _Tee(sys.__stderr__, f) print(f"# log: {path} started {datetime.now().isoformat(timespec='seconds')}", flush=True) return f def cmd_add_terms(args): workflow_terms.run(seed_files=args.files, all_seeds=args.all) def cmd_add_images(args): workflow_images.run( source=args.source, verify=not args.no_verify, max_wikimedia=args.max_wikimedia, wikimedia_category=args.wikimedia_category, wikimedia_model=args.wikimedia_model, delay=args.delay, max_openi=args.max_openi, openi_delay=args.openi_delay, openi_category=args.openi_category, openi_model=args.openi_model, max_figshare=args.max_figshare, figshare_category=args.figshare_category, figshare_model=args.figshare_model, figshare_delay=args.figshare_delay, max_pmc=args.max_pmc, pmc_category=args.pmc_category, pmc_model=args.pmc_model, pmc_delay=args.pmc_delay, max_vcu=args.max_vcu, vcu_category=args.vcu_category, vcu_model=args.vcu_model, vcu_delay=args.vcu_delay, max_oralsd=args.max_oralsd, oralsd_category=args.oralsd_category, oralsd_model=args.oralsd_model, oralsd_delay=args.oralsd_delay, max_akudental=args.max_akudental, akudental_category=args.akudental_category, akudental_model=args.akudental_model, akudental_delay=args.akudental_delay, max_code=args.max_code, code_category=args.code_category, code_llm_match=args.code_llm_match, max_stock=args.max_stock, stock_category=args.stock_category, stock_model=args.stock_model, stock_delay=args.stock_delay, ) def cmd_verify(args): workflow_verify.verify_db( category=args.category, limit=args.limit, fix=args.fix, model=args.model, ) def cmd_localize(args): workflow_images.bulk_localize( limit=args.limit, category=args.category, dry_run=args.dry_run ) def cmd_stats(args): conn = connect_db() total, imgs = conn.execute( "SELECT COUNT(*), SUM(CASE WHEN image_url IS NOT NULL AND image_url != '' " "THEN 1 ELSE 0 END) FROM terms" ).fetchone() print(f"Terms: {total} Images: {imgs} ({100.0 * imgs / total:.1f}%)\n") print(f"{'category':<16} {'terms':>6} {'images':>7} {'cov':>7}") print("-" * 40) for cat, t, n, pct in conn.execute(""" SELECT category AS cat, COUNT(*) AS t, SUM(CASE WHEN image_url IS NOT NULL AND image_url != '' THEN 1 ELSE 0 END) AS n, ROUND(100.0 * SUM(CASE WHEN image_url IS NOT NULL AND image_url != '' THEN 1 ELSE 0 END) / COUNT(*), 1) AS pct FROM terms GROUP BY cat ORDER BY pct ASC, t DESC """): print(f"{cat:<16} {t:>6} {n:>7} {pct:>6}%") conn.close() def build_parser(): p = argparse.ArgumentParser(prog="manage.py", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) sub = p.add_subparsers(dest="command", required=True) # add-terms t = sub.add_parser("add-terms", help="Import terms from JSON seed files") t.add_argument("files", nargs="*", help="Seed file names (relative to seeds/) or paths") t.add_argument("--all", action="store_true", help="Import every *.json in seeds/") t.set_defaults(func=cmd_add_terms) # add-images i = sub.add_parser("add-images", help="Map terms to images from a source") i.add_argument("--source", default="all", help="Image source: all, 단일 소스, 또는 쉼표 구분 복수 소스 (default: all)") i.add_argument("--no-verify", action="store_true", help="Skip vision-model verification before saving " "(faster bulk runs; openi/wikimedia always verify)") i.add_argument("--max-wikimedia", type=int, default=50) i.add_argument("--wikimedia-category", default=None) i.add_argument("--wikimedia-model", default=workflow_verify.VERIFY_MODEL, help="Wikimedia 이미지 검증 모델 (default: VERIFY_MODEL)") i.add_argument("--delay", type=int, default=5, help="Delay between Wikimedia requests (s)") i.add_argument("--max-openi", type=int, default=50, help="Open-i 한 회 처리 term 수 (default 50)") i.add_argument("--openi-delay", type=float, default=0.5, help="Open-i API 요청 간 지연 (s)") i.add_argument("--openi-category", default=None, help="Open-i 카테고리 필터 (예: 구강병리)") i.add_argument("--openi-model", default=workflow_verify.VERIFY_MODEL, help="Open-i 이미지 검증 모델 (default: VERIFY_MODEL)") # Figshare i.add_argument("--max-figshare", type=int, default=50) i.add_argument("--figshare-category", default=None) i.add_argument("--figshare-model", default=workflow_verify.VERIFY_MODEL) i.add_argument("--figshare-delay", type=float, default=1.0, help="Figshare API 요청 간 지연 (s)") # PMC Entrez i.add_argument("--max-pmc", type=int, default=50) i.add_argument("--pmc-category", default=None) i.add_argument("--pmc-model", default=workflow_verify.VERIFY_MODEL) i.add_argument("--pmc-delay", type=float, default=1.0, help="PMC Entrez 요청 간 지연 (s)") # VCU Oral Pathology Review i.add_argument("--max-vcu", type=int, default=60) i.add_argument("--vcu-category", default=None) i.add_argument("--vcu-model", default=workflow_verify.VERIFY_MODEL) i.add_argument("--vcu-delay", type=float, default=1.0, help="VCU 요청 간 지연 (s)") # OralSDv1 (GitHub) i.add_argument("--max-oralsd", type=int, default=50) i.add_argument("--oralsd-category", default=None) i.add_argument("--oralsd-model", default=workflow_verify.VERIFY_MODEL) i.add_argument("--oralsd-delay", type=float, default=0.5, help="OralSDv1 요청 간 지연 (s)") # AKUDENTAL 파노라마 (GitHub) i.add_argument("--max-akudental", type=int, default=20) i.add_argument("--akudental-category", default=None) i.add_argument("--akudental-model", default=workflow_verify.VERIFY_MODEL) i.add_argument("--akudental-delay", type=float, default=0.5, help="AKUDENTAL 요청 간 지연 (s)") # COde (로컬 다운로드 데이터셋) i.add_argument("--max-code", type=int, default=None, help="COde 최대 매핑 term 수 (default=전체)") i.add_argument("--code-category", default=None, help="COde 카테고리 필터 (해당 카테고리 term만 매핑)") i.add_argument("--code-llm-match", action="store_true", default=False, help="COde: 문자열 매칭 실패 term을 LLM 의미론적 라벨 매칭으로 보완") # Stock Image APIs (Pixabay CC0 / Pexels / Unsplash) i.add_argument("--max-stock", type=int, default=50, help="Stock 최대 매핑 term 수 (default=50)") i.add_argument("--stock-category", default=None, help="Stock 카테고리 필터") i.add_argument("--stock-model", default=workflow_verify.VERIFY_MODEL) i.add_argument("--stock-delay", type=float, default=0.5, help="Stock API 요청 간 지연 (s)") i.add_argument("--log", default=None, help="stdout/stderr 를 이 파일에도 동시 기록 (예: scratch/run.log)") i.set_defaults(func=cmd_add_images) # verify v = sub.add_parser("verify", help="Re-check that mapped images fit their terms") v.add_argument("--category", default=None, help="Limit to one category") v.add_argument("--limit", type=int, default=None, help="Max terms to check") v.add_argument("--fix", action="store_true", help="Clear bad mappings (image_url -> NULL)") v.add_argument("--model", default=workflow_verify.VERIFY_MODEL) v.add_argument("--log", default=None, help="stdout/stderr 를 이 파일에도 동시 기록") v.set_defaults(func=cmd_verify) # localize lo = sub.add_parser("localize", help="Download http:// image_urls to local remote/ dir") lo.add_argument("--limit", type=int, default=None, help="최대 처리 수") lo.add_argument("--category", default=None, help="카테고리 필터") lo.add_argument("--dry-run", action="store_true", help="실제 다운로드 없이 목록만 출력") lo.set_defaults(func=cmd_localize) # stats s = sub.add_parser("stats", help="Per-category coverage report") s.set_defaults(func=cmd_stats) return p def main(argv=None): args = build_parser().parse_args(argv) log_fh = _open_log(args.log) if getattr(args, "log", None) else None try: args.func(args) finally: if log_fh: print(f"# log end {datetime.now().isoformat(timespec='seconds')}", flush=True) log_fh.close() if __name__ == "__main__": sys.exit(main())