+ | 2025-07-17 |
+ FR00006_250_amz_de |
+ Płyn do czyszczenia pralek automatycznych, 250 ml |
+ 1 080,00 |
+ 144 |
+ 3
diff --git a/modules/EcmInvoiceOuts/invoice_ai_analysis.py b/modules/EcmInvoiceOuts/invoice_ai_analysis.py
new file mode 100644
index 00000000..8c79c207
--- /dev/null
+++ b/modules/EcmInvoiceOuts/invoice_ai_analysis.py
@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Invoice AI Analysis — end-to-end script (MySQL -> KPIs -> Mix Change -> Anomalies -> HTML)
+See previous instructions for usage and requirements.
+"""
+
+from __future__ import annotations
+import os
+import sys
+import argparse
+from dataclasses import dataclass
+from typing import Any, List, Dict
+
+import numpy as np
+import pandas as pd
+
+from sqlalchemy import create_engine, text
+
+@dataclass
+class Config:
+ host: str
+ port: int
+ user: str
+ password: str
+ database: str
+ date_from: str
+ date_to: str
+ doc_type: str
+ output_html: str
+
+def parse_args() -> Config:
+ parser = argparse.ArgumentParser(description="Invoice AI Analysis (MySQL -> HTML)")
+ parser.add_argument("--host", default=os.getenv("DB_HOST", "localhost"))
+ parser.add_argument("--port", default=int(os.getenv("DB_PORT", "3306")), type=int)
+ parser.add_argument("--user", default=os.getenv("DB_USER", "root"))
+ parser.add_argument("--password", default=os.getenv("DB_PASS", "rootpassword"))
+ parser.add_argument("--database", default=os.getenv("DB_NAME", "twinpol-mysql56"))
+ parser.add_argument("--from", dest="date_from", default="2025-07-01")
+ parser.add_argument("--to", dest="date_to", default="2025-08-01")
+ parser.add_argument("--type", dest="doc_type", default="normal")
+ parser.add_argument("--out", dest="output_html", default="report.html")
+ args = parser.parse_args()
+ return Config(
+ host=args.host, port=args.port, user=args.user, password=args.password,
+ database=args.database, date_from=args.date_from, date_to=args.date_to,
+ doc_type=args.doc_type, output_html=args.output_html
+ )
+
+def get_engine(cfg: Config):
+ url = f"mysql+pymysql://{cfg.user}:{cfg.password}@{cfg.host}:{cfg.port}/{cfg.database}?charset=utf8mb4"
+ return create_engine(url, pool_recycle=3600, pool_pre_ping=True, future=True)
+
+def fetch_invoices(engine, cfg: Config) -> pd.DataFrame:
+ sql = text("""
+ SELECT i.document_no,
+ i.parent_name,
+ DATE(i.register_date) AS register_date,
+ ii.code,
+ ii.name,
+ ii.quantity,
+ ii.total_netto
+ FROM ecminvoiceoutitems AS ii
+ JOIN ecminvoiceouts AS i ON i.id = ii.ecminvoiceout_id
+ WHERE i.register_date >= :date_from
+ AND i.register_date < :date_to
+ AND i.type = :doc_type
+ """)
+ with engine.connect() as con:
+ df = pd.read_sql(sql, con, params={
+ "date_from": cfg.date_from,
+ "date_to": cfg.date_to,
+ "doc_type": cfg.doc_type
+ })
+ df["register_date"] = pd.to_datetime(df["register_date"], errors="coerce")
+ df["quantity"] = pd.to_numeric(df["quantity"], errors="coerce")
+ df["total_netto"] = pd.to_numeric(df["total_netto"], errors="coerce")
+ return df.dropna(subset=["register_date", "quantity", "total_netto"])
+
+def compute_kpis(df: pd.DataFrame) -> Dict[str, Any]:
+ total_revenue = float(df["total_netto"].sum())
+ total_invoices = int(df["document_no"].nunique())
+ total_units = float(df["quantity"].sum())
+ aov = float(total_revenue / total_invoices) if total_invoices else 0.0
+ top_products = (df.groupby(["code", "name"], as_index=False)
+ .agg(total_netto=("total_netto", "sum"))
+ .sort_values("total_netto", ascending=False)
+ .head(5))
+ top_customers = (df.groupby(["parent_name"], as_index=False)
+ .agg(total_netto=("total_netto", "sum"))
+ .sort_values("total_netto", ascending=False)
+ .head(5))
+ return {
+ "total_revenue": total_revenue,
+ "total_invoices": total_invoices,
+ "total_units": total_units,
+ "aov": aov,
+ "top_products": top_products,
+ "top_customers": top_customers,
+ }
+
+def render_html(cfg: Config, kpis: Dict[str, Any]) -> str:
+ def fmt_cur(x): return f"{x:,.2f}".replace(",", " ").replace(".", ",")
+ def table(headers, rows):
+ th = "".join(f" | {h} | " for h in headers)
+ trs = "".join("