642 lines
28 KiB
Python
642 lines
28 KiB
Python
|
"""Core repositories and workflows for the manual migration centre."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import csv
|
||
|
|
import hashlib
|
||
|
|
import io
|
||
|
|
import json
|
||
|
|
from datetime import date, datetime, timedelta, timezone
|
||
|
|
from decimal import Decimal
|
||
|
|
from difflib import SequenceMatcher
|
||
|
|
from typing import Any, Dict, Iterable, List, Optional
|
||
|
|
|
||
|
|
import jwt
|
||
|
|
from fastapi import HTTPException, Request
|
||
|
|
from psycopg2.extras import Json, RealDictCursor
|
||
|
|
|
||
|
|
from app.core.config import settings
|
||
|
|
from app.core.database import (
|
||
|
|
execute_query,
|
||
|
|
execute_query_single,
|
||
|
|
get_db_connection,
|
||
|
|
release_db_connection,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
MUTABLE_LOCK_STATES = {"unlocked", "lock_failed"}
|
||
|
|
|
||
|
|
|
||
|
|
def subscription_like_item_sql(alias: str = "migration_center_session_items") -> str:
|
||
|
|
"""Fast predicate using the classification refreshed when snapshots change."""
|
||
|
|
p = f"{alias}." if alias else ""
|
||
|
|
return f"{p}subscription_like=TRUE"
|
||
|
|
|
||
|
|
|
||
|
|
def refresh_subscription_relevance(session_id: int) -> Dict[str, int]:
|
||
|
|
"""Classify imported invoice lines once so all views share a fast, consistent filter."""
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
UPDATE migration_center_session_items
|
||
|
|
SET subscription_like=TRUE, subscription_relevance_reason='crm_subscription'
|
||
|
|
WHERE session_id=%s AND source_system<>'economic'
|
||
|
|
""",
|
||
|
|
(session_id,), fetch=False,
|
||
|
|
)
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
UPDATE migration_center_session_items
|
||
|
|
SET subscription_like=FALSE, subscription_relevance_reason='one_off'
|
||
|
|
WHERE session_id=%s AND source_system='economic'
|
||
|
|
""",
|
||
|
|
(session_id,), fetch=False,
|
||
|
|
)
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
UPDATE migration_center_session_items
|
||
|
|
SET subscription_relevance_reason='excluded_charge'
|
||
|
|
WHERE session_id=%s AND source_system='economic'
|
||
|
|
AND (
|
||
|
|
POSITION('gebyr' IN LOWER(COALESCE(product_name,'')))>0
|
||
|
|
OR POSITION('fragt' IN LOWER(COALESCE(product_name,'')))>0
|
||
|
|
OR POSITION('porto' IN LOWER(COALESCE(product_name,'')))>0
|
||
|
|
)
|
||
|
|
""",
|
||
|
|
(session_id,), fetch=False,
|
||
|
|
)
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
UPDATE migration_center_session_items
|
||
|
|
SET subscription_like=TRUE, subscription_relevance_reason='subscription_keyword'
|
||
|
|
WHERE session_id=%s AND source_system='economic'
|
||
|
|
AND subscription_relevance_reason<>'excluded_charge'
|
||
|
|
AND LOWER(COALESCE(product_name,'')) ~
|
||
|
|
'(abonnement|subscription|måned|kvartal|årlig|licens|license|fiber|internet|bredbånd|hosting|domæne|domain|cloud|microsoft|office[ ]?365|backup|supportaftale|driftsaftale|udlejning|leje|telefoni|simkort|eset)'
|
||
|
|
""",
|
||
|
|
(session_id,), fetch=False,
|
||
|
|
)
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
WITH recurring AS (
|
||
|
|
SELECT
|
||
|
|
COALESCE(NULLIF(customer_no,''),source_customer_id,customer_name) AS customer_key,
|
||
|
|
COALESCE(NULLIF(product_code,''),LOWER(REGEXP_REPLACE(product_name,'\\s+',' ','g'))) AS product_key
|
||
|
|
FROM migration_center_session_items
|
||
|
|
WHERE session_id=%s AND source_system='economic'
|
||
|
|
AND subscription_relevance_reason<>'excluded_charge'
|
||
|
|
AND invoice_date >= (DATE_TRUNC('month',CURRENT_DATE)-INTERVAL '12 months')::date
|
||
|
|
GROUP BY 1,2
|
||
|
|
HAVING COUNT(DISTINCT invoice_no)>=2
|
||
|
|
AND COUNT(DISTINCT DATE_TRUNC('month',invoice_date))>=2
|
||
|
|
)
|
||
|
|
UPDATE migration_center_session_items item
|
||
|
|
SET subscription_like=TRUE, subscription_relevance_reason='recurring_invoice'
|
||
|
|
FROM recurring
|
||
|
|
WHERE item.session_id=%s AND item.source_system='economic'
|
||
|
|
AND item.subscription_relevance_reason<>'excluded_charge'
|
||
|
|
AND COALESCE(NULLIF(item.customer_no,''),item.source_customer_id,item.customer_name)
|
||
|
|
IS NOT DISTINCT FROM recurring.customer_key
|
||
|
|
AND COALESCE(NULLIF(item.product_code,''),LOWER(REGEXP_REPLACE(item.product_name,'\\s+',' ','g')))
|
||
|
|
IS NOT DISTINCT FROM recurring.product_key
|
||
|
|
""",
|
||
|
|
(session_id, session_id), fetch=False,
|
||
|
|
)
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
UPDATE migration_center_session_items item
|
||
|
|
SET subscription_like=TRUE, subscription_relevance_reason='existing_subscription'
|
||
|
|
WHERE item.session_id=%s AND item.source_system='economic'
|
||
|
|
AND item.subscription_relevance_reason<>'excluded_charge'
|
||
|
|
AND EXISTS (
|
||
|
|
SELECT 1 FROM sag_subscriptions subscription
|
||
|
|
WHERE subscription.customer_id=item.hub_customer_id
|
||
|
|
AND subscription.status<>'cancelled'
|
||
|
|
AND (
|
||
|
|
LOWER(TRIM(COALESCE(subscription.product_name,'')))=
|
||
|
|
LOWER(TRIM(COALESCE(item.product_name,'')))
|
||
|
|
OR ABS(COALESCE(subscription.price,0)-COALESCE(item.amount,0))<=0.01
|
||
|
|
)
|
||
|
|
)
|
||
|
|
""",
|
||
|
|
(session_id,), fetch=False,
|
||
|
|
)
|
||
|
|
row = execute_query_single(
|
||
|
|
"""
|
||
|
|
SELECT COUNT(*) FILTER (WHERE subscription_like) AS visible,
|
||
|
|
COUNT(*) FILTER (WHERE NOT subscription_like) AS hidden
|
||
|
|
FROM migration_center_session_items WHERE session_id=%s
|
||
|
|
""",
|
||
|
|
(session_id,),
|
||
|
|
)
|
||
|
|
return {"visible": int(row["visible"]), "hidden": int(row["hidden"])}
|
||
|
|
|
||
|
|
|
||
|
|
def json_value(value: Any) -> Any:
|
||
|
|
if isinstance(value, (datetime, date)):
|
||
|
|
return value.isoformat()
|
||
|
|
if isinstance(value, Decimal):
|
||
|
|
return float(value)
|
||
|
|
if isinstance(value, dict):
|
||
|
|
return {str(k): json_value(v) for k, v in value.items()}
|
||
|
|
if isinstance(value, (list, tuple)):
|
||
|
|
return [json_value(v) for v in value]
|
||
|
|
return value
|
||
|
|
|
||
|
|
|
||
|
|
def snapshot_hash(payload: Dict[str, Any]) -> str:
|
||
|
|
packed = json.dumps(json_value(payload), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||
|
|
return hashlib.sha256(packed.encode("utf-8")).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def user_id(current_user: Dict[str, Any]) -> Optional[int]:
|
||
|
|
value = current_user.get("id") or current_user.get("user_id")
|
||
|
|
return int(value) if value is not None else None
|
||
|
|
|
||
|
|
|
||
|
|
def audit(
|
||
|
|
*,
|
||
|
|
request: Request,
|
||
|
|
current_user: Dict[str, Any],
|
||
|
|
action: str,
|
||
|
|
entity_type: str,
|
||
|
|
entity_id: Any = None,
|
||
|
|
session_id: Optional[int] = None,
|
||
|
|
item_id: Optional[int] = None,
|
||
|
|
old_value: Any = None,
|
||
|
|
new_value: Any = None,
|
||
|
|
source_hash_value: Optional[str] = None,
|
||
|
|
success: bool = True,
|
||
|
|
error_message: Optional[str] = None,
|
||
|
|
) -> None:
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
INSERT INTO migration_center_audit_log
|
||
|
|
(session_id, session_item_id, entity_type, entity_id, action, old_value, new_value,
|
||
|
|
source_hash, performed_by_user_id, ip_address, success, error_message)
|
||
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
session_id, item_id, entity_type, str(entity_id) if entity_id is not None else None,
|
||
|
|
action, Json(json_value(old_value)) if old_value is not None else None,
|
||
|
|
Json(json_value(new_value)) if new_value is not None else None,
|
||
|
|
source_hash_value, user_id(current_user),
|
||
|
|
request.client.host if request.client else None, success, error_message,
|
||
|
|
),
|
||
|
|
fetch=False,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def ensure_writable(session_id: Optional[int] = None) -> None:
|
||
|
|
if getattr(settings, "MIGRATION_CENTER_READ_ONLY", False):
|
||
|
|
raise HTTPException(status_code=423, detail="Migreringscenteret er i read-only tilstand")
|
||
|
|
if session_id:
|
||
|
|
session = execute_query_single(
|
||
|
|
"SELECT read_only, status FROM migration_center_sessions WHERE id = %s", (session_id,)
|
||
|
|
)
|
||
|
|
if not session:
|
||
|
|
raise HTTPException(status_code=404, detail="Kontrolsession blev ikke fundet")
|
||
|
|
if session["read_only"] or session["status"] in {"completed", "archived"}:
|
||
|
|
raise HTTPException(status_code=423, detail="Kontrolsessionen er skrivebeskyttet")
|
||
|
|
|
||
|
|
|
||
|
|
class EconomicSnapshotRepository:
|
||
|
|
"""Read-only access to the Invoice Error Finder invoice snapshot."""
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def latest_run() -> Optional[Dict[str, Any]]:
|
||
|
|
return execute_query_single(
|
||
|
|
"""
|
||
|
|
SELECT r.id, r.source_type, r.started_at, r.completed_at, r.status,
|
||
|
|
r.records_imported, r.records_failed, COUNT(i.id) AS invoice_count
|
||
|
|
FROM invoice_error_finder_import_runs r
|
||
|
|
JOIN invoice_error_finder_economic_invoices i ON i.import_run_id = r.id
|
||
|
|
WHERE r.source_type = 'economic_invoices'
|
||
|
|
AND r.status IN ('success', 'partial')
|
||
|
|
AND r.completed_at IS NOT NULL
|
||
|
|
GROUP BY r.id
|
||
|
|
HAVING COUNT(i.id) > 0
|
||
|
|
ORDER BY r.completed_at DESC, r.id DESC
|
||
|
|
LIMIT 1
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def lines(run_id: int) -> List[Dict[str, Any]]:
|
||
|
|
# DISTINCT ON makes the source identity stable even if an API endpoint repeats a line.
|
||
|
|
return execute_query(
|
||
|
|
"""
|
||
|
|
SELECT DISTINCT ON (
|
||
|
|
COALESCE(i.source_invoice_number, i.id::text),
|
||
|
|
i.source_type,
|
||
|
|
COALESCE(l.line_number, l.id)
|
||
|
|
)
|
||
|
|
i.id AS invoice_id, i.source_invoice_number, i.source_type, i.customer_number,
|
||
|
|
i.customer_name, i.invoice_date, i.currency, i.source_raw AS invoice_raw,
|
||
|
|
l.id AS invoice_line_id, l.line_number, l.product_number, l.product_name,
|
||
|
|
l.description, l.quantity, l.unit_price, l.line_net_amount, l.source_raw AS line_raw
|
||
|
|
FROM invoice_error_finder_economic_invoices i
|
||
|
|
JOIN invoice_error_finder_economic_invoice_lines l ON l.invoice_id = i.id
|
||
|
|
WHERE i.import_run_id = %s
|
||
|
|
AND i.invoice_date >= (DATE_TRUNC('month', CURRENT_DATE) - INTERVAL '12 months')::date
|
||
|
|
AND i.invoice_date <= CURRENT_DATE
|
||
|
|
AND LOWER(
|
||
|
|
COALESCE(l.product_name,'') || ' ' ||
|
||
|
|
COALESCE(l.description,'') || ' ' ||
|
||
|
|
COALESCE(l.product_number,'')
|
||
|
|
) NOT SIMILAR TO '%%(gebyr|fragt|porto)%%'
|
||
|
|
ORDER BY
|
||
|
|
COALESCE(i.source_invoice_number, i.id::text),
|
||
|
|
i.source_type,
|
||
|
|
COALESCE(l.line_number, l.id),
|
||
|
|
l.id DESC
|
||
|
|
""",
|
||
|
|
(run_id,),
|
||
|
|
) or []
|
||
|
|
|
||
|
|
|
||
|
|
def attach_economic_snapshot(session_id: int, run: Dict[str, Any]) -> Dict[str, int]:
|
||
|
|
"""Attach the selected usable 13-month IEF snapshot to an explicit session."""
|
||
|
|
counts = {"created": 0, "unchanged": 0}
|
||
|
|
conn = get_db_connection()
|
||
|
|
try:
|
||
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
|
||
|
|
cursor.execute(
|
||
|
|
"""
|
||
|
|
UPDATE migration_center_sessions
|
||
|
|
SET economic_import_run_id=%s, economic_snapshot_at=%s, updated_at=CURRENT_TIMESTAMP
|
||
|
|
WHERE id=%s
|
||
|
|
""",
|
||
|
|
(run["id"], run["completed_at"], session_id),
|
||
|
|
)
|
||
|
|
for row in EconomicSnapshotRepository.lines(run["id"]):
|
||
|
|
item = _normalize_economic_line(dict(row))
|
||
|
|
cursor.execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO migration_center_session_items
|
||
|
|
(session_id, entity_type, source_system, source_record_id, source_customer_id,
|
||
|
|
customer_no, customer_name, product_code, product_name, amount, quantity,
|
||
|
|
billing_frequency, period_from, period_to, invoice_no, invoice_date,
|
||
|
|
source_payload, source_hash)
|
||
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
|
|
ON CONFLICT (session_id, entity_type, source_system, source_record_id) DO NOTHING
|
||
|
|
RETURNING id
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
session_id, item["entity_type"], item["source_system"], item["source_record_id"],
|
||
|
|
item["source_customer_id"], item["customer_no"], item["customer_name"],
|
||
|
|
item["product_code"], item["product_name"], item["amount"], item["quantity"],
|
||
|
|
item["billing_frequency"], item["period_from"], item["period_to"], item["invoice_no"],
|
||
|
|
item["invoice_date"], Json(json_value(item["source_payload"])), item["source_hash"],
|
||
|
|
),
|
||
|
|
)
|
||
|
|
inserted = cursor.fetchone()
|
||
|
|
counts["created" if inserted else "unchanged"] += 1
|
||
|
|
conn.commit()
|
||
|
|
except Exception:
|
||
|
|
conn.rollback()
|
||
|
|
raise
|
||
|
|
finally:
|
||
|
|
release_db_connection(conn)
|
||
|
|
return counts
|
||
|
|
|
||
|
|
|
||
|
|
def _normalize_economic_line(row: Dict[str, Any]) -> Dict[str, Any]:
|
||
|
|
source_id = f"{row.get('source_type')}:{row.get('source_invoice_number') or row['invoice_id']}:{row.get('line_number') or row['invoice_line_id']}"
|
||
|
|
raw = {"invoice": row.get("invoice_raw") or {}, "line": row.get("line_raw") or {}}
|
||
|
|
normalized = {
|
||
|
|
"entity_type": "invoice_line",
|
||
|
|
"source_system": "economic",
|
||
|
|
"source_record_id": source_id,
|
||
|
|
"source_customer_id": str(row.get("customer_number") or ""),
|
||
|
|
"customer_no": str(row.get("customer_number") or ""),
|
||
|
|
"customer_name": str(row.get("customer_name") or "")[:255] or None,
|
||
|
|
"product_code": str(row.get("product_number") or "")[:100] or None,
|
||
|
|
"product_name": str(row.get("product_name") or row.get("description") or "Fakturalinje")[:500],
|
||
|
|
"amount": row.get("line_net_amount") or 0,
|
||
|
|
"quantity": row.get("quantity") or 1,
|
||
|
|
"billing_frequency": None,
|
||
|
|
"period_from": row.get("invoice_date"),
|
||
|
|
"period_to": None,
|
||
|
|
"invoice_no": row.get("source_invoice_number"),
|
||
|
|
"invoice_date": row.get("invoice_date"),
|
||
|
|
"source_payload": raw,
|
||
|
|
}
|
||
|
|
normalized["source_hash"] = snapshot_hash(normalized)
|
||
|
|
return normalized
|
||
|
|
|
||
|
|
|
||
|
|
def _customer_candidates(item: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||
|
|
rows = execute_query(
|
||
|
|
"""
|
||
|
|
SELECT id, name, cvr_number, email, email_domain, economic_customer_number
|
||
|
|
FROM customers
|
||
|
|
WHERE deleted_at IS NULL
|
||
|
|
ORDER BY id
|
||
|
|
"""
|
||
|
|
) or []
|
||
|
|
source_no = str(item.get("customer_no") or "").strip().lower()
|
||
|
|
source_name = str(item.get("customer_name") or "").strip().lower()
|
||
|
|
source_cvr = str((item.get("source_payload") or {}).get("customer_cvr") or "").replace(" ", "").strip()
|
||
|
|
candidates = []
|
||
|
|
for row in rows:
|
||
|
|
rules: List[str] = []
|
||
|
|
score = 0.0
|
||
|
|
if source_no and str(row.get("economic_customer_number") or "").strip().lower() == source_no:
|
||
|
|
score += 0.75
|
||
|
|
rules.append("Kundenummer stemmer")
|
||
|
|
if source_cvr and str(row.get("cvr_number") or "").replace(" ", "").strip() == source_cvr:
|
||
|
|
score += 0.90
|
||
|
|
rules.append("CVR stemmer")
|
||
|
|
name_ratio = SequenceMatcher(None, source_name, str(row.get("name") or "").strip().lower()).ratio()
|
||
|
|
if source_name and name_ratio >= 0.70:
|
||
|
|
score += min(0.25, name_ratio * 0.25)
|
||
|
|
rules.append(f"Firmanavn ligner ({round(name_ratio * 100)} %)")
|
||
|
|
if score:
|
||
|
|
candidates.append({"id": row["id"], "score": min(score, 1.0), "rules": rules})
|
||
|
|
return sorted(candidates, key=lambda value: value["score"], reverse=True)
|
||
|
|
|
||
|
|
|
||
|
|
def _resolve_source_customer(item: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||
|
|
"""Resolve authoritative CRM mappings before heuristic customer matching."""
|
||
|
|
source_id = str(item.get("source_customer_id") or "").strip()
|
||
|
|
if not source_id:
|
||
|
|
return None
|
||
|
|
if item.get("source_system") == "simply":
|
||
|
|
row = execute_query_single(
|
||
|
|
"""
|
||
|
|
SELECT source_customer_name AS customer_name, source_customer_cvr AS cvr,
|
||
|
|
hub_customer_id
|
||
|
|
FROM simply_subscription_staging
|
||
|
|
WHERE source_account_id=%s
|
||
|
|
ORDER BY (hub_customer_id IS NOT NULL) DESC, updated_at DESC, id DESC
|
||
|
|
LIMIT 1
|
||
|
|
""",
|
||
|
|
(source_id,),
|
||
|
|
)
|
||
|
|
if row:
|
||
|
|
result = dict(row)
|
||
|
|
result["rule"] = "Eksisterende Simply-kundemapping"
|
||
|
|
return result
|
||
|
|
if item.get("source_system") == "vtiger":
|
||
|
|
row = execute_query_single(
|
||
|
|
"""
|
||
|
|
SELECT id AS hub_customer_id, name AS customer_name, cvr_number AS cvr
|
||
|
|
FROM customers WHERE vtiger_id=%s AND deleted_at IS NULL LIMIT 1
|
||
|
|
""",
|
||
|
|
(source_id,),
|
||
|
|
)
|
||
|
|
if row:
|
||
|
|
result = dict(row)
|
||
|
|
result["rule"] = "Vtiger-konto-id stemmer"
|
||
|
|
return result
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def _subscription_candidates(item: Dict[str, Any], customer_id: Optional[int]) -> List[Dict[str, Any]]:
|
||
|
|
if not customer_id:
|
||
|
|
return []
|
||
|
|
rows = execute_query(
|
||
|
|
"""
|
||
|
|
SELECT id, product_name, price, billing_interval, start_date, end_date
|
||
|
|
FROM sag_subscriptions
|
||
|
|
WHERE customer_id = %s AND status <> 'cancelled'
|
||
|
|
ORDER BY updated_at DESC, id DESC
|
||
|
|
""",
|
||
|
|
(customer_id,),
|
||
|
|
) or []
|
||
|
|
source_name = str(item.get("product_name") or "").strip().lower()
|
||
|
|
source_amount = Decimal(str(item.get("amount") or 0))
|
||
|
|
candidates = []
|
||
|
|
for row in rows:
|
||
|
|
rules: List[str] = []
|
||
|
|
score = 0.0
|
||
|
|
ratio = SequenceMatcher(None, source_name, str(row.get("product_name") or "").strip().lower()).ratio()
|
||
|
|
if ratio >= 0.55:
|
||
|
|
score += ratio * 0.55
|
||
|
|
rules.append(f"Produktnavn ligner ({round(ratio * 100)} %)")
|
||
|
|
hub_amount = Decimal(str(row.get("price") or 0))
|
||
|
|
if abs(source_amount - hub_amount) <= Decimal("0.01"):
|
||
|
|
score += 0.35
|
||
|
|
rules.append("Beløb stemmer")
|
||
|
|
elif max(abs(source_amount), Decimal("1")) and abs(source_amount - hub_amount) / max(abs(source_amount), Decimal("1")) <= Decimal("0.10"):
|
||
|
|
score += 0.15
|
||
|
|
rules.append("Beløb afviger højst 10 %")
|
||
|
|
if item.get("billing_frequency") and item["billing_frequency"] == row.get("billing_interval"):
|
||
|
|
score += 0.10
|
||
|
|
rules.append("Frekvens stemmer")
|
||
|
|
if score >= 0.35:
|
||
|
|
candidates.append({"id": row["id"], "score": min(score, 1.0), "rules": rules})
|
||
|
|
return sorted(candidates, key=lambda value: value["score"], reverse=True)
|
||
|
|
|
||
|
|
|
||
|
|
def match_item(item_id: int) -> Dict[str, Any]:
|
||
|
|
item = execute_query_single("SELECT * FROM migration_center_session_items WHERE id = %s", (item_id,))
|
||
|
|
if not item:
|
||
|
|
raise HTTPException(status_code=404, detail="Post blev ikke fundet")
|
||
|
|
authoritative = _resolve_source_customer(item)
|
||
|
|
if authoritative:
|
||
|
|
enriched_payload = dict(item.get("source_payload") or {})
|
||
|
|
if authoritative.get("cvr"):
|
||
|
|
enriched_payload["customer_cvr"] = authoritative["cvr"]
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
UPDATE migration_center_session_items
|
||
|
|
SET customer_name=COALESCE(NULLIF(%s,''),customer_name),
|
||
|
|
hub_customer_id=COALESCE(%s,hub_customer_id), source_payload=%s,
|
||
|
|
updated_at=CURRENT_TIMESTAMP
|
||
|
|
WHERE id=%s AND lock_status IN ('unlocked','lock_failed')
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
authoritative.get("customer_name"), authoritative.get("hub_customer_id"),
|
||
|
|
Json(json_value(enriched_payload)), item_id,
|
||
|
|
), fetch=False,
|
||
|
|
)
|
||
|
|
item = execute_query_single("SELECT * FROM migration_center_session_items WHERE id=%s", (item_id,))
|
||
|
|
customer_matches = _customer_candidates(item)
|
||
|
|
customer_id = item.get("hub_customer_id") or (customer_matches[0]["id"] if customer_matches else None)
|
||
|
|
sub_matches = _subscription_candidates(item, customer_id)
|
||
|
|
best = sub_matches[0] if sub_matches else None
|
||
|
|
explanations = []
|
||
|
|
if authoritative:
|
||
|
|
explanations.append(authoritative["rule"])
|
||
|
|
if customer_matches:
|
||
|
|
explanations.extend(customer_matches[0]["rules"])
|
||
|
|
explanations.extend(best["rules"] if best else ["Intet sikkert abonnement-match"])
|
||
|
|
confidence = best["score"] if best else (customer_matches[0]["score"] * 0.4 if customer_matches else 0)
|
||
|
|
status = "match_found" if best and confidence >= 0.60 else ("manual_review" if confidence else "no_match")
|
||
|
|
hub_status = item["hub_status"]
|
||
|
|
if hub_status == "not_created" and not best:
|
||
|
|
hub_status = "ready_for_creation" if customer_id else "not_created"
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
UPDATE migration_center_session_items
|
||
|
|
SET hub_customer_id = COALESCE(hub_customer_id, %s), suggested_hub_record_id = %s,
|
||
|
|
match_confidence = %s, match_explanation = %s, match_status = %s,
|
||
|
|
hub_status = %s, updated_at = CURRENT_TIMESTAMP
|
||
|
|
WHERE id = %s AND lock_status IN ('unlocked','lock_failed')
|
||
|
|
""",
|
||
|
|
(customer_id, best["id"] if best else None, confidence, Json(explanations), status, hub_status, item_id),
|
||
|
|
fetch=False,
|
||
|
|
)
|
||
|
|
execute_query(
|
||
|
|
"DELETE FROM migration_center_matches WHERE session_item_id=%s AND approved IS NULL",
|
||
|
|
(item_id,), fetch=False,
|
||
|
|
)
|
||
|
|
for candidate in customer_matches[:5]:
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
INSERT INTO migration_center_matches
|
||
|
|
(session_item_id, matched_entity_type, matched_hub_id, confidence, rules)
|
||
|
|
VALUES (%s,'customer',%s,%s,%s)
|
||
|
|
ON CONFLICT (session_item_id, matched_entity_type, matched_hub_id)
|
||
|
|
DO UPDATE SET confidence=EXCLUDED.confidence, rules=EXCLUDED.rules
|
||
|
|
""",
|
||
|
|
(item_id, candidate["id"], candidate["score"], Json(candidate["rules"])),
|
||
|
|
fetch=False,
|
||
|
|
)
|
||
|
|
for candidate in sub_matches[:5]:
|
||
|
|
execute_query(
|
||
|
|
"""
|
||
|
|
INSERT INTO migration_center_matches
|
||
|
|
(session_item_id, matched_entity_type, matched_hub_id, confidence, rules)
|
||
|
|
VALUES (%s,'subscription',%s,%s,%s)
|
||
|
|
ON CONFLICT (session_item_id, matched_entity_type, matched_hub_id)
|
||
|
|
DO UPDATE SET confidence=EXCLUDED.confidence, rules=EXCLUDED.rules
|
||
|
|
""",
|
||
|
|
(item_id, candidate["id"], candidate["score"], Json(candidate["rules"])),
|
||
|
|
fetch=False,
|
||
|
|
)
|
||
|
|
return execute_query_single("SELECT * FROM migration_center_session_items WHERE id = %s", (item_id,))
|
||
|
|
|
||
|
|
|
||
|
|
def create_session(name: str, current_user: Dict[str, Any], request: Request) -> Dict[str, Any]:
|
||
|
|
ensure_writable()
|
||
|
|
run = EconomicSnapshotRepository.latest_run()
|
||
|
|
if not run:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=409,
|
||
|
|
detail="Ingen færdig e-conomic-import findes i Faktura-fejl-finder",
|
||
|
|
)
|
||
|
|
conn = get_db_connection()
|
||
|
|
try:
|
||
|
|
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
|
||
|
|
cursor.execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO migration_center_sessions
|
||
|
|
(name, status, economic_import_run_id, economic_snapshot_at, created_by_user_id)
|
||
|
|
VALUES (%s, 'active', %s, %s, %s) RETURNING *
|
||
|
|
""",
|
||
|
|
(name.strip(), run["id"], run["completed_at"], user_id(current_user)),
|
||
|
|
)
|
||
|
|
session = dict(cursor.fetchone())
|
||
|
|
for row in EconomicSnapshotRepository.lines(run["id"]):
|
||
|
|
item = _normalize_economic_line(dict(row))
|
||
|
|
if item["source_customer_id"]:
|
||
|
|
customer_payload = {
|
||
|
|
"source_customer_id": item["source_customer_id"],
|
||
|
|
"customer_no": item["customer_no"],
|
||
|
|
"customer_name": item["customer_name"] or item["source_customer_id"],
|
||
|
|
}
|
||
|
|
cursor.execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO migration_center_source_customers
|
||
|
|
(source_system, source_customer_id, customer_no, customer_name, raw_payload, snapshot_hash)
|
||
|
|
VALUES ('economic',%s,%s,%s,%s,%s)
|
||
|
|
ON CONFLICT (source_system, source_customer_id) DO UPDATE SET
|
||
|
|
customer_no=EXCLUDED.customer_no, customer_name=EXCLUDED.customer_name,
|
||
|
|
raw_payload=EXCLUDED.raw_payload, snapshot_hash=EXCLUDED.snapshot_hash,
|
||
|
|
updated_at=CURRENT_TIMESTAMP
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
item["source_customer_id"], item["customer_no"], customer_payload["customer_name"],
|
||
|
|
Json(customer_payload), snapshot_hash(customer_payload),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
cursor.execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO migration_center_session_items
|
||
|
|
(session_id, entity_type, source_system, source_record_id, source_customer_id,
|
||
|
|
customer_no, customer_name, product_code, product_name, amount, quantity,
|
||
|
|
billing_frequency, period_from, period_to, invoice_no, invoice_date,
|
||
|
|
source_payload, source_hash)
|
||
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
|
|
ON CONFLICT (session_id, entity_type, source_system, source_record_id) DO NOTHING
|
||
|
|
""",
|
||
|
|
(
|
||
|
|
session["id"], item["entity_type"], item["source_system"], item["source_record_id"],
|
||
|
|
item["source_customer_id"], item["customer_no"], item["customer_name"],
|
||
|
|
item["product_code"], item["product_name"], item["amount"], item["quantity"],
|
||
|
|
item["billing_frequency"], item["period_from"], item["period_to"], item["invoice_no"],
|
||
|
|
item["invoice_date"], Json(json_value(item["source_payload"])), item["source_hash"],
|
||
|
|
),
|
||
|
|
)
|
||
|
|
conn.commit()
|
||
|
|
except Exception:
|
||
|
|
conn.rollback()
|
||
|
|
raise
|
||
|
|
finally:
|
||
|
|
release_db_connection(conn)
|
||
|
|
audit(
|
||
|
|
request=request, current_user=current_user, action="session_created",
|
||
|
|
entity_type="session", entity_id=session["id"], session_id=session["id"], new_value=session,
|
||
|
|
)
|
||
|
|
# Matching is deliberately outside the snapshot transaction and can be re-run safely.
|
||
|
|
ids = execute_query("SELECT id FROM migration_center_session_items WHERE session_id = %s", (session["id"],)) or []
|
||
|
|
for row in ids:
|
||
|
|
match_item(int(row["id"]))
|
||
|
|
refresh_subscription_relevance(int(session["id"]))
|
||
|
|
return session
|
||
|
|
|
||
|
|
|
||
|
|
def preflight_token(item: Dict[str, Any], current_user: Dict[str, Any]) -> str:
|
||
|
|
payload = {
|
||
|
|
"purpose": "migration_center_create",
|
||
|
|
"item_id": item["id"],
|
||
|
|
"source_hash": item["source_hash"],
|
||
|
|
"user_id": user_id(current_user),
|
||
|
|
"exp": datetime.now(timezone.utc) + timedelta(minutes=5),
|
||
|
|
}
|
||
|
|
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
|
||
|
|
|
||
|
|
|
||
|
|
def verify_preflight(token: str, item: Dict[str, Any], current_user: Dict[str, Any]) -> None:
|
||
|
|
try:
|
||
|
|
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
|
||
|
|
except jwt.PyJWTError as exc:
|
||
|
|
raise HTTPException(status_code=409, detail="Preflight er udløbet eller ugyldig") from exc
|
||
|
|
if (
|
||
|
|
payload.get("purpose") != "migration_center_create"
|
||
|
|
or int(payload.get("item_id", 0)) != int(item["id"])
|
||
|
|
or payload.get("source_hash") != item["source_hash"]
|
||
|
|
or int(payload.get("user_id", 0)) != int(user_id(current_user) or 0)
|
||
|
|
):
|
||
|
|
raise HTTPException(status_code=409, detail="Kildedata eller bruger er ændret siden preflight")
|
||
|
|
|
||
|
|
|
||
|
|
def report_csv(session_id: int) -> str:
|
||
|
|
session = execute_query_single("SELECT * FROM migration_center_sessions WHERE id = %s", (session_id,))
|
||
|
|
if not session:
|
||
|
|
raise HTTPException(status_code=404, detail="Kontrolsession blev ikke fundet")
|
||
|
|
rows = execute_query(
|
||
|
|
f"""
|
||
|
|
SELECT entity_type, source_system, source_record_id, invoice_no, customer_no, customer_name,
|
||
|
|
product_code, product_name, amount, quantity, match_status, approval_status,
|
||
|
|
hub_status, lock_status, hub_customer_id, hub_sag_id, hub_record_id,
|
||
|
|
ignore_reason, verified_at, locked_at
|
||
|
|
FROM migration_center_session_items
|
||
|
|
WHERE session_id = %s
|
||
|
|
AND {subscription_like_item_sql()}
|
||
|
|
ORDER BY id
|
||
|
|
""",
|
||
|
|
(session_id,),
|
||
|
|
) or []
|
||
|
|
output = io.StringIO()
|
||
|
|
fields = list(rows[0].keys()) if rows else [
|
||
|
|
"entity_type", "source_system", "source_record_id", "match_status",
|
||
|
|
"approval_status", "hub_status", "lock_status",
|
||
|
|
]
|
||
|
|
writer = csv.DictWriter(output, fieldnames=fields, extrasaction="ignore")
|
||
|
|
writer.writeheader()
|
||
|
|
for row in rows:
|
||
|
|
writer.writerow({key: json_value(value) for key, value in dict(row).items()})
|
||
|
|
return output.getvalue()
|