287 lines
9.8 KiB
Python
287 lines
9.8 KiB
Python
|
import logging
|
||
|
|
from typing import Any, Dict, List, Optional
|
||
|
|
|
||
|
|
from app.core.database import execute_query
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
def _to_float(value: Any, default: float = 0.0) -> float:
|
||
|
|
if value is None:
|
||
|
|
return default
|
||
|
|
try:
|
||
|
|
return float(value)
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
return default
|
||
|
|
|
||
|
|
|
||
|
|
def _apply_common_filters(
|
||
|
|
base_query: str,
|
||
|
|
params: List[Any],
|
||
|
|
customer_id: Optional[int],
|
||
|
|
sag_id: Optional[int],
|
||
|
|
q: Optional[str],
|
||
|
|
customer_alias: str,
|
||
|
|
sag_alias: str,
|
||
|
|
description_alias: str,
|
||
|
|
) -> tuple[str, List[Any]]:
|
||
|
|
query = base_query
|
||
|
|
|
||
|
|
if customer_id:
|
||
|
|
query += f" AND {customer_alias}.id = %s"
|
||
|
|
params.append(customer_id)
|
||
|
|
if sag_id:
|
||
|
|
query += f" AND {sag_alias}.id = %s"
|
||
|
|
params.append(sag_id)
|
||
|
|
if q:
|
||
|
|
like = f"%{q.lower()}%"
|
||
|
|
query += (
|
||
|
|
f" AND (LOWER({description_alias}) LIKE %s"
|
||
|
|
f" OR LOWER(COALESCE({sag_alias}.titel, '')) LIKE %s"
|
||
|
|
f" OR LOWER(COALESCE({customer_alias}.name, '')) LIKE %s)"
|
||
|
|
)
|
||
|
|
params.extend([like, like, like])
|
||
|
|
|
||
|
|
return query, params
|
||
|
|
|
||
|
|
|
||
|
|
def _fetch_sales_lines(customer_id: Optional[int], sag_id: Optional[int], q: Optional[str]) -> List[Dict[str, Any]]:
|
||
|
|
query = """
|
||
|
|
SELECT
|
||
|
|
si.id,
|
||
|
|
si.sag_id,
|
||
|
|
s.titel AS sag_title,
|
||
|
|
s.customer_id,
|
||
|
|
c.name AS customer_name,
|
||
|
|
si.description,
|
||
|
|
si.quantity,
|
||
|
|
si.unit,
|
||
|
|
si.unit_price,
|
||
|
|
si.amount,
|
||
|
|
si.currency,
|
||
|
|
si.status,
|
||
|
|
si.line_date,
|
||
|
|
si.product_id
|
||
|
|
FROM sag_salgsvarer si
|
||
|
|
JOIN sag_sager s ON s.id = si.sag_id
|
||
|
|
LEFT JOIN customers c ON c.id = s.customer_id
|
||
|
|
WHERE s.deleted_at IS NULL
|
||
|
|
AND LOWER(si.type) = 'sale'
|
||
|
|
AND LOWER(si.status) != 'cancelled'
|
||
|
|
"""
|
||
|
|
params: List[Any] = []
|
||
|
|
query, params = _apply_common_filters(query, params, customer_id, sag_id, q, "c", "s", "si.description")
|
||
|
|
query += " ORDER BY si.line_date DESC NULLS LAST, si.id DESC"
|
||
|
|
|
||
|
|
rows = execute_query(query, tuple(params)) or []
|
||
|
|
lines: List[Dict[str, Any]] = []
|
||
|
|
for row in rows:
|
||
|
|
qty = _to_float(row.get("quantity"), 0.0)
|
||
|
|
unit_price = _to_float(row.get("unit_price"), 0.0)
|
||
|
|
amount = _to_float(row.get("amount"), qty * unit_price)
|
||
|
|
lines.append(
|
||
|
|
{
|
||
|
|
"line_key": f"sale:{row['id']}",
|
||
|
|
"source_type": "sale",
|
||
|
|
"source_id": row["id"],
|
||
|
|
"reference_id": row["id"],
|
||
|
|
"sag_id": row.get("sag_id"),
|
||
|
|
"sag_title": row.get("sag_title"),
|
||
|
|
"customer_id": row.get("customer_id"),
|
||
|
|
"customer_name": row.get("customer_name"),
|
||
|
|
"description": row.get("description") or "Salgslinje",
|
||
|
|
"quantity": qty if qty > 0 else 1.0,
|
||
|
|
"unit": row.get("unit") or "stk",
|
||
|
|
"unit_price": unit_price,
|
||
|
|
"discount_percentage": 0.0,
|
||
|
|
"amount": amount,
|
||
|
|
"currency": row.get("currency") or "DKK",
|
||
|
|
"status": row.get("status") or "draft",
|
||
|
|
"line_date": str(row.get("line_date")) if row.get("line_date") else None,
|
||
|
|
"product_id": row.get("product_id"),
|
||
|
|
"selected": True,
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
return lines
|
||
|
|
|
||
|
|
|
||
|
|
def _fetch_subscription_lines(customer_id: Optional[int], sag_id: Optional[int], q: Optional[str]) -> List[Dict[str, Any]]:
|
||
|
|
query = """
|
||
|
|
SELECT
|
||
|
|
i.id,
|
||
|
|
i.subscription_id,
|
||
|
|
i.line_no,
|
||
|
|
i.product_id,
|
||
|
|
i.description,
|
||
|
|
i.quantity,
|
||
|
|
i.unit_price,
|
||
|
|
i.line_total,
|
||
|
|
s.id AS sub_id,
|
||
|
|
s.subscription_number,
|
||
|
|
s.status AS subscription_status,
|
||
|
|
s.billing_interval,
|
||
|
|
s.sag_id,
|
||
|
|
sg.titel AS sag_title,
|
||
|
|
s.customer_id,
|
||
|
|
c.name AS customer_name
|
||
|
|
FROM sag_subscription_items i
|
||
|
|
JOIN sag_subscriptions s ON s.id = i.subscription_id
|
||
|
|
JOIN sag_sager sg ON sg.id = s.sag_id
|
||
|
|
LEFT JOIN customers c ON c.id = s.customer_id
|
||
|
|
WHERE sg.deleted_at IS NULL
|
||
|
|
AND LOWER(s.status) IN ('draft', 'active', 'paused')
|
||
|
|
"""
|
||
|
|
params: List[Any] = []
|
||
|
|
query, params = _apply_common_filters(query, params, customer_id, sag_id, q, "c", "sg", "i.description")
|
||
|
|
query += " ORDER BY s.id DESC, i.line_no ASC, i.id ASC"
|
||
|
|
|
||
|
|
rows = execute_query(query, tuple(params)) or []
|
||
|
|
lines: List[Dict[str, Any]] = []
|
||
|
|
for row in rows:
|
||
|
|
qty = _to_float(row.get("quantity"), 1.0)
|
||
|
|
unit_price = _to_float(row.get("unit_price"), 0.0)
|
||
|
|
amount = _to_float(row.get("line_total"), qty * unit_price)
|
||
|
|
lines.append(
|
||
|
|
{
|
||
|
|
"line_key": f"subscription:{row['id']}",
|
||
|
|
"source_type": "subscription",
|
||
|
|
"source_id": row["id"],
|
||
|
|
"reference_id": row.get("subscription_id"),
|
||
|
|
"subscription_number": row.get("subscription_number"),
|
||
|
|
"sag_id": row.get("sag_id"),
|
||
|
|
"sag_title": row.get("sag_title"),
|
||
|
|
"customer_id": row.get("customer_id"),
|
||
|
|
"customer_name": row.get("customer_name"),
|
||
|
|
"description": row.get("description") or "Abonnementslinje",
|
||
|
|
"quantity": qty if qty > 0 else 1.0,
|
||
|
|
"unit": "stk",
|
||
|
|
"unit_price": unit_price,
|
||
|
|
"discount_percentage": 0.0,
|
||
|
|
"amount": amount,
|
||
|
|
"currency": "DKK",
|
||
|
|
"status": row.get("subscription_status") or "draft",
|
||
|
|
"line_date": None,
|
||
|
|
"product_id": row.get("product_id"),
|
||
|
|
"selected": True,
|
||
|
|
"meta": {
|
||
|
|
"billing_interval": row.get("billing_interval"),
|
||
|
|
"line_no": row.get("line_no"),
|
||
|
|
},
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
return lines
|
||
|
|
|
||
|
|
|
||
|
|
def _fetch_hardware_lines(customer_id: Optional[int], sag_id: Optional[int], q: Optional[str]) -> List[Dict[str, Any]]:
|
||
|
|
query = """
|
||
|
|
SELECT
|
||
|
|
sh.id AS relation_id,
|
||
|
|
sh.sag_id,
|
||
|
|
sh.note,
|
||
|
|
s.titel AS sag_title,
|
||
|
|
s.customer_id,
|
||
|
|
c.name AS customer_name,
|
||
|
|
h.id AS hardware_id,
|
||
|
|
h.asset_type,
|
||
|
|
h.brand,
|
||
|
|
h.model,
|
||
|
|
h.serial_number,
|
||
|
|
h.status AS hardware_status
|
||
|
|
FROM sag_hardware sh
|
||
|
|
JOIN sag_sager s ON s.id = sh.sag_id
|
||
|
|
JOIN hardware_assets h ON h.id = sh.hardware_id
|
||
|
|
LEFT JOIN customers c ON c.id = s.customer_id
|
||
|
|
WHERE sh.deleted_at IS NULL
|
||
|
|
AND s.deleted_at IS NULL
|
||
|
|
AND h.deleted_at IS NULL
|
||
|
|
"""
|
||
|
|
params: List[Any] = []
|
||
|
|
query, params = _apply_common_filters(
|
||
|
|
query,
|
||
|
|
params,
|
||
|
|
customer_id,
|
||
|
|
sag_id,
|
||
|
|
q,
|
||
|
|
"c",
|
||
|
|
"s",
|
||
|
|
"CONCAT(COALESCE(h.brand, ''), ' ', COALESCE(h.model, ''), ' ', COALESCE(h.serial_number, ''))",
|
||
|
|
)
|
||
|
|
query += " ORDER BY sh.id DESC"
|
||
|
|
|
||
|
|
rows = execute_query(query, tuple(params)) or []
|
||
|
|
lines: List[Dict[str, Any]] = []
|
||
|
|
for row in rows:
|
||
|
|
serial = row.get("serial_number")
|
||
|
|
serial_part = f" (S/N: {serial})" if serial else ""
|
||
|
|
brand_model = " ".join([part for part in [row.get("brand"), row.get("model")] if part]).strip()
|
||
|
|
label = brand_model or row.get("asset_type") or "Hardware"
|
||
|
|
desc = f"Hardware: {label}{serial_part}"
|
||
|
|
if row.get("note"):
|
||
|
|
desc = f"{desc} - {row['note']}"
|
||
|
|
|
||
|
|
lines.append(
|
||
|
|
{
|
||
|
|
"line_key": f"hardware:{row['relation_id']}",
|
||
|
|
"source_type": "hardware",
|
||
|
|
"source_id": row["relation_id"],
|
||
|
|
"reference_id": row.get("hardware_id"),
|
||
|
|
"sag_id": row.get("sag_id"),
|
||
|
|
"sag_title": row.get("sag_title"),
|
||
|
|
"customer_id": row.get("customer_id"),
|
||
|
|
"customer_name": row.get("customer_name"),
|
||
|
|
"description": desc,
|
||
|
|
"quantity": 1.0,
|
||
|
|
"unit": "stk",
|
||
|
|
"unit_price": 0.0,
|
||
|
|
"discount_percentage": 0.0,
|
||
|
|
"amount": 0.0,
|
||
|
|
"currency": "DKK",
|
||
|
|
"status": row.get("hardware_status") or "active",
|
||
|
|
"line_date": None,
|
||
|
|
"product_id": None,
|
||
|
|
"selected": True,
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
return lines
|
||
|
|
|
||
|
|
|
||
|
|
def aggregate_order_lines(
|
||
|
|
customer_id: Optional[int] = None,
|
||
|
|
sag_id: Optional[int] = None,
|
||
|
|
q: Optional[str] = None,
|
||
|
|
) -> Dict[str, Any]:
|
||
|
|
"""Aggregate order-ready lines across sale, subscription and hardware sources."""
|
||
|
|
sales_lines = _fetch_sales_lines(customer_id=customer_id, sag_id=sag_id, q=q)
|
||
|
|
subscription_lines = _fetch_subscription_lines(customer_id=customer_id, sag_id=sag_id, q=q)
|
||
|
|
hardware_lines = _fetch_hardware_lines(customer_id=customer_id, sag_id=sag_id, q=q)
|
||
|
|
|
||
|
|
all_lines = sales_lines + subscription_lines + hardware_lines
|
||
|
|
|
||
|
|
total_amount = sum(_to_float(line.get("amount")) for line in all_lines)
|
||
|
|
selected_amount = sum(_to_float(line.get("amount")) for line in all_lines if line.get("selected"))
|
||
|
|
|
||
|
|
customer_ids = sorted(
|
||
|
|
{
|
||
|
|
int(line["customer_id"])
|
||
|
|
for line in all_lines
|
||
|
|
if line.get("customer_id") is not None
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"lines": all_lines,
|
||
|
|
"summary": {
|
||
|
|
"line_count": len(all_lines),
|
||
|
|
"line_count_sales": len(sales_lines),
|
||
|
|
"line_count_subscriptions": len(subscription_lines),
|
||
|
|
"line_count_hardware": len(hardware_lines),
|
||
|
|
"customer_count": len(customer_ids),
|
||
|
|
"total_amount": round(total_amount, 2),
|
||
|
|
"selected_amount": round(selected_amount, 2),
|
||
|
|
"currency": "DKK",
|
||
|
|
},
|
||
|
|
}
|