Ajoute le graphique du nombre de minifigs par set

lib/rebrickable/minifig_counts.py

@@ -0,0 +1,71 @@
+"""Comptage des minifigs par set filtré."""
+
+import csv
+from pathlib import Path
+from typing import Dict, Iterable, List, Sequence, Set
+
+from lib.filesystem import ensure_parent_dir
+from lib.rebrickable.minifigs_by_set import load_parts_catalog, select_head_parts
+from lib.rebrickable.stats import read_rows
+
+
+def load_sets(path: Path) -> List[dict]:
+    """Charge les sets enrichis depuis un CSV."""
+    return read_rows(path)
+
+
+def load_parts_filtered(path: Path) -> List[dict]:
+    """Charge parts_filtered.csv en mémoire."""
+    return read_rows(path)
+
+
+def count_heads_by_set(
+    sets_rows: Iterable[dict],
+    parts_rows: Iterable[dict],
+    head_parts: Set[str],
+) -> List[dict]:
+    """Compte les têtes de minifigs présentes dans chaque set (hors rechanges)."""
+    counts: Dict[str, int] = {row["set_num"]: 0 for row in sets_rows}
+    for row in parts_rows:
+        if row["part_num"] not in head_parts:
+            continue
+        if row["is_spare"] == "true":
+            continue
+        counts[row["set_num"]] += int(row["quantity_in_set"])
+    results: List[dict] = []
+    for row in sets_rows:
+        results.append(
+            {
+                "set_num": row["set_num"],
+                "set_id": row["set_id"],
+                "name": row["name"],
+                "year": row["year"],
+                "minifig_count": counts[row["set_num"]],
+            }
+        )
+    results.sort(key=lambda r: (-r["minifig_count"], r["set_num"]))
+    return results
+
+
+def build_minifig_counts_by_set(
+    sets_path: Path,
+    parts_filtered_path: Path,
+    parts_catalog_path: Path,
+) -> List[dict]:
+    """Construit la liste des sets avec leur nombre de minifigs."""
+    sets_rows = load_sets(sets_path)
+    parts_rows = load_parts_filtered(parts_filtered_path)
+    catalog = load_parts_catalog(parts_catalog_path)
+    head_parts = select_head_parts(catalog)
+    return count_heads_by_set(sets_rows, parts_rows, head_parts)
+
+
+def write_minifig_counts(destination_path: Path, rows: Sequence[dict]) -> None:
+    """Écrit le CSV listant le nombre de minifigs par set."""
+    ensure_parent_dir(destination_path)
+    fieldnames = ["set_num", "set_id", "name", "year", "minifig_count"]
+    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)