2598 lines
100 KiB
Python
2598 lines
100 KiB
Python
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()
|