from datetime import datetime, timedelta from typing import List, Optional from models import ( Customer, Supplier, Product, OrderLine, SalesOrder, PurchaseOrder, OrderStatus, Invoice, InvoiceLine, DocumentType, DocumentStatus, InventoryReason, InventoryMovement, InventoryMovementLine, MovementSign, PaymentSchedule, PaymentScheduleType, DOCUMENT_TYPE_LABELS, ) def _today() -> datetime: now = datetime.now() return datetime(now.year, now.month, now.day) def _round(v: float) -> float: return round(v + 1e-9, 2) class FakeDatabase: def __init__(self): self.customers = self._init_customers() self.suppliers = self._init_suppliers() self.products = self._init_products() self.sales_orders = self._init_sales_orders() self.purchase_orders = self._init_purchase_orders() self.inventory_reasons = self._init_inventory_reasons() self.invoices = self._init_invoices() self.inventory_movements = self._init_inventory_movements() self.payment_schedules = self._init_payment_schedules() self.next_customer_id = max((c.id for c in self.customers), default=0) + 1 self.next_supplier_id = max((s.id for s in self.suppliers), default=0) + 1 self.next_product_id = max((p.id for p in self.products), default=0) + 1 self.next_so_id = max((o.id for o in self.sales_orders), default=0) + 1 self.next_po_id = max((o.id for o in self.purchase_orders), default=0) + 1 self.next_invoice_id = max((i.id for i in self.invoices), default=0) + 1 self.next_movement_id = max((m.id for m in self.inventory_movements), default=0) + 1 self.next_schedule_id = max((s.id for s in self.payment_schedules), default=0) + 1 # ---------- Seed: Anagrafiche ---------- def _init_customers(self) -> List[Customer]: return [ Customer(id=1, code="CUST001", name="Acme Corporation S.p.A.", email="info@acme.it", phone="+39 02 1234567", address="Via Roma 10", city="Milano", country="IT", vat_number="IT01234567890", fiscal_code="CMRACM80A01F205X", payment_terms="RB 30 gg FM", credit_limit=50000, balance=15000), Customer(id=2, code="CUST002", name="TechFlow Industries GmbH", email="contact@techflow.de", phone="+49 30 555888", address="Hauptstrasse 45", city="Berlin", country="DE", vat_number="DE811234567", payment_terms="Bonifico 60 gg", credit_limit=75000, balance=32000), Customer(id=3, code="CUST003", name="European Logistics SARL", email="sales@eulog.fr", phone="+33 1 4520 3580", address="Avenue des Champs 20", city="Paris", country="FR", vat_number="FR40123456789", payment_terms="Bonifico 30 gg", credit_limit=100000, balance=8500), Customer(id=4, code="CUST004", name="Alpine Manufacturing AG", email="orders@alpmfg.ch", phone="+41 44 5678900", address="Industriestrasse 12", city="Zurich", country="CH", vat_number="CHE116281456", payment_terms="Rimessa diretta", credit_limit=60000, balance=22000), ] def _init_suppliers(self) -> List[Supplier]: return [ Supplier(id=1, code="SUPP001", name="Global Components Ltd", email="sales@globalcomp.com", phone="+44 20 7946 0958", address="123 Oxford Street", city="London", country="UK", vat_number="GB123456789", payment_terms_days=30, balance=5000), Supplier(id=2, code="SUPP002", name="Asia Electronics Co.", email="export@asiaelec.com", phone="+886 2 8797 3000", address="Taipei Industrial Park", city="Taipei", country="TW", vat_number="", payment_terms_days=45, balance=18000), Supplier(id=3, code="SUPP003", name="Nordic Materials AB", email="info@nordicmat.se", phone="+46 8 123 4567", address="Industrigatan 5", city="Stockholm", country="SE", vat_number="SE556677889901", payment_terms_days=60, balance=3500), ] def _init_products(self) -> List[Product]: return [ Product(id=1, code="PROD001", description="Industrial Widget A", category="Electronics", uom="NR", unit_price=125.50, purchase_price=85.00, tax_rate=22.0, quantity_in_stock=250, reorder_level=50, max_stock=500, supplier_id=1), Product(id=2, code="PROD002", description="Premium Connector Set", category="Hardware", uom="NR", unit_price=45.00, purchase_price=28.00, tax_rate=22.0, quantity_in_stock=180, reorder_level=30, max_stock=300, supplier_id=2), Product(id=3, code="PROD003", description="Steel Bracket Type B", category="Structural", uom="KG", unit_price=22.75, purchase_price=14.50, tax_rate=22.0, quantity_in_stock=45, reorder_level=100, max_stock=600, supplier_id=3), Product(id=4, code="PROD004", description="Precision Bearing Set", category="Mechanical", uom="NR", unit_price=89.99, purchase_price=55.00, tax_rate=22.0, quantity_in_stock=120, reorder_level=40, max_stock=250, supplier_id=1), Product(id=5, code="PROD005", description="Control Module MCU-32", category="Electronics", uom="NR", unit_price=234.50, purchase_price=160.00, tax_rate=22.0, quantity_in_stock=18, reorder_level=20, max_stock=100, supplier_id=2), ] def _init_sales_orders(self) -> List[SalesOrder]: now = datetime.now() return [ SalesOrder(id=1, order_number="SO-2024-0001", order_date=now - timedelta(days=10), customer_id=1, customer_name="Acme Corporation S.p.A.", status=OrderStatus.CONFIRMED, lines=[ OrderLine(id=1, product_id=1, product_code="PROD001", product_description="Industrial Widget A", quantity=10, unit_price=125.50, total_amount=1255.00), OrderLine(id=2, product_id=4, product_code="PROD004", product_description="Precision Bearing Set", quantity=5, unit_price=89.99, total_amount=449.95), ], total_amount=1704.95, delivery_date=now + timedelta(days=15)), SalesOrder(id=2, order_number="SO-2024-0002", order_date=now - timedelta(days=5), customer_id=2, customer_name="TechFlow Industries GmbH", status=OrderStatus.CONFIRMED, lines=[ OrderLine(id=3, product_id=5, product_code="PROD005", product_description="Control Module MCU-32", quantity=8, unit_price=234.50, total_amount=1876.00), ], total_amount=1876.00, delivery_date=now + timedelta(days=10)), SalesOrder(id=3, order_number="SO-2024-0003", order_date=now - timedelta(days=2), customer_id=3, customer_name="European Logistics SARL", status=OrderStatus.DRAFT, lines=[ OrderLine(id=4, product_id=2, product_code="PROD002", product_description="Premium Connector Set", quantity=20, unit_price=45.00, total_amount=900.00), OrderLine(id=5, product_id=3, product_code="PROD003", product_description="Steel Bracket Type B", quantity=15, unit_price=22.75, total_amount=341.25), ], total_amount=1241.25, delivery_date=now + timedelta(days=20)), ] def _init_purchase_orders(self) -> List[PurchaseOrder]: now = datetime.now() return [ PurchaseOrder(id=1, order_number="PO-2024-0001", order_date=now - timedelta(days=15), supplier_id=1, supplier_name="Global Components Ltd", status=OrderStatus.PARTIALLY_RECEIVED, lines=[OrderLine(id=1, product_id=1, product_code="PROD001", product_description="Industrial Widget A", quantity=100, unit_price=85.00, total_amount=8500.00)], total_amount=8500.00, expected_delivery_date=now + timedelta(days=5)), PurchaseOrder(id=2, order_number="PO-2024-0002", order_date=now - timedelta(days=8), supplier_id=2, supplier_name="Asia Electronics Co.", status=OrderStatus.CONFIRMED, lines=[OrderLine(id=2, product_id=5, product_code="PROD005", product_description="Control Module MCU-32", quantity=50, unit_price=160.00, total_amount=8000.00)], total_amount=8000.00, expected_delivery_date=now + timedelta(days=25)), PurchaseOrder(id=3, order_number="PO-2024-0003", order_date=now - timedelta(days=1), supplier_id=3, supplier_name="Nordic Materials AB", status=OrderStatus.DRAFT, lines=[OrderLine(id=3, product_id=3, product_code="PROD003", product_description="Steel Bracket Type B", quantity=200, unit_price=14.50, total_amount=2900.00)], total_amount=2900.00, expected_delivery_date=now + timedelta(days=30)), ] # ---------- Seed: Magazzino ---------- def _init_inventory_reasons(self) -> List[InventoryReason]: return [ InventoryReason(code="CAR-ACQ", description="Carico da acquisto", sign=MovementSign.LOAD, fiscal=True), InventoryReason(code="CAR-RET", description="Reso da cliente", sign=MovementSign.LOAD, fiscal=True), InventoryReason(code="CAR-RET+", description="Rettifica inventariale positiva", sign=MovementSign.LOAD, fiscal=False), InventoryReason(code="SCA-VEN", description="Scarico per vendita", sign=MovementSign.UNLOAD, fiscal=True), InventoryReason(code="SCA-PRO", description="Scarico per produzione", sign=MovementSign.UNLOAD, fiscal=False), InventoryReason(code="SCA-RET-", description="Rettifica inventariale negativa", sign=MovementSign.UNLOAD, fiscal=False), ] def _init_inventory_movements(self) -> List[InventoryMovement]: now = datetime.now() return [ InventoryMovement( id=1, reason_code="CAR-ACQ", reason_description="Carico da acquisto", sign=MovementSign.LOAD, doc_no="MOV-0001", doc_date=now - timedelta(days=12), posting_date=now - timedelta(days=12), storage="MAG01", custsupp_name="Global Components Ltd", lines=[InventoryMovementLine(line_no=1, item_code="PROD001", description="Industrial Widget A", uom="NR", quantity=100, unit_value=85.00, line_amount=8500.00, location="A-01-03")], total_amount=8500.00, posted=True, notes="Carico merce da OdA PO-2024-0001"), InventoryMovement( id=2, reason_code="SCA-VEN", reason_description="Scarico per vendita", sign=MovementSign.UNLOAD, doc_no="MOV-0002", doc_date=now - timedelta(days=6), posting_date=now - timedelta(days=6), storage="MAG01", custsupp_name="Acme Corporation S.p.A.", lines=[InventoryMovementLine(line_no=1, item_code="PROD001", description="Industrial Widget A", uom="NR", quantity=10, unit_value=125.50, line_amount=1255.00, location="A-01-03")], total_amount=1255.00, posted=True, notes="Scarico per fattura FT-2024-0001"), InventoryMovement( id=3, reason_code="CAR-RET+", reason_description="Rettifica inventariale positiva", sign=MovementSign.LOAD, doc_no="MOV-0003", doc_date=now - timedelta(days=1), posting_date=now - timedelta(days=1), storage="MAG01", custsupp_name="", lines=[InventoryMovementLine(line_no=1, item_code="PROD003", description="Steel Bracket Type B", uom="KG", quantity=5, unit_value=14.50, line_amount=72.50, location="B-02-01")], total_amount=72.50, posted=False, notes="Da confermare"), ] # ---------- Seed: Fatture ---------- def _build_invoice_lines(self, raw_lines) -> List[InvoiceLine]: lines = [] for i, (code, desc, uom, qty, val, disc, rate) in enumerate(raw_lines, start=1): net_price = _round(val * (1 - disc / 100)) taxable = _round(net_price * qty) tax = _round(taxable * rate / 100) lines.append(InvoiceLine(line_no=i, item_code=code, description=desc, uom=uom, quantity=qty, unit_value=val, discount_pct=disc, net_price=net_price, taxable_amount=taxable, tax_rate=rate, tax_amount=tax)) return lines def _finalize_invoice(self, inv: Invoice) -> Invoice: inv.taxable_total = _round(sum(l.taxable_amount for l in inv.lines)) inv.tax_total = _round(sum(l.tax_amount for l in inv.lines)) inv.total = _round(inv.taxable_total + inv.tax_total) return inv def _init_invoices(self) -> List[Invoice]: now = datetime.now() invoices = [ Invoice(id=1, doc_type=DocumentType.IMMEDIATE_INVOICE, doc_no="FT-2024-0001", doc_date=now - timedelta(days=6), posting_date=now - timedelta(days=6), customer_id=1, customer_name="Acme Corporation S.p.A.", customer_vat="IT01234567890", your_reference="ODA 4521", payment_terms="RB 30 gg FM", salesperson="Rossi M.", lines=self._build_invoice_lines([ ("PROD001", "Industrial Widget A", "NR", 10, 125.50, 0, 22.0), ("PROD004", "Precision Bearing Set", "NR", 5, 89.99, 5, 22.0), ]), status=DocumentStatus.ISSUED, printed=True, issued=True, posted_to_inventory=True), Invoice(id=2, doc_type=DocumentType.DELIVERY_NOTE, doc_no="DDT-2024-0014", doc_date=now - timedelta(days=3), posting_date=now - timedelta(days=3), customer_id=2, customer_name="TechFlow Industries GmbH", customer_vat="DE811234567", payment_terms="Bonifico 60 gg", salesperson="Bianchi L.", lines=self._build_invoice_lines([ ("PROD005", "Control Module MCU-32", "NR", 8, 234.50, 0, 22.0), ]), status=DocumentStatus.PRINTED, printed=True), Invoice(id=3, doc_type=DocumentType.IMMEDIATE_INVOICE, doc_no="FT-2024-0002", doc_date=now - timedelta(days=1), posting_date=now - timedelta(days=1), customer_id=3, customer_name="European Logistics SARL", customer_vat="FR40123456789", payment_terms="Bonifico 30 gg", salesperson="Rossi M.", lines=self._build_invoice_lines([ ("PROD002", "Premium Connector Set", "NR", 20, 45.00, 10, 22.0), ("PROD003", "Steel Bracket Type B", "KG", 15, 22.75, 0, 22.0), ]), status=DocumentStatus.DRAFT), Invoice(id=4, doc_type=DocumentType.CREDIT_NOTE, doc_no="NC-2024-0003", doc_date=now - timedelta(days=2), posting_date=now - timedelta(days=2), customer_id=1, customer_name="Acme Corporation S.p.A.", customer_vat="IT01234567890", payment_terms="RB 30 gg FM", salesperson="Rossi M.", notes="Reso merce difettosa", lines=self._build_invoice_lines([ ("PROD004", "Precision Bearing Set", "NR", 2, 89.99, 0, 22.0), ]), status=DocumentStatus.ISSUED, printed=True, issued=True), ] return [self._finalize_invoice(i) for i in invoices] # ---------- Seed: Scadenzario (partite aperte) ---------- def _init_payment_schedules(self) -> List[PaymentSchedule]: t = _today() R, P = PaymentScheduleType.RECEIVABLE, PaymentScheduleType.PAYABLE return [ # --- Incassi (receivables) da fatture clienti --- PaymentSchedule(id=1, type=R, custsupp_name="Acme Corporation S.p.A.", doc_no="FT-2024-0001", doc_date=t - timedelta(days=36), due_date=t - timedelta(days=6), payment_method="RB", amount=2052.59, bank="Intesa Sanpaolo", closed=False, notes="Scaduta — da sollecitare"), PaymentSchedule(id=2, type=R, custsupp_name="European Logistics SARL", doc_no="FT-2024-0002", doc_date=t - timedelta(days=1), due_date=t + timedelta(days=29), payment_method="Bonifico", amount=1404.53, bank="Intesa Sanpaolo", closed=False), PaymentSchedule(id=3, type=R, custsupp_name="TechFlow Industries GmbH", doc_no="FT-2024-0009", doc_date=t - timedelta(days=50), due_date=t + timedelta(days=4), payment_method="RID", amount=3680.00, bank="Unicredit", closed=False), PaymentSchedule(id=4, type=R, custsupp_name="Acme Corporation S.p.A.", doc_no="NC-2024-0003", doc_date=t - timedelta(days=2), due_date=t + timedelta(days=10), payment_method="RB", amount=219.58, is_credit_note=True, closed=False, notes="Nota di credito — riduce il credito"), PaymentSchedule(id=5, type=R, custsupp_name="Alpine Manufacturing AG", doc_no="FT-2024-0007", doc_date=t - timedelta(days=40), due_date=t - timedelta(days=10), payment_method="Bonifico", amount=5430.00, bank="Unicredit", closed=False, notes="Scaduta da 10 giorni"), PaymentSchedule(id=6, type=R, custsupp_name="European Logistics SARL", doc_no="FT-2024-0005", doc_date=t - timedelta(days=60), due_date=t - timedelta(days=30), payment_method="RB", amount=2100.00, bank="Intesa Sanpaolo", closed=True, notes="Incassata"), # --- Pagamenti (payables) verso fornitori --- PaymentSchedule(id=7, type=P, custsupp_name="Global Components Ltd", doc_no="FTACQ-1187", doc_date=t - timedelta(days=33), due_date=t - timedelta(days=3), payment_method="Bonifico", amount=8500.00, closed=False, notes="Scaduta — pagamento urgente"), PaymentSchedule(id=8, type=P, custsupp_name="Asia Electronics Co.", doc_no="FTACQ-2204", doc_date=t - timedelta(days=12), due_date=t + timedelta(days=2), payment_method="Bonifico", amount=8000.00, closed=False), PaymentSchedule(id=9, type=P, custsupp_name="Nordic Materials AB", doc_no="FTACQ-0991", doc_date=t - timedelta(days=5), due_date=t + timedelta(days=25), payment_method="Bonifico", amount=2900.00, closed=False), PaymentSchedule(id=10, type=P, custsupp_name="Global Components Ltd", doc_no="FTACQ-1150", doc_date=t - timedelta(days=70), due_date=t - timedelta(days=40), payment_method="RIBA", amount=3200.00, closed=True, notes="Pagata"), ] # ==================== CRUD: Anagrafiche ==================== def get_customers(self): return self.customers def get_customer(self, cid): return next((c for c in self.customers if c.id == cid), None) def create_customer(self, customer: Customer) -> Customer: customer.id = self.next_customer_id self.next_customer_id += 1 self.customers.append(customer) return customer def update_customer(self, cid, data: dict): c = self.get_customer(cid) if c: for k, v in data.items(): if hasattr(c, k) and v is not None: setattr(c, k, v) return c def get_suppliers(self): return self.suppliers def get_supplier(self, sid): return next((s for s in self.suppliers if s.id == sid), None) def create_supplier(self, supplier: Supplier) -> Supplier: supplier.id = self.next_supplier_id self.next_supplier_id += 1 self.suppliers.append(supplier) return supplier def get_products(self): return self.products def get_product(self, pid): return next((p for p in self.products if p.id == pid), None) def get_product_by_code(self, code): return next((p for p in self.products if p.code == code), None) def create_product(self, product: Product) -> Product: product.id = self.next_product_id self.next_product_id += 1 self.products.append(product) return product # ==================== CRUD: Ordini ==================== def get_sales_orders(self): return self.sales_orders def get_sales_order(self, oid): return next((o for o in self.sales_orders if o.id == oid), None) def create_sales_order(self, order: SalesOrder) -> SalesOrder: order.id = self.next_so_id self.next_so_id += 1 self.sales_orders.append(order) return order def get_purchase_orders(self): return self.purchase_orders def get_purchase_order(self, oid): return next((o for o in self.purchase_orders if o.id == oid), None) def create_purchase_order(self, order: PurchaseOrder) -> PurchaseOrder: order.id = self.next_po_id self.next_po_id += 1 self.purchase_orders.append(order) return order # ==================== CRUD: Fatture ==================== def get_invoices(self): return self.invoices def get_invoice(self, iid): return next((i for i in self.invoices if i.id == iid), None) def get_invoices_by_type(self, doc_type: DocumentType): return [i for i in self.invoices if i.doc_type == doc_type] def create_invoice(self, invoice: Invoice) -> Invoice: invoice.id = self.next_invoice_id self.next_invoice_id += 1 self._finalize_invoice(invoice) self.invoices.append(invoice) return invoice # ==================== CRUD: Magazzino ==================== def get_inventory_reasons(self): return self.inventory_reasons def get_inventory_reason(self, code): return next((r for r in self.inventory_reasons if r.code == code), None) def get_inventory_movements(self): return self.inventory_movements def get_inventory_movement(self, mid): return next((m for m in self.inventory_movements if m.id == mid), None) def create_inventory_movement(self, movement: InventoryMovement) -> InventoryMovement: movement.id = self.next_movement_id self.next_movement_id += 1 movement.total_amount = _round(sum(l.line_amount for l in movement.lines)) self.inventory_movements.append(movement) if movement.posted: self._apply_movement_to_balances(movement) return movement def post_movement(self, mid) -> Optional[InventoryMovement]: """Conferma un movimento e aggiorna le giacenze (Mago4: PostedToInventory).""" m = self.get_inventory_movement(mid) if m and not m.posted: m.posted = True self._apply_movement_to_balances(m) return m def _apply_movement_to_balances(self, movement: InventoryMovement): """Applica il movimento alle giacenze articoli secondo il segno della causale.""" delta_sign = 1 if movement.sign == MovementSign.LOAD else -1 for line in movement.lines: product = self.get_product_by_code(line.item_code) if product: product.quantity_in_stock += int(delta_sign * line.quantity) # ==================== Scadenzario (AP_AR) ==================== def get_payment_schedules(self): return self.payment_schedules def get_payment_schedule(self, sid): return next((s for s in self.payment_schedules if s.id == sid), None) def schedule_status(self, s: PaymentSchedule) -> str: """Stato derivato: paid / overdue / due_soon / open""" if s.closed: return "paid" days = (s.due_date - _today()).days if days < 0: return "overdue" if days <= 7: return "due_soon" return "open" def schedule_days_to_due(self, s: PaymentSchedule) -> int: return (s.due_date - _today()).days def settle_schedule(self, sid) -> Optional[PaymentSchedule]: """Salda/incassa una partita (Mago4: Closed/Collected).""" s = self.get_payment_schedule(sid) if s and not s.closed: s.closed = True return s def get_open_schedules(self, stype: Optional[PaymentScheduleType] = None): items = [s for s in self.payment_schedules if not s.closed] if stype: items = [s for s in items if s.type == stype] return items def get_overdue_schedules(self, stype: Optional[PaymentScheduleType] = None): return [s for s in self.get_open_schedules(stype) if self.schedule_status(s) == "overdue"] def get_upcoming_schedules(self, days: int = 30): """Partite aperte in scadenza nei prossimi 'days' giorni (incluse scadute).""" limit = _today() + timedelta(days=days) items = [s for s in self.get_open_schedules() if s.due_date <= limit] return sorted(items, key=lambda s: s.due_date) def get_credit_exceeded_customers(self): """Clienti che hanno superato il limite di fido (modulo CreditLimit).""" return [c for c in self.customers if c.credit_limit > 0 and c.balance > c.credit_limit] def get_todo_items(self): """Aggrega le 'cose da fare' per la dashboard.""" draft_invoices = [i for i in self.invoices if i.status == DocumentStatus.DRAFT] draft_movements = [m for m in self.inventory_movements if not m.posted] orders_to_fulfill = [o for o in self.sales_orders if o.status in (OrderStatus.CONFIRMED, OrderStatus.DRAFT)] low_stock = [p for p in self.products if p.quantity_in_stock <= p.reorder_level] overdue_recv = self.get_overdue_schedules(PaymentScheduleType.RECEIVABLE) overdue_pay = self.get_overdue_schedules(PaymentScheduleType.PAYABLE) credit_exceeded = self.get_credit_exceeded_customers() todos = [] if overdue_pay: todos.append({"icon": "payments", "color": "red", "priority": 1, "text": f"{len(overdue_pay)} pagamenti scaduti da effettuare", "amount": sum(s.amount for s in overdue_pay), "link": "/scadenzario?type=payable", "action": "Paga"}) if overdue_recv: todos.append({"icon": "request_quote", "color": "orange", "priority": 2, "text": f"{len(overdue_recv)} incassi scaduti da sollecitare", "amount": sum(s.amount for s in overdue_recv), "link": "/scadenzario?type=receivable", "action": "Sollecita"}) if low_stock: todos.append({"icon": "production_quantity_limits", "color": "red", "priority": 3, "text": f"{len(low_stock)} articoli sotto scorta da ordinare", "amount": None, "link": "/warehouse", "action": "Ordina"}) if draft_invoices: todos.append({"icon": "edit_document", "color": "blue", "priority": 4, "text": f"{len(draft_invoices)} documenti in bozza da confermare", "amount": None, "link": "/invoices", "action": "Apri"}) if orders_to_fulfill: todos.append({"icon": "local_shipping", "color": "blue", "priority": 5, "text": f"{len(orders_to_fulfill)} ordini cliente da evadere", "amount": None, "link": "/sales-orders", "action": "Apri"}) if draft_movements: todos.append({"icon": "swap_vert", "color": "teal", "priority": 6, "text": f"{len(draft_movements)} movimenti di magazzino da confermare", "amount": None, "link": "/inventory-movements", "action": "Apri"}) if credit_exceeded: todos.append({"icon": "credit_card_off", "color": "red", "priority": 7, "text": f"{len(credit_exceeded)} clienti con fido superato", "amount": None, "link": "/customers", "action": "Verifica"}) return sorted(todos, key=lambda x: x["priority"]) def get_cashflow_forecast(self, days: int = 30): """Previsione di cassa a 'days' giorni: incassi vs pagamenti previsti.""" upcoming = self.get_upcoming_schedules(days) inflow = sum(s.signed_amount for s in upcoming if s.type == PaymentScheduleType.RECEIVABLE) outflow = sum(s.amount for s in upcoming if s.type == PaymentScheduleType.PAYABLE) return {"days": days, "inflow": _round(inflow), "outflow": _round(outflow), "net": _round(inflow - outflow)} # ==================== Statistiche ==================== def get_dashboard_stats(self): sales_to_fulfill = sum(1 for o in self.sales_orders if o.status in (OrderStatus.CONFIRMED, OrderStatus.DRAFT)) purchase_to_receive = sum(1 for o in self.purchase_orders if o.status in (OrderStatus.CONFIRMED, OrderStatus.DRAFT, OrderStatus.PARTIALLY_RECEIVED)) total_receivable = sum(c.balance for c in self.customers) total_payable = sum(s.balance for s in self.suppliers) low_stock_items = [p for p in self.products if p.quantity_in_stock <= p.reorder_level] # Fatturato emesso (fatture/note credito issued) invoiced_total = sum( (i.total if i.doc_type != DocumentType.CREDIT_NOTE else -i.total) for i in self.invoices if i.issued ) pending_invoices = sum(1 for i in self.invoices if i.status == DocumentStatus.DRAFT) warehouse_value = sum(p.quantity_in_stock * p.purchase_price for p in self.products) # Scadenzario overdue_recv = self.get_overdue_schedules(PaymentScheduleType.RECEIVABLE) overdue_pay = self.get_overdue_schedules(PaymentScheduleType.PAYABLE) open_recv = self.get_open_schedules(PaymentScheduleType.RECEIVABLE) open_pay = self.get_open_schedules(PaymentScheduleType.PAYABLE) return { "sales_to_fulfill": sales_to_fulfill, "purchase_to_receive": purchase_to_receive, "total_receivable": total_receivable, "total_payable": total_payable, "low_stock_count": len(low_stock_items), "low_stock_items": low_stock_items, "total_customers": len(self.customers), "total_suppliers": len(self.suppliers), "total_products": len(self.products), "invoiced_total": invoiced_total, "pending_invoices": pending_invoices, "warehouse_value": warehouse_value, "total_invoices": len(self.invoices), "total_movements": len(self.inventory_movements), # Scadenze "overdue_recv_count": len(overdue_recv), "overdue_recv_amount": _round(sum(s.amount for s in overdue_recv)), "overdue_pay_count": len(overdue_pay), "overdue_pay_amount": _round(sum(s.amount for s in overdue_pay)), "open_recv_amount": _round(sum(s.signed_amount for s in open_recv)), "open_pay_amount": _round(sum(s.amount for s in open_pay)), "credit_exceeded_count": len(self.get_credit_exceeded_customers()), } # Global database instance db = FakeDatabase()