Ajoute l’étape 26 pièces/minifigs

lib/rebrickable/minifig_parts_correlation.py

@@ -0,0 +1,96 @@
+"""Prépare les données de corrélation pièces/minifigs par set."""
+
+import csv
+from pathlib import Path
+from typing import Dict, List, Sequence
+
+from lib.filesystem import ensure_parent_dir
+from lib.rebrickable.parts_inventory import index_inventory_minifigs_by_inventory, select_latest_inventories
+from lib.rebrickable.stats import read_rows
+
+
+def load_minifig_counts_by_set(path: Path) -> Dict[str, int]:
+    """Indexe le nombre de minifigs par set filtré."""
+    lookup: Dict[str, int] = {}
+    with path.open() as csv_file:
+        reader = csv.DictReader(csv_file)
+        for row in reader:
+            lookup[row["set_num"]] = int(row["minifig_count"])
+    return lookup
+
+
+def load_num_parts(path: Path) -> Dict[str, int]:
+    """Indexe le nombre de pièces par set."""
+    lookup: Dict[str, int] = {}
+    with path.open() as csv_file:
+        reader = csv.DictReader(csv_file)
+        for row in reader:
+            lookup[row["set_num"]] = int(row["num_parts"])
+    return lookup
+
+
+def build_global_minifig_counts(inventories_path: Path, inventory_minifigs_path: Path) -> Dict[str, int]:
+    """Calcule le nombre de minifigs par set pour le catalogue complet."""
+    inventories = select_latest_inventories(inventories_path)
+    minifigs_by_inventory = index_inventory_minifigs_by_inventory(inventory_minifigs_path)
+    counts: Dict[str, int] = {}
+    for set_num, inventory in inventories.items():
+        total = 0
+        for row in minifigs_by_inventory.get(inventory["id"], []):
+            total += int(row["quantity"])
+        counts[set_num] = total
+    return counts
+
+
+def build_correlation_rows(
+    filtered_counts_path: Path,
+    filtered_sets_path: Path,
+    all_sets_path: Path,
+    inventories_path: Path,
+    inventory_minifigs_path: Path,
+) -> List[dict]:
+    """Construit les lignes de corrélation pièces/minifigs pour sets filtrés et catalogue."""
+    filtered_counts = load_minifig_counts_by_set(filtered_counts_path)
+    filtered_parts = load_num_parts(filtered_sets_path)
+    rows: List[dict] = []
+    for set_num, minifig_count in filtered_counts.items():
+        num_parts = filtered_parts[set_num]
+        rows.append(
+            {
+                "scope": "filtered",
+                "set_num": set_num,
+                "num_parts": str(num_parts),
+                "minifig_count": str(minifig_count),
+            }
+        )
+    global_parts = load_num_parts(all_sets_path)
+    global_minifigs = build_global_minifig_counts(inventories_path, inventory_minifigs_path)
+    for set_num, num_parts in global_parts.items():
+        if num_parts <= 0:
+            continue
+        minifig_count = global_minifigs.get(set_num, 0)
+        rows.append(
+            {
+                "scope": "catalog",
+                "set_num": set_num,
+                "num_parts": str(num_parts),
+                "minifig_count": str(minifig_count),
+            }
+        )
+    return rows
+
+
+def write_correlation_rows(path: Path, rows: Sequence[dict]) -> None:
+    """Écrit les lignes de corrélation pièces/minifigs."""
+    ensure_parent_dir(path)
+    fieldnames = ["scope", "set_num", "num_parts", "minifig_count"]
+    with path.open("w", newline="") as csv_file:
+        writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
+        writer.writeheader()
+        for row in rows:
+            writer.writerow(row)
+
+
+def load_correlation_rows(path: Path) -> List[dict]:
+    """Charge le CSV de corrélation pièces/minifigs."""
+    return read_rows(path)