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

README.md

@@ -216,3 +216,10 @@ Ces scripts lisent les CSV bruts du catalogue complet (`data/raw/inventories.csv
 
 Le script relit les sets (`data/raw/themes.csv`, `data/raw/sets.csv`, `data/intermediate/sets_filtered.csv`, `data/intermediate/sets_enriched.csv`) ainsi que les inventaires (`data/raw/inventories.csv`, `data/raw/inventory_minifigs.csv`), recalcule toutes les statistiques de base puis régénère `data/final/stats.csv` en y ajoutant le libellé « Nombre total de minifigs (thèmes filtrés) ».
 Cette étape se lance après le téléchargement des données d'inventaire (étape 8) et doit être rejouée si les sets filtrés ou les inventaires sont mis à jour.
+
+### Étape 20 : lister les têtes de minifigs par set
+
+1. `source .venv/bin/activate`
+2. `python -m scripts.compute_minifigs_by_set`
+
+Le script lit l'inventaire agrégé `data/intermediate/parts_filtered.csv` ainsi que le catalogue des pièces (`data/raw/parts.csv`). Il sélectionne les têtes de minifigs (catégorie 59), ignore les rechanges et dédoublonne par set et référence. Le CSV `data/intermediate/minifigs_by_set.csv` contient une ligne par set et par référence de tête : `set_num`, `part_num`, `part_name`.

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)

scripts/compute_minifigs_by_set.py

@@ -0,0 +1,23 @@
+"""Liste les têtes de minifigs présentes dans chaque set filtré."""
+
+from pathlib import Path
+
+from lib.rebrickable.minifigs_by_set import build_minifigs_by_set
+
+
+PARTS_FILTERED_PATH = Path("data/intermediate/parts_filtered.csv")
+PARTS_CATALOG_PATH = Path("data/raw/parts.csv")
+DESTINATION_PATH = Path("data/intermediate/minifigs_by_set.csv")
+
+
+def main() -> None:
+    """Construit le CSV listant les têtes de minifigs par set."""
+    build_minifigs_by_set(
+        PARTS_FILTERED_PATH,
+        PARTS_CATALOG_PATH,
+        DESTINATION_PATH,
+    )
+
+
+if __name__ == "__main__":
+    main()

tests/test_minifigs_by_set.py

@@ -0,0 +1,46 @@
+"""Tests de l'extraction des têtes de minifigs par set."""
+
+from pathlib import Path
+
+from lib.rebrickable.minifigs_by_set import build_minifigs_by_set
+
+
+def write_csv(path: Path, content: str) -> None:
+    """Écrit un CSV brut."""
+    path.write_text(content)
+
+
+def test_build_minifigs_by_set_filters_spares_and_deduplicates(tmp_path) -> None:
+    """Identifie les têtes de minifigs par set en dédupliquant et en excluant les rechanges."""
+    parts_filtered_path = tmp_path / "parts_filtered.csv"
+    write_csv(
+        parts_filtered_path,
+        "part_num,color_rgb,is_translucent,set_num,set_id,year,quantity_in_set,is_spare,is_minifig_part\n"
+        "head-a,ffffff,false,123-1,123,2020,1,false,true\n"
+        "head-b,ffffff,false,123-1,123,2020,2,false,true\n"
+        "head-b,ffffff,false,123-1,123,2020,1,true,true\n"
+        "head-b,ffffff,false,124-1,124,2021,1,false,true\n"
+        "other,000000,false,123-1,123,2020,1,false,false\n",
+    )
+    parts_catalog_path = tmp_path / "parts.csv"
+    write_csv(
+        parts_catalog_path,
+        "part_num,name,part_cat_id\n"
+        "head-a,Head A,59\n"
+        "head-b,Head B,59\n"
+        "other,Other,1\n",
+    )
+    destination_path = tmp_path / "minifigs_by_set.csv"
+
+    build_minifigs_by_set(
+        parts_filtered_path,
+        parts_catalog_path,
+        destination_path,
+    )
+
+    assert destination_path.read_text() == (
+        "set_num,part_num,part_name\n"
+        "123-1,head-a,Head A\n"
+        "123-1,head-b,Head B\n"
+        "124-1,head-b,Head B\n"
+    )