d81a8f41b4
- Implemented a comprehensive end-to-end testing script for the Sag module's HTTP API, covering various functionalities including case creation, updates, and file uploads. - Introduced a safe HTML sanitizer utility to ensure safe rendering of HTML content in the BMC Hub UI. - Added database migrations for new features including WAN connection marking for wall outlets, permanent audit trails for supplier invoices, and dedicated permissions for the Sag module. - Created a migration center for manual subscription and invoice migrations with relevant tables and indexes. - Added tests for migration center functionalities, ensuring stability and correctness of the new features.
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()
|