"""Rapport des écarts entre catalogue et inventaire agrégé.""" import csv from pathlib import Path from typing import Dict, Iterable, List from lib.filesystem import ensure_parent_dir def load_sets(sets_path: Path) -> List[dict]: """Charge les sets filtrés pour l'analyse.""" with sets_path.open() as sets_file: reader = csv.DictReader(sets_file) return list(reader) def index_sets_by_num(sets: Iterable[dict]) -> Dict[str, dict]: """Crée un index des sets par numéro complet.""" return {row["set_num"]: row for row in sets} def compute_inventory_totals(parts_path: Path, include_spares: bool) -> Dict[str, int]: """Calcule le total de pièces par set, avec ou sans rechanges.""" totals: Dict[str, int] = {} with parts_path.open() as parts_file: reader = csv.DictReader(parts_file) for row in reader: if not include_spares and row["is_spare"] == "true": continue set_num = row["set_num"] totals[set_num] = totals.get(set_num, 0) + int(row["quantity_in_set"]) return totals def compute_inventory_gaps(sets_path: Path, parts_path: Path) -> List[dict]: """Liste les sets dont le total de pièces diffère du catalogue.""" sets = load_sets(sets_path) totals_with_spares = compute_inventory_totals(parts_path, include_spares=True) totals_without_spares = compute_inventory_totals(parts_path, include_spares=False) gaps: List[dict] = [] for set_row in sets: expected_parts = int(set_row["num_parts"]) inventory_parts_with_spares = totals_with_spares[set_row["set_num"]] inventory_parts_non_spare = totals_without_spares[set_row["set_num"]] if expected_parts != inventory_parts_with_spares: gaps.append( { "set_num": set_row["set_num"], "set_id": set_row["set_id"], "expected_parts": expected_parts, "inventory_parts": inventory_parts_with_spares, "inventory_parts_non_spare": inventory_parts_non_spare, "delta": abs(expected_parts - inventory_parts_with_spares), "delta_non_spare": abs(expected_parts - inventory_parts_non_spare), "in_collection": set_row["in_collection"], } ) return gaps def write_inventory_gaps_csv(destination_path: Path, gaps: Iterable[dict]) -> None: """Écrit un CSV listant les sets en écart d'inventaire.""" ensure_parent_dir(destination_path) with destination_path.open("w", newline="") as csv_file: fieldnames = [ "set_num", "set_id", "expected_parts", "inventory_parts", "inventory_parts_non_spare", "delta", "delta_non_spare", "in_collection", ] writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writeheader() for row in gaps: writer.writerow(row) def build_instructions_url(set_id: str) -> str: """Construit un lien direct vers la page d'instructions LEGO du set.""" return f"https://www.lego.com/service/buildinginstructions/{set_id}" def write_inventory_gaps_markdown( destination_path: Path, gaps: Iterable[dict], sets_by_num: Dict[str, dict], ) -> None: """Génère un tableau Markdown listant les sets en écart d'inventaire.""" ensure_parent_dir(destination_path) with destination_path.open("w") as markdown_file: markdown_file.write( "| set_id | name | year | delta (spares inclus) | delta (spares exclus) | expected_parts | inventory_parts | inventory_parts_non_spare | in_collection | instructions |\n" ) markdown_file.write("| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |\n") for row in gaps: if row["delta_non_spare"] == 0: continue set_row = sets_by_num[row["set_num"]] set_link = f"[{row['set_id']}]({set_row['rebrickable_url']})" instructions_link = f"[PDF]({build_instructions_url(row['set_id'])})" markdown_file.write( f"| {set_link} | {set_row['name']} | {set_row['year']} | {row['delta']} | {row['delta_non_spare']} | " f"{row['expected_parts']} | {row['inventory_parts']} | {row['inventory_parts_non_spare']} | " f"{row['in_collection']} | {instructions_link} |\n" )