Extrait les têtes de minifigs depuis l'inventaire agrégé

lib/rebrickable/minifigs_by_set.py

@@ -0,0 +1,82 @@
+"""Extraction des têtes de minifigs présentes dans chaque set filtré."""
+
+import csv
+from pathlib import Path
+from typing import Dict, Iterable, List, Sequence, Set, Tuple
+
+from lib.filesystem import ensure_parent_dir
+from lib.rebrickable.minifig_heads import HEAD_CATEGORIES
+from lib.rebrickable.stats import read_rows
+
+
+def load_parts_filtered(path: Path) -> List[dict]:
+    """Charge parts_filtered.csv en mémoire."""
+    return read_rows(path)
+
+
+def load_parts_catalog(path: Path) -> Dict[str, dict]:
+    """Construit un index des pièces avec leur catégorie et leur nom."""
+    catalog: Dict[str, dict] = {}
+    with path.open() as catalog_file:
+        reader = csv.DictReader(catalog_file)
+        for row in reader:
+            catalog[row["part_num"]] = row
+    return catalog
+
+
+def select_head_parts(catalog: Dict[str, dict]) -> Set[str]:
+    """Sélectionne les références de têtes via leur catégorie."""
+    return {part_num for part_num, row in catalog.items() if row["part_cat_id"] in HEAD_CATEGORIES}
+
+
+def aggregate_heads_by_set(
+    parts_rows: Iterable[dict],
+    catalog: Dict[str, dict],
+    head_parts: Set[str],
+) -> List[dict]:
+    """Agrège les têtes de minifigs par set en éliminant les rechanges et doublons."""
+    seen: Set[Tuple[str, str]] = set()
+    heads: List[dict] = []
+    for row in parts_rows:
+        if row["part_num"] not in head_parts:
+            continue
+        if row["is_spare"] == "true":
+            continue
+        key = (row["set_num"], row["part_num"])
+        if key in seen:
+            continue
+        part = catalog[row["part_num"]]
+        heads.append(
+            {
+                "set_num": row["set_num"],
+                "part_num": row["part_num"],
+                "part_name": part["name"],
+            }
+        )
+        seen.add(key)
+    heads.sort(key=lambda row: (row["set_num"], row["part_num"]))
+    return heads
+
+
+def write_heads_by_set(destination_path: Path, rows: Sequence[dict]) -> None:
+    """Écrit le CSV intermédiaire listant les têtes de minifigs par set."""
+    ensure_parent_dir(destination_path)
+    fieldnames = ["set_num", "part_num", "part_name"]
+    with destination_path.open("w", newline="") as csv_file:
+        writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
+        writer.writeheader()
+        for row in rows:
+            writer.writerow(row)
+
+
+def build_minifigs_by_set(
+    parts_filtered_path: Path,
+    parts_catalog_path: Path,
+    destination_path: Path,
+) -> None:
+    """Construit le CSV listant les têtes de minifigs présentes par set."""
+    parts_rows = load_parts_filtered(parts_filtered_path)
+    parts_catalog = load_parts_catalog(parts_catalog_path)
+    head_parts = select_head_parts(parts_catalog)
+    heads = aggregate_heads_by_set(parts_rows, parts_catalog, head_parts)
+    write_heads_by_set(destination_path, heads)