631fc6cdca
- Added migration to create economic_change_requests table for tracking change requests for customers and products. - Introduced new permissions for requesting and approving changes in e-conomic. - Developed frontend JavaScript functionality for managing economic catalog, including handling change requests and displaying their statuses. - Created browser tests to validate UI interactions related to product management and change requests. - Added unit tests for backend logic to ensure proper handling of product numbers, price rules, and economic write policies.
324 lines
14 KiB
Python
324 lines
14 KiB
Python
import json
|
|
import logging
|
|
from datetime import date
|
|
from typing import Any, Dict, List, Optional
|
|
from urllib.parse import quote
|
|
|
|
import aiohttp
|
|
from fastapi import HTTPException
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import execute_query, execute_query_single
|
|
from app.core.economic_write_policy import assert_economic_write_allowed
|
|
|
|
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,
|
|
document_key: Optional[str] = None,
|
|
currency: str = 'DKK',
|
|
) -> Dict[str, Any]:
|
|
from app.products.backend.economic_documents import active_connection, export_document, preflight, unsaved_key
|
|
connection = active_connection()
|
|
if connection:
|
|
if not self._check_write_permission(f'Export ordre for customer {customer_id}') or settings.ECONOMIC_READ_ONLY or settings.ECONOMIC_DRY_RUN:
|
|
checked = await preflight(connection, customer_id, lines, layout_number, currency, notes)
|
|
return {'success': True, 'dry_run': True, 'message': 'Safety mode: valideret uden ekstern skrivning', 'details': checked}
|
|
return await export_document(connection, 'order', document_key or unsaved_key(customer_id, lines, notes, layout_number),
|
|
customer_id, lines, layout_number, currency, notes, user_id)
|
|
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",
|
|
)
|
|
|
|
product_numbers = {
|
|
str(line.get("product", {}).get("productNumber") or "").strip()
|
|
for line in economic_lines
|
|
if line.get("product")
|
|
}
|
|
valid_product_numbers = set()
|
|
for product_number in product_numbers:
|
|
if not product_number:
|
|
continue
|
|
async with session.get(
|
|
f"{self.api_url}/products/{quote(product_number, safe='')}",
|
|
headers=self._headers(),
|
|
timeout=aiohttp.ClientTimeout(total=30),
|
|
) as product_response:
|
|
if product_response.status == 200:
|
|
valid_product_numbers.add(product_number)
|
|
elif product_response.status == 404:
|
|
logger.warning(
|
|
"e-conomic product %s does not exist; exporting matching lines as free text",
|
|
product_number,
|
|
)
|
|
else:
|
|
product_text = await product_response.text()
|
|
raise HTTPException(
|
|
status_code=502,
|
|
detail=_economic_error_message(product_response.status, product_text),
|
|
)
|
|
|
|
skipped_product_numbers = sorted(product_numbers - valid_product_numbers)
|
|
for line in economic_lines:
|
|
product_number = str(line.get("product", {}).get("productNumber") or "").strip()
|
|
if product_number and product_number not in valid_product_numbers:
|
|
line.pop("product", None)
|
|
|
|
customer_layout = economic_customer.get("layout") or {}
|
|
resolved_layout_number = customer_layout.get("layoutNumber") or self.default_layout
|
|
|
|
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(resolved_layout_number),
|
|
},
|
|
"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,
|
|
"skipped_product_numbers": skipped_product_numbers,
|
|
"payload": payload,
|
|
},
|
|
}
|
|
|
|
logger.info("📤 Sending ordre payload to e-conomic: %s", json.dumps(payload, default=str))
|
|
|
|
assert_economic_write_allowed("POST", "/orders/drafts")
|
|
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,
|
|
"skipped_product_numbers": skipped_product_numbers,
|
|
"economic_response": export_result,
|
|
},
|
|
}
|
|
|
|
|
|
ordre_economic_export_service = OrdreEconomicExportService()
|