559 lines
20 KiB
Python
559 lines
20 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import uuid
|
|
from datetime import datetime
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from crm import nextcloud
|
|
from crm.quotation_models import (
|
|
QuotationCreate,
|
|
QuotationInDB,
|
|
QuotationItemCreate,
|
|
QuotationItemInDB,
|
|
QuotationListItem,
|
|
QuotationUpdate,
|
|
)
|
|
from crm.service import get_customer
|
|
import database as mqtt_db
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Path to Jinja2 templates directory (relative to this file)
|
|
_TEMPLATES_DIR = Path(__file__).parent.parent / "templates"
|
|
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
def _d(value) -> Decimal:
|
|
"""Convert to Decimal safely."""
|
|
return Decimal(str(value if value is not None else 0))
|
|
|
|
|
|
def _float(d: Decimal) -> float:
|
|
"""Round Decimal to 2dp and return as float for storage."""
|
|
return float(d.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
|
|
|
|
|
|
def _calculate_totals(
|
|
items: list,
|
|
global_discount_percent: float,
|
|
shipping_cost: float,
|
|
shipping_cost_discount: float,
|
|
install_cost: float,
|
|
install_cost_discount: float,
|
|
extras_cost: float,
|
|
) -> dict:
|
|
"""
|
|
Calculate all monetary totals using Decimal arithmetic (ROUND_HALF_UP).
|
|
VAT is computed per-item from each item's vat_percent field.
|
|
Shipping and install costs carry 0% VAT.
|
|
Returns a dict of floats ready for DB storage.
|
|
"""
|
|
# Per-line totals and per-item VAT
|
|
item_totals = []
|
|
item_vat = Decimal(0)
|
|
for item in items:
|
|
cost = _d(item.get("unit_cost", 0))
|
|
qty = _d(item.get("quantity", 1))
|
|
disc = _d(item.get("discount_percent", 0))
|
|
net = cost * qty * (1 - disc / 100)
|
|
item_totals.append(net)
|
|
vat_pct = _d(item.get("vat_percent", 24))
|
|
item_vat += net * (vat_pct / 100)
|
|
|
|
# Shipping net (VAT = 0%)
|
|
ship_gross = _d(shipping_cost)
|
|
ship_disc = _d(shipping_cost_discount)
|
|
ship_net = ship_gross * (1 - ship_disc / 100)
|
|
|
|
# Install net (VAT = 0%)
|
|
install_gross = _d(install_cost)
|
|
install_disc = _d(install_cost_discount)
|
|
install_net = install_gross * (1 - install_disc / 100)
|
|
|
|
subtotal = sum(item_totals, Decimal(0)) + ship_net + install_net
|
|
|
|
global_disc_pct = _d(global_discount_percent)
|
|
global_disc_amount = subtotal * (global_disc_pct / 100)
|
|
new_subtotal = subtotal - global_disc_amount
|
|
|
|
# Global discount proportionally reduces VAT too
|
|
if subtotal > 0:
|
|
disc_ratio = new_subtotal / subtotal
|
|
vat_amount = item_vat * disc_ratio
|
|
else:
|
|
vat_amount = Decimal(0)
|
|
|
|
extras = _d(extras_cost)
|
|
final_total = new_subtotal + vat_amount + extras
|
|
|
|
return {
|
|
"subtotal_before_discount": _float(subtotal),
|
|
"global_discount_amount": _float(global_disc_amount),
|
|
"new_subtotal": _float(new_subtotal),
|
|
"vat_amount": _float(vat_amount),
|
|
"final_total": _float(final_total),
|
|
}
|
|
|
|
|
|
def _calc_line_total(item) -> float:
|
|
cost = _d(item.get("unit_cost", 0))
|
|
qty = _d(item.get("quantity", 1))
|
|
disc = _d(item.get("discount_percent", 0))
|
|
return _float(cost * qty * (1 - disc / 100))
|
|
|
|
|
|
async def _generate_quotation_number(db) -> str:
|
|
year = datetime.utcnow().year
|
|
prefix = f"QT-{year}-"
|
|
rows = await db.execute_fetchall(
|
|
"SELECT quotation_number FROM crm_quotations WHERE quotation_number LIKE ? ORDER BY quotation_number DESC LIMIT 1",
|
|
(f"{prefix}%",),
|
|
)
|
|
if rows:
|
|
last_num = rows[0][0] # e.g. "QT-2026-012"
|
|
try:
|
|
seq = int(last_num[len(prefix):]) + 1
|
|
except ValueError:
|
|
seq = 1
|
|
else:
|
|
seq = 1
|
|
return f"{prefix}{seq:03d}"
|
|
|
|
|
|
def _row_to_quotation(row: dict, items: list[dict]) -> QuotationInDB:
|
|
row = dict(row)
|
|
row["comments"] = json.loads(row.get("comments") or "[]")
|
|
row["quick_notes"] = json.loads(row.get("quick_notes") or "{}")
|
|
item_models = [QuotationItemInDB(**{k: v for k, v in i.items() if k in QuotationItemInDB.model_fields}) for i in items]
|
|
return QuotationInDB(**{k: v for k, v in row.items() if k in QuotationInDB.model_fields}, items=item_models)
|
|
|
|
|
|
def _row_to_list_item(row: dict) -> QuotationListItem:
|
|
return QuotationListItem(**{k: v for k, v in dict(row).items() if k in QuotationListItem.model_fields})
|
|
|
|
|
|
async def _fetch_items(db, quotation_id: str) -> list[dict]:
|
|
rows = await db.execute_fetchall(
|
|
"SELECT * FROM crm_quotation_items WHERE quotation_id = ? ORDER BY sort_order ASC",
|
|
(quotation_id,),
|
|
)
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
# ── Public API ────────────────────────────────────────────────────────────────
|
|
|
|
async def get_next_number() -> str:
|
|
db = await mqtt_db.get_db()
|
|
return await _generate_quotation_number(db)
|
|
|
|
|
|
async def list_all_quotations() -> list[dict]:
|
|
"""Return all quotations across all customers, with customer_name injected."""
|
|
from shared.firebase import get_db as get_firestore
|
|
db = await mqtt_db.get_db()
|
|
rows = await db.execute_fetchall(
|
|
"SELECT id, quotation_number, title, customer_id, status, final_total, created_at, updated_at, "
|
|
"nextcloud_pdf_url, is_legacy, legacy_date, legacy_pdf_path "
|
|
"FROM crm_quotations ORDER BY created_at DESC",
|
|
(),
|
|
)
|
|
items = [dict(r) for r in rows]
|
|
# Fetch unique customer names from Firestore in one pass
|
|
customer_ids = {i["customer_id"] for i in items if i.get("customer_id")}
|
|
customer_names: dict[str, str] = {}
|
|
if customer_ids:
|
|
fstore = get_firestore()
|
|
for cid in customer_ids:
|
|
try:
|
|
doc = fstore.collection("crm_customers").document(cid).get()
|
|
if doc.exists:
|
|
d = doc.to_dict()
|
|
parts = [d.get("name", ""), d.get("surname", ""), d.get("organization", "")]
|
|
label = " ".join(p for p in parts if p).strip()
|
|
customer_names[cid] = label or cid
|
|
except Exception:
|
|
customer_names[cid] = cid
|
|
for item in items:
|
|
item["customer_name"] = customer_names.get(item["customer_id"], "")
|
|
return items
|
|
|
|
|
|
async def list_quotations(customer_id: str) -> list[QuotationListItem]:
|
|
db = await mqtt_db.get_db()
|
|
rows = await db.execute_fetchall(
|
|
"SELECT id, quotation_number, title, customer_id, status, final_total, created_at, updated_at, "
|
|
"nextcloud_pdf_url, is_legacy, legacy_date, legacy_pdf_path "
|
|
"FROM crm_quotations WHERE customer_id = ? ORDER BY created_at DESC",
|
|
(customer_id,),
|
|
)
|
|
return [_row_to_list_item(dict(r)) for r in rows]
|
|
|
|
|
|
async def get_quotation(quotation_id: str) -> QuotationInDB:
|
|
db = await mqtt_db.get_db()
|
|
rows = await db.execute_fetchall(
|
|
"SELECT * FROM crm_quotations WHERE id = ?", (quotation_id,)
|
|
)
|
|
if not rows:
|
|
raise HTTPException(status_code=404, detail="Quotation not found")
|
|
items = await _fetch_items(db, quotation_id)
|
|
return _row_to_quotation(dict(rows[0]), items)
|
|
|
|
|
|
async def create_quotation(data: QuotationCreate, generate_pdf: bool = False) -> QuotationInDB:
|
|
db = await mqtt_db.get_db()
|
|
now = datetime.utcnow().isoformat()
|
|
qid = str(uuid.uuid4())
|
|
quotation_number = await _generate_quotation_number(db)
|
|
|
|
# Build items list for calculation
|
|
items_raw = [item.model_dump() for item in data.items]
|
|
|
|
# Calculate per-item line totals
|
|
for item in items_raw:
|
|
item["line_total"] = _calc_line_total(item)
|
|
|
|
totals = _calculate_totals(
|
|
items_raw,
|
|
data.global_discount_percent,
|
|
data.shipping_cost,
|
|
data.shipping_cost_discount,
|
|
data.install_cost,
|
|
data.install_cost_discount,
|
|
data.extras_cost,
|
|
)
|
|
|
|
comments_json = json.dumps(data.comments)
|
|
quick_notes_json = json.dumps(data.quick_notes or {})
|
|
|
|
await db.execute(
|
|
"""INSERT INTO crm_quotations (
|
|
id, quotation_number, title, subtitle, customer_id,
|
|
language, status, order_type, shipping_method, estimated_shipping_date,
|
|
global_discount_label, global_discount_percent,
|
|
shipping_cost, shipping_cost_discount, install_cost, install_cost_discount,
|
|
extras_label, extras_cost, comments, quick_notes,
|
|
subtotal_before_discount, global_discount_amount, new_subtotal, vat_amount, final_total,
|
|
nextcloud_pdf_path, nextcloud_pdf_url,
|
|
client_org, client_name, client_location, client_phone, client_email,
|
|
is_legacy, legacy_date, legacy_pdf_path,
|
|
created_at, updated_at
|
|
) VALUES (
|
|
?, ?, ?, ?, ?,
|
|
?, 'draft', ?, ?, ?,
|
|
?, ?,
|
|
?, ?, ?, ?,
|
|
?, ?, ?, ?,
|
|
?, ?, ?, ?, ?,
|
|
NULL, NULL,
|
|
?, ?, ?, ?, ?,
|
|
?, ?, ?,
|
|
?, ?
|
|
)""",
|
|
(
|
|
qid, quotation_number, data.title, data.subtitle, data.customer_id,
|
|
data.language, data.order_type, data.shipping_method, data.estimated_shipping_date,
|
|
data.global_discount_label, data.global_discount_percent,
|
|
data.shipping_cost, data.shipping_cost_discount, data.install_cost, data.install_cost_discount,
|
|
data.extras_label, data.extras_cost, comments_json, quick_notes_json,
|
|
totals["subtotal_before_discount"], totals["global_discount_amount"],
|
|
totals["new_subtotal"], totals["vat_amount"], totals["final_total"],
|
|
data.client_org, data.client_name, data.client_location, data.client_phone, data.client_email,
|
|
1 if data.is_legacy else 0, data.legacy_date, data.legacy_pdf_path,
|
|
now, now,
|
|
),
|
|
)
|
|
|
|
# Insert items
|
|
for i, item in enumerate(items_raw):
|
|
item_id = str(uuid.uuid4())
|
|
await db.execute(
|
|
"""INSERT INTO crm_quotation_items
|
|
(id, quotation_id, product_id, description, description_en, description_gr,
|
|
unit_type, unit_cost, discount_percent, quantity, vat_percent, line_total, sort_order)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
item_id, qid, item.get("product_id"), item.get("description"),
|
|
item.get("description_en"), item.get("description_gr"),
|
|
item.get("unit_type", "pcs"), item.get("unit_cost", 0),
|
|
item.get("discount_percent", 0), item.get("quantity", 1),
|
|
item.get("vat_percent", 24), item["line_total"], item.get("sort_order", i),
|
|
),
|
|
)
|
|
|
|
await db.commit()
|
|
|
|
quotation = await get_quotation(qid)
|
|
|
|
if generate_pdf and not data.is_legacy:
|
|
quotation = await _do_generate_and_upload_pdf(quotation)
|
|
|
|
return quotation
|
|
|
|
|
|
async def update_quotation(quotation_id: str, data: QuotationUpdate, generate_pdf: bool = False) -> QuotationInDB:
|
|
db = await mqtt_db.get_db()
|
|
rows = await db.execute_fetchall(
|
|
"SELECT * FROM crm_quotations WHERE id = ?", (quotation_id,)
|
|
)
|
|
if not rows:
|
|
raise HTTPException(status_code=404, detail="Quotation not found")
|
|
|
|
existing = dict(rows[0])
|
|
now = datetime.utcnow().isoformat()
|
|
|
|
# Merge update into existing values
|
|
update_fields = data.model_dump(exclude_none=True)
|
|
|
|
# Build SET clause — handle comments JSON separately
|
|
set_parts = []
|
|
params = []
|
|
|
|
scalar_fields = [
|
|
"title", "subtitle", "language", "status", "order_type", "shipping_method",
|
|
"estimated_shipping_date", "global_discount_label", "global_discount_percent",
|
|
"shipping_cost", "shipping_cost_discount", "install_cost",
|
|
"install_cost_discount", "extras_label", "extras_cost",
|
|
"client_org", "client_name", "client_location", "client_phone", "client_email",
|
|
"legacy_date", "legacy_pdf_path",
|
|
]
|
|
|
|
for field in scalar_fields:
|
|
if field in update_fields:
|
|
set_parts.append(f"{field} = ?")
|
|
params.append(update_fields[field])
|
|
|
|
if "comments" in update_fields:
|
|
set_parts.append("comments = ?")
|
|
params.append(json.dumps(update_fields["comments"]))
|
|
|
|
if "quick_notes" in update_fields:
|
|
set_parts.append("quick_notes = ?")
|
|
params.append(json.dumps(update_fields["quick_notes"] or {}))
|
|
|
|
# Recalculate totals using merged values
|
|
merged = {**existing, **{k: update_fields.get(k, existing.get(k)) for k in scalar_fields}}
|
|
|
|
# If items are being updated, recalculate with new items; otherwise use existing items
|
|
if "items" in update_fields:
|
|
items_raw = [item.model_dump() for item in data.items]
|
|
for item in items_raw:
|
|
item["line_total"] = _calc_line_total(item)
|
|
else:
|
|
existing_items = await _fetch_items(db, quotation_id)
|
|
items_raw = existing_items
|
|
|
|
totals = _calculate_totals(
|
|
items_raw,
|
|
float(merged.get("global_discount_percent", 0)),
|
|
float(merged.get("shipping_cost", 0)),
|
|
float(merged.get("shipping_cost_discount", 0)),
|
|
float(merged.get("install_cost", 0)),
|
|
float(merged.get("install_cost_discount", 0)),
|
|
float(merged.get("extras_cost", 0)),
|
|
)
|
|
|
|
for field, val in totals.items():
|
|
set_parts.append(f"{field} = ?")
|
|
params.append(val)
|
|
|
|
set_parts.append("updated_at = ?")
|
|
params.append(now)
|
|
params.append(quotation_id)
|
|
|
|
if set_parts:
|
|
await db.execute(
|
|
f"UPDATE crm_quotations SET {', '.join(set_parts)} WHERE id = ?",
|
|
params,
|
|
)
|
|
|
|
# Replace items if provided
|
|
if "items" in update_fields:
|
|
await db.execute("DELETE FROM crm_quotation_items WHERE quotation_id = ?", (quotation_id,))
|
|
for i, item in enumerate(items_raw):
|
|
item_id = str(uuid.uuid4())
|
|
await db.execute(
|
|
"""INSERT INTO crm_quotation_items
|
|
(id, quotation_id, product_id, description, description_en, description_gr,
|
|
unit_type, unit_cost, discount_percent, quantity, vat_percent, line_total, sort_order)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
item_id, quotation_id, item.get("product_id"), item.get("description"),
|
|
item.get("description_en"), item.get("description_gr"),
|
|
item.get("unit_type", "pcs"), item.get("unit_cost", 0),
|
|
item.get("discount_percent", 0), item.get("quantity", 1),
|
|
item.get("vat_percent", 24), item["line_total"], item.get("sort_order", i),
|
|
),
|
|
)
|
|
|
|
await db.commit()
|
|
|
|
quotation = await get_quotation(quotation_id)
|
|
|
|
if generate_pdf:
|
|
quotation = await _do_generate_and_upload_pdf(quotation)
|
|
|
|
return quotation
|
|
|
|
|
|
async def delete_quotation(quotation_id: str) -> None:
|
|
db = await mqtt_db.get_db()
|
|
rows = await db.execute_fetchall(
|
|
"SELECT nextcloud_pdf_path FROM crm_quotations WHERE id = ?", (quotation_id,)
|
|
)
|
|
if not rows:
|
|
raise HTTPException(status_code=404, detail="Quotation not found")
|
|
|
|
pdf_path = dict(rows[0]).get("nextcloud_pdf_path")
|
|
|
|
await db.execute("DELETE FROM crm_quotation_items WHERE quotation_id = ?", (quotation_id,))
|
|
await db.execute("DELETE FROM crm_quotations WHERE id = ?", (quotation_id,))
|
|
await db.commit()
|
|
|
|
# Remove PDF from Nextcloud (best-effort)
|
|
if pdf_path:
|
|
try:
|
|
await nextcloud.delete_file(pdf_path)
|
|
except Exception as e:
|
|
logger.warning("Failed to delete PDF from Nextcloud (%s): %s", pdf_path, e)
|
|
|
|
|
|
# ── PDF Generation ─────────────────────────────────────────────────────────────
|
|
|
|
async def _do_generate_and_upload_pdf(quotation: QuotationInDB) -> QuotationInDB:
|
|
"""Generate PDF, upload to Nextcloud, update DB record. Returns updated quotation."""
|
|
try:
|
|
customer = get_customer(quotation.customer_id)
|
|
except Exception as e:
|
|
logger.error("Cannot generate PDF — customer not found: %s", e)
|
|
return quotation
|
|
|
|
try:
|
|
pdf_bytes = await _generate_pdf_bytes(quotation, customer)
|
|
except Exception as e:
|
|
logger.error("PDF generation failed for quotation %s: %s", quotation.id, e)
|
|
return quotation
|
|
|
|
# Delete old PDF if present
|
|
if quotation.nextcloud_pdf_path:
|
|
try:
|
|
await nextcloud.delete_file(quotation.nextcloud_pdf_path)
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
pdf_path, pdf_url = await _upload_pdf(customer, quotation, pdf_bytes)
|
|
except Exception as e:
|
|
logger.error("PDF upload failed for quotation %s: %s", quotation.id, e)
|
|
return quotation
|
|
|
|
# Persist paths
|
|
db = await mqtt_db.get_db()
|
|
await db.execute(
|
|
"UPDATE crm_quotations SET nextcloud_pdf_path = ?, nextcloud_pdf_url = ? WHERE id = ?",
|
|
(pdf_path, pdf_url, quotation.id),
|
|
)
|
|
await db.commit()
|
|
|
|
return await get_quotation(quotation.id)
|
|
|
|
|
|
async def _generate_pdf_bytes(quotation: QuotationInDB, customer) -> bytes:
|
|
"""Render Jinja2 template and convert to PDF via WeasyPrint."""
|
|
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
|
import weasyprint
|
|
|
|
env = Environment(
|
|
loader=FileSystemLoader(str(_TEMPLATES_DIR)),
|
|
autoescape=select_autoescape(["html"]),
|
|
)
|
|
|
|
def format_money(value):
|
|
try:
|
|
f = float(value)
|
|
# Greek-style: dot thousands separator, comma decimal
|
|
formatted = f"{f:,.2f}".replace(",", "X").replace(".", ",").replace("X", ".")
|
|
return f"{formatted} €"
|
|
except (TypeError, ValueError):
|
|
return "0,00 €"
|
|
|
|
env.filters["format_money"] = format_money
|
|
|
|
template = env.get_template("quotation.html")
|
|
|
|
html_str = template.render(
|
|
quotation=quotation,
|
|
customer=customer,
|
|
lang=quotation.language,
|
|
)
|
|
|
|
pdf = weasyprint.HTML(string=html_str, base_url=str(_TEMPLATES_DIR)).write_pdf()
|
|
return pdf
|
|
|
|
|
|
async def _upload_pdf(customer, quotation: QuotationInDB, pdf_bytes: bytes) -> tuple[str, str]:
|
|
"""Upload PDF to Nextcloud, return (relative_path, public_url)."""
|
|
from crm.service import get_customer_nc_path
|
|
from config import settings
|
|
|
|
nc_folder = get_customer_nc_path(customer)
|
|
date_str = datetime.utcnow().strftime("%Y-%m-%d")
|
|
filename = f"Quotation-{quotation.quotation_number}-{date_str}.pdf"
|
|
rel_path = f"customers/{nc_folder}/quotations/{filename}"
|
|
|
|
await nextcloud.ensure_folder(f"customers/{nc_folder}/quotations")
|
|
await nextcloud.upload_file(rel_path, pdf_bytes, "application/pdf")
|
|
|
|
# Construct a direct WebDAV download URL
|
|
from crm.nextcloud import _full_url
|
|
pdf_url = _full_url(rel_path)
|
|
|
|
return rel_path, pdf_url
|
|
|
|
|
|
async def regenerate_pdf(quotation_id: str) -> QuotationInDB:
|
|
quotation = await get_quotation(quotation_id)
|
|
return await _do_generate_and_upload_pdf(quotation)
|
|
|
|
|
|
async def get_quotation_pdf_bytes(quotation_id: str) -> bytes:
|
|
"""Download the PDF for a quotation from Nextcloud and return raw bytes."""
|
|
from fastapi import HTTPException
|
|
quotation = await get_quotation(quotation_id)
|
|
# For legacy quotations, the PDF is at legacy_pdf_path
|
|
path = quotation.legacy_pdf_path if quotation.is_legacy else quotation.nextcloud_pdf_path
|
|
if not path:
|
|
raise HTTPException(status_code=404, detail="No PDF available for this quotation")
|
|
pdf_bytes, _ = await nextcloud.download_file(path)
|
|
return pdf_bytes
|
|
|
|
|
|
async def upload_legacy_pdf(quotation_id: str, pdf_bytes: bytes, filename: str) -> QuotationInDB:
|
|
"""Upload a legacy PDF to Nextcloud and store its path in the quotation record."""
|
|
quotation = await get_quotation(quotation_id)
|
|
if not quotation.is_legacy:
|
|
raise HTTPException(status_code=400, detail="This quotation is not a legacy quotation")
|
|
|
|
from crm.service import get_customer, get_customer_nc_path
|
|
customer = get_customer(quotation.customer_id)
|
|
nc_folder = get_customer_nc_path(customer)
|
|
|
|
await nextcloud.ensure_folder(f"customers/{nc_folder}/quotations")
|
|
rel_path = f"customers/{nc_folder}/quotations/{filename}"
|
|
await nextcloud.upload_file(rel_path, pdf_bytes, "application/pdf")
|
|
|
|
db = await mqtt_db.get_db()
|
|
now = datetime.utcnow().isoformat()
|
|
await db.execute(
|
|
"UPDATE crm_quotations SET legacy_pdf_path = ?, updated_at = ? WHERE id = ?",
|
|
(rel_path, now, quotation_id),
|
|
)
|
|
await db.commit()
|
|
return await get_quotation(quotation_id)
|