from __future__ import annotations import argparse import json from .config import ProjectPaths, TrainingConfig from .console import console def _print_summary(summary: dict, *, as_json: bool = False, title: str = "Summary") -> None: if as_json: print(json.dumps(summary, indent=2, sort_keys=True)) return console.summary(title, summary) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="AniRec backend pipeline") parser.add_argument("--json", action="store_true", help="Print final summary as JSON") subparsers = parser.add_subparsers(dest="command", required=True) scrape_parser = subparsers.add_parser("scrape-usernames", help="Scrape MAL public usernames") scrape_parser.add_argument("--pages", type=int, default=123) scrape_parser.add_argument("--delay", type=float, default=1.0) scrape_parser.add_argument("--max-usernames", type=int, default=5000) fetch_users_parser = subparsers.add_parser( "fetch-user-interactions", help="Fetch MAL anime lists for scraped usernames", ) fetch_users_parser.add_argument("--max-new-public-users", type=int) fetch_users_parser.add_argument("--sleep", type=float, default=0.0) catalog_parser = subparsers.add_parser( "refresh-anime-catalog", help="Refresh catalog metadata and pull in recent seasonal anime", ) catalog_parser.add_argument("--years-back", type=int, default=2) catalog_parser.add_argument("--sleep", type=float, default=0.0) catalog_parser.add_argument("--refresh-existing", action="store_true") relationships_parser = subparsers.add_parser( "refresh-relationships", help="Populate sequel/season relationship labels", ) relationships_parser.add_argument("--sleep", type=float, default=0.0) features_parser = subparsers.add_parser( "build-user-genre-features", help="Derive user genre features from interactions and the anime catalog", ) train_parser = subparsers.add_parser("train", help="Train the LightFM model") train_parser.add_argument("--min-user-interactions", type=int, default=5) train_parser.add_argument("--min-item-interactions", type=int, default=3) train_parser.add_argument("--test-fraction", type=float, default=0.1) train_parser.add_argument("--components", type=int, default=30) train_parser.add_argument("--learning-rate", type=float, default=0.05) train_parser.add_argument("--loss", default="warp") train_parser.add_argument("--epochs", type=int, default=10) train_parser.add_argument("--threads", type=int, default=1) atlas_parser = subparsers.add_parser("build-atlas", help="Generate atlas coordinates") atlas_parser.add_argument("--perplexity", type=int, default=30) refresh_all_parser = subparsers.add_parser( "refresh-all", help="Run the refresh pipeline: scrape, fetch, catalog, relationships, train, atlas", ) refresh_all_parser.add_argument("--pages", type=int, default=300) refresh_all_parser.add_argument("--years-back", type=int, default=2) refresh_all_parser.add_argument("--scrape-delay", type=float, default=1.0) refresh_all_parser.add_argument("--api-sleep", type=float, default=1.0) refresh_all_parser.add_argument("--refresh-existing-catalog", action="store_true") refresh_all_parser.add_argument("--max-usernames", type=int, default=5000) refresh_all_parser.add_argument("--start-step", type=int, default=1, help="Step number to start from (skip earlier steps)") serve_parser = subparsers.add_parser("serve", help="Run the FastAPI server") return parser def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) paths = ProjectPaths() json_output = bool(args.json) if args.command == "scrape-usernames": from .pipeline import scrape_usernames with console.step("Run scrape-usernames"): summary = scrape_usernames(paths, num_pages=args.pages, delay_seconds=args.delay, max_usernames=args.max_usernames) _print_summary(summary, as_json=json_output, title="scrape-usernames") return 0 if args.command == "fetch-user-interactions": from .pipeline import fetch_user_interactions with console.step("Run fetch-user-interactions"): summary = fetch_user_interactions( paths, refresh_existing_users=True, max_new_public_users=args.max_new_public_users, sleep_seconds=args.sleep, ) _print_summary(summary, as_json=json_output, title="fetch-user-interactions") return 0 if args.command == "refresh-anime-catalog": from .catalog import refresh_anime_catalog with console.step("Run refresh-anime-catalog"): summary = refresh_anime_catalog( paths, years_back=args.years_back, sleep_seconds=args.sleep, refresh_existing=args.refresh_existing, ) _print_summary(summary, as_json=json_output, title="refresh-anime-catalog") return 0 if args.command == "refresh-relationships": from .pipeline import refresh_relationships with console.step("Run refresh-relationships"): summary = refresh_relationships(paths, sleep_seconds=args.sleep) _print_summary(summary, as_json=json_output, title="refresh-relationships") return 0 if args.command == "build-user-genre-features": from .pipeline import build_user_genre_features with console.step("Run build-user-genre-features"): summary = build_user_genre_features(paths) _print_summary(summary, as_json=json_output, title="build-user-genre-features") return 0 if args.command == "train": from .training import train_model training_config = TrainingConfig( min_interactions_per_user=args.min_user_interactions, min_interactions_per_item=args.min_item_interactions, test_set_fraction=args.test_fraction, n_components=args.components, learning_rate=args.learning_rate, loss_function=args.loss, epochs=args.epochs, threads=args.threads, ) with console.step("Run train"): summary = train_model(paths, training_config) _print_summary(summary, as_json=json_output, title="train") return 0 if args.command == "build-atlas": from .atlas import generate_atlas with console.step("Run build-atlas"): summary = generate_atlas(paths, perplexity=args.perplexity) _print_summary(summary, as_json=json_output, title="build-atlas") return 0 if args.command == "refresh-all": from .pipeline import refresh_all console.banner( "AniRec Refresh Pipeline", f"Command: refresh-all | years-back={args.years_back} | pages={args.pages}", ) summary = refresh_all( paths, username_pages=args.pages, years_back=args.years_back, scrape_delay_seconds=args.scrape_delay, api_delay_seconds=args.api_sleep, refresh_existing_catalog=args.refresh_existing_catalog, max_usernames=args.max_usernames, start_step=args.start_step, ) _print_summary(summary, as_json=json_output, title="refresh-all") return 0 if args.command == "serve": from .api import run run() return 0 parser.error(f"Unsupported command: {args.command}") return 1 if __name__ == "__main__": raise SystemExit(main())