#!/usr/bin/env python3 """Normalize vendor CSV into Mobile Medic inventory template.""" from __future__ import annotations import csv import re from collections import Counter, defaultdict from dataclasses import dataclass from decimal import Decimal, ROUND_HALF_UP from pathlib import Path import subprocess from typing import Dict, Tuple BASE_DIR = Path(__file__).resolve().parents[0] REPO_ROOT = BASE_DIR.parent INPUT_CSV = REPO_ROOT / "imports" / "mobile_medic_parts_pricing_full.csv" OUTPUT_CSV = REPO_ROOT / "imports" / "mobile_medic_parts_normalized.csv" DEVICE_REF = REPO_ROOT / "exports" / "device_ids_reference.csv" MANUFACTURER_CODES: Dict[str, str] = { "Apple": "AP", "Samsung": "SA", "Google": "GO", "Motorola": "MO", "LG": "LG", "OnePlus": "OP", "Xiaomi": "XI", "Oppo": "OP", "Vivo": "VI", "Huawei": "HU", "Sony": "SO", "Asus": "AS", "Microsoft": "MS", "ZTE": "ZT", "Kyocera": "KY", "TCL": "TC", "Nokia": "NO", "BLU": "BL", "BlackBerry": "BB", "Essential": "ES", "Nothing": "NO", # reuse for Nothing Phone "Cricket": "CR", "Cat": "CA", "Sonim": "SO", } FAMILY_OVERRIDES: Dict[str, str] = { "10 Pro Series": "OnePlus", "10T Series": "OnePlus", "11 Series": "OnePlus", "12 Series": "OnePlus", "13 Ultra Series": "Xiaomi", "7 Pro Series": "OnePlus", "7T Series": "OnePlus", "8 Pro Series": "OnePlus", "8 Series": "OnePlus", "8 V Series": "Nokia", "9 Pro Series": "OnePlus", "9 Series": "OnePlus", "6T Series": "OnePlus", "3V Series": "TCL", "4 Series": "Nokia", "5 Series": "Nokia", "G6 Series": "LG", "G90 Series": "BLU", "Icon 3 Series": "Cricket", "Legacy Series": "Cricket", "Legacy Brisa Series": "Cricket", "Phone (2018) Series": "Nokia", "Phone (PH-1) Series": "Essential", "Phone 2 Series": "Nothing", "Ride Series": "Cricket", "S61 Series": "Cat", "S62 Series": "Cat", "Stylus 5G Series": "TCL", "U12+ Series": "HTC", "Wonder Series": "BLU", "X80 Pro Series": "Vivo", "X90 Pro Series": "Vivo", "XP8 Series": "Sonim", "30 XE Series": "TCL", "Edge (2021) Series": "Motorola", "Edge+ (2023) Series": "Motorola", "Key2 Series": "BlackBerry", } KEYWORD_MANUFACTURERS: Tuple[Tuple[str, str], ...] = ( ("iphone", "Apple"), ("ipad", "Apple"), ("galaxy", "Samsung"), ("pixel", "Google"), ("moto", "Motorola"), ("razr", "Motorola"), ("thinkphone", "Motorola"), ("nord", "OnePlus"), ("oneplus", "OnePlus"), ("find x", "Oppo"), ("reno", "Oppo"), ("poco", "Xiaomi"), ("redmi", "Xiaomi"), ("mi ", "Xiaomi"), ("mate", "Huawei"), ("p20", "Huawei"), ("p30", "Huawei"), ("p40", "Huawei"), ("axon", "ZTE"), ("blade", "ZTE"), ("dura", "Kyocera"), ("k51", "LG"), ("k92", "LG"), ("aristo", "LG"), ("stylo", "LG"), ("velvet", "LG"), ("wing", "LG"), ("thinq", "LG"), ("zenfone", "Asus"), ("rog phone", "Asus"), ("xperia", "Sony"), ("surface duo", "Microsoft"), ("mate", "Huawei"), ("xr20", "Nokia"), ("c200", "Nokia"), ("g20", "Nokia"), ("g50", "Nokia"), ("cat", "Cat"), ("sonim", "Sonim"), ("vivo", "Vivo"), ("tcl", "TCL"), ) PART_TYPE_CODES: Dict[str, str] = { "battery": "BAT", "charging_port": "CHG", "screen_lcd": "LCD", "screen_oled": "OLD", "back_glass": "BGL", "back_camera": "BCM", "front_camera": "FCM", "loud_speaker": "LSP", "fingerprint_scanner": "FPS", "home_button": "HMB", } DEVICE_TYPE_CODES = { "phone": "P", "tablet": "T", } PN_REGEX = re.compile(r"^[A-Z]{3}[A-Z0-9]{3}\d+$") @dataclass class DeviceRef: id: int model_name: str family_name: str def letters_only(value: str) -> str: return "".join(ch for ch in value if ch.isalpha()) def detect_manufacturer(family: str, model: str) -> str: if family in FAMILY_OVERRIDES: return FAMILY_OVERRIDES[family] text = f"{family} {model}".lower() for keyword, manufacturer in KEYWORD_MANUFACTURERS: if keyword in text: return manufacturer return family or "Unknown" def manufacturer_code(name: str) -> str: normalized = name.strip() or "Unknown" if normalized in MANUFACTURER_CODES: return MANUFACTURER_CODES[normalized] letters = letters_only(normalized.upper()) if len(letters) >= 2: return letters[:2] if letters: return letters[0] * 2 return "UN" def infer_part_template(service_type: str, raw_label: str) -> Tuple[str, str]: st = service_type.lower() label = raw_label.lower() if st == "screen": if "oled" in label: suffix = "OLED Display Assembly" if "hard" in label: suffix = "Hard OLED Display Assembly" elif "soft" in label: suffix = "Soft OLED Display Assembly" return "screen_oled", suffix return "screen_lcd", "LCD Screen Assembly" if st == "battery": return "battery", "Battery" if st == "charging_port": return "charging_port", "Charging Port" if st == "back_glass": return "back_glass", "Back Glass Panel" if st == "back_camera": return "back_camera", "Back Camera Module" if st == "front_camera": return "front_camera", "Front Camera Module" if st == "loud_speaker": return "loud_speaker", "Loud Speaker" if st == "fingerprint_scanner": return "fingerprint_scanner", "Fingerprint Scanner" if st == "home_button": return "home_button", "Home Button" raise ValueError(f"Unsupported service_type '{service_type}'") def load_device_lookup() -> Tuple[Dict[str, DeviceRef], Dict[str, DeviceRef]]: model_lookup: Dict[str, DeviceRef] = {} family_lookup: Dict[str, DeviceRef] = {} if not DEVICE_REF.exists(): return model_lookup, family_lookup with DEVICE_REF.open() as handle: reader = csv.DictReader(handle) for row in reader: ref = DeviceRef( id=int(row["id"]), model_name=row["model_name"].strip(), family_name=row["family_name"].strip(), ) model_lookup[ref.model_name.lower()] = ref family_lookup[ref.family_name.lower()] = ref return model_lookup, family_lookup def load_existing_parts() -> Tuple[Dict[str, int], Dict[Tuple[str, str], str], set[str]]: query = "SELECT service_type, name, part_number FROM parts" result = subprocess.run( [ "sudo", "-u", "postgres", "psql", "-d", "medicore", "-A", "-F", "\t", "-t", "-c", query, ], capture_output=True, text=True, check=True, ) seq_map: Dict[str, int] = defaultdict(lambda: 99) existing_parts: Dict[Tuple[str, str], str] = {} existing_part_numbers: set[str] = set() for line in result.stdout.strip().splitlines(): parts = line.split("\t") if len(parts) != 3: continue service_type, name, part_number = parts if service_type and name and part_number: existing_parts[(service_type.lower(), name.lower())] = part_number if part_number and len(part_number) >= 7: part_code = part_number[3:6] if part_code.isalpha(): try: seq_value = int(part_number[6:]) except ValueError: seq_value = 99 seq_map[part_code] = max(seq_map[part_code], seq_value) existing_part_numbers.add(part_number) return seq_map, existing_parts, existing_part_numbers def format_decimal(value: str) -> str: if not value: return "0.00" amount = Decimal(value) return f"{amount.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)}" def next_sequence(seq_map: Dict[str, int], code: str) -> int: seq_map[code] = seq_map.get(code, 99) + 1 return seq_map[code] def main() -> None: device_by_model, device_by_family = load_device_lookup() seq_map, existing_parts, existing_part_numbers = load_existing_parts() counters = Counter() generated_rows: list[list[str]] = [] seen_generated: set[Tuple[str, str]] = set() with INPUT_CSV.open() as handle: reader = csv.DictReader(handle) for row in reader: counters["total"] += 1 model = (row.get("model_name") or "").strip() family = (row.get("family_name") or "").strip() service_type = (row.get("service_type") or "").strip().lower() raw_label = row.get("name") or "" if not model or not service_type: counters["missing_fields"] += 1 continue part_type_key, suffix = infer_part_template(service_type, raw_label) part_type_code = PART_TYPE_CODES[part_type_key] part_name = f"{model} {suffix}".strip() dedupe_key = (service_type, part_name.lower()) if dedupe_key in seen_generated: counters["skipped_duplicate"] += 1 continue manufacturer = detect_manufacturer(family, model) mfg_code = manufacturer_code(manufacturer) if dedupe_key in existing_parts: part_number = existing_parts[dedupe_key] is_new_part = False else: is_new_part = True device_letter = DEVICE_TYPE_CODES.get(row.get("device_type_name", "").lower(), "X") sequence = next_sequence(seq_map, part_type_code) part_number = f"{mfg_code}{device_letter}{part_type_code}{sequence}" while part_number in existing_part_numbers: sequence = next_sequence(seq_map, part_type_code) part_number = f"{mfg_code}{device_letter}{part_type_code}{sequence}" existing_part_numbers.add(part_number) existing_parts[dedupe_key] = part_number sku = part_number grade = (row.get("grade") or "aftermarket").strip().lower() cost = format_decimal(row.get("cost") or "0") quantity_on_hand = int(float(row.get("quantity_on_hand") or 0)) description = (row.get("part_description") or "").strip() if not description: description = f"{raw_label} for {model}" device_id = "" model_key = model.lower() if model_key in device_by_model: device_id = str(device_by_model[model_key].id) elif family and family.lower() in device_by_family: device_id = str(device_by_family[family.lower()].id) else: counters["missing_device"] += 1 notes = f"{family} | {raw_label}".strip(" |") row_out = [ part_number, part_name, service_type, row.get("category") or "main_component", description, "", sku, row.get("color") or "", grade, cost, str(quantity_on_hand), "0", "2", "4", "", "TRUE", device_id, notes, ] generated_rows.append(row_out) seen_generated.add(dedupe_key) if is_new_part: counters["created"] += 1 else: counters["skipped_existing"] += 1 with OUTPUT_CSV.open("w", newline="") as out_handle: writer = csv.writer(out_handle) writer.writerow( [ "part_number", "part_name", "service_type", "category", "part_description", "manufacturer_part_number", "sku", "color", "grade", "cost", "quantity_on_hand", "quantity_reserved", "reorder_threshold", "reorder_quantity", "supplier_id", "active", "compatible_device_ids", "notes", ] ) writer.writerows(generated_rows) print("Processed rows:", counters["total"]) print("Created rows:", counters["created"]) print("Skipped existing parts:", counters["skipped_existing"]) print("Skipped duplicates in CSV:", counters["skipped_duplicate"]) print("Rows missing required fields:", counters["missing_fields"]) print("Rows missing device matches:", counters["missing_device"]) print(f"Output written to {OUTPUT_CSV}") if __name__ == "__main__": main()