fix(msp): Hornet-Run-Vergleich gegen Legacy bit-genau + Run-Backlinks/UI-Totals

Validierung der Hornet-Pipeline gegen drei historische Monate
(10.2025 / 11.2025 / 02.2026, je 16 Sales Invoices in der Test-DB).
Pro Customer und in Summe identisch zu den vom Legacy-adnconnect-
Modul erzeugten Belegen (z. B. Südsee-Camp 634,50 €, Saxlund 234 €,
KFS Fensterbau 162 €, Total 1.836,36 €).

Notwendige Korrekturen unterwegs:

- Hornet-Parser: ENDKUNDE_REFERENCE ist in den realen ADN-Exporten
  praktisch immer leer; die Mail-Domain steht in ENDKUNDE. Legacy-
  Verhalten (adnconnect.adn_hornet_import) gespiegelt: bei leerem
  Reference-Feld auf ENDKUNDE als Domain zurückfallen, sofern es wie
  eine Domain aussieht.

- Sniff inhaltsbasiert: ADN exportiert für Microsoft und Hornet
  dieselbe Spaltenstruktur (inkl. BILLINGPLAN/MSERP/VERTRAGSDAUER) —
  rein kolumnenbasierte Discrimination versagt. Beide Handler
  inspizieren jetzt die ersten Datenzeilen: Hornet gewinnt, wenn die
  MS-Spalten leer sind, MS gewinnt, wenn sie Werte tragen. Mit echten
  CSVs verifiziert (Hornet 1.00/MS 0.55 vs. MS 1.00/Hornet 0.33).

- Title-Template kennt jetzt {invoice_month} (CSV-DATUM, MM.YYYY) —
  ADN stellt für Hornet stets im Folgemonat in Rechnung; Legacy nutzt
  den Rechnungsmonat im Title, nicht die Wartungsperiode. Hornet-
  Profil-Default angepasst.

- DocumentBuilder annotiert erfolgreiche Outcomes mit (target_doctype,
  target_name, resolved_customer). Der Run-Orchestrator-Persist
  verlinkt Supplier Import Lines damit auch dann zurück, wenn keine
  Supply-Subscription-Events erzeugt wurden (Hornet ohne Vertrag-
  Spalte). Behebt 'customer=null'-Anzeige im Wizard-Step-3 und die
  leere Dokumenten-Tabelle.

- run_summary / run_documents zeigen jetzt das Verkaufs-Total der
  erzeugten Sales-Invoice/Delivery-Note (vorher: ADN-Einkaufspreis aus
  Supplier Import Line.amount, was bei Hornet ≠ Verkaufspreis ist).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

msp/fixtures/supplier_import_profile.json

@@ -50,8 +50,8 @@
   "supplier": "SUPP-00887",
   "title_prefix": "Abrechnung Hornetsecurity ",
   "title_prefix_credit_note": "Gutschrift Hornetsecurity ",
-  "title_template": "{prefix} {period}",
-  "title_template_credit_note": "{prefix} {period}",
+  "title_template": "{prefix} {invoice_month}",
+  "title_template_credit_note": "{prefix} {invoice_month}",
   "vendor": "Hornetsecurity"
  }
 ]

msp/importers/adn_hornet_csv.py

@@ -117,7 +117,15 @@ class ADNHornetCSVParser(BaseSupplierParser):
 		return out
 
 	def _to_canonical(self, d: dict[str, str], source_row: int) -> CanonicalRow | None:
+		# ADN-Hornet-CSVs lassen ENDKUNDE_REFERENCE in der Praxis fast immer
+		# leer und tragen die Mail-Domain stattdessen direkt in ENDKUNDE ein.
+		# Legacy-Verhalten (adnconnect.adn_hornet_import.get_invoice_dict_from_csv)
+		# fällt in dem Fall auf ENDKUNDE als Domain zurück.
 		raw_ref = (d.get("ENDKUNDE_REFERENCE") or "").strip() or None
+		endkunde = (d.get("ENDKUNDE") or "").strip() or None
+		if not raw_ref and endkunde and "." in endkunde and " " not in endkunde:
+			# Sieht nach einer Domain aus → als Hornet-Lookup-Schlüssel nutzen.
+			raw_ref = endkunde
 
 		raw_qty = parse_german_decimal(d.get("MENGE")) or 0.0
 		list_price = parse_german_decimal(d.get("LISTPREIS"))

