# Add parent directory to Python path so we can import helpers import sys from pathlib import Path parent_dir = Path(__file__).parent.parent sys.path.insert(0, str(parent_dir)) import csv from proteus import Model from helpers.config import ( PRICE_INDEXES_CSV, connect_to_tryton, ) CSV_FILE_PATH = PRICE_INDEXES_CSV # --------------------------------------------------------------------------- # Main import function # --------------------------------------------------------------------------- def import_price_indexes(csv_file): PriceIndex = Model.get('price.price') ProductMonth = Model.get('product.month') Currency = Model.get('currency.currency') PriceFixType = Model.get('price.fixtype') ProductUom = Model.get('product.uom') print(f"{'='*70}") print(f"IMPORTING PRICE INDEXES FROM CSV") print(f"{'='*70}\n") print(f"Reading from: {csv_file}\n") imported_count = 0 skipped_count = 0 error_count = 0 ignored = [] # list of (row_num, price_index, reason) errors = [] # list of error strings # Build lookup maps upfront to avoid repeated DB calls print("Loading existing price indexes...") all_indexes = PriceIndex.find([]) index_map = {pi.price_index: pi for pi in all_indexes} print(f" Found {len(index_map)} price indexes in database\n") print("Loading product months...") all_months = ProductMonth.find([]) month_map = {pm.month_name: pm for pm in all_months} print(f" Found {len(month_map)} product months in database\n") print("Loading reference data (currencies, price types, units)...") currency_map = {c.id: c for c in Currency.find([])} fixtype_map = {pft.id: pft for pft in PriceFixType.find([])} uom_map = {u.id: u for u in ProductUom.find([])} print() row_num = 0 try: with open(csv_file, 'r', encoding='utf-8-sig') as file: reader = csv.DictReader(file) for row in reader: row_num += 1 index_name = row.get('price_index', '').strip() try: print(f"{'='*70}") print(f"Row {row_num}: {index_name}") print(f"{'='*70}") if not index_name: raise ValueError("price_index is empty") # Already exists → skip silently if index_name in index_map: print(f" ⏭ Already exists in database — skipping\n") ignored.append((row_num, index_name, "Already exists in price_price")) skipped_count += 1 continue # Lookup pricing period in product_month period_label = row.get('pricing_period_label', '').strip() if not period_label: msg = "pricing_period_label is empty" print(f" ✗ {msg} — not imported\n") ignored.append((row_num, index_name, msg)) skipped_count += 1 continue product_month = month_map.get(period_label) if not product_month: # Case-insensitive fallback for k, v in month_map.items(): if k and k.upper() == period_label.upper(): product_month = v break if not product_month: msg = f"pricing_period_label '{period_label}' not found in product_month" print(f" ✗ {msg} — not imported\n") ignored.append((row_num, index_name, msg)) skipped_count += 1 continue # Resolve FK: currency raw_currency = row.get('price_currency', '').strip() currency = None if raw_currency: try: currency = currency_map.get(int(raw_currency)) except (ValueError, TypeError): pass if not currency: raise ValueError(f"price_currency id '{raw_currency}' not found in currency_currency") # Resolve FK: price type (optional) raw_type = row.get('price_type', '').strip() price_type = None if raw_type: try: price_type = fixtype_map.get(int(raw_type)) except (ValueError, TypeError): pass # Resolve FK: unit of measure (optional) raw_unit = row.get('price_unit', '').strip() price_unit = None if raw_unit: try: price_unit = uom_map.get(int(raw_unit)) except (ValueError, TypeError): pass # Scalar fields price_desc = row.get('price_desc', '').strip() price_curve_type = row.get('price_curve_type', '').strip() active_raw = row.get('active', 'TRUE').strip().upper() active = active_raw in ('TRUE', '1', 'YES') # Create the record print(f" Creating price index...") record = PriceIndex() record.price_index = index_name record.price_desc = price_desc record.price_currency = currency record.price_curve_type = price_curve_type record.price_period = product_month record.active = active if price_type: record.price_type = price_type if price_unit: record.price_unit = price_unit record.save() print(f" ✓ Created (ID: {record.id})") print(f" Index : {index_name}") print(f" Period : {period_label} (ID: {product_month.id})") print(f" Desc : {price_desc}") print() imported_count += 1 except Exception as e: error_msg = f"Row {row_num} - {index_name or 'Unknown'}: {str(e)}" errors.append(error_msg) error_count += 1 print(f" ✗ Error on row {row_num}: {e}\n") import traceback traceback.print_exc() except FileNotFoundError: print(f"✗ Error: CSV file not found at {csv_file}") print(f"Please update PRICE_INDEXES_CSV in helpers/config.py with the correct path.") return except Exception as e: print(f"✗ Fatal error: {e}") import traceback traceback.print_exc() return # Summary print(f"\n{'='*70}") print("IMPORT SUMMARY") print(f"{'='*70}") print(f"Total rows processed : {row_num}") print(f"Successfully imported : {imported_count}") print(f"Skipped / Not imported : {skipped_count}") print(f"Errors : {error_count}") if ignored: print(f"\nIgnored / Not imported ({len(ignored)}):") for r, name, reason in ignored: print(f" - Row {r:>4} [{name}]: {reason}") if errors: print(f"\nErrors ({len(errors)}):") for err in errors: print(f" - {err}") print(f"\n{'='*70}") # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main(): print("=" * 70) print("TRYTON PRICE INDEX IMPORT SCRIPT") print("Using Proteus with XML-RPC Connection") print("=" * 70) print() if not connect_to_tryton(): return 1 import_price_indexes(CSV_FILE_PATH) return 0 if __name__ == '__main__': exit(main())