import hashlib import json import logging import csv import io import unicodedata import zipfile from datetime import datetime, timedelta from decimal import Decimal from pathlib import Path import re import xml.etree.ElementTree as ET from typing import Any, Dict, List, Optional from fastapi import HTTPException import httpx from app.core.config import settings from app.core.database import execute_query, execute_query_single, execute_update, get_db_connection, release_db_connection, table_has_column from psycopg2.extras import RealDictCursor from app.modules.also.models.schemas import ( AlsoCompanyMappingUpsert, AlsoImportJobCreate, AlsoImportLinesRequest, AlsoProductMappingUpsert, ) logger = logging.getLogger(__name__) ALSO_EFFECTIVE_COST_SQL = ( "COALESCE(" "l.cost_amount, " "CASE " "WHEN l.matched_product_id IS NOT NULL THEN " "COALESCE(p.supplier_price, 0) * COALESCE(NULLIF(l.billable_parameters, 0), 1) " "ELSE 0 " "END, " "0)" ) def _json_default(value: Any) -> Any: if isinstance(value, Decimal): return float(value) if isinstance(value, datetime): return value.isoformat() return str(value) def _json_dumps(value: Any) -> str: return json.dumps(value, ensure_ascii=False, default=_json_default) def _normalized_text(value: Optional[Any]) -> str: return str(value or "").strip() def _normalize_match_key(value: Optional[Any]) -> str: text = _normalized_text(value) if not text: return "" text = unicodedata.normalize("NFKD", text) text = "".join(ch for ch in text if not unicodedata.combining(ch)) text = re.sub(r"[^a-zA-Z0-9]+", "", text).lower() return text def _to_decimal(value: Any, default: Decimal = Decimal("0")) -> Decimal: if value is None or value == "": return default try: return Decimal(str(value)) except Exception: return default def _normalize_cvr(vat_value: Optional[str]) -> str: digits = re.sub(r"[^0-9]", "", _normalized_text(vat_value)) return digits def _normalize_download_url(download_url: str) -> str: raw = _normalized_text(download_url) if not raw: return raw match = re.match(r"^(https://marketplace\.also\.[^/]+)/#/Redirect/(.+)$", raw, re.IGNORECASE) if match: return f"{match.group(1)}/{match.group(2)}" return raw def _normalize_header(value: Optional[str]) -> str: return re.sub(r"[^a-z0-9]+", "_", _normalized_text(value).strip().lower()).strip("_") def _parse_decimal_candidate(value: Any) -> Optional[Decimal]: if value is None: return None if isinstance(value, Decimal): return value if isinstance(value, (int, float)): return Decimal(str(value)) raw = str(value).strip() if not raw: return None raw = raw.replace("\u00a0", "").replace(" ", "") raw = raw.replace("DKK", "").replace("EUR", "").replace("USD", "") raw = raw.replace("%", "") if raw.count(",") == 1 and raw.count(".") > 1: raw = raw.replace(".", "").replace(",", ".") elif raw.count(",") == 1 and raw.count(".") == 0: raw = raw.replace(",", ".") elif raw.count(",") > 1 and raw.count(".") == 0: raw = raw.replace(",", "") elif raw.count(".") > 1 and raw.count(",") == 0: raw = raw.replace(".", "") elif "," in raw and "." in raw: if raw.rfind(",") > raw.rfind("."): raw = raw.replace(".", "").replace(",", ".") else: raw = raw.replace(",", "") try: return Decimal(raw) except Exception: return None def _parse_date_candidate(value: Any) -> Optional[str]: if value is None: return None if isinstance(value, datetime): return value.date().isoformat() if hasattr(value, "isoformat"): try: iso = value.isoformat() if isinstance(iso, str) and iso: return iso[:10] except Exception: pass raw = str(value).strip() if not raw: return None numeric_candidate = _parse_decimal_candidate(raw) if numeric_candidate is not None: try: numeric_float = float(numeric_candidate) if 20000 <= numeric_float <= 80000: base = datetime(1899, 12, 30) return (base + timedelta(days=numeric_float)).date().isoformat() except Exception: pass for fmt in ("%Y-%m-%d", "%d-%m-%Y", "%d/%m/%Y", "%Y/%m/%d", "%m/%d/%Y", "%d.%m.%Y"): try: return datetime.strptime(raw, fmt).date().isoformat() except ValueError: continue try: return datetime.fromisoformat(raw.replace("Z", "+00:00")).date().isoformat() except Exception: return None def _flatten_json_object(value: Dict[str, Any], prefix: str = "") -> Dict[str, Any]: flattened: Dict[str, Any] = {} for raw_key, raw_val in value.items(): key = _normalize_header(raw_key) if not key: continue target_key = f"{prefix}_{key}" if prefix else key if isinstance(raw_val, dict): flattened.update(_flatten_json_object(raw_val, target_key)) elif isinstance(raw_val, list): continue else: flattened[target_key] = raw_val if key not in flattened: flattened[key] = raw_val return flattened def _derive_file_context(file_name: str) -> Dict[str, Any]: stem = Path(file_name or "").stem context: Dict[str, Any] = {} date_match = re.search(r"(.+?)_(\d{2}-\d{2}-\d{4})_(\d{2}-\d{2}-\d{4})$", stem) if date_match: company = date_match.group(1).replace("_", " ").strip() if company: context["company"] = company start = _parse_date_candidate(date_match.group(2)) end = _parse_date_candidate(date_match.group(3)) if start: context["billing_start"] = start if end: context["billing_end"] = end return context if stem: context["company"] = stem.replace("_", " ").strip() return context def _has_candidate_fields(row: Dict[str, Any]) -> bool: candidate_keys = ( "company", "customer", "customer_name", "tenant", "companydisplayname", "product", "product_name", "productdisplayname", "productname", "subscription", "description", "material", "material_number", "sku", "amount", "total", "charge", "price", "sales_price", "cost_amount", "billing_start", "period_start", "startdate", ) return any(_normalized_text(row.get(key)) for key in candidate_keys) def _is_probable_leaf_json_row(row: Dict[str, Any], has_nested_children: bool) -> bool: product_keys = ("product_name", "productdisplayname", "productname", "material_number", "sku", "description") financial_keys = ("charge", "amount", "total", "total_price", "sales_price", "price") period_keys = ("billing_start", "period_start", "startdate") has_product = any(_normalized_text(row.get(key)) for key in product_keys) has_financial = any(_normalized_text(row.get(key)) for key in financial_keys) has_period = any(_normalized_text(row.get(key)) for key in period_keys) if has_product: return True if has_financial and has_period: return True if not has_nested_children and _has_candidate_fields(row): return True return False def _extract_quantity_from_field_values(value: Any) -> Optional[Decimal]: text = _normalized_text(value) if not text: return None match = re.search(r"Quantity\s*=\s*([0-9]+(?:[.,][0-9]+)?)", text, re.IGNORECASE) if not match: return None return _parse_decimal_candidate(match.group(1)) def _extract_period_start(value: Any) -> Optional[str]: text = _normalized_text(value) if not text: return None match = re.match(r"\s*(\d{2}[./-]\d{2}[./-]\d{4})\s*-\s*(\d{2}[./-]\d{2}[./-]\d{4})\s*$", text) if not match: return None return _parse_date_candidate(match.group(1)) def _is_zero_value_tenant_line(line: Dict[str, Any]) -> bool: product_name = _normalize_match_key(line.get("product_name")) if "microsoftorganizationtenant" not in product_name: return False total_price = _to_decimal(line.get("total_price"), default=Decimal("0")) unit_price = _to_decimal(line.get("unit_price"), default=Decimal("0")) return total_price == 0 and unit_price == 0 def _line_hash(job_id: int, line_payload: Dict[str, Any]) -> str: source_ref = _normalized_text(line_payload.get("source_line_ref")) if source_ref: key = f"{job_id}|{source_ref}" else: key = "|".join( [ str(job_id), _normalized_text(line_payload.get("company")).lower(), _normalized_text(line_payload.get("customer_id")), _normalized_text(line_payload.get("material_number")), _normalized_text(line_payload.get("vendor")).lower(), _normalized_text(line_payload.get("billing_start")), _normalized_text(line_payload.get("total_price")), ] ) return hashlib.sha256(key.encode("utf-8")).hexdigest() class AlsoService: HEADER_ALIASES = { "company": ["company", "customer", "customer_name", "firma", "company_name", "kunde", "tenant", "billing_customer", "name", "tenant_name", "customer_company_name", "account_name", "company_display_name", "companydisplayname"], "customer_id": ["customer_id", "customerid", "kunde_id", "account_customer_id", "external_customer_id"], "account_id": ["account_id", "accountid", "tenant_id", "subscription_id", "company_account_id", "companyaccountid"], "vat": ["vat", "vat_number", "cvr", "cvr_number", "orgnr", "organization_number", "customer_vat_id", "department_vat_id"], "also_company_id": ["also_company_id", "company_id", "cloud_company_id", "reseller_customer_id", "company_account_id", "companyaccountid"], "material_number": ["material_number", "material", "material_no", "sku", "item_number", "product_code", "varenummer", "part_number", "article_number", "article_no", "service_code"], "product_name": ["product_name", "product", "description", "service_name", "item_description", "subscription_name", "product_description", "service", "subscription", "offer_name", "license_name", "item_name", "product_display_name", "productdisplayname", "productname"], "vendor": ["vendor", "manufacturer", "brand", "leverandor", "publisher", "supplier"], "cost_amount": ["cost_amount", "cost", "costs", "purchase_price", "buy_price", "costprice", "indkob", "cost_total"], "sales_price": ["sales_price", "sales", "sell_price", "list_price", "omsaetning", "revenue", "sales_total"], "unit_price": ["unit_price", "price", "unit_cost", "price_per_unit", "monthly_price", "sales_price_per_unit", "unit_sales_price", "sales_price_of_unit", "charge"], "total_price": ["total_price", "amount", "line_total", "net_amount", "subtotal", "extended_price", "total", "sales_price_total", "sales_price", "total_amount", "charge"], "currency": ["currency", "valuta"], "billing_start": ["billing_start", "period_start", "start_date", "billing_start_date", "invoice_date", "service_period_start", "billing_month", "period_from", "billing_from", "valid_from", "from_date", "start_date", "startdate"], "charge_interval": ["charge_interval", "actual_charge_interval", "actualchargeinterval", "term", "commitment", "period_type", "contract_term"], "billing_interval": ["billing_interval", "interval", "billing_cycle", "frequency", "charge_frequency"], "billable_parameters": ["billable_parameters", "billableparameters", "quantity", "qty", "udrc_value", "licenses", "seats", "users", "units", "antal", "license_count", "unit_count", "count"], "source_line_ref": ["source_line_ref", "line_id", "line_ref", "id", "reference"], "line_no": ["line_no", "line_number", "lineno", "row_number"], } @property def enabled(self) -> bool: return bool(settings.ALSO_ENABLED) @property def read_only(self) -> bool: return bool(settings.ALSO_READ_ONLY) @property def dry_run(self) -> bool: return bool(settings.ALSO_DRY_RUN) def _assert_enabled(self) -> None: if not self.enabled: raise HTTPException(status_code=503, detail="ALSO integration is disabled") def get_config(self) -> Dict[str, Any]: return { "enabled": self.enabled, "read_only": self.read_only, "dry_run": self.dry_run, "api_base_url": settings.ALSO_API_BASE_URL, "preferred_import_order": ["api", "xml", "json_export", "xml_export", "csv"], } def _mark_import_job_failed(self, job_id: int, error_message: str) -> None: execute_update( """ UPDATE also_import_jobs SET status = 'failed', finished_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP, log_json = COALESCE(log_json, '[]'::jsonb) || %s::jsonb WHERE id = %s """, ( _json_dumps( [ { "timestamp": datetime.utcnow().isoformat(), "level": "error", "message": error_message[:1000], } ] ), job_id, ), ) def _complete_import_job( self, job_id: int, *, status_hint: Optional[str] = None, ) -> Dict[str, Any]: queue_totals = execute_query_single( """ SELECT COUNT(*)::INTEGER AS total_lines, COUNT(*) FILTER (WHERE queue_status = 'approved')::INTEGER AS approved_lines, COUNT(*) FILTER (WHERE queue_status = 'ready_for_approval')::INTEGER AS ready_lines, COUNT(*) FILTER (WHERE queue_status = 'error')::INTEGER AS error_lines, COUNT(*) FILTER (WHERE matched_customer_id IS NULL)::INTEGER AS unmatched_customers, COUNT(*) FILTER (WHERE matched_product_id IS NULL)::INTEGER AS unmatched_products FROM also_import_lines WHERE import_job_id = %s """, (job_id,), ) or {} final_status = status_hint or "completed" if ( int(queue_totals.get("error_lines") or 0) > 0 or int(queue_totals.get("unmatched_customers") or 0) > 0 or int(queue_totals.get("unmatched_products") or 0) > 0 ): final_status = "partial" execute_update( """ UPDATE also_import_jobs SET status = %s, finished_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = %s """, (final_status, job_id), ) return {"status": final_status, "queue_totals": queue_totals} def _resolve_customer_match(self, line: Dict[str, Any]) -> Optional[int]: also_company_id = _normalized_text(line.get("also_company_id")) if also_company_id: mapped = execute_query_single( """ SELECT customer_id FROM also_company_mapping WHERE also_company_id = %s AND is_active = true LIMIT 1 """, (also_company_id,), ) if mapped and mapped.get("customer_id"): return int(mapped["customer_id"]) external_customer_id = _normalized_text(line.get("customer_id")) if external_customer_id: row = execute_query_single( "SELECT id FROM customers WHERE vtiger_id = %s LIMIT 1", (external_customer_id,), ) if row and row.get("id"): return int(row["id"]) vat = _normalize_cvr(line.get("vat")) if vat: cvr_column = "cvr_number" if table_has_column("customers", "cvr_number") else None if cvr_column: row = execute_query_single( f"SELECT id FROM customers WHERE REPLACE(REPLACE(COALESCE({cvr_column}, ''), ' ', ''), '-', '') = %s LIMIT 1", (vat,), ) if row and row.get("id"): return int(row["id"]) company_name = _normalized_text(line.get("company")) if company_name: row = execute_query_single( "SELECT id FROM customers WHERE LOWER(name) = LOWER(%s) LIMIT 1", (company_name,), ) if row and row.get("id"): return int(row["id"]) normalized_company = _normalize_match_key(company_name) if normalized_company: candidate_rows = execute_query( """ SELECT id, name FROM customers WHERE name ILIKE %s ORDER BY CHAR_LENGTH(name) ASC, id ASC LIMIT 50 """, (f"%{company_name}%",), ) or [] for candidate in candidate_rows: candidate_name = candidate.get("name") if _normalize_match_key(candidate_name) == normalized_company: return int(candidate["id"]) for candidate in candidate_rows: candidate_key = _normalize_match_key(candidate.get("name")) if candidate_key.startswith(normalized_company) or normalized_company.startswith(candidate_key): return int(candidate["id"]) fallback_candidates = execute_query( """ SELECT id, name FROM customers ORDER BY id ASC LIMIT 5000 """, (), ) or [] for candidate in fallback_candidates: candidate_key = _normalize_match_key(candidate.get("name")) if candidate_key == normalized_company: return int(candidate["id"]) for candidate in fallback_candidates: candidate_key = _normalize_match_key(candidate.get("name")) if normalized_company and (normalized_company in candidate_key or candidate_key in normalized_company): return int(candidate["id"]) return None def _resolve_product_match(self, line: Dict[str, Any]) -> Optional[int]: material_number = _normalized_text(line.get("material_number")) vendor = _normalized_text(line.get("vendor")) product_name = _normalized_text(line.get("product_name")) if material_number and vendor: mapped = execute_query_single( """ SELECT hub_product_id FROM also_product_mapping WHERE material_number = %s AND LOWER(vendor) = LOWER(%s) AND is_active = true LIMIT 1 """, (material_number, vendor), ) if mapped and mapped.get("hub_product_id"): return int(mapped["hub_product_id"]) elif material_number: mapped_without_vendor = execute_query_single( """ SELECT hub_product_id FROM also_product_mapping WHERE material_number = %s AND is_active = true ORDER BY id ASC LIMIT 1 """, (material_number,), ) if mapped_without_vendor and mapped_without_vendor.get("hub_product_id"): return int(mapped_without_vendor["hub_product_id"]) if material_number: direct_product = execute_query_single( """ SELECT id FROM products WHERE COALESCE(deleted_at, NULL) IS NULL AND ( sku_internal = %s OR supplier_sku = %s OR LOWER(REGEXP_REPLACE(COALESCE(sku_internal, ''), '[^a-zA-Z0-9]', '', 'g')) = %s OR LOWER(REGEXP_REPLACE(COALESCE(supplier_sku, ''), '[^a-zA-Z0-9]', '', 'g')) = %s ) ORDER BY id ASC LIMIT 1 """, ( material_number, material_number, _normalize_match_key(material_number), _normalize_match_key(material_number), ), ) if direct_product and direct_product.get("id"): return int(direct_product["id"]) supplier = execute_query_single( """ SELECT product_id FROM product_suppliers WHERE supplier_sku = %s AND ( %s = '' OR LOWER(COALESCE(supplier_name, '')) = LOWER(%s) OR LOWER(COALESCE(supplier_code, '')) = LOWER(%s) ) ORDER BY id ASC LIMIT 1 """, (material_number, vendor, vendor, vendor), ) if supplier and supplier.get("product_id"): return int(supplier["product_id"]) supplier_without_vendor = execute_query_single( """ SELECT product_id FROM product_suppliers WHERE supplier_sku = %s ORDER BY id ASC LIMIT 1 """, (material_number,), ) if supplier_without_vendor and supplier_without_vendor.get("product_id"): return int(supplier_without_vendor["product_id"]) if product_name: row = execute_query_single( "SELECT id FROM products WHERE LOWER(name) = LOWER(%s) LIMIT 1", (product_name,), ) if row and row.get("id"): return int(row["id"]) fuzzy = execute_query( """ SELECT id, name FROM products WHERE name ILIKE %s ORDER BY CHAR_LENGTH(name) ASC, id ASC LIMIT 25 """, (f"%{product_name}%",), ) or [] normalized_product = _normalize_match_key(product_name) for candidate in fuzzy: candidate_key = _normalize_match_key(candidate.get("name")) if candidate_key == normalized_product: return int(candidate["id"]) for candidate in fuzzy: candidate_key = _normalize_match_key(candidate.get("name")) if normalized_product and (normalized_product in candidate_key or candidate_key in normalized_product): return int(candidate["id"]) return None def _auto_map_customer_for_line(self, line: Dict[str, Any]) -> bool: also_company_id = _normalized_text(line.get("also_company_id")) if not also_company_id: return False existing = execute_query_single( """ SELECT id FROM also_company_mapping WHERE also_company_id = %s AND is_active = true LIMIT 1 """, (also_company_id,), ) if existing: return False candidate_customer_id = self._resolve_customer_match(line) if not candidate_customer_id: return False execute_query( """ INSERT INTO also_company_mapping ( also_company_id, also_customer_id, customer_id, match_confidence, notes, is_active ) VALUES (%s, %s, %s, %s, %s, true) ON CONFLICT (also_company_id) DO UPDATE SET customer_id = EXCLUDED.customer_id, also_customer_id = EXCLUDED.also_customer_id, match_confidence = EXCLUDED.match_confidence, notes = EXCLUDED.notes, is_active = true, updated_at = CURRENT_TIMESTAMP """, ( also_company_id, _normalized_text(line.get("customer_id")) or _normalized_text(line.get("account_id")) or None, candidate_customer_id, Decimal("0.85"), "Auto-created from ALSO import", ), ) return True def _auto_map_product_for_line(self, line: Dict[str, Any]) -> bool: material_number = _normalized_text(line.get("material_number")) if not material_number or _is_zero_value_tenant_line(line): return False vendor = _normalized_text(line.get("vendor")) or "ALSO" existing = execute_query_single( """ SELECT id FROM also_product_mapping WHERE material_number = %s AND LOWER(vendor) = LOWER(%s) AND is_active = true LIMIT 1 """, (material_number, vendor), ) if existing: return False candidate_product_id = self._resolve_product_match( { **line, "vendor": line.get("vendor") or vendor, } ) if not candidate_product_id: return False execute_query( """ INSERT INTO also_product_mapping ( material_number, vendor, hub_product_id, product_name_snapshot, is_active ) VALUES (%s, %s, %s, %s, true) ON CONFLICT (material_number, vendor) DO UPDATE SET hub_product_id = EXCLUDED.hub_product_id, product_name_snapshot = EXCLUDED.product_name_snapshot, is_active = true, updated_at = CURRENT_TIMESTAMP """, ( material_number, vendor, candidate_product_id, _normalized_text(line.get("product_name")) or None, ), ) return True def _create_local_product_for_line(self, line: Dict[str, Any], vendor: str) -> Optional[int]: material_number = _normalized_text(line.get("material_number")) product_name = _normalized_text(line.get("product_name")) if not material_number or not product_name: return None existing = execute_query_single( """ SELECT id FROM products WHERE deleted_at IS NULL AND ( sku_internal = %s OR supplier_sku = %s ) ORDER BY id ASC LIMIT 1 """, (material_number, material_number), ) if existing and existing.get("id"): return int(existing["id"]) manufacturer = vendor if not manufacturer or manufacturer == "ALSO": if "microsoft" in _normalize_match_key(product_name): manufacturer = "Microsoft" else: manufacturer = "ALSO" sales_price = _to_decimal(line.get("unit_price"), default=Decimal("0")) created = execute_query( """ INSERT INTO products ( name, short_description, type, status, sku_internal, manufacturer, supplier_name, supplier_sku, supplier_price, supplier_currency, sales_price, vat_rate, billable ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) RETURNING id """, ( product_name, "Auto-oprettet fra ALSO Cloud Marketplace", "subscription", "active", material_number, manufacturer, "ALSO Cloud Marketplace", material_number, sales_price, _normalized_text(line.get("currency")) or "DKK", sales_price, Decimal("25.00"), True, ), ) or [] if not created: return None product_id = int(created[0]["id"]) execute_query( """ INSERT INTO product_suppliers ( product_id, supplier_name, supplier_code, supplier_sku, supplier_price, supplier_currency, source, last_updated_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP) RETURNING id """, ( product_id, "ALSO Cloud Marketplace", vendor or "ALSO", material_number, sales_price, _normalized_text(line.get("currency")) or "DKK", "also_cloud", ), ) return product_id def auto_map_import_job(self, job_id: int) -> Dict[str, Any]: self._assert_enabled() lines = self.get_import_job_lines(job_id=job_id, status=None, limit=5000) if not lines: raise HTTPException(status_code=404, detail="No lines found for import job") customer_mappings_created = 0 product_mappings_created = 0 products_created = 0 for line in lines: if self._auto_map_customer_for_line(line): customer_mappings_created += 1 if self._auto_map_product_for_line(line): product_mappings_created += 1 continue if _is_zero_value_tenant_line(line): continue if line.get("matched_product_id"): continue material_number = _normalized_text(line.get("material_number")) if not material_number: continue vendor = _normalized_text(line.get("vendor")) or "ALSO" created_product_id = self._create_local_product_for_line(line, vendor=vendor) if not created_product_id: continue products_created += 1 execute_query( """ INSERT INTO also_product_mapping ( material_number, vendor, hub_product_id, product_name_snapshot, is_active ) VALUES (%s, %s, %s, %s, true) ON CONFLICT (material_number, vendor) DO UPDATE SET hub_product_id = EXCLUDED.hub_product_id, product_name_snapshot = EXCLUDED.product_name_snapshot, is_active = true, updated_at = CURRENT_TIMESTAMP """, ( material_number, vendor, created_product_id, _normalized_text(line.get("product_name")) or None, ), ) product_mappings_created += 1 matching_result = self.run_matching(import_job_id=job_id, line_ids=[], limit=5000) validation_result = self.run_validation(import_job_id=job_id, line_ids=[], limit=5000) approval_result: Dict[str, Any] = {"approved_lines": 0, "created_drafts": 0, "draft_ids": []} ready_count = int(validation_result.get("ready_for_approval") or 0) if ready_count > 0: approval_result = self.approve_lines_to_drafts( import_job_id=job_id, line_ids=[], approved_by_user_id=None, ) completion = self._complete_import_job(job_id) return { "job_id": job_id, "customer_mappings_created": customer_mappings_created, "product_mappings_created": product_mappings_created, "products_created": products_created, "matching_result": matching_result, "validation_result": validation_result, "approval_result": approval_result, "queue_totals": completion["queue_totals"], "status": completion["status"], } def _derive_queue_status(self, matched_customer_id: Optional[int], matched_product_id: Optional[int], has_errors: bool) -> str: if has_errors: return "error" if matched_customer_id and matched_product_id: return "ready_for_approval" if not matched_customer_id and matched_product_id: return "matching_customers" if matched_customer_id and not matched_product_id: return "matching_products" return "new" def _build_validation_errors(self, line: Dict[str, Any], duplicate_ids: set[int]) -> List[Dict[str, Any]]: if _is_zero_value_tenant_line(line): return [] errors: List[Dict[str, Any]] = [] if not line.get("matched_customer_id"): errors.append({"code": "customer_not_found", "message": "Kunde ikke fundet/matchet"}) if not line.get("matched_product_id"): errors.append({"code": "product_not_found", "message": "Produkt ikke fundet/matchet"}) total_price = _to_decimal(line.get("total_price"), default=_to_decimal(line.get("sales_price"), Decimal("0"))) if total_price < 0: errors.append({"code": "negative_price", "message": "Negativ pris fundet"}) if total_price == 0: errors.append({"code": "zero_price", "message": "Pris er 0"}) if not line.get("billing_start"): errors.append({"code": "missing_period", "message": "Manglende billing_start/periode"}) currency = _normalized_text(line.get("currency")) if not currency: errors.append({"code": "missing_currency", "message": "Valuta mangler"}) if int(line.get("id") or 0) in duplicate_ids: errors.append({"code": "duplicate_line", "message": "Dublet-linje fundet i samme import"}) if line.get("matched_customer_id") is None and _normalized_text(line.get("also_company_id")): has_company_mapping = execute_query_single( "SELECT id FROM also_company_mapping WHERE also_company_id = %s AND is_active = true LIMIT 1", (_normalized_text(line.get("also_company_id")),), ) if not has_company_mapping: errors.append({"code": "missing_company_mapping", "message": "Manglende company mapping"}) if line.get("matched_product_id") is None and _normalized_text(line.get("material_number")) and _normalized_text(line.get("vendor")): has_product_mapping = execute_query_single( """ SELECT id FROM also_product_mapping WHERE material_number = %s AND LOWER(vendor) = LOWER(%s) AND is_active = true LIMIT 1 """, (_normalized_text(line.get("material_number")), _normalized_text(line.get("vendor"))), ) if not has_product_mapping: errors.append({"code": "missing_product_mapping", "message": "Manglende product mapping"}) return errors def _fetch_process_lines(self, import_job_id: Optional[int], line_ids: Optional[List[int]], limit: int) -> List[Dict[str, Any]]: params: List[Any] = [] where: List[str] = ["queue_status <> 'invoiced'"] if import_job_id: where.append("import_job_id = %s") params.append(import_job_id) if line_ids: placeholders = ",".join(["%s"] * len(line_ids)) where.append(f"id IN ({placeholders})") params.extend(line_ids) params.append(max(1, min(limit, 5000))) return execute_query( f""" SELECT * FROM also_import_lines WHERE {' AND '.join(where)} ORDER BY id ASC LIMIT %s """, tuple(params), ) or [] def create_import_job(self, payload: AlsoImportJobCreate, imported_by_user_id: Optional[int]) -> Dict[str, Any]: self._assert_enabled() rows = execute_query( """ INSERT INTO also_import_jobs ( source_type, source_label, status, file_name, import_version, imported_by_user_id, raw_payload_json, log_json, started_at, imported_at ) VALUES (%s, %s, 'new', %s, %s, %s, %s::jsonb, '[]'::jsonb, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) RETURNING * """, ( payload.source_type, payload.source_label, payload.file_name, payload.import_version, imported_by_user_id, _json_dumps(payload.raw_payload), ), ) return rows[0] def _pick_value(self, row: Dict[str, Any], field_name: str) -> Any: for alias in self.HEADER_ALIASES.get(field_name, []): if alias in row and row[alias] not in (None, ""): return row[alias] return None def _normalize_import_row( self, row: Dict[str, Any], source_line_ref: str, file_context: Optional[Dict[str, Any]] = None, ) -> Optional[Dict[str, Any]]: merged_row = dict(file_context or {}) merged_row.update(row) normalized = {_normalize_header(key): value for key, value in merged_row.items()} customer_id_value = self._pick_value(normalized, "customer_id") account_id_value = self._pick_value(normalized, "account_id") also_company_id_value = self._pick_value(normalized, "also_company_id") product_name = ( normalized.get("productdisplayname") or normalized.get("product_display_name") or self._pick_value(normalized, "product_name") ) material_number = self._pick_value(normalized, "material_number") or normalized.get("productname") company = ( normalized.get("companydisplayname") or normalized.get("company_display_name") or self._pick_value(normalized, "company") ) explicit_charge = _parse_decimal_candidate(normalized.get("charge")) total_price = explicit_charge if explicit_charge is not None else _parse_decimal_candidate(self._pick_value(normalized, "total_price")) sales_price = _parse_decimal_candidate(self._pick_value(normalized, "sales_price")) cost_amount = _parse_decimal_candidate(self._pick_value(normalized, "cost_amount")) unit_price = explicit_charge if explicit_charge is not None else _parse_decimal_candidate(self._pick_value(normalized, "unit_price")) quantity = ( _extract_quantity_from_field_values(normalized.get("fieldvalues")) or _extract_quantity_from_field_values(normalized.get("billableparameters")) or _parse_decimal_candidate(self._pick_value(normalized, "billable_parameters")) ) charge_interval_value = self._pick_value(normalized, "charge_interval") billing_start_value = ( _extract_period_start(charge_interval_value) or normalized.get("startdate") or normalized.get("start_date") or self._pick_value(normalized, "billing_start") ) if total_price is None and sales_price is not None: total_price = sales_price if not any([product_name, material_number, company, total_price, sales_price, cost_amount]): return None return { "line_no": self._pick_value(normalized, "line_no"), "source_line_ref": self._pick_value(normalized, "source_line_ref") or source_line_ref, "company": company, "customer_id": str(customer_id_value) if customer_id_value not in (None, "") else None, "account_id": str(account_id_value) if account_id_value not in (None, "") else None, "vat": self._pick_value(normalized, "vat"), "also_company_id": str(also_company_id_value) if also_company_id_value not in (None, "") else None, "material_number": material_number, "product_name": product_name, "vendor": self._pick_value(normalized, "vendor"), "cost_amount": str(cost_amount) if cost_amount is not None else None, "sales_price": str(sales_price) if sales_price is not None else None, "unit_price": str(unit_price) if unit_price is not None else None, "total_price": str(total_price) if total_price is not None else None, "currency": self._pick_value(normalized, "currency") or "DKK", "billing_start": _parse_date_candidate(billing_start_value) or _parse_date_candidate((file_context or {}).get("billing_start")), "charge_interval": charge_interval_value, "billing_interval": self._pick_value(normalized, "billing_interval"), "billable_parameters": str(quantity) if quantity is not None else None, "raw_line": merged_row, } def _parse_delimited_file(self, payload: bytes) -> List[Dict[str, Any]]: text = payload.decode("utf-8-sig", errors="ignore") if not text.strip(): return [] delimiter = "," sample = text[:5000] try: delimiter = csv.Sniffer().sniff(sample, delimiters=",;\t|").delimiter except Exception: if ";" in sample: delimiter = ";" elif "\t" in sample: delimiter = "\t" reader = csv.DictReader(io.StringIO(text), delimiter=delimiter) return [dict(row) for row in reader if any(_normalized_text(value) for value in row.values())] def _parse_json_file(self, payload: bytes, file_name: Optional[str] = None) -> List[Dict[str, Any]]: decoded = json.loads(payload.decode("utf-8", errors="ignore")) file_context = _derive_file_context(file_name or "") rows: List[Dict[str, Any]] = [] seen: set[str] = set() def add_row(candidate: Dict[str, Any], inherited: Optional[Dict[str, Any]] = None) -> None: merged = dict(file_context) if inherited: merged.update(inherited) merged.update(candidate) if not _has_candidate_fields(merged): return key = json.dumps(merged, ensure_ascii=False, default=str, sort_keys=True) if key in seen: return seen.add(key) rows.append(merged) def walk(node: Any, inherited: Optional[Dict[str, Any]] = None) -> None: if isinstance(node, list): for item in node: walk(item, inherited) return if not isinstance(node, dict): return flat = _flatten_json_object(node) merged_inherited = dict(inherited or {}) for key, value in flat.items(): if value in (None, "", [], {}): continue if key not in merged_inherited: merged_inherited[key] = value has_nested_children = any(isinstance(value, (dict, list)) for value in node.values()) if _is_probable_leaf_json_row(flat, has_nested_children=has_nested_children): add_row(flat, inherited) for value in node.values(): if isinstance(value, dict): walk(value, merged_inherited) elif isinstance(value, list): for item in value: walk(item, merged_inherited) walk(decoded, {}) return rows def _parse_xml_file(self, payload: bytes) -> List[Dict[str, Any]]: root = ET.fromstring(payload) rows: List[Dict[str, Any]] = [] for candidate in root.findall(".//row") + root.findall(".//line") + root.findall(".//item") + root.findall(".//record"): row: Dict[str, Any] = {} for child in list(candidate): row[_normalize_header(child.tag)] = child.text if row: rows.append(row) return rows def _parse_excel_file(self, payload: bytes) -> List[Dict[str, Any]]: if not zipfile.is_zipfile(io.BytesIO(payload)): raise HTTPException(status_code=422, detail="Excel-filen kunne ikke læses som XLSX") ns = { "a": "http://schemas.openxmlformats.org/spreadsheetml/2006/main", "r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships", "pr": "http://schemas.openxmlformats.org/package/2006/relationships", } def _column_index(ref: str) -> int: letters = "".join(ch for ch in ref if ch.isalpha()).upper() value = 0 for ch in letters: value = (value * 26) + (ord(ch) - 64) return max(value - 1, 0) with zipfile.ZipFile(io.BytesIO(payload)) as workbook_zip: shared_strings: List[str] = [] if "xl/sharedStrings.xml" in workbook_zip.namelist(): shared_root = ET.fromstring(workbook_zip.read("xl/sharedStrings.xml")) for si in shared_root.findall("a:si", ns): fragments = [node.text or "" for node in si.iterfind(".//a:t", ns)] shared_strings.append("".join(fragments)) workbook_root = ET.fromstring(workbook_zip.read("xl/workbook.xml")) rel_root = ET.fromstring(workbook_zip.read("xl/_rels/workbook.xml.rels")) relationship_map = { rel.attrib["Id"]: rel.attrib["Target"] for rel in rel_root.findall("pr:Relationship", ns) } rows: List[Dict[str, Any]] = [] for sheet in workbook_root.find("a:sheets", ns) or []: relation_id = sheet.attrib.get("{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id") target = relationship_map.get(relation_id or "") if not target: continue sheet_path = target if target.startswith("xl/") else f"xl/{target}" sheet_root = ET.fromstring(workbook_zip.read(sheet_path)) header: Optional[List[str]] = None for row_node in sheet_root.findall(".//a:sheetData/a:row", ns): cell_map: Dict[int, Any] = {} for cell in row_node.findall("a:c", ns): ref = cell.attrib.get("r", "") idx = _column_index(ref) value_node = cell.find("a:v", ns) cell_type = cell.attrib.get("t") if cell_type == "inlineStr": inline_node = cell.find("a:is", ns) value = "".join(node.text or "" for node in inline_node.iterfind(".//a:t", ns)) if inline_node is not None else "" elif value_node is None: value = "" else: raw_value = value_node.text or "" if cell_type == "s": try: value = shared_strings[int(raw_value)] except Exception: value = raw_value else: value = raw_value cell_map[idx] = value if not cell_map: continue max_idx = max(cell_map) values = [cell_map.get(i, "") for i in range(max_idx + 1)] if header is None: candidate_header = [_normalized_text(v) for v in values] if not any(candidate_header): continue header = candidate_header continue if not any(_normalized_text(v) for v in values): continue row: Dict[str, Any] = {} for idx, column_name in enumerate(header): key = column_name.strip() if not key: continue row[key] = values[idx] if idx < len(values) else "" if any(_normalized_text(v) for v in row.values()): rows.append(row) return rows def _extract_zip_rows(self, zip_bytes: bytes) -> Dict[str, Any]: lines: List[Dict[str, Any]] = [] extracted_files: List[str] = [] source_type = "csv" with zipfile.ZipFile(io.BytesIO(zip_bytes)) as archive: for member in archive.infolist(): if member.is_dir(): continue name = member.filename ext = Path(name).suffix.lower() if ext not in {".csv", ".tsv", ".txt", ".json", ".xml", ".xlsx", ".xls"}: continue extracted_files.append(name) payload = archive.read(member) source_rows: List[Dict[str, Any]] = [] if ext in {".csv", ".tsv", ".txt"}: source_rows = self._parse_delimited_file(payload) source_type = "csv" elif ext == ".json": source_rows = self._parse_json_file(payload, file_name=name) source_type = "json_export" elif ext == ".xml": source_rows = self._parse_xml_file(payload) source_type = "xml_export" elif ext in {".xlsx", ".xls"}: source_rows = self._parse_excel_file(payload) source_type = "csv" file_context = _derive_file_context(name) for index, row in enumerate(source_rows, start=1): normalized = self._normalize_import_row( row, source_line_ref=f"{name}:{index}", file_context=file_context, ) if normalized: lines.append(normalized) return { "source_type": source_type, "lines": lines, "files": extracted_files, } async def import_billing_zip_from_url( self, download_url: str, imported_by_user_id: Optional[int] = None, email_id: Optional[int] = None, source_label: Optional[str] = None, ) -> Dict[str, Any]: self._assert_enabled() normalized_url = _normalize_download_url(download_url) timeout = httpx.Timeout( connect=min(float(settings.ALSO_TIMEOUT_SECONDS or 20), 20.0), read=max(float(settings.ALSO_TIMEOUT_SECONDS or 20), 20.0), write=20.0, pool=20.0, ) async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: response = await client.get(normalized_url) if response.headers.get("content-type", "").lower().startswith("application/json"): try: payload = response.json() except Exception: payload = {} if payload.get("error") and "authenticate header" in _normalized_text(payload.get("text")).lower(): raise HTTPException( status_code=502, detail="ALSO download kræver gyldig Authenticate-header/session. Linket alene er ikke nok fra server-side workflow.", ) response.raise_for_status() zip_bytes = response.content if not zipfile.is_zipfile(io.BytesIO(zip_bytes)): raise HTTPException(status_code=422, detail="ALSO download returnerede ikke en ZIP-fil") parsed = self._extract_zip_rows(zip_bytes) normalized_lines = parsed.get("lines") or [] if not normalized_lines: raise HTTPException(status_code=422, detail="ALSO billing ZIP contained no recognizable billing rows") file_name = normalized_url.rstrip("/").split("/")[-1] or "also-billing.zip" job = self.create_import_job( AlsoImportJobCreate( source_type=parsed.get("source_type") or "csv", source_label=source_label or "Email workflow", file_name=file_name, import_version="email_workflow_v1", raw_payload={ "email_id": email_id, "download_url": normalized_url, "zip_entries": parsed.get("files") or [], "line_count": len(normalized_lines), }, ), imported_by_user_id=imported_by_user_id, ) job_id = int(job["id"]) try: import_result = self.import_lines( job_id=job_id, payload=AlsoImportLinesRequest(lines=normalized_lines), ) matching_result = self.run_matching(import_job_id=job_id, line_ids=[], limit=5000) validation_result = self.run_validation(import_job_id=job_id, line_ids=[], limit=5000) approval_result: Dict[str, Any] = {"approved_lines": 0, "created_drafts": 0, "draft_ids": []} ready_count = int(validation_result.get("ready_for_approval") or 0) if ready_count > 0: approval_result = self.approve_lines_to_drafts( import_job_id=job_id, line_ids=[], approved_by_user_id=imported_by_user_id, ) completion = self._complete_import_job(job_id) final_status = completion["status"] queue_totals = completion["queue_totals"] except Exception as e: self._mark_import_job_failed(job_id, str(e)) raise if email_id: execute_update( """ UPDATE email_messages SET status = 'processed', folder = 'Processed', processed_at = CURRENT_TIMESTAMP, auto_processed = true, updated_at = CURRENT_TIMESTAMP WHERE id = %s """, (email_id,), ) return { "job_id": int(job["id"]), "status": final_status, "download_url": normalized_url, "zip_entries": parsed.get("files") or [], "import_result": import_result, "matching_result": matching_result, "validation_result": validation_result, "approval_result": approval_result, "queue_totals": queue_totals, } def import_billing_upload( self, *, file_name: str, file_bytes: bytes, imported_by_user_id: Optional[int] = None, source_label: Optional[str] = None, ) -> Dict[str, Any]: self._assert_enabled() safe_file_name = Path(file_name or "also-billing.xlsx").name ext = Path(safe_file_name).suffix.lower() if ext == ".xlsx": source_rows = self._parse_excel_file(file_bytes) parsed = { "source_type": "csv", "lines": [ self._normalize_import_row( row, source_line_ref=f"{safe_file_name}:{index}", file_context=None, ) for index, row in enumerate(source_rows, start=1) ], "files": [safe_file_name], } parsed["lines"] = [row for row in (parsed.get("lines") or []) if row] elif zipfile.is_zipfile(io.BytesIO(file_bytes)): parsed = self._extract_zip_rows(file_bytes) else: raise HTTPException(status_code=422, detail="Upload skal være en XLSX-fil eller en gyldig ZIP-fil") normalized_lines = parsed.get("lines") or [] if not normalized_lines: raise HTTPException(status_code=422, detail="Filen indeholdt ingen genkendelige billing-linjer") job = self.create_import_job( AlsoImportJobCreate( source_type=parsed.get("source_type") or "csv", source_label=source_label or "Manual upload", file_name=safe_file_name, import_version="manual_upload_v1", raw_payload={ "upload_filename": safe_file_name, "uploaded_entries": parsed.get("files") or [], "upload_format": ext.lstrip(".") or "zip", "line_count": len(normalized_lines), }, ), imported_by_user_id=imported_by_user_id, ) job_id = int(job["id"]) try: import_result = self.import_lines( job_id=job_id, payload=AlsoImportLinesRequest(lines=normalized_lines), ) matching_result = self.run_matching(import_job_id=job_id, line_ids=[], limit=5000) validation_result = self.run_validation(import_job_id=job_id, line_ids=[], limit=5000) approval_result: Dict[str, Any] = {"approved_lines": 0, "created_drafts": 0, "draft_ids": []} ready_count = int(validation_result.get("ready_for_approval") or 0) if ready_count > 0: approval_result = self.approve_lines_to_drafts( import_job_id=job_id, line_ids=[], approved_by_user_id=imported_by_user_id, ) completion = self._complete_import_job(job_id) return { "job_id": job_id, "status": completion["status"], "zip_entries": parsed.get("files") or [], "import_result": import_result, "matching_result": matching_result, "validation_result": validation_result, "approval_result": approval_result, "queue_totals": completion["queue_totals"], } except Exception as e: self._mark_import_job_failed(job_id, str(e)) raise def list_import_jobs(self, status: Optional[str], limit: int) -> List[Dict[str, Any]]: self._assert_enabled() if status: return execute_query( """ SELECT j.*, COALESCE(l.line_count, 0) AS line_count FROM also_import_jobs j LEFT JOIN ( SELECT import_job_id, COUNT(*) AS line_count FROM also_import_lines GROUP BY import_job_id ) l ON l.import_job_id = j.id WHERE j.status = %s ORDER BY j.imported_at DESC, j.id DESC LIMIT %s """, (status, max(1, min(limit, 500))), ) or [] return execute_query( """ SELECT j.*, COALESCE(l.line_count, 0) AS line_count FROM also_import_jobs j LEFT JOIN ( SELECT import_job_id, COUNT(*) AS line_count FROM also_import_lines GROUP BY import_job_id ) l ON l.import_job_id = j.id ORDER BY j.imported_at DESC, j.id DESC LIMIT %s """, (max(1, min(limit, 500)),), ) or [] def get_import_job(self, job_id: int) -> Dict[str, Any]: self._assert_enabled() row = execute_query_single( """ SELECT j.*, COALESCE(l.line_count, 0) AS line_count FROM also_import_jobs j LEFT JOIN ( SELECT import_job_id, COUNT(*) AS line_count FROM also_import_lines WHERE import_job_id = %s GROUP BY import_job_id ) l ON l.import_job_id = j.id WHERE j.id = %s """, (job_id, job_id), ) if not row: raise HTTPException(status_code=404, detail="Import job not found") return row def import_lines(self, job_id: int, payload: AlsoImportLinesRequest) -> Dict[str, Any]: self._assert_enabled() job = execute_query_single("SELECT id FROM also_import_jobs WHERE id = %s", (job_id,)) if not job: raise HTTPException(status_code=404, detail="Import job not found") inserted = 0 duplicates = 0 for idx, line in enumerate(payload.lines, start=1): line_data = line.model_dump() line_no = line_data.get("line_no") or idx hash_value = _line_hash(job_id, line_data) existing = execute_query_single( "SELECT id FROM also_import_lines WHERE import_job_id = %s AND line_hash = %s", (job_id, hash_value), ) if existing: duplicates += 1 continue execute_query( """ INSERT INTO also_import_lines ( import_job_id, line_no, queue_status, source_line_ref, line_hash, also_company_id, company, customer_id, account_id, vat, material_number, product_name, vendor, cost_amount, sales_price, unit_price, total_price, currency, billing_start, charge_interval, billing_interval, billable_parameters, raw_line_json, validation_errors_json ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, '[]'::jsonb ) """, ( job_id, line_no, line_data.get("queue_status") or "new", line_data.get("source_line_ref"), hash_value, line_data.get("also_company_id"), line_data.get("company"), line_data.get("customer_id"), line_data.get("account_id"), line_data.get("vat"), line_data.get("material_number"), line_data.get("product_name"), line_data.get("vendor"), line_data.get("cost_amount"), line_data.get("sales_price"), line_data.get("unit_price"), line_data.get("total_price"), line_data.get("currency"), line_data.get("billing_start"), line_data.get("charge_interval"), line_data.get("billing_interval"), line_data.get("billable_parameters"), _json_dumps(line_data.get("raw_line") or {}), ), ) inserted += 1 execute_query( """ UPDATE also_import_jobs SET status = CASE WHEN %s > 0 THEN 'lines_imported' ELSE status END, finished_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = %s """, (inserted, job_id), ) return { "job_id": job_id, "received": len(payload.lines), "inserted": inserted, "duplicates": duplicates, } def get_queue(self, status: Optional[str], limit: int) -> List[Dict[str, Any]]: self._assert_enabled() if status: return execute_query( f""" SELECT l.*, {ALSO_EFFECTIVE_COST_SQL} AS effective_cost_amount, COALESCE(l.total_price, l.sales_price, 0) - ({ALSO_EFFECTIVE_COST_SQL}) AS effective_margin_amount FROM also_import_lines l LEFT JOIN products p ON p.id = l.matched_product_id WHERE l.queue_status = %s ORDER BY l.id DESC LIMIT %s """, (status, max(1, min(limit, 1000))), ) or [] return execute_query( f""" SELECT l.*, {ALSO_EFFECTIVE_COST_SQL} AS effective_cost_amount, COALESCE(l.total_price, l.sales_price, 0) - ({ALSO_EFFECTIVE_COST_SQL}) AS effective_margin_amount FROM also_import_lines l LEFT JOIN products p ON p.id = l.matched_product_id ORDER BY l.id DESC LIMIT %s """, (max(1, min(limit, 1000)),), ) or [] def get_import_job_lines(self, job_id: int, status: Optional[str], limit: int) -> List[Dict[str, Any]]: self._assert_enabled() params: List[Any] = [job_id] where = ["l.import_job_id = %s"] if status: where.append("l.queue_status = %s") params.append(status) params.append(max(1, min(limit, 2000))) return execute_query( f""" SELECT l.*, c.name AS matched_customer_name, p.name AS matched_product_name, {ALSO_EFFECTIVE_COST_SQL} AS effective_cost_amount, COALESCE(l.total_price, l.sales_price, 0) - ({ALSO_EFFECTIVE_COST_SQL}) AS effective_margin_amount, d.title AS order_draft_title, d.sync_status AS order_draft_sync_status, d.economic_order_number, d.economic_invoice_number FROM also_import_lines l LEFT JOIN customers c ON c.id = l.matched_customer_id LEFT JOIN products p ON p.id = l.matched_product_id LEFT JOIN ordre_drafts d ON d.id = l.order_draft_id WHERE {' AND '.join(where)} ORDER BY l.line_no ASC NULLS LAST, l.id ASC LIMIT %s """, tuple(params), ) or [] def delete_import_job(self, job_id: int, *, delete_order_drafts: bool = False) -> Dict[str, Any]: self._assert_enabled() conn = get_db_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cursor: cursor.execute( """ SELECT j.*, COALESCE(l.line_count, 0) AS line_count FROM also_import_jobs j LEFT JOIN ( SELECT import_job_id, COUNT(*) AS line_count FROM also_import_lines WHERE import_job_id = %s GROUP BY import_job_id ) l ON l.import_job_id = j.id WHERE j.id = %s """, (job_id, job_id), ) job = cursor.fetchone() if not job: raise HTTPException(status_code=404, detail="Import job not found") cursor.execute( """ SELECT d.id, d.title, d.sync_status, d.economic_order_number, d.economic_invoice_number FROM ordre_drafts d JOIN ( SELECT DISTINCT order_draft_id FROM also_import_lines WHERE import_job_id = %s AND order_draft_id IS NOT NULL ) x ON x.order_draft_id = d.id ORDER BY d.id ASC """, (job_id,), ) drafts = cursor.fetchall() or [] blocked_statuses = {"exported", "posted", "paid"} blocked_drafts = [dict(row) for row in drafts if str(row.get("sync_status") or "").strip().lower() in blocked_statuses] if drafts and not delete_order_drafts: raise HTTPException( status_code=409, detail={ "message": "Import job has linked ordre drafts. Confirm deletion with delete_order_drafts=true if drafts should be removed too.", "draft_count": len(drafts), "blocked_draft_count": len(blocked_drafts), "drafts": [dict(row) for row in drafts], }, ) if blocked_drafts: raise HTTPException( status_code=409, detail={ "message": "One or more linked ordre drafts are already exported/posted/paid and cannot be auto-deleted.", "blocked_drafts": blocked_drafts, }, ) deleted_draft_ids: List[int] = [] if drafts and delete_order_drafts: draft_ids = [int(row["id"]) for row in drafts if row.get("id") is not None] if draft_ids: placeholders = ",".join(["%s"] * len(draft_ids)) cursor.execute( f"DELETE FROM ordre_drafts WHERE id IN ({placeholders})", tuple(draft_ids), ) deleted_draft_ids = draft_ids cursor.execute( "DELETE FROM also_import_jobs WHERE id = %s RETURNING id", (job_id,), ) deleted = cursor.fetchone() if not deleted: raise HTTPException(status_code=404, detail="Import job not found") conn.commit() return { "success": True, "deleted_job_id": job_id, "deleted_line_count": int(job.get("line_count") or 0), "deleted_order_draft_ids": deleted_draft_ids, } except HTTPException: conn.rollback() raise except Exception: conn.rollback() raise finally: release_db_connection(conn) def run_matching(self, import_job_id: Optional[int], line_ids: List[int], limit: int) -> Dict[str, Any]: self._assert_enabled() lines = self._fetch_process_lines(import_job_id=import_job_id, line_ids=line_ids, limit=limit) updated = 0 ready = 0 errored = 0 for line in lines: matched_customer_id = self._resolve_customer_match(line) matched_product_id = self._resolve_product_match(line) current_errors = line.get("validation_errors_json") or [] has_errors = bool(current_errors) if _is_zero_value_tenant_line(line): new_status = "approved" else: new_status = self._derive_queue_status(matched_customer_id, matched_product_id, has_errors) execute_query( """ UPDATE also_import_lines SET matched_customer_id = %s, matched_product_id = %s, queue_status = %s, matching_checked_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = %s """, (matched_customer_id, matched_product_id, new_status, line["id"]), ) updated += 1 if new_status == "ready_for_approval": ready += 1 if new_status == "error": errored += 1 return { "processed": len(lines), "updated": updated, "ready_for_approval": ready, "errored": errored, } def run_validation(self, import_job_id: Optional[int], line_ids: List[int], limit: int) -> Dict[str, Any]: self._assert_enabled() lines = self._fetch_process_lines(import_job_id=import_job_id, line_ids=line_ids, limit=limit) duplicate_rows = execute_query( """ SELECT id FROM ( SELECT id, COUNT(*) OVER ( PARTITION BY import_job_id, COALESCE(company, ''), COALESCE(material_number, ''), COALESCE(vendor, ''), COALESCE(billing_start::text, ''), COALESCE(total_price::text, '') ) AS dup_count FROM also_import_lines WHERE queue_status <> 'invoiced' AND (%s::INTEGER IS NULL OR import_job_id = %s) ) q WHERE q.dup_count > 1 """, (import_job_id, import_job_id), ) or [] duplicate_ids = {int(row["id"]) for row in duplicate_rows} updated = 0 ready = 0 errored = 0 for line in lines: errors = self._build_validation_errors(line, duplicate_ids=duplicate_ids) if _is_zero_value_tenant_line(line): errors = [] new_status = "approved" else: new_status = self._derive_queue_status(line.get("matched_customer_id"), line.get("matched_product_id"), bool(errors)) execute_query( """ UPDATE also_import_lines SET validation_errors_json = %s::jsonb, queue_status = %s, validation_checked_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = %s """, (_json_dumps(errors), new_status, line["id"]), ) updated += 1 if new_status == "ready_for_approval": ready += 1 if new_status == "error": errored += 1 return { "processed": len(lines), "updated": updated, "ready_for_approval": ready, "errored": errored, } def approve_lines_to_drafts(self, import_job_id: Optional[int], line_ids: List[int], approved_by_user_id: Optional[int]) -> Dict[str, Any]: self._assert_enabled() params: List[Any] = [] where = ["l.queue_status = 'ready_for_approval'", "l.matched_customer_id IS NOT NULL", "l.matched_product_id IS NOT NULL"] if import_job_id: where.append("l.import_job_id = %s") params.append(import_job_id) if line_ids: placeholders = ",".join(["%s"] * len(line_ids)) where.append(f"l.id IN ({placeholders})") params.extend(line_ids) lines = execute_query( f""" SELECT l.*, c.name AS matched_customer_name, p.name AS matched_product_name FROM also_import_lines l LEFT JOIN customers c ON c.id = l.matched_customer_id LEFT JOIN products p ON p.id = l.matched_product_id WHERE {' AND '.join(where)} ORDER BY l.matched_customer_id, l.currency, COALESCE(l.billing_start, CURRENT_DATE), l.id """, tuple(params), ) or [] if not lines: raise HTTPException(status_code=400, detail="No ready-for-approval lines found") grouped: Dict[str, List[Dict[str, Any]]] = {} for line in lines: period_key = str(line.get("billing_start") or datetime.utcnow().date())[:7] key = f"{line.get('matched_customer_id')}|{_normalized_text(line.get('currency')) or 'DKK'}|{period_key}" grouped.setdefault(key, []).append(line) draft_ids: List[int] = [] approved_lines = 0 for group_key, group_lines in grouped.items(): first = group_lines[0] customer_id = int(first["matched_customer_id"]) customer_name = first.get("matched_customer_name") or f"Kunde {customer_id}" currency = _normalized_text(first.get("currency")) or "DKK" period_key = str(first.get("billing_start") or datetime.utcnow().date())[:7] draft_lines: List[Dict[str, Any]] = [] for line in group_lines: quantity = _to_decimal(line.get("billable_parameters"), Decimal("1")) if quantity <= 0: quantity = Decimal("1") unit_price = _to_decimal(line.get("unit_price"), _to_decimal(line.get("sales_price"), Decimal("0"))) amount = _to_decimal(line.get("total_price"), default=(quantity * unit_price)) draft_lines.append( { "line_key": f"also:{line['id']}", "source_type": "also_cloud", "source_id": int(line["id"]), "reference_id": int(line["import_job_id"]), "description": line.get("product_name") or line.get("matched_product_name") or "Cloud abonnement", "quantity": float(quantity), "unit": "stk", "unit_price": float(unit_price), "discount_percentage": 0.0, "amount": float(amount), "currency": currency, "status": "approved", "line_date": str(line.get("billing_start")) if line.get("billing_start") else None, "product_id": int(line["matched_product_id"]), "customer_id": customer_id, "customer_name": customer_name, "selected": True, "meta": { "also_material_number": line.get("material_number"), "also_vendor": line.get("vendor"), "also_import_job_id": int(line.get("import_job_id")), }, } ) draft = execute_query_single( """ INSERT INTO ordre_drafts ( title, customer_id, lines_json, notes, layout_number, created_by_user_id, sync_status, export_status_json, invoice_aggregate_key, updated_at ) VALUES (%s, %s, %s::jsonb, %s, %s, %s, 'pending', %s::jsonb, %s, CURRENT_TIMESTAMP) RETURNING id """, ( f"ALSO Cloud {customer_name} - {period_key}", customer_id, _json_dumps(draft_lines), "Genereret fra ALSO Cloud Billing approval", 1, approved_by_user_id, _json_dumps({"source": "also_cloud_billing"}), f"also-cloud-{customer_id}-{period_key}", ), ) draft_id = int(draft["id"]) if draft and draft.get("id") else None if not draft_id: raise HTTPException(status_code=500, detail="Failed creating ordre draft from ALSO approval") draft_ids.append(draft_id) line_id_values = [int(line["id"]) for line in group_lines] placeholders = ",".join(["%s"] * len(line_id_values)) execute_query( f""" UPDATE also_import_lines SET queue_status = 'approved', approved_at = CURRENT_TIMESTAMP, approved_by_user_id = %s, order_draft_id = %s, updated_at = CURRENT_TIMESTAMP WHERE id IN ({placeholders}) """, tuple([approved_by_user_id, draft_id] + line_id_values), ) approved_lines += len(group_lines) return { "approved_lines": approved_lines, "created_drafts": len(draft_ids), "draft_ids": draft_ids, } def get_dashboard_summary(self) -> Dict[str, Any]: self._assert_enabled() row = execute_query_single( f""" WITH month_lines AS ( SELECT l.*, {ALSO_EFFECTIVE_COST_SQL} AS effective_cost_amount FROM also_import_lines l LEFT JOIN products p ON p.id = l.matched_product_id WHERE date_trunc('month', COALESCE(l.billing_start::timestamp, l.created_at)) = date_trunc('month', CURRENT_DATE) ) SELECT COALESCE(SUM(COALESCE(total_price, sales_price, 0)), 0) AS monthly_revenue, COALESCE(SUM(COALESCE(effective_cost_amount, 0)), 0) AS monthly_cost, COALESCE(SUM(COALESCE(total_price, sales_price, 0) - COALESCE(effective_cost_amount, 0)), 0) AS monthly_margin, COUNT(*) FILTER (WHERE matched_product_id IS NULL) AS unmatched_products, COUNT(*) FILTER (WHERE matched_customer_id IS NULL) AS unmatched_customers, COUNT(*) FILTER (WHERE queue_status = 'ready_for_approval') AS pending_approvals, COUNT(DISTINCT matched_customer_id) FILTER (WHERE queue_status = 'invoiced' AND matched_customer_id IS NOT NULL) AS invoiced_customers FROM month_lines """, (), ) or {} return { "monthly_revenue": row.get("monthly_revenue") or Decimal("0"), "monthly_cost": row.get("monthly_cost") or Decimal("0"), "monthly_margin": row.get("monthly_margin") or Decimal("0"), "unmatched_products": int(row.get("unmatched_products") or 0), "unmatched_customers": int(row.get("unmatched_customers") or 0), "pending_approvals": int(row.get("pending_approvals") or 0), "invoiced_customers": int(row.get("invoiced_customers") or 0), } def get_monthly_differences(self, limit: int = 50) -> List[Dict[str, Any]]: self._assert_enabled() rows = execute_query( """ WITH current_month AS ( SELECT matched_customer_id, material_number, vendor, product_name, SUM(COALESCE(billable_parameters, 1)) AS qty FROM also_import_lines WHERE date_trunc('month', COALESCE(billing_start::timestamp, created_at)) = date_trunc('month', CURRENT_DATE) GROUP BY matched_customer_id, material_number, vendor, product_name ), previous_month AS ( SELECT matched_customer_id, material_number, vendor, product_name, SUM(COALESCE(billable_parameters, 1)) AS qty FROM also_import_lines WHERE date_trunc('month', COALESCE(billing_start::timestamp, created_at)) = date_trunc('month', CURRENT_DATE - INTERVAL '1 month') GROUP BY matched_customer_id, material_number, vendor, product_name ), merged AS ( SELECT COALESCE(c.matched_customer_id, p.matched_customer_id) AS customer_id, COALESCE(c.material_number, p.material_number) AS material_number, COALESCE(c.vendor, p.vendor) AS vendor, COALESCE(c.product_name, p.product_name) AS product_name, COALESCE(p.qty, 0) AS previous_month_qty, COALESCE(c.qty, 0) AS current_month_qty FROM current_month c FULL OUTER JOIN previous_month p ON COALESCE(c.matched_customer_id, 0) = COALESCE(p.matched_customer_id, 0) AND COALESCE(c.material_number, '') = COALESCE(p.material_number, '') AND COALESCE(c.vendor, '') = COALESCE(p.vendor, '') AND COALESCE(c.product_name, '') = COALESCE(p.product_name, '') ) SELECT m.*, cu.name AS customer_name, CASE WHEN m.previous_month_qty = 0 THEN NULL ELSE ROUND(((m.current_month_qty - m.previous_month_qty) / NULLIF(m.previous_month_qty, 0)) * 100, 2) END AS change_percent FROM merged m LEFT JOIN customers cu ON cu.id = m.customer_id WHERE m.previous_month_qty <> m.current_month_qty ORDER BY ABS(COALESCE( CASE WHEN m.previous_month_qty = 0 THEN NULL ELSE ((m.current_month_qty - m.previous_month_qty) / NULLIF(m.previous_month_qty, 0)) * 100 END, 0 )) DESC, m.current_month_qty DESC LIMIT %s """, (max(1, min(limit, 200)),), ) or [] results: List[Dict[str, Any]] = [] for row in rows: change_percent = row.get("change_percent") warning = False if change_percent is not None: try: warning = abs(Decimal(str(change_percent))) >= Decimal("50") except Exception: warning = False results.append( { "customer_id": row.get("customer_id"), "customer_name": row.get("customer_name"), "material_number": row.get("material_number"), "product_name": row.get("product_name"), "vendor": row.get("vendor"), "previous_month_qty": row.get("previous_month_qty") or Decimal("0"), "current_month_qty": row.get("current_month_qty") or Decimal("0"), "change_percent": change_percent, "warning": warning, } ) return results def get_status_breakdown(self) -> Dict[str, Any]: self._assert_enabled() line_rows = execute_query( """ SELECT queue_status, COUNT(*)::INTEGER AS total_count, COUNT(*) FILTER ( WHERE date_trunc('month', COALESCE(billing_start::timestamp, created_at)) = date_trunc('month', CURRENT_DATE) )::INTEGER AS current_month_count FROM also_import_lines GROUP BY queue_status ORDER BY total_count DESC, queue_status ASC """, (), ) or [] job_rows = execute_query( """ SELECT status, COUNT(*)::INTEGER AS total_count FROM also_import_jobs GROUP BY status ORDER BY total_count DESC, status ASC """, (), ) or [] return { "line_statuses": line_rows, "job_statuses": job_rows, } def get_monthly_history(self, months: int = 6) -> List[Dict[str, Any]]: self._assert_enabled() rows = execute_query( """ WITH month_series AS ( SELECT generate_series( date_trunc('month', CURRENT_DATE) - (%s::INTEGER - 1) * INTERVAL '1 month', date_trunc('month', CURRENT_DATE), INTERVAL '1 month' ) AS month_start ), line_agg AS ( SELECT date_trunc('month', COALESCE(l.billing_start::timestamp, l.created_at)) AS month_start, COALESCE(SUM(COALESCE(l.total_price, l.sales_price, 0)), 0) AS revenue, COALESCE(SUM( COALESCE( l.cost_amount, CASE WHEN l.matched_product_id IS NOT NULL THEN COALESCE(p.supplier_price, 0) * COALESCE(NULLIF(l.billable_parameters, 0), 1) ELSE 0 END, 0 ) ), 0) AS cost, COUNT(*)::INTEGER AS line_count, COUNT(DISTINCT l.matched_customer_id)::INTEGER AS customer_count FROM also_import_lines l LEFT JOIN products p ON p.id = l.matched_product_id GROUP BY 1 ), job_agg AS ( SELECT date_trunc('month', imported_at) AS month_start, COUNT(*)::INTEGER AS job_count FROM also_import_jobs GROUP BY 1 ) SELECT ms.month_start::date AS month_start, COALESCE(la.revenue, 0) AS revenue, COALESCE(la.cost, 0) AS cost, COALESCE(la.revenue, 0) - COALESCE(la.cost, 0) AS margin, COALESCE(la.line_count, 0) AS line_count, COALESCE(la.customer_count, 0) AS customer_count, COALESCE(ja.job_count, 0) AS job_count FROM month_series ms LEFT JOIN line_agg la ON la.month_start = ms.month_start LEFT JOIN job_agg ja ON ja.month_start = ms.month_start ORDER BY ms.month_start DESC """, (max(1, min(months, 24)),), ) or [] return rows def upsert_company_mapping(self, payload: AlsoCompanyMappingUpsert) -> Dict[str, Any]: self._assert_enabled() rows = execute_query( """ INSERT INTO also_company_mapping ( also_company_id, also_customer_id, customer_id, match_confidence, notes, is_active ) VALUES (%s, %s, %s, %s, %s, true) ON CONFLICT (also_company_id) DO UPDATE SET also_customer_id = EXCLUDED.also_customer_id, customer_id = EXCLUDED.customer_id, match_confidence = EXCLUDED.match_confidence, notes = EXCLUDED.notes, is_active = true, updated_at = CURRENT_TIMESTAMP RETURNING * """, ( payload.also_company_id, payload.also_customer_id, payload.customer_id, payload.match_confidence, payload.notes, ), ) return rows[0] def manual_map_company_for_line( self, *, line_id: int, customer_id: int, notes: Optional[str], updated_by_user_id: Optional[int], ) -> Dict[str, Any]: self._assert_enabled() line = execute_query_single( """ SELECT * FROM also_import_lines WHERE id = %s """, (line_id,), ) if not line: raise HTTPException(status_code=404, detail="ALSO import line not found") also_company_id = _normalized_text(line.get("also_company_id")) if not also_company_id: raise HTTPException(status_code=400, detail="Line is missing also_company_id and cannot be mapped manually") customer = execute_query_single( """ SELECT id, name FROM customers WHERE id = %s AND deleted_at IS NULL AND is_active = true """, (customer_id,), ) if not customer: raise HTTPException(status_code=404, detail="Customer not found or inactive") mapping = self.upsert_company_mapping( AlsoCompanyMappingUpsert( also_company_id=also_company_id, also_customer_id=_normalized_text(line.get("customer_id")) or _normalized_text(line.get("account_id")) or None, customer_id=customer_id, match_confidence=Decimal("1.00"), notes=notes or "Manual mapping from ALSO Cloud Marketplace", ) ) related_rows = execute_query( """ SELECT id FROM also_import_lines WHERE import_job_id = %s AND also_company_id = %s AND order_draft_id IS NULL AND queue_status <> 'invoiced' ORDER BY id ASC """, (line.get("import_job_id"), also_company_id), ) or [] affected_line_ids = [int(row["id"]) for row in related_rows if row.get("id") is not None] if not affected_line_ids: affected_line_ids = [line_id] line_limit = max(len(affected_line_ids), 1) matching_result = self.run_matching(import_job_id=None, line_ids=affected_line_ids, limit=line_limit) validation_result = self.run_validation(import_job_id=None, line_ids=affected_line_ids, limit=line_limit) try: approval_result = self.approve_lines_to_drafts( import_job_id=None, line_ids=affected_line_ids, approved_by_user_id=updated_by_user_id, ) except HTTPException as exc: if exc.status_code == 400 and exc.detail == "No ready-for-approval lines found": approval_result = { "approved_lines": 0, "created_drafts": 0, "draft_ids": [], } else: raise refreshed_line = execute_query_single( """ SELECT l.*, c.name AS matched_customer_name FROM also_import_lines l LEFT JOIN customers c ON c.id = l.matched_customer_id WHERE l.id = %s """, (line_id,), ) return { "success": True, "line_id": line_id, "import_job_id": int(line["import_job_id"]), "also_company_id": also_company_id, "customer_id": int(customer["id"]), "customer_name": customer.get("name"), "affected_line_ids": affected_line_ids, "affected_line_count": len(affected_line_ids), "mapping": mapping, "matching_result": matching_result, "validation_result": validation_result, "approval_result": approval_result, "line": refreshed_line, } def manual_map_product_for_line( self, *, line_id: int, product_id: int, notes: Optional[str], updated_by_user_id: Optional[int], ) -> Dict[str, Any]: self._assert_enabled() line = execute_query_single( """ SELECT * FROM also_import_lines WHERE id = %s """, (line_id,), ) if not line: raise HTTPException(status_code=404, detail="ALSO import line not found") material_number = _normalized_text(line.get("material_number")) if not material_number: raise HTTPException(status_code=400, detail="Line is missing material_number and cannot be mapped manually") vendor = _normalized_text(line.get("vendor")) or "ALSO" product = execute_query_single( """ SELECT id, name FROM products WHERE id = %s AND deleted_at IS NULL """, (product_id,), ) if not product: raise HTTPException(status_code=404, detail="Product not found") mapping = self.upsert_product_mapping( AlsoProductMappingUpsert( material_number=material_number, vendor=vendor, hub_product_id=product_id, product_name_snapshot=_normalized_text(line.get("product_name")) or None, ) ) related_rows = execute_query( """ SELECT id FROM also_import_lines WHERE import_job_id = %s AND material_number = %s AND LOWER(COALESCE(vendor, 'ALSO')) = LOWER(%s) AND order_draft_id IS NULL AND queue_status <> 'invoiced' ORDER BY id ASC """, (line.get("import_job_id"), material_number, vendor), ) or [] affected_line_ids = [int(row["id"]) for row in related_rows if row.get("id") is not None] if not affected_line_ids: affected_line_ids = [line_id] line_limit = max(len(affected_line_ids), 1) matching_result = self.run_matching(import_job_id=None, line_ids=affected_line_ids, limit=line_limit) validation_result = self.run_validation(import_job_id=None, line_ids=affected_line_ids, limit=line_limit) try: approval_result = self.approve_lines_to_drafts( import_job_id=None, line_ids=affected_line_ids, approved_by_user_id=updated_by_user_id, ) except HTTPException as exc: if exc.status_code == 400 and exc.detail == "No ready-for-approval lines found": approval_result = { "approved_lines": 0, "created_drafts": 0, "draft_ids": [], } else: raise refreshed_line = execute_query_single( """ SELECT l.*, p.name AS matched_product_name FROM also_import_lines l LEFT JOIN products p ON p.id = l.matched_product_id WHERE l.id = %s """, (line_id,), ) return { "success": True, "line_id": line_id, "import_job_id": int(line["import_job_id"]), "material_number": material_number, "vendor": vendor, "product_id": int(product["id"]), "product_name": product.get("name"), "affected_line_ids": affected_line_ids, "affected_line_count": len(affected_line_ids), "mapping": mapping, "notes": notes, "matching_result": matching_result, "validation_result": validation_result, "approval_result": approval_result, "line": refreshed_line, } def upsert_product_mapping(self, payload: AlsoProductMappingUpsert) -> Dict[str, Any]: self._assert_enabled() rows = execute_query( """ INSERT INTO also_product_mapping ( material_number, vendor, hub_product_id, product_name_snapshot, is_active ) VALUES (%s, %s, %s, %s, true) ON CONFLICT (material_number, vendor) DO UPDATE SET hub_product_id = EXCLUDED.hub_product_id, product_name_snapshot = EXCLUDED.product_name_snapshot, is_active = true, updated_at = CURRENT_TIMESTAMP RETURNING * """, ( payload.material_number, payload.vendor, payload.hub_product_id, payload.product_name_snapshot, ), ) return rows[0] also_service = AlsoService()