42 lines
1.5 KiB
Python
42 lines
1.5 KiB
Python
"""Agrégation des minifigs par personnage représenté."""
|
|
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Dict, Iterable, List, Sequence
|
|
|
|
from lib.rebrickable.stats import read_rows
|
|
from lib.filesystem import ensure_parent_dir
|
|
import csv
|
|
|
|
|
|
def load_minifigs_by_set(path: Path) -> List[dict]:
|
|
"""Charge le CSV minifigs_by_set."""
|
|
return read_rows(path)
|
|
|
|
|
|
def aggregate_by_character(rows: Iterable[dict]) -> List[dict]:
|
|
"""Compte les minifigs distinctes par personnage (fig_num unique)."""
|
|
fig_nums_by_character: Dict[str, set] = defaultdict(set)
|
|
for row in rows:
|
|
character = row["known_character"].strip()
|
|
fig_num = row["fig_num"].strip()
|
|
if character == "" or fig_num == "":
|
|
continue
|
|
fig_nums_by_character[character].add(fig_num)
|
|
aggregates: List[dict] = []
|
|
for character, fig_nums in fig_nums_by_character.items():
|
|
aggregates.append({"known_character": character, "minifig_count": len(fig_nums)})
|
|
aggregates.sort(key=lambda r: (-r["minifig_count"], r["known_character"]))
|
|
return aggregates
|
|
|
|
|
|
def write_character_counts(path: Path, rows: Sequence[dict]) -> None:
|
|
"""Écrit le CSV des comptes par personnage."""
|
|
ensure_parent_dir(path)
|
|
fieldnames = ["known_character", "minifig_count"]
|
|
with path.open("w", newline="") as csv_file:
|
|
writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
for row in rows:
|
|
writer.writerow(row)
|