a604a3cc44
- Implemented test scripts for vTiger account retrieval, contact data, and modules. - Created a detailed test suite for the ticket module, covering database schema validation, ticket number generation, prepaid card constraints, and service logic. - Added tests for vTiger field inspection and various queries related to accounts and sales orders. - Introduced SQL migration scripts for the ALSO Cloud Billing foundation, including import jobs, lines, and mapping tables with necessary constraints and indices. - Enhanced workflow columns in the import lines table to support matching and validation timestamps.
908 lines
34 KiB
Python
908 lines
34 KiB
Python
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()
|