CBAM kafka, NCTS Fremd(Tobb), realyhub

VERAG_PROG_ALLGEMEIN/Schnittstellen/ATEZ/GREENPULSE/cATEZ_Greenpulse_KafkaDecs.vb

@@ -23,8 +23,8 @@ Public Class cATEZ_Greenpulse_KafkaDecs
     '== Kafka: Konfiguration (Klassenebene)
     '========================
     Public Shared BootstrapServers As String = "192.168.85.250:9092" 'http://192.168.85.250:8888
-    Public Shared TopicName As String = "greenpulse.declarationdata.v1"
-    '  Public Shared TopicName As String = "dev.greenpulse.declarationdata.v1"
+    ' Public Shared TopicName As String = "greenpulse.declarationdata.v1"
+    Public Shared TopicName As String = "dev.greenpulse.declarationdata.v1"
     ' Falls SASL/TLS benötigt:
     Public Shared UseSasl As Boolean = False
     Public Shared SaslUsername As String = ""
@@ -66,6 +66,11 @@ Public Class cATEZ_Greenpulse_KafkaDecs
     <JsonProperty("importerDetails")>
     Public Property ImporterDetails As ImporterDetailsNode
 
+    '--- documents ---
+    <JsonProperty("documents")>
+    Public Property Documents As List(Of DocumentNode)
+
+
     '--- declaration ---
     Public Class DeclarationNode
         <JsonProperty("declarationsourceId")>
@@ -201,6 +206,19 @@ Public Class cATEZ_Greenpulse_KafkaDecs
         <JsonProperty("importerCoordinateLatitudeY")>
         Public Property ImporterCoordinateLatitudeY As String
     End Class
+    Public Class DocumentNode
+        <JsonProperty("reference")>
+        Public Property Reference As String
+
+        <JsonProperty("doc-type")>
+        Public Property DocType As String
+
+        <JsonProperty("mime-type")>
+        Public Property MimeType As String
+
+        <JsonProperty("blob")>
+        Public Property Blob As String
+    End Class
 
     '========================
     '== Serialisierung
@@ -267,7 +285,8 @@ Public Class cATEZ_Greenpulse_KafkaDecs
           .ImporterPoBox = "PO DCL-123",
           .ImporterCoordinateLongitudeX = "41.0091982",
           .ImporterCoordinateLatitudeY = "28.9662187"
-        }
+        },
+         .Documents = New List(Of cATEZ_Greenpulse_KafkaDecs.DocumentNode)()
       }
     End Function
 
@@ -298,6 +317,8 @@ Public Class cATEZ_Greenpulse_KafkaDecs
     .MaxInFlight = 5,
     .MessageTimeoutMs = Math.Max(waitMs, 60000),
     .RequestTimeoutMs = 30000,
+    .CompressionType = Confluent.Kafka.CompressionType.Zstd,  ' gute Kompression
+    .MessageMaxBytes = 20971520,           ' ≈ 20 MB – darf Topic/Broker nicht übersteigen
     .EnableDeliveryReports = True,
     .AllowAutoCreateTopics = True
   }
@@ -379,7 +400,7 @@ Public Class cATEZ_Greenpulse_KafkaDecsBuilder_DAKOSY
 
             Dim obj As New cATEZ_Greenpulse_KafkaDecs() With {
         .Declaration = New cATEZ_Greenpulse_KafkaDecs.DeclarationNode() With {
-          .DeclarationSourceId = SafeStr(head("Bezugsnummer_LRN")),
+          .DeclarationSourceId = SafeStr(head("Registriernummer_MRN")),
           .DeclarationNo = SafeStr(head("Registriernummer_MRN")),
           .DeclarationDate = FirstNonEmptyDateStr(head, {"Annahmedatum", "Überlassungsdatum"}),
           .RequestedProcedure = SafeStr(head("Verfahren")),
@@ -412,7 +433,8 @@ Public Class cATEZ_Greenpulse_KafkaDecsBuilder_DAKOSY
           .ImporterPoBox = "",
           .ImporterCoordinateLongitudeX = "",
           .ImporterCoordinateLatitudeY = ""
-        }
+        },
+         .Documents = New List(Of cATEZ_Greenpulse_KafkaDecs.DocumentNode)()
       }
 
             ' 2) Commercial (Rechnung) – aus Unterlagen N380, falls vorhanden
@@ -424,6 +446,38 @@ Public Class cATEZ_Greenpulse_KafkaDecsBuilder_DAKOSY
         .DefaultIfEmpty(Nothing) _
         .FirstOrDefault()
 
+            ' --- Dokumente aus Unterlagen übernehmen ---
+            Dim SQLS As New VERAG_PROG_ALLGEMEIN.SQL
+            Dim SenungsId = SQLS.getValueTxtBySql("SELECT dy_SendungsId from [tblDakosy_Zollanmeldungen]  where dy_BezugsNr=''", "FMZOLL",,, Nothing)
+
+
+            If SenungsId IsNot Nothing Then
+                If IsNumeric(SenungsId) AndAlso SenungsId > 0 Then
+                    Dim ANH_LIST As New List(Of cAvisoAnhaenge)
+                    cAvisoAnhaenge.LOAD_LIST_BySendung(ANH_LIST, SenungsId)
+
+
+                    For Each doc In ANH_LIST
+                        Select Case doc.anh_Art
+                            Case "Rechnung", "eFatura"
+
+                                Dim dateiBytes As Byte() = System.IO.File.ReadAllBytes(VERAG_PROG_ALLGEMEIN.cDATENSERVER.GET_PDFPath_BY_DocID(doc.anh_docId))
+
+                                Dim d As New cATEZ_Greenpulse_KafkaDecs.DocumentNode With {
+                                        .Reference = doc.anh_Name,
+                                        .DocType = "invoice",
+                                        .MimeType = cATEZ_Greenpulse_KafkaDecsBuilder_DAKOSY.GuessMimeTypeFromNumber(doc.anh_Typ),
+                                        .Blob = Convert.ToBase64String(dateiBytes)
+                                    }
+                                obj.Documents.Add(d)
+                        End Select
+
+                    Next
+                End If
+
+            End If
+
+
             If invRow IsNot Nothing Then
                 obj.Commercial.InvoiceNumbers = SafeStr(invRow("Unterlagennummer"))
                 obj.Commercial.InvoiceDate = SafeDateStr(invRow("Unterlagendatum"))
@@ -519,4 +573,15 @@ Public Class cATEZ_Greenpulse_KafkaDecsBuilder_DAKOSY
         Return String.IsNullOrWhiteSpace(Convert.ToString(value))
     End Function
 
+
+    Public Shared Function GuessMimeTypeFromNumber(num As Object) As String
+        ' Wenn du Dateiendungen erkennst (z. B. .pdf oder .jpg im Namen)
+        Dim s As String = SafeStr(num).ToLowerInvariant()
+        If s.EndsWith(".pdf") Or s.ToLower = "PDF" Then Return "application/pdf"
+        If s.EndsWith(".jpg") Or s.EndsWith(".jpeg") Or s.ToLower = "JPG" Or s.ToLower = "JPEG" Then Return "image/jpeg"
+        If s.EndsWith(".png") Or s.ToLower = "PNG" Then Return "image/png"
+        Return "application/octet-stream"
+    End Function
+
+
 End Class