import json import logging from datetime import date from typing import Any, Dict, List, Optional import aiohttp from fastapi import HTTPException from app.core.config import settings from app.core.database import execute_query, execute_query_single logger = logging.getLogger(__name__) def _economic_error_message(status: int, response_text: str) -> str: """Return a useful, bounded e-conomic validation error without headers or credentials.""" raw = str(response_text or "").strip() messages: List[str] = [] try: payload = json.loads(raw) except (TypeError, ValueError): payload = None def collect(value: Any, prefix: str = "") -> None: if isinstance(value, dict): for key, child in value.items(): if str(key).lower() in {"developerhint", "logid", "httpstatuscode"}: continue collect(child, f"{prefix}.{key}".strip(".")) elif isinstance(value, list): for child in value: collect(child, prefix) elif value not in (None, "") and str(value) not in messages: label = f"{prefix}: " if prefix else "" messages.append(f"{label}{value}") if payload is not None: collect(payload) elif raw: messages.append(raw) detail = " · ".join(messages)[:1200] return f"e-conomic afviste ordren ({status})" + (f": {detail}" if detail else "") class OrdreEconomicExportService: """e-conomic export service for global ordre page.""" def __init__(self): self.api_url = settings.ECONOMIC_API_URL self.app_secret_token = settings.ECONOMIC_APP_SECRET_TOKEN self.agreement_grant_token = settings.ECONOMIC_AGREEMENT_GRANT_TOKEN configured_fields = getattr(settings, "model_fields_set", set()) self.read_only = ( settings.ORDRE_ECONOMIC_READ_ONLY if "ORDRE_ECONOMIC_READ_ONLY" in configured_fields else settings.ECONOMIC_READ_ONLY ) self.dry_run = ( settings.ORDRE_ECONOMIC_DRY_RUN if "ORDRE_ECONOMIC_DRY_RUN" in configured_fields else settings.ECONOMIC_DRY_RUN ) self.default_layout = settings.ORDRE_ECONOMIC_LAYOUT self.default_product = settings.ORDRE_ECONOMIC_PRODUCT if self.read_only: logger.warning("🔒 ORDRE e-conomic READ-ONLY mode: Enabled") if self.dry_run: logger.warning("🏃 ORDRE e-conomic DRY-RUN mode: Enabled") if not self.read_only: logger.error("⚠️ WARNING: ORDRE e-conomic READ-ONLY disabled!") def _headers(self) -> Dict[str, str]: return { "X-AppSecretToken": self.app_secret_token, "X-AgreementGrantToken": self.agreement_grant_token, "Content-Type": "application/json", } def _check_write_permission(self, operation: str) -> bool: if self.read_only: logger.error("🚫 BLOCKED: %s - READ_ONLY mode enabled", operation) return False if self.dry_run: logger.warning("🏃 DRY-RUN: %s - Would execute but not sending", operation) return False logger.warning("⚠️ EXECUTING WRITE: %s", operation) return True async def export_order( self, customer_id: int, lines: List[Dict[str, Any]], notes: Optional[str] = None, layout_number: Optional[int] = None, user_id: Optional[int] = None, ) -> Dict[str, Any]: customer = execute_query_single( "SELECT id, name, economic_customer_number FROM customers WHERE id = %s", (customer_id,), ) if not customer: raise HTTPException(status_code=404, detail="Customer not found") if not customer.get("economic_customer_number"): raise HTTPException( status_code=400, detail=( f"Kan ikke overfoere ordre til e-conomic: Customer '{customer.get('name')}' " "mangler e-conomic kundenummer. Lokal ordre er bevaret." ), ) selected_lines = [line for line in lines if bool(line.get("selected", True))] if not selected_lines: raise HTTPException(status_code=400, detail="Ingen linjer valgt til eksport") product_ids = [int(line["product_id"]) for line in selected_lines if line.get("product_id")] product_map: Dict[int, str] = {} if product_ids: product_rows = execute_query( "SELECT id, sku_internal FROM products WHERE id = ANY(%s)", (product_ids,), ) or [] product_map = { int(row["id"]): str(row["sku_internal"]) for row in product_rows if row.get("sku_internal") } economic_lines: List[Dict[str, Any]] = [] for line in selected_lines: try: quantity = float(line.get("quantity") or 0) unit_price = float(line.get("unit_price") or 0) discount = float(line.get("discount_percentage") or 0) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Ugyldige tal i linjer") if quantity <= 0: raise HTTPException(status_code=400, detail="Linje quantity skal være > 0") if unit_price < 0: raise HTTPException(status_code=400, detail="Linje unit_price skal være >= 0") line_payload: Dict[str, Any] = { "description": line.get("description") or "Ordrelinje", "quantity": quantity, "unitNetPrice": unit_price, } product_id = line.get("product_id") product_number = None if product_id is not None: try: product_number = product_map.get(int(product_id)) except (TypeError, ValueError): product_number = None if not product_number: product_number = self.default_product if product_number: line_payload["product"] = {"productNumber": str(product_number)} if discount > 0: line_payload["discountPercentage"] = discount economic_lines.append(line_payload) operation = f"Export ordre for customer {customer_id} to e-conomic" write_allowed = self._check_write_permission(operation) async with aiohttp.ClientSession() as session: customer_number = int(customer["economic_customer_number"]) async with session.get( f"{self.api_url}/customers/{customer_number}", headers=self._headers(), timeout=aiohttp.ClientTimeout(total=30), ) as customer_response: customer_text = await customer_response.text() if customer_response.status != 200: logger.error("❌ e-conomic customer lookup failed (%s): %s", customer_response.status, customer_text) raise HTTPException( status_code=502, detail=_economic_error_message(customer_response.status, customer_text), ) economic_customer = await customer_response.json(content_type=None) payment_terms = economic_customer.get("paymentTerms") vat_zone = economic_customer.get("vatZone") if not payment_terms or not vat_zone: raise HTTPException( status_code=502, detail="e-conomic-kunden mangler betalingsbetingelser eller momszone", ) payload: Dict[str, Any] = { "date": date.today().isoformat(), "currency": str(economic_customer.get("currency") or "DKK"), "customer": {"customerNumber": customer_number}, "paymentTerms": payment_terms, "recipient": { "name": str(economic_customer.get("name") or customer.get("name") or "Kunde"), "address": str(economic_customer.get("address") or ""), "zip": str(economic_customer.get("zip") or ""), "city": str(economic_customer.get("city") or ""), "country": str(economic_customer.get("country") or ""), "vatZone": vat_zone, }, "layout": { "layoutNumber": int(layout_number or self.default_layout), }, "lines": economic_lines, } if notes: payload["notes"] = {"textLine1": str(notes)[:1000]} if not write_allowed: return { "success": True, "dry_run": True, "message": "DRY-RUN: Export blocked by safety flags", "details": { "customer_id": customer_id, "customer_name": customer.get("name"), "selected_line_count": len(selected_lines), "read_only": self.read_only, "dry_run": self.dry_run, "user_id": user_id, "payload": payload, }, } logger.info("📤 Sending ordre payload to e-conomic: %s", json.dumps(payload, default=str)) async with session.post( f"{self.api_url}/orders/drafts", headers=self._headers(), json=payload, timeout=aiohttp.ClientTimeout(total=30), ) as response: response_text = await response.text() if response.status not in [200, 201]: logger.error("❌ e-conomic export failed (%s): %s", response.status, response_text) raise HTTPException( status_code=502, detail=_economic_error_message(response.status, response_text), ) export_result = await response.json(content_type=None) draft_number = export_result.get("draftOrderNumber") or export_result.get("orderNumber") logger.info("✅ Ordre exported to e-conomic draft %s", draft_number) return { "success": True, "dry_run": False, "message": f"Ordre eksporteret til e-conomic draft {draft_number}", "economic_draft_id": draft_number, "details": { "customer_id": customer_id, "customer_name": customer.get("name"), "selected_line_count": len(selected_lines), "user_id": user_id, "economic_response": export_result, }, } ordre_economic_export_service = OrdreEconomicExportService()