#!/usr/bin/env python3 """Run a small manual MVP classification batch. Examples: python scripts/run_small_classification.py --fake python scripts/run_small_classification.py --provider openai --model your-model """ from __future__ import annotations import argparse import csv import os import sys from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[1] SRC_PATH = PROJECT_ROOT / "src" if SRC_PATH.exists() and str(SRC_PATH) not in sys.path: sys.path.insert(0, str(SRC_PATH)) from gcmd_classifier.articles import load_articles # noqa: E402 from gcmd_classifier.config import ModelSettings # noqa: E402 from gcmd_classifier.llm import FakeModelClient # noqa: E402 from gcmd_classifier.llm.base import ModelClient # noqa: E402 from gcmd_classifier.llm.openai_provider import OpenAIModelClient # noqa: E402 from gcmd_classifier.models import ArticleLoadResult, ArticleRecord, ArticleResult # noqa: E402 from gcmd_classifier.persistence import ArticleResultCache, JsonResultStore # noqa: E402 from gcmd_classifier.pipeline import run_batch # noqa: E402 from gcmd_classifier.vocabulary import VocabularyIndex, load_vocabulary # noqa: E402 DEFAULT_HIERARCHY_PATH = PROJECT_ROOT / "data" / "gcmd_hierarchy.json" DEFAULT_ARTICLES_PATH = PROJECT_ROOT / "data" / "articles.json" DEFAULT_OUTPUT_DIR = PROJECT_ROOT / "outputs" / "small_run" def parse_args(argv: list[str] | None = None) -> argparse.Namespace: """Parse command-line arguments for the small manual run.""" parser = argparse.ArgumentParser(description="Run a small GCMD classifier MVP batch.") parser.add_argument("--articles", type=Path, default=DEFAULT_ARTICLES_PATH) parser.add_argument("--limit", type=int, default=10) parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR) parser.add_argument("--provider", type=str, default=None) parser.add_argument("--model", type=str, default=None) parser.add_argument("--fake", action="store_true") return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: """Run the small classification batch and print a concise console summary.""" args = parse_args(argv) if args.limit < 1: raise SystemExit("--limit must be at least 1.") vocabulary = load_vocabulary(DEFAULT_HIERARCHY_PATH) article_load = load_articles(args.articles) limited_load = limit_article_load(article_load, args.limit) settings = build_settings(args) model_client = build_model_client( settings=settings, fake=args.fake, vocabulary=vocabulary, article_count=len(limited_load.articles), ) store = JsonResultStore(args.output_dir) cache = ArticleResultCache(args.output_dir / "cache") batch = run_batch( article_load_result=limited_load, vocabulary=vocabulary, model_client=model_client, settings=settings, store=store, cache=cache, run_id="manual-small-run", relevant_config={"limit": args.limit, "mode": "fake" if args.fake else "live"}, on_article_start=print_article_start, on_article_finish=print_article_finish, ) review_csv_path = write_review_csv(batch.results, args.output_dir) print_console_summary(batch.results) print( "\nSummary: " f"processed={batch.summary.processed_articles}, " f"classified={batch.summary.accepted_classifications}, " f"not_classified={batch.summary.articles_not_classified}, " f"failed={batch.summary.articles_failed}, " f"invalid_source_records={batch.summary.invalid_source_records}, " f"output={store.consolidated_path}, " f"review_csv={review_csv_path}" ) return 0 def build_settings(args: argparse.Namespace) -> ModelSettings: """Build model settings from environment plus CLI overrides.""" settings = ModelSettings.from_environment() updates: dict[str, object] = {} if args.fake: updates["provider"] = "fake" updates["model_name"] = args.model or "fake-model" else: updates["provider"] = args.provider or settings.provider if args.model is not None: updates["model_name"] = args.model return settings.model_copy(update=updates) def build_model_client( *, settings: ModelSettings, fake: bool, vocabulary: VocabularyIndex, article_count: int, ) -> ModelClient: """Create the configured model client for a manual run.""" if fake: return FakeModelClient(fake_scripted_responses(vocabulary, article_count)) if settings.provider != "openai": raise SystemExit("Live mode requires --provider openai, or use --fake for demo mode.") if not os.environ.get(settings.api_key_env_var): raise SystemExit( f"Live mode requires {settings.api_key_env_var} to be set. Use --fake for demo mode." ) return OpenAIModelClient(settings) def limit_article_load(article_load: ArticleLoadResult, limit: int) -> ArticleLoadResult: """Return an ArticleLoadResult containing the first N valid articles.""" return article_load.model_copy(update={"articles": article_load.articles[:limit]}) def fake_scripted_responses(vocabulary: VocabularyIndex, article_count: int) -> list[dict]: """Build deterministic fake responses matching the MVP smoke-test pattern.""" if article_count < 1: return [] topic_position, _, term_position, term = first_term_with_variables(vocabulary) actions = [ topic_response(f"topic_{topic_position:04d}"), term_response(f"term_{term_position:04d}"), ] parent = term while parent.level != "Variable_Level_3": children = vocabulary.variables_for_parent(parent.UUID) if not children: break actions.append(variable_response("variable_0001")) parent = children[0] actions.extend(no_topic_response() for _ in range(max(article_count - 1, 0))) return actions def first_term_with_variables(vocabulary: VocabularyIndex): """Return the first Topic/Term position pair whose Term has Variable children.""" for topic_position, topic in enumerate(vocabulary.topics(), start=1): for term_position, term in enumerate(vocabulary.terms_for_topic(topic.UUID), start=1): if vocabulary.variables_for_parent(term.UUID): return topic_position, topic, term_position, term raise RuntimeError("Vocabulary did not contain a Term with Variable children.") def decision(candidate_id: str) -> dict: """Return one fake structured candidate decision.""" return { "candidate_id": candidate_id, "confidence": 0.9, "evidence": f"Manual fake-run evidence for {candidate_id}.", "support_type": "explicit", "reason": f"Manual fake-run selected {candidate_id}.", } def topic_response(candidate_id: str) -> dict: """Return a fake TopicResponse-shaped payload.""" return {"selected": [decision(candidate_id)]} def term_response(candidate_id: str) -> dict: """Return a fake TermResponse-shaped payload.""" return {"selected": [decision(candidate_id)], "stop_at_parent": False} def variable_response(candidate_id: str) -> dict: """Return a fake VariableResponse-shaped payload.""" return {"selected": [decision(candidate_id)], "stop_at_parent": False} def no_topic_response() -> dict: """Return a fake no-classification TopicResponse-shaped payload.""" return { "selected": [], "no_selection_reason": "Manual fake run did not select a GCMD Topic.", } REVIEW_CSV_FIELDS = ( "DOI", "Title", "Year", "classification_outcome", "processing_status", "level", "UUID", "canonical_path", "topic", "term", "support_type", "confidence_final", "classifier_evidence", "reason_for_stopping", "deterministic_valid", "review_required", "warnings", "no_classification_reason", "errors", ) def write_review_csv(results: tuple[ArticleResult, ...], output_dir: Path) -> Path: """Write a review-friendly CSV beside the consolidated small-run JSON output.""" output_dir.mkdir(parents=True, exist_ok=True) csv_path = output_dir / "review_table.csv" with csv_path.open("w", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=REVIEW_CSV_FIELDS) writer.writeheader() for result in results: for row in review_rows_for_result(result): writer.writerow(row) return csv_path def review_rows_for_result(result: ArticleResult) -> list[dict[str, object]]: """Return one CSV row per accepted classification, or one article-level row.""" outcome = None if result.classification_outcome is None else result.classification_outcome.value common = { "DOI": result.DOI, "Title": result.Title, "Year": result.Year, "classification_outcome": outcome or "", "processing_status": result.processing_status.value, } if result.classifications: rows: list[dict[str, object]] = [] for record in result.classifications: rows.append( { **common, "level": record.level, "UUID": record.UUID, "canonical_path": record.canonical_path, "topic": record.topic, "term": record.term or "", "support_type": "" if record.support_type is None else record.support_type.value, "confidence_final": "" if record.confidence is None or record.confidence.final is None else record.confidence.final, "classifier_evidence": record.classifier_evidence or "", "reason_for_stopping": record.reason_for_stopping or "", "deterministic_valid": record.deterministic_validation.valid, "review_required": record.review_required, "warnings": classification_warnings_text(record), "no_classification_reason": "", "errors": "", } ) return rows return [ { **common, "level": "", "UUID": "", "canonical_path": "", "topic": "", "term": "", "support_type": "", "confidence_final": "", "classifier_evidence": "", "reason_for_stopping": "", "deterministic_valid": "", "review_required": "", "warnings": article_warnings_text(result), "no_classification_reason": result.no_classification_reason or "", "errors": errors_text(result), } ] def classification_warnings_text(record) -> str: """Format structured classification warnings compactly for review CSV output.""" return "; ".join(f"{warning.code}: {warning.message}" for warning in record.warnings) def article_warnings_text(result: ArticleResult) -> str: """Format structured article warnings compactly for review CSV output.""" return "; ".join(f"{warning.code}: {warning.message}" for warning in result.warnings) def errors_text(result: ArticleResult) -> str: """Format structured article errors compactly for review CSV output.""" return "; ".join(f"{error.code}: {error.message}" for error in result.errors) def print_article_start(index: int, total: int, article: ArticleRecord) -> None: """Print safe progress before one article starts.""" print( f"Starting article {index + 1}/{total}: " f"DOI={article.DOI} title={truncate_title(article.Title)!r}", flush=True, ) def print_article_finish( index: int, total: int, result: ArticleResult, elapsed_seconds: float, ) -> None: """Print safe progress after one article finishes.""" outcome = None if result.classification_outcome is None else result.classification_outcome.value print( f"Finished article {index + 1}/{total}: " f"processing_status={result.processing_status.value} " f"classification_outcome={outcome} " f"accepted_classifications={len(result.classifications)} " f"elapsed_seconds={elapsed_seconds:.2f}", flush=True, ) def truncate_title(title: str, max_length: int = 80) -> str: """Return a one-line title truncated for safe progress output.""" compact = one_line(title) if len(compact) <= max_length: return compact return f"{compact[: max_length - 3]}..." def print_console_summary(results: tuple[ArticleResult, ...]) -> None: """Print DOI, title, status, outcome, and final classification details.""" print("DOI\tTitle\tprocessing_status\tclassification_outcome\tlevel\tUUID\tcanonical_path") for result in results: outcome = ( None if result.classification_outcome is None else result.classification_outcome.value ) if result.classifications: for record in result.classifications: print( "\t".join( [ result.DOI, one_line(result.Title), result.processing_status.value, str(outcome), record.level, record.UUID, record.canonical_path, ] ) ) else: print( "\t".join( [ result.DOI, one_line(result.Title), result.processing_status.value, str(outcome), "", "", result.no_classification_reason or "", ] ) ) print_messages(result) def print_messages(result: ArticleResult) -> None: """Print structured warnings and errors for one article.""" for warning in result.warnings: print(f" WARNING {warning.code}: {warning.message}") for error in result.errors: print(f" ERROR {error.code}: {error.message}") def one_line(value: str) -> str: """Keep console rows tab-safe and compact.""" return " ".join(value.split()) if __name__ == "__main__": raise SystemExit(main())