1
etude_lego_jurassic_world/lib/rebrickable/inventory_reconciliation.py

108 lines
4.4 KiB
Python

"""Rapport des écarts entre catalogue et inventaire agrégé."""
import csv
from pathlib import Path
from typing import Dict, Iterable, List
from lib.filesystem import ensure_parent_dir
def load_sets(sets_path: Path) -> List[dict]:
"""Charge les sets filtrés pour l'analyse."""
with sets_path.open() as sets_file:
reader = csv.DictReader(sets_file)
return list(reader)
def index_sets_by_num(sets: Iterable[dict]) -> Dict[str, dict]:
"""Crée un index des sets par numéro complet."""
return {row["set_num"]: row for row in sets}
def compute_inventory_totals(parts_path: Path, include_spares: bool) -> Dict[str, int]:
"""Calcule le total de pièces par set, avec ou sans rechanges."""
totals: Dict[str, int] = {}
with parts_path.open() as parts_file:
reader = csv.DictReader(parts_file)
for row in reader:
if not include_spares and row["is_spare"] == "true":
continue
set_num = row["set_num"]
totals[set_num] = totals.get(set_num, 0) + int(row["quantity_in_set"])
return totals
def compute_inventory_gaps(sets_path: Path, parts_path: Path) -> List[dict]:
"""Liste les sets dont le total de pièces diffère du catalogue."""
sets = load_sets(sets_path)
totals_with_spares = compute_inventory_totals(parts_path, include_spares=True)
totals_without_spares = compute_inventory_totals(parts_path, include_spares=False)
gaps: List[dict] = []
for set_row in sets:
expected_parts = int(set_row["num_parts"])
inventory_parts_with_spares = totals_with_spares[set_row["set_num"]]
inventory_parts_non_spare = totals_without_spares[set_row["set_num"]]
if expected_parts != inventory_parts_with_spares:
gaps.append(
{
"set_num": set_row["set_num"],
"set_id": set_row["set_id"],
"expected_parts": expected_parts,
"inventory_parts": inventory_parts_with_spares,
"inventory_parts_non_spare": inventory_parts_non_spare,
"delta": abs(expected_parts - inventory_parts_with_spares),
"delta_non_spare": abs(expected_parts - inventory_parts_non_spare),
"in_collection": set_row["in_collection"],
}
)
return gaps
def write_inventory_gaps_csv(destination_path: Path, gaps: Iterable[dict]) -> None:
"""Écrit un CSV listant les sets en écart d'inventaire."""
ensure_parent_dir(destination_path)
with destination_path.open("w", newline="") as csv_file:
fieldnames = [
"set_num",
"set_id",
"expected_parts",
"inventory_parts",
"inventory_parts_non_spare",
"delta",
"delta_non_spare",
"in_collection",
]
writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
writer.writeheader()
for row in gaps:
writer.writerow(row)
def build_instructions_url(set_id: str) -> str:
"""Construit un lien direct vers la page d'instructions LEGO du set."""
return f"https://www.lego.com/service/buildinginstructions/{set_id}"
def write_inventory_gaps_markdown(
destination_path: Path,
gaps: Iterable[dict],
sets_by_num: Dict[str, dict],
) -> None:
"""Génère un tableau Markdown listant les sets en écart d'inventaire."""
ensure_parent_dir(destination_path)
with destination_path.open("w") as markdown_file:
markdown_file.write(
"| set_id | name | year | delta (spares inclus) | delta (spares exclus) | expected_parts | inventory_parts | inventory_parts_non_spare | in_collection | instructions |\n"
)
markdown_file.write("| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |\n")
for row in gaps:
if row["delta_non_spare"] == 0:
continue
set_row = sets_by_num[row["set_num"]]
set_link = f"[{row['set_id']}]({set_row['rebrickable_url']})"
instructions_link = f"[PDF]({build_instructions_url(row['set_id'])})"
markdown_file.write(
f"| {set_link} | {set_row['name']} | {set_row['year']} | {row['delta']} | {row['delta_non_spare']} | "
f"{row['expected_parts']} | {row['inventory_parts']} | {row['inventory_parts_non_spare']} | "
f"{row['in_collection']} | {instructions_link} |\n"
)