import hashlib import json import logging from datetime import datetime from decimal import Decimal import re from typing import Any, Dict, List, Optional from fastapi import HTTPException from app.core.config import settings from app.core.database import execute_query, execute_query_single, table_has_column from app.modules.also.models.schemas import ( AlsoCompanyMappingUpsert, AlsoImportJobCreate, AlsoImportLinesRequest, AlsoProductMappingUpsert, ) logger = logging.getLogger(__name__) 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 _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 _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: @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 _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"]) 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"]) if material_number: 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"]) 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"]) return None 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]]: 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 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( """ SELECT * FROM also_import_lines WHERE queue_status = %s ORDER BY id DESC LIMIT %s """, (status, max(1, min(limit, 1000))), ) or [] return execute_query( """ SELECT * FROM also_import_lines ORDER BY id DESC LIMIT %s """, (max(1, min(limit, 1000)),), ) or [] 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) 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) 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( """ WITH month_lines AS ( SELECT * FROM also_import_lines WHERE date_trunc('month', COALESCE(billing_start::timestamp, created_at)) = date_trunc('month', CURRENT_DATE) ) SELECT COALESCE(SUM(COALESCE(total_price, sales_price, 0)), 0) AS monthly_revenue, COALESCE(SUM(COALESCE(cost_amount, 0)), 0) AS monthly_cost, COALESCE(SUM(COALESCE(total_price, sales_price, 0) - COALESCE(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 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 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()