import html import json import logging import oracledb from datetime import datetime, timedelta from .. import database logger = logging.getLogger(__name__) # Diacritics to ASCII mapping (Romanian) _DIACRITICS = str.maketrans({ '\u0103': 'a', # ă '\u00e2': 'a', # â '\u00ee': 'i', # î '\u0219': 's', # ș '\u021b': 't', # ț '\u0102': 'A', # Ă '\u00c2': 'A', # Â '\u00ce': 'I', # Î '\u0218': 'S', # Ș '\u021a': 'T', # Ț # Older Unicode variants '\u015f': 's', # ş (cedilla) '\u0163': 't', # ţ (cedilla) '\u015e': 'S', # Ş '\u0162': 'T', # Ţ }) def clean_web_text(text: str) -> str: """Port of VFP CleanWebText: unescape HTML entities + diacritics to ASCII.""" if not text: return "" result = html.unescape(text) result = result.translate(_DIACRITICS) # Remove any remaining
tags for br in ('
', '
', '
'): result = result.replace(br, ' ') return result.strip() def convert_web_date(date_str: str) -> datetime: """Port of VFP ConvertWebDate: parse web date to datetime.""" if not date_str: return datetime.now() try: return datetime.strptime(date_str.strip(), '%Y-%m-%d %H:%M:%S') except ValueError: try: return datetime.strptime(date_str.strip()[:10], '%Y-%m-%d') except ValueError: return datetime.now() def format_address_for_oracle(address: str, city: str, region: str) -> str: """Port of VFP FormatAddressForOracle.""" region_clean = clean_web_text(region) city_clean = clean_web_text(city) address_clean = clean_web_text(address) return f"JUD:{region_clean};{city_clean};{address_clean}" def build_articles_json(items, order=None, settings=None) -> str: """Build JSON string for Oracle PACK_IMPORT_COMENZI.importa_comanda. Includes transport and discount as extra articles if configured.""" articles = [] for item in items: articles.append({ "sku": item.sku, "quantity": str(item.quantity), "price": str(item.price), "vat": str(item.vat), "name": clean_web_text(item.name) }) if order and settings: transport_codmat = settings.get("transport_codmat", "") transport_vat = settings.get("transport_vat", "21") discount_codmat = settings.get("discount_codmat", "") # Transport as article with quantity +1 if order.delivery_cost > 0 and transport_codmat: article_dict = { "sku": transport_codmat, "quantity": "1", "price": str(order.delivery_cost), "vat": transport_vat, "name": "Transport" } if settings.get("transport_id_pol"): article_dict["id_pol"] = settings["transport_id_pol"] articles.append(article_dict) # Discount total with quantity -1 (positive price) if order.discount_total > 0 and discount_codmat: discount_vat = settings.get("discount_vat", "19") article_dict = { "sku": discount_codmat, "quantity": "-1", "price": str(order.discount_total), "vat": discount_vat, "name": "Discount" } if settings.get("discount_id_pol"): article_dict["id_pol"] = settings["discount_id_pol"] articles.append(article_dict) return json.dumps(articles) def import_single_order(order, id_pol: int = None, id_sectie: int = None, app_settings: dict = None) -> dict: """Import a single order into Oracle ROA. Returns dict with: success: bool id_comanda: int or None id_partener: int or None id_adresa_facturare: int or None id_adresa_livrare: int or None error: str or None """ result = { "success": False, "id_comanda": None, "id_partener": None, "id_adresa_facturare": None, "id_adresa_livrare": None, "error": None } try: order_number = clean_web_text(order.number) order_date = convert_web_date(order.date) logger.info( f"Order {order.number}: raw date={order.date!r} → " f"parsed={order_date.strftime('%Y-%m-%d %H:%M:%S')}" ) if database.pool is None: raise RuntimeError("Oracle pool not initialized") with database.pool.acquire() as conn: with conn.cursor() as cur: # Step 1: Process partner — use shipping person data for name id_partener = cur.var(oracledb.DB_TYPE_NUMBER) if order.billing.is_company: denumire = clean_web_text(order.billing.company_name).upper() cod_fiscal = clean_web_text(order.billing.company_code) or None registru = clean_web_text(order.billing.company_reg) or None is_pj = 1 else: # Use shipping person for partner name (person on shipping label) if order.shipping and (order.shipping.lastname or order.shipping.firstname): denumire = clean_web_text( f"{order.shipping.lastname} {order.shipping.firstname}" ).upper() else: denumire = clean_web_text( f"{order.billing.lastname} {order.billing.firstname}" ).upper() cod_fiscal = None registru = None is_pj = 0 cur.callproc("PACK_IMPORT_PARTENERI.cauta_sau_creeaza_partener", [ cod_fiscal, denumire, registru, is_pj, id_partener ]) partner_id = id_partener.getvalue() if not partner_id or partner_id <= 0: result["error"] = f"Partner creation failed for {denumire}" return result result["id_partener"] = int(partner_id) # Determine if billing and shipping are different persons billing_name = clean_web_text( f"{order.billing.lastname} {order.billing.firstname}" ).strip().upper() shipping_name = "" if order.shipping: shipping_name = clean_web_text( f"{order.shipping.lastname} {order.shipping.firstname}" ).strip().upper() different_person = bool( shipping_name and billing_name and shipping_name != billing_name ) # Step 2: Process shipping address (primary — person on shipping label) # Use shipping person phone/email for partner contact shipping_phone = "" shipping_email = "" if order.shipping: shipping_phone = order.shipping.phone or "" shipping_email = order.shipping.email or "" if not shipping_phone: shipping_phone = order.billing.phone or "" if not shipping_email: shipping_email = order.billing.email or "" addr_livr_id = None if order.shipping: id_adresa_livr = cur.var(oracledb.DB_TYPE_NUMBER) shipping_addr = format_address_for_oracle( order.shipping.address, order.shipping.city, order.shipping.region ) cur.callproc("PACK_IMPORT_PARTENERI.cauta_sau_creeaza_adresa", [ partner_id, shipping_addr, shipping_phone, shipping_email, id_adresa_livr ]) addr_livr_id = id_adresa_livr.getvalue() # Step 3: Process billing address if different_person: # Different person: use shipping address for BOTH billing and shipping in ROA addr_fact_id = addr_livr_id else: # Same person: use billing address as-is id_adresa_fact = cur.var(oracledb.DB_TYPE_NUMBER) billing_addr = format_address_for_oracle( order.billing.address, order.billing.city, order.billing.region ) cur.callproc("PACK_IMPORT_PARTENERI.cauta_sau_creeaza_adresa", [ partner_id, billing_addr, order.billing.phone or "", order.billing.email or "", id_adresa_fact ]) addr_fact_id = id_adresa_fact.getvalue() if addr_fact_id is not None: result["id_adresa_facturare"] = int(addr_fact_id) if addr_livr_id is not None: result["id_adresa_livrare"] = int(addr_livr_id) # Step 4: Build articles JSON and import order articles_json = build_articles_json(order.items, order, app_settings) # Use CLOB for the JSON clob_var = cur.var(oracledb.DB_TYPE_CLOB) clob_var.setvalue(0, articles_json) id_comanda = cur.var(oracledb.DB_TYPE_NUMBER) cur.callproc("PACK_IMPORT_COMENZI.importa_comanda", [ order_number, # p_nr_comanda_ext order_date, # p_data_comanda partner_id, # p_id_partener clob_var, # p_json_articole (CLOB) addr_livr_id, # p_id_adresa_livrare addr_fact_id, # p_id_adresa_facturare id_pol, # p_id_pol id_sectie, # p_id_sectie id_comanda # v_id_comanda (OUT) ]) comanda_id = id_comanda.getvalue() if comanda_id and comanda_id > 0: conn.commit() result["success"] = True result["id_comanda"] = int(comanda_id) logger.info(f"Order {order_number} imported: ID={comanda_id}") else: conn.rollback() result["error"] = "importa_comanda returned invalid ID" except oracledb.DatabaseError as e: error_msg = str(e) result["error"] = error_msg logger.error(f"Oracle error importing order {order.number}: {error_msg}") except Exception as e: result["error"] = str(e) logger.error(f"Error importing order {order.number}: {e}") return result