msp/importers/assistant.py

@@ -103,11 +103,21 @@ def run_documents(run_name: str) -> list[dict]:
 	``frappe.client.get_list`` auf Child-Doctypes die `fields` ignoriert."""
 	if not run_name:
 		return []
+	# Beträge aus dem erzeugten Ziel-Doc nehmen (Sales Invoice / Delivery Note),
+	# nicht aus l.amount — das ist der ADN-Einkaufspreis. Der Endkunde sieht
+	# den Verkaufspreis aus dem SI/DN.
 	rows = frappe.db.sql(
 		"""
 		SELECT l.customer, c.customer_name,
 		       l.target_doc_type, l.target_doc_name,
-		       COUNT(*) AS line_count, SUM(COALESCE(l.amount, 0)) AS total_amount
+		       COUNT(*) AS line_count,
+		       CASE
+		         WHEN l.target_doc_type = 'Sales Invoice'
+		           THEN COALESCE((SELECT total FROM `tabSales Invoice` WHERE name = l.target_doc_name), 0)
+		         WHEN l.target_doc_type = 'Delivery Note'
+		           THEN COALESCE((SELECT total FROM `tabDelivery Note` WHERE name = l.target_doc_name), 0)
+		         ELSE 0
+		       END AS total_amount
 		FROM `tabSupplier Import Line` l
 		LEFT JOIN `tabCustomer` c ON c.name = l.customer
 		WHERE l.parent = %(run)s
@@ -290,28 +300,67 @@ def run_summary(run_name: str, top_n: int = 5) -> dict:
 		return {}
 	totals = frappe.db.sql(
 		"""
-		SELECT COUNT(DISTINCT customer) AS customers,
-		       COUNT(DISTINCT target_doc_name) AS documents,
-		       COALESCE(SUM(CASE WHEN line_status = 'Created' THEN amount END), 0) AS total_created,
-		       SUM(CASE WHEN line_status = 'Created' THEN 1 ELSE 0 END) AS lines_created,
-		       SUM(CASE WHEN line_status = 'Skipped' THEN 1 ELSE 0 END) AS lines_skipped,
-		       SUM(CASE WHEN line_status = 'Error' THEN 1 ELSE 0 END) AS lines_error
-		FROM `tabSupplier Import Line`
-		WHERE parent = %(run)s
+		SELECT COUNT(DISTINCT l.customer) AS customers,
+		       COUNT(DISTINCT l.target_doc_name) AS documents,
+		       SUM(CASE WHEN l.line_status = 'Created' THEN 1 ELSE 0 END) AS lines_created,
+		       SUM(CASE WHEN l.line_status = 'Skipped' THEN 1 ELSE 0 END) AS lines_skipped,
+		       SUM(CASE WHEN l.line_status = 'Error' THEN 1 ELSE 0 END) AS lines_error
+		FROM `tabSupplier Import Line` l
+		WHERE l.parent = %(run)s
 		""",
 		{"run": run_name},
 		as_dict=True,
 	)
+	# Verkaufs-Total über alle erzeugten Ziel-Docs (DISTINCT, damit Multi-Line-
+	# SIs nicht vervielfacht werden).
+	totals_sum = frappe.db.sql(
+		"""
+		SELECT COALESCE(SUM(t.doc_total), 0) AS total_created
+		FROM (
+		  SELECT DISTINCT l.target_doc_type, l.target_doc_name,
+		    CASE
+		      WHEN l.target_doc_type = 'Sales Invoice'
+		        THEN COALESCE((SELECT total FROM `tabSales Invoice` si WHERE si.name = l.target_doc_name), 0)
+		      WHEN l.target_doc_type = 'Delivery Note'
+		        THEN COALESCE((SELECT total FROM `tabDelivery Note` dn WHERE dn.name = l.target_doc_name), 0)
+		      ELSE 0
+		    END AS doc_total
+		  FROM `tabSupplier Import Line` l
+		  WHERE l.parent = %(run)s AND l.line_status = 'Created'
+		    AND l.target_doc_name IS NOT NULL AND l.target_doc_name != ''
+		) t
+		""",
+		{"run": run_name},
+		as_dict=True,
+	)
+	if totals and totals_sum:
+		totals[0]["total_created"] = float(totals_sum[0]["total_created"] or 0)
+
+	# Top-Kunden: erst pro (customer, target_doc) das Doc-Total holen,
+	# dann pro Customer summieren.
 	top = frappe.db.sql(
 		"""
-		SELECT l.customer, c.customer_name,
-		       COUNT(DISTINCT l.target_doc_name) AS documents,
-		       COUNT(*) AS line_count,
-		       COALESCE(SUM(l.amount), 0) AS total_amount
-		FROM `tabSupplier Import Line` l
-		LEFT JOIN `tabCustomer` c ON c.name = l.customer
-		WHERE l.parent = %(run)s AND l.line_status = 'Created'
-		GROUP BY l.customer, c.customer_name
+		SELECT customer, MAX(customer_name) AS customer_name,
+		       COUNT(DISTINCT target_doc_name) AS documents,
+		       SUM(line_count) AS line_count,
+		       SUM(doc_total) AS total_amount
+		FROM (
+		  SELECT l.customer, c.customer_name, l.target_doc_type, l.target_doc_name,
+		    COUNT(*) AS line_count,
+		    CASE
+		      WHEN l.target_doc_type = 'Sales Invoice'
+		        THEN COALESCE((SELECT total FROM `tabSales Invoice` si WHERE si.name = l.target_doc_name), 0)
+		      WHEN l.target_doc_type = 'Delivery Note'
+		        THEN COALESCE((SELECT total FROM `tabDelivery Note` dn WHERE dn.name = l.target_doc_name), 0)
+		      ELSE 0
+		    END AS doc_total
+		  FROM `tabSupplier Import Line` l
+		  LEFT JOIN `tabCustomer` c ON c.name = l.customer
+		  WHERE l.parent = %(run)s AND l.line_status = 'Created'
+		    AND l.target_doc_name IS NOT NULL AND l.target_doc_name != ''
+		  GROUP BY l.customer, c.customer_name, l.target_doc_type, l.target_doc_name
+		) t
+		GROUP BY customer
 		ORDER BY total_amount DESC
 		LIMIT %(limit)s
 		""",

msp/importers/builder.py

@@ -106,6 +106,16 @@ class DocumentBuilder:
 
 		result.created_documents.append((doc.doctype, doc.name))
 
+		# Erfolgreiche Outcomes mit dem Ziel-Dokument annotieren — damit der
+		# Persist-Schritt im Run-Orchestrator die Zeilen rückverlinken kann,
+		# auch wenn keine Subscription-Events erzeugt wurden (z. B. Hornet
+		# ohne Vertrag-Spalte).
+		for o in outcomes_this_group:
+			if o.get("status") == "Created":
+				o.setdefault("target_doctype", doc.doctype)
+				o.setdefault("target_name", doc.name)
+				o.setdefault("resolved_customer", customer)
+
 		# Subscription-Events erst nach erfolgreichem Insert, damit wir die
 		# echten Item-Row-IDs nutzen können.
 		self._post_create_subscriptions(doc, rows, result)
@@ -327,10 +337,14 @@ class DocumentBuilder:
 			return f"{prefix.strip()} {period}".strip()
 
 		customer_name = frappe.db.get_value("Customer", customer, "customer_name") or customer
+		invoice_month = (
+			posting_date.strftime("%m.%Y") if hasattr(posting_date, "strftime") else ""
+		)
 		try:
 			rendered = template.format(
 				prefix=prefix.strip(),
 				period=period,
+				invoice_month=invoice_month,
 				customer=customer,
 				customer_name=customer_name,
 				vendor=profile.vendor or "",

msp/importers/handlers/adn_hornet_csv_handler.py

@@ -42,21 +42,18 @@ class ADNHornetCSVHandler(BaseFileHandler):
 		"ENDKUNDE_REFERENCE", "VERTRAG",
 		"WARTUNGSBEGINN", "WARTUNGSENDE",
 	}
-	# Spalten, die exklusiv im Microsoft-Format vorkommen — sind sie da,
-	# spricht das gegen Hornet.
-	_NEGATIVE_COLUMNS = {"BILLINGPLAN", "VERTRAGSDAUER", "MSERP", "ADDITIONALID"}
 
 	@classmethod
 	def sniff(cls, sample_bytes: bytes, filename: str) -> float:
 		name = (filename or "").lower()
 
 		if name.endswith(".zip"):
-			first_line = cls._peek_first_line_in_zip(sample_bytes)
-			if first_line is None:
+			lines = cls._peek_lines_in_zip(sample_bytes)
+			if not lines:
 				if "hornet" in name:
 					return 0.65
 				return 0.0
-			score = cls._score_first_line(first_line)
+			score = cls._score_lines(lines)
 			if "hornet" in name:
 				score = min(1.0, score + 0.1)
 			return score
@@ -69,46 +66,71 @@ class ADNHornetCSVHandler(BaseFileHandler):
 		except Exception:
 			return 0.0
 
-		first_line = head.split("\n", 1)[0]
-		score = cls._score_first_line(first_line)
+		lines = head.split("\n", 5)[:5]
+		score = cls._score_lines(lines)
 		if "hornet" in name:
 			score = min(1.0, score + 0.1)
 		return score
 
 	@classmethod
-	def _score_first_line(cls, first_line: str) -> float:
-		first_line = (first_line or "").lstrip("\ufeff").strip()
-		if not first_line:
+	def _score_lines(cls, lines: list[str]) -> float:
+		if not lines:
 			return 0.0
-		cols = {c.strip().upper() for c in first_line.split(";")}
-		matching = cls._SIGNATURE_COLUMNS & cols
+		header_line = (lines[0] or "").lstrip("\ufeff").strip()
+		if not header_line:
+			return 0.0
+		header_cols = [c.strip() for c in header_line.split(";")]
+		header_upper = {c.upper() for c in header_cols if c}
+		matching = cls._SIGNATURE_COLUMNS & header_upper
 		if not matching:
 			return 0.0
 		ratio = len(matching) / len(cls._SIGNATURE_COLUMNS)
 		base = 0.4 + 0.55 * ratio  # max ≈ 0.95
-		# Hard penalty wenn MS-Discriminatoren da sind: dann ist's Microsoft.
-		if cls._NEGATIVE_COLUMNS & cols:
-			base = base * 0.4
+
+		# Inhaltsbasierter Discriminator: wenn die MS-spezifischen Spalten
+		# (BILLINGPLAN/MSERP/VERTRAGSDAUER) zwar im Header stehen, aber in den
+		# ersten Datenzeilen leer bleiben → Hornet-Variante.
+		ms_payload = cls._has_ms_payload(header_cols, lines[1:])
+		if ms_payload:
+			# MS-Werte vorhanden — wir sind hier nicht zuständig.
+			base = base * 0.35
+		else:
+			base = min(1.0, base + 0.1)
 		return min(1.0, base)
 
+	@staticmethod
+	def _has_ms_payload(header_cols: list[str], data_lines: list[str]) -> bool:
+		header_upper = [c.strip().upper() for c in header_cols]
+		ms_indices = [i for i, c in enumerate(header_upper)
+		              if c in {"BILLINGPLAN", "MSERP", "VERTRAGSDAUER"}]
+		if not ms_indices:
+			return False
+		for line in data_lines:
+			if not line or not line.strip():
+				continue
+			cells = [c.strip().strip('"') for c in line.split(";")]
+			if any(i < len(cells) and cells[i] for i in ms_indices):
+				return True
+		return False
+
 	@classmethod
-	def _peek_first_line_in_zip(cls, sample_bytes: bytes) -> str | None:
+	def _peek_lines_in_zip(cls, sample_bytes: bytes) -> list[str]:
 		import io
 		import zipfile
 
 		if not sample_bytes:
-			return None
+			return []
 		try:
 			with zipfile.ZipFile(io.BytesIO(sample_bytes)) as z:
 				csvs = [m for m in z.namelist()
 				        if m.lower().endswith(".csv") and not m.endswith("/")]
 				if not csvs:
-					return None
+					return []
 				with z.open(csvs[0]) as fh:
-					head = fh.read(2048).decode("utf-8", errors="replace")
-					return head.split("\n", 1)[0]
+					head = fh.read(8192).decode("utf-8", errors="replace")
+					return head.split("\n", 5)[:5]
 		except (zipfile.BadZipFile, EOFError, KeyError):
-			return None
+			return []
 
 	# ------------------------------------------------------------------
 	# Preview

msp/importers/handlers/adn_monthly_csv_handler.py

@@ -47,72 +47,88 @@ class ADNMonthlyCSVHandler(BaseFileHandler):
 	def sniff(cls, sample_bytes: bytes, filename: str) -> float:
 		name = (filename or "").lower()
 
-		# ZIP: Inhalt auspacken und die erste CSV-Kopfzeile prüfen. So funktioniert
-		# die Erkennung auch bei umbenannten ZIPs.
+		# ZIP: Inhalt auspacken und die ersten Zeilen prüfen.
 		if name.endswith(".zip"):
-			first_line = cls._peek_first_line_in_zip(sample_bytes)
-			if first_line is None:
-				# Kein CSV im ZIP zu sehen (vielleicht verschlüsselt oder zu groß
-				# im Sample) — greifen zurück auf Namens-Heuristik.
+			lines = cls._peek_lines_in_zip(sample_bytes)
+			if not lines:
 				if "rechnungen" in name or "433148" in name:
 					return 0.65
 				return 0.0
-			return cls._score_first_line(first_line)
+			return cls._score_lines(lines)
 
 		if not name.endswith(".csv"):
 			return 0.0
 
-		# CSV-Header inspizieren: erste Zeile (bis 8 KiB reicht)
 		try:
 			head = sample_bytes.decode("utf-8", errors="replace")
 		except Exception:
 			return 0.0
 
-		first_line = head.split("\n", 1)[0]
-		return cls._score_first_line(first_line)
+		lines = head.split("\n", 5)[:5]
+		return cls._score_lines(lines)
 
 	@classmethod
-	def _score_first_line(cls, first_line: str) -> float:
-		first_line = (first_line or "").lstrip("\ufeff").strip()
-		if not first_line:
+	def _score_lines(cls, lines: list[str]) -> float:
+		if not lines:
 			return 0.0
-		cols = {c.strip().upper() for c in first_line.split(";")}
-		matching = cls._SIGNATURE_COLUMNS & cols
+		header_line = (lines[0] or "").lstrip("\ufeff").strip()
+		if not header_line:
+			return 0.0
+		header_cols = [c.strip() for c in header_line.split(";")]
+		header_upper = {c.upper() for c in header_cols if c}
+		matching = cls._SIGNATURE_COLUMNS & header_upper
 		if not matching:
 			return 0.0
 		ratio = len(matching) / len(cls._SIGNATURE_COLUMNS)
 		base = 0.4 + 0.55 * ratio  # max ≈ 0.95
-		# MS-Discriminatoren heben uns sicher von Hornet ab, das nur Vertrag
-		# kennt: voller Bonus, wenn alle drei Spalten da sind, sonst anteilig.
-		discriminators = cls._DISCRIMINATOR_COLUMNS & cols
-		if discriminators:
-			base = min(1.0, base + 0.05 * len(discriminators))
+
+		# Inhaltsbasierter Discriminator: ADN exportiert für Hornet dieselbe
+		# Spaltenstruktur (inkl. BILLINGPLAN/MSERP), füllt die MS-spezifischen
+		# Spalten dort aber nicht. Wir prüfen darum die ersten Datenzeilen auf
+		# Werte in BILLINGPLAN / MSERP / VERTRAGSDAUER.
+		ms_evidence = cls._has_ms_payload(header_cols, lines[1:])
+		if ms_evidence:
+			base = min(1.0, base + 0.15)
 		else:
-			# Ohne MS-Discriminatoren ist es entweder Hornet oder ein älterer
-			# MS-Export — wir reichen unter 1.0, damit Hornet entscheiden kann.
-			base = min(0.7, base)
+			# Spalten da, Werte aber leer → typisch Hornet, MS verliert.
+			base = min(0.55, base)
 		return min(1.0, base)
 
+	@staticmethod
+	def _has_ms_payload(header_cols: list[str], data_lines: list[str]) -> bool:
+		header_upper = [c.strip().upper() for c in header_cols]
+		ms_indices = [i for i, c in enumerate(header_upper)
+		              if c in {"BILLINGPLAN", "MSERP", "VERTRAGSDAUER"}]
+		if not ms_indices:
+			return False
+		for line in data_lines:
+			if not line or not line.strip():
+				continue
+			cells = [c.strip().strip('"') for c in line.split(";")]
+			if any(i < len(cells) and cells[i] for i in ms_indices):
+				return True
+		return False
+
 	@classmethod
-	def _peek_first_line_in_zip(cls, sample_bytes: bytes) -> str | None:
-		"""Öffnet das ZIP in-memory und gibt die erste Zeile der enthaltenen CSV
-		zurück. None, wenn keine CSV gefunden oder der Sample zu klein ist."""
+	def _peek_lines_in_zip(cls, sample_bytes: bytes) -> list[str]:
+		"""Öffnet das ZIP in-memory und gibt die ersten ~5 Zeilen der enthaltenen
+		CSV zurück. Leere Liste, wenn keine CSV gefunden oder Sample zu klein."""
 		import io
 		import zipfile
 
 		if not sample_bytes:
-			return None
+			return []
 		try:
 			with zipfile.ZipFile(io.BytesIO(sample_bytes)) as z:
 				csvs = [m for m in z.namelist()
 				        if m.lower().endswith(".csv") and not m.endswith("/")]
 				if not csvs:
-					return None
+					return []
 				with z.open(csvs[0]) as fh:
-					head = fh.read(2048).decode("utf-8", errors="replace")
-					return head.split("\n", 1)[0]
+					head = fh.read(8192).decode("utf-8", errors="replace")
+					return head.split("\n", 5)[:5]
 		except (zipfile.BadZipFile, EOFError, KeyError):
-			return None
+			return []
 
 	# ------------------------------------------------------------------
 	# Preview

msp/importers/run_orchestrator.py

@@ -181,11 +181,12 @@ def _persist_lines(run, outcomes: Iterable[dict]) -> None:
 	run.set("lines", [])
 	for o in outcomes:
 		row: CanonicalRow = o["row"]
-		target_doctype = None
-		target_name = None
+		target_doctype = o.get("target_doctype")
+		target_name = o.get("target_name")
 		target_row = None
 		if o.get("status") == "Created":
-			# Aus Subscription-Event das Ziel rückgewinnen (best effort)
+			# Aus Subscription-Event das Ziel rückgewinnen (überschreibt das vom
+			# Builder gesetzte Backup, falls genauer — z. B. Item-Row-ID).
 			event_name = o.get("event")
 			if event_name:
 				ev = frappe.db.get_value(
@@ -204,9 +205,13 @@ def _persist_lines(run, outcomes: Iterable[dict]) -> None:
 						target_name = ev["delivery_note"]
 						target_row = ev.get("delivery_note_item_row")
 
+		# Customer-Ref: bei Lookup-Field-Profilen (Hornet) ist die Ref eine
+		# Domain — der Builder hat dann die echte CUST-ID resolved.
+		customer_field = o.get("resolved_customer") or _customer_or_null(row.customer_external_ref)
+
 		run.append("lines", {
 			"source_row": row.source_row,
-			"customer": _customer_or_null(row.customer_external_ref),
+			"customer": customer_field,
 			"customer_external_ref": row.customer_external_ref,
 			"vendor_product_id": row.vendor_product_id,
 			"qty": row.qty or 0,

msp/msp/doctype/supplier_import_profile/supplier_import_profile.json

@@ -177,7 +177,7 @@
    "fieldname": "title_template",
    "fieldtype": "Data",
    "label": "Title Template (Invoice)",
-   "description": "Optionales Format-String für den Beleg-Titel. Platzhalter: {prefix}, {period}, {customer}, {customer_name}, {vendor}, {posting_date}, {document_type}. Leer = '{prefix} {period}'."
+   "description": "Optionales Format-String für den Beleg-Titel. Platzhalter: {prefix}, {period}, {invoice_month}, {customer}, {customer_name}, {vendor}, {posting_date}, {document_type}. Leer = '{prefix} {period}'."
   },
   {
    "fieldname": "title_template_credit_note",