bmc_hub/app/modules/also/backend/service.py

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()