Ajoute l'analyse des pièces rares

lib/rebrickable/rare_parts.py

@@ -0,0 +1,204 @@
+"""Identification des pièces rares (variantes exclusives à un set)."""
+
+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.stats import read_rows
+
+
+def load_parts_catalog(path: Path) -> Dict[str, dict]:
+    """Charge le catalogue des pièces et l'indexe par référence."""
+    catalog: Dict[str, dict] = {}
+    with path.open() as csv_file:
+        reader = csv.DictReader(csv_file)
+        for row in reader:
+            catalog[row["part_num"]] = row
+    return catalog
+
+
+def load_colors_lookup(path: Path) -> Dict[Tuple[str, str], str]:
+    """Associe un couple (rgb, is_trans) au nom de couleur."""
+    lookup: Dict[Tuple[str, str], str] = {}
+    for row in read_rows(path):
+        lookup[(row["rgb"], row["is_trans"].lower())] = row["name"]
+    return lookup
+
+
+def load_sets_enriched(path: Path) -> Dict[str, dict]:
+    """Indexe les sets enrichis par numéro complet."""
+    sets: Dict[str, dict] = {}
+    for row in read_rows(path):
+        sets[row["set_num"]] = row
+    return sets
+
+
+def aggregate_non_spare_parts(rows: Iterable[dict]) -> List[dict]:
+    """Agrège les pièces hors rechanges par set et variation couleur."""
+    aggregated: Dict[Tuple[str, str, str, str, str, str, str], int] = {}
+    for row in rows:
+        if row["is_spare"] == "true":
+            continue
+        key = (
+            row["set_num"],
+            row["part_num"],
+            row["color_rgb"],
+            row["is_translucent"],
+            row["is_minifig_part"],
+            row["set_id"],
+            row["year"],
+        )
+        aggregated[key] = aggregated.get(key, 0) + int(row["quantity_in_set"])
+    result: List[dict] = []
+    for key, quantity in aggregated.items():
+        set_num, part_num, color_rgb, is_translucent, is_minifig_part, set_id, year = key
+        result.append(
+            {
+                "set_num": set_num,
+                "part_num": part_num,
+                "color_rgb": color_rgb,
+                "is_translucent": is_translucent,
+                "is_minifig_part": is_minifig_part,
+                "set_id": set_id,
+                "year": year,
+                "quantity_in_set": str(quantity),
+            }
+        )
+    result.sort(key=lambda row: (row["set_num"], row["part_num"], row["color_rgb"]))
+    return result
+
+
+def compute_combo_set_counts(rows: Iterable[dict]) -> Dict[Tuple[str, str, str], Set[str]]:
+    """Compte les sets distincts par combinaison pièce+couleur."""
+    combos: Dict[Tuple[str, str, str], Set[str]] = {}
+    for row in rows:
+        key = (row["part_num"], row["color_rgb"], row["is_translucent"])
+        if key not in combos:
+            combos[key] = set()
+        combos[key].add(row["set_num"])
+    return combos
+
+
+def build_rare_parts(
+    parts_filtered_path: Path,
+    sets_enriched_path: Path,
+    parts_catalog_path: Path,
+    colors_path: Path,
+) -> Tuple[List[dict], List[dict]]:
+    """Construit les listes des pièces rares et leur répartition par set."""
+    parts_rows = read_rows(parts_filtered_path)
+    aggregated = aggregate_non_spare_parts(parts_rows)
+    combo_sets = compute_combo_set_counts(aggregated)
+    parts_catalog = load_parts_catalog(parts_catalog_path)
+    color_names = load_colors_lookup(colors_path)
+    sets_lookup = load_sets_enriched(sets_enriched_path)
+
+    rare_parts: List[dict] = []
+    for row in aggregated:
+        combo_key = (row["part_num"], row["color_rgb"], row["is_translucent"])
+        if len(combo_sets[combo_key]) != 1:
+            continue
+        set_row = sets_lookup[row["set_num"]]
+        part = parts_catalog[row["part_num"]]
+        color_name = color_names[(row["color_rgb"], row["is_translucent"])]
+        rare_parts.append(
+            {
+                "set_num": row["set_num"],
+                "set_id": row["set_id"],
+                "set_name": set_row["name"],
+                "year": set_row["year"],
+                "part_num": row["part_num"],
+                "part_name": part["name"],
+                "part_cat_id": part["part_cat_id"],
+                "color_rgb": row["color_rgb"],
+                "color_name": color_name,
+                "is_translucent": row["is_translucent"],
+                "is_minifig_part": row["is_minifig_part"],
+                "quantity_in_set": row["quantity_in_set"],
+                "in_collection": set_row["in_collection"],
+            }
+        )
+    rare_parts.sort(key=lambda row: (row["set_num"], row["part_num"], row["color_rgb"]))
+
+    rare_by_set: Dict[str, dict] = {}
+    for row in rare_parts:
+        record = rare_by_set.get(row["set_num"])
+        if record is None:
+            record = {
+                "set_num": row["set_num"],
+                "set_id": row["set_id"],
+                "name": row["set_name"],
+                "year": row["year"],
+                "in_collection": row["in_collection"],
+                "rare_parts_distinct": 0,
+                "rare_parts_quantity": 0,
+                "rare_minifig_parts_distinct": 0,
+                "rare_minifig_quantity": 0,
+            }
+            rare_by_set[row["set_num"]] = record
+        record["rare_parts_distinct"] += 1
+        record["rare_parts_quantity"] += int(row["quantity_in_set"])
+        if row["is_minifig_part"] == "true":
+            record["rare_minifig_parts_distinct"] += 1
+            record["rare_minifig_quantity"] += int(row["quantity_in_set"])
+    rare_by_set_rows = list(rare_by_set.values())
+    rare_by_set_rows.sort(
+        key=lambda row: (
+            -row["rare_parts_distinct"],
+            -row["rare_parts_quantity"],
+            row["set_num"],
+        )
+    )
+    for row in rare_by_set_rows:
+        row["rare_parts_distinct"] = str(row["rare_parts_distinct"])
+        row["rare_parts_quantity"] = str(row["rare_parts_quantity"])
+        row["rare_minifig_parts_distinct"] = str(row["rare_minifig_parts_distinct"])
+        row["rare_minifig_quantity"] = str(row["rare_minifig_quantity"])
+    return rare_parts, rare_by_set_rows
+
+
+def write_rare_parts_list(destination_path: Path, rows: Sequence[dict]) -> None:
+    """Écrit le détail des pièces rares avec leur set et leur couleur."""
+    ensure_parent_dir(destination_path)
+    fieldnames = [
+        "set_num",
+        "set_id",
+        "set_name",
+        "year",
+        "part_num",
+        "part_name",
+        "part_cat_id",
+        "color_rgb",
+        "color_name",
+        "is_translucent",
+        "is_minifig_part",
+        "quantity_in_set",
+        "in_collection",
+    ]
+    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 write_rare_parts_by_set(destination_path: Path, rows: Sequence[dict]) -> None:
+    """Écrit l'agrégat des pièces rares par set."""
+    ensure_parent_dir(destination_path)
+    fieldnames = [
+        "set_num",
+        "set_id",
+        "name",
+        "year",
+        "in_collection",
+        "rare_parts_distinct",
+        "rare_parts_quantity",
+        "rare_minifig_parts_distinct",
+        "rare_minifig_quantity",
+    ]
+    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)