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__) 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 self.read_only = settings.ORDRE_ECONOMIC_READ_ONLY self.dry_run = settings.ORDRE_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="Kunden mangler e-conomic kundenummer i Customers modulet", ) 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) payload: Dict[str, Any] = { "date": date.today().isoformat(), "currency": "DKK", "customer": { "customerNumber": int(customer["economic_customer_number"]), }, "layout": { "layoutNumber": int(layout_number or self.default_layout), }, "lines": economic_lines, } if notes: payload["notes"] = {"textLine1": str(notes)[:250]} operation = f"Export ordre for customer {customer_id} to e-conomic" if not self._check_write_permission(operation): 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 aiohttp.ClientSession() as session: 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=f"e-conomic export fejlede ({response.status})", ) 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()