83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
"""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)
|