from __future__ import annotations import argparse import csv import json import math import re import unicodedata from datetime import datetime from pathlib import Path import pandas as pd DEFAULT_INPUT = Path( r"C:\wamp64\www\atelier-com-site\racine\atelier-com-platform\outputs\pricing_fill_full_grid\modele_import_tarifs_depuis_sql_full_20260526_095640.xlsx" ) DEFAULT_META = Path( r"C:\wamp64\www\atelier-com-site\racine\atelier-com-platform\outputs\pricing_import\pricing_product_meta.json" ) DEFAULT_OUTPUT_DIR = Path( r"C:\wamp64\www\atelier-com-site\racine\atelier-com-platform\outputs\pricing_import" ) SKU_ALIASES = { "EVT-TOURS-DE-COU": "EVT-TOUR-DE-COU", "PAP-CLASSEUR-A-LEVIER": "PAP-CLASSEUR", "OBJ-STYLO-PUBLICITAIRE-RIOCOLOR": "OBJ-STYLO-PUBLICITAIRE", } def normalize_key(value: str) -> str: text = unicodedata.normalize("NFKD", str(value)) text = "".join(ch for ch in text if not unicodedata.combining(ch)) text = text.lower().strip() text = re.sub(r"[^a-z0-9]+", "", text) return text def cell_text(value) -> str: if value is None: return "" if isinstance(value, float) and math.isnan(value): return "" return str(value).strip() def parse_series(value: str) -> list[str]: return [item.strip() for item in cell_text(value).split(";") if item.strip()] def format_price(value: str) -> str: raw = cell_text(value).replace(",", ".") try: number = float(raw) except ValueError: return raw if number.is_integer(): return f"{int(number)}" return f"{number:.2f}" def timestamp() -> str: return datetime.now().strftime("%Y%m%d_%H%M%S") def workbook_column_mapping(columns: list[str]) -> tuple[dict[str, str], dict[str, str]]: exact = {str(col): str(col) for col in columns} normalized = {} for col in columns: normalized.setdefault(normalize_key(str(col)), str(col)) return exact, normalized def resolve_column(code: str, exact: dict[str, str], normalized: dict[str, str]) -> str | None: candidates = [ code, code.replace("_", "-"), code.replace("-", "_"), code.replace("matiere", "mati-ere"), code.replace("mati-ere", "matiere"), ] for candidate in candidates: if candidate in exact: return exact[candidate] wanted = normalize_key(code) return normalized.get(wanted) def load_meta(meta_path: Path) -> dict[str, dict]: rows = json.loads(meta_path.read_text(encoding="utf-8-sig")) return {str(item["sku"]): item for item in rows} def build_rows(df: pd.DataFrame, meta_by_sku: dict[str, dict]) -> tuple[list[str], list[dict], dict]: exact_columns, normalized_columns = workbook_column_mapping(df.columns.tolist()) base_headers = ["sku", "lot_quantity", "buy_price_ht"] used_attribute_codes: list[str] = [] for item in meta_by_sku.values(): for attribute in item.get("attributes", []): if int(attribute.get("is_pricing_quantity", 0)) == 1: continue if int(attribute.get("affects_pricing", 1)) != 1: continue code = str(attribute.get("code", "")).strip() if code and code not in used_attribute_codes: used_attribute_codes.append(code) output_rows: list[dict] = [] missing_skus: set[str] = set() unresolved_columns: dict[str, set[str]] = {} sku_col = exact_columns.get("sku") or normalized_columns.get("sku") qty_col = exact_columns.get("Quantité") or normalized_columns.get(normalize_key("Quantité")) cost_col = exact_columns.get("Coût achat HT") or normalized_columns.get(normalize_key("Coût achat HT")) if not sku_col or not qty_col or not cost_col: raise RuntimeError("Colonnes obligatoires introuvables dans le classeur source.") for _, row in df.iterrows(): sku = cell_text(row.get(sku_col)) if sku == "": continue target_sku = SKU_ALIASES.get(sku, sku) meta = meta_by_sku.get(target_sku) if meta is None: missing_skus.add(sku) continue quantities = parse_series(row.get(qty_col)) prices = parse_series(row.get(cost_col)) if not quantities or not prices: continue pair_count = min(len(quantities), len(prices)) if pair_count == 0: continue attribute_map: dict[str, str] = {} for attribute in meta.get("attributes", []): if int(attribute.get("is_pricing_quantity", 0)) == 1: continue if int(attribute.get("affects_pricing", 1)) != 1: continue code = str(attribute.get("code", "")).strip() if code == "": continue column = resolve_column(code, exact_columns, normalized_columns) if column is None: unresolved_columns.setdefault(sku, set()).add(code) continue value = cell_text(row.get(column)) if value != "": attribute_map[code] = value for index in range(pair_count): output = { "sku": target_sku, "lot_quantity": re.sub(r"[^\d]", "", quantities[index]) or quantities[index], "buy_price_ht": format_price(prices[index]), } for code in used_attribute_codes: output[code] = attribute_map.get(code, "") output_rows.append(output) summary = { "source_rows": int(len(df)), "csv_rows": int(len(output_rows)), "missing_skus": sorted(missing_skus), "unresolved_columns": {sku: sorted(codes) for sku, codes in unresolved_columns.items()}, "products_in_meta": int(len(meta_by_sku)), } return base_headers + used_attribute_codes, output_rows, summary def write_csv(headers: list[str], rows: list[dict], output_path: Path) -> None: output_path.parent.mkdir(parents=True, exist_ok=True) with output_path.open("w", encoding="utf-8-sig", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=headers, delimiter=";") writer.writeheader() for row in rows: writer.writerow(row) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--input", default=str(DEFAULT_INPUT)) parser.add_argument("--meta", default=str(DEFAULT_META)) parser.add_argument("--output-dir", default=str(DEFAULT_OUTPUT_DIR)) args = parser.parse_args() input_path = Path(args.input) meta_path = Path(args.meta) output_dir = Path(args.output_dir) df = pd.read_excel(input_path, sheet_name="Import_Tarifs") meta_by_sku = load_meta(meta_path) headers, rows, summary = build_rows(df, meta_by_sku) output_path = output_dir / f"pricing_import_{timestamp()}.csv" write_csv(headers, rows, output_path) summary_path = output_dir / f"{output_path.stem}_summary.json" summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({"output": str(output_path), "summary": str(summary_path), **summary}, ensure_ascii=False)) if __name__ == "__main__": main()