feat: add invoice generation

server/vbv_lernwelt/shop/invoice/abacus.py

@@ -0,0 +1,93 @@
+import datetime
+from typing import List
+from uuid import uuid4
+from xml.dom import minidom
+from xml.etree.ElementTree import Element, SubElement, tostring
+
+from vbv_lernwelt.shop.invoice.creator import InvoiceCreator, Item
+from vbv_lernwelt.shop.invoice.repositories import InvoiceRepository
+
+
+class AbacusInvoiceCreator(InvoiceCreator):
+    def __init__(self, repository: InvoiceRepository):
+        self.repository = repository
+
+    def create_invoice(
+        self,
+        customer_number: str,
+        purchase_order_date: datetime.date,
+        delivery_date: datetime.date,
+        reference_purchase_order: str,
+        unic_id: str,
+        items: List[Item],
+        filename: str = None,
+    ):
+        invoice = self.render_invoice(
+            customer_number,
+            purchase_order_date,
+            delivery_date,
+            reference_purchase_order,
+            unic_id,
+            items,
+        )
+
+        if filename is None:
+            filename = f"vbv-vv-{uuid4().hex}.xml"
+
+        self.repository.upload_invoice(invoice, filename)
+
+    @staticmethod
+    def render_invoice(
+        customer_number: str,
+        purchase_order_date: datetime.date,
+        delivery_date: datetime.date,
+        reference_purchase_order: str,
+        unic_id: str,
+        items: List[Item],
+    ) -> str:
+        container = Element("AbaConnectContainer")
+        task = SubElement(container, "Task")
+        parameter = SubElement(task, "Parameter")
+        SubElement(parameter, "Application").text = "ORDE"
+        SubElement(parameter, "Id").text = "Verkaufsauftrag"
+        SubElement(parameter, "MapId").text = "AbaDefault"
+        SubElement(parameter, "Version").text = "2022.00"
+
+        transaction = SubElement(task, "Transaction")
+        sales_order_header = SubElement(transaction, "SalesOrderHeader", mode="SAVE")
+        sales_order_header_fields = SubElement(
+            sales_order_header, "SalesOrderHeaderFields", mode="SAVE"
+        )
+
+        SubElement(sales_order_header_fields, "CustomerNumber").text = customer_number
+        SubElement(
+            sales_order_header_fields, "PurchaseOrderDate"
+        ).text = purchase_order_date.isoformat()
+        SubElement(
+            sales_order_header_fields, "DeliveryDate"
+        ).text = delivery_date.isoformat()
+        SubElement(
+            sales_order_header_fields, "ReferencePurchaseOrder"
+        ).text = reference_purchase_order
+        SubElement(sales_order_header_fields, "UnicId").text = unic_id
+
+        for index, item in enumerate(items, start=1):
+            item_element = SubElement(sales_order_header, "Item", mode="SAVE")
+            item_fields = SubElement(item_element, "ItemFields", mode="SAVE")
+            SubElement(item_fields, "ItemNumber").text = str(index)
+            SubElement(item_fields, "ProductNumber").text = item.product_number
+            SubElement(item_fields, "QuantityOrdered").text = item.quantity
+
+            item_text = SubElement(item_element, "ItemText", mode="SAVE")
+            item_text_fields = SubElement(item_text, "ItemTextFields", mode="SAVE")
+            SubElement(item_text_fields, "Text").text = item.description
+
+        return AbacusInvoiceCreator.create_xml_string(container)
+
+    @staticmethod
+    def create_xml_string(container: Element, encoding: str = "UTF-8") -> str:
+        xml_bytes = tostring(container, encoding)
+        xml_pretty_str = minidom.parseString(xml_bytes).toprettyxml(
+            indent="  ", encoding=encoding
+        )
+        return xml_pretty_str.decode(encoding)

server/vbv_lernwelt/shop/invoice/creator.py

@@ -0,0 +1,25 @@
+import datetime
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import List
+
+
+@dataclass
+class Item:
+    product_number: str
+    quantity: str
+    description: str
+
+
+class InvoiceCreator(ABC):
+    @abstractmethod
+    def create_invoice(
+        self,
+        customer_number: str,
+        purchase_order_date: datetime.date,
+        delivery_date: datetime.date,
+        reference_purchase_order: str,
+        unic_id: str,
+        items: List[Item],
+    ):
+        pass

server/vbv_lernwelt/shop/invoice/repositories.py

@@ -0,0 +1,40 @@
+from abc import ABC, abstractmethod
+
+
+class InvoiceRepository(ABC):
+    @abstractmethod
+    def upload_invoice(self, invoice: str, filename: str):
+        pass
+
+
+class SFTPInvoiceRepository(InvoiceRepository):
+    def __init__(self, hostname: str, username: str, password: str, port: int = 22):
+        self.hostname = hostname
+        self.username = username
+        self.password = password
+        self.port = port
+
+    def upload_invoice(self, invoice: str, filename: str) -> None:
+        from io import BytesIO
+
+        from paramiko import AutoAddPolicy, SSHClient
+
+        invoice_io = BytesIO(invoice.encode("utf-8"))
+        ssh_client = SSHClient()
+
+        try:
+            ssh_client.set_missing_host_key_policy(AutoAddPolicy())
+            ssh_client.connect(
+                self.hostname,
+                port=self.port,
+                username=self.username,
+                password=self.password,
+            )
+
+            with ssh_client.open_sftp() as sftp_client:
+                sftp_client.putfo(invoice_io, f"uploads/{filename}")
+
+        except Exception as e:
+            print(f"An error occurred: {e}")
+        finally:
+            ssh_client.close()

server/vbv_lernwelt/shop/tests/test_invoice.py

@@ -0,0 +1,71 @@
+from datetime import date
+from unittest.mock import create_autospec
+
+from vbv_lernwelt.shop.invoice.abacus import AbacusInvoiceCreator
+from vbv_lernwelt.shop.invoice.creator import Item
+from vbv_lernwelt.shop.invoice.repositories import InvoiceRepository
+
+
+def test_render_invoice():
+    # GIVEN
+    creator = AbacusInvoiceCreator(repository=create_autospec(InvoiceRepository))
+    items = [Item(product_number="001", quantity="10", description="Test Item")]
+    customer_number = "12345"
+    purchase_order_date = date(2023, 1, 1)
+    delivery_date = date(2023, 1, 10)
+    reference_purchase_order = "PO12345678"
+    unic_id = "UNIC001"
+
+    # WHEN
+    invoice_xml = creator.render_invoice(
+        customer_number,
+        purchase_order_date,
+        delivery_date,
+        reference_purchase_order,
+        unic_id,
+        items,
+    )
+
+    # THEN
+    assert "<CustomerNumber>12345</CustomerNumber>" in invoice_xml
+    assert "<ItemNumber>1</ItemNumber>" in invoice_xml
+    assert "<ProductNumber>001</ProductNumber>" in invoice_xml
+    assert "<QuantityOrdered>10</QuantityOrdered>" in invoice_xml
+    assert "<Text>Test Item</Text>" in invoice_xml
+
+
+def test_create_invoice_calls_upload():
+    # GIVEN
+    repository_mock = create_autospec(InvoiceRepository)
+
+    creator = AbacusInvoiceCreator(repository=repository_mock)
+    items = [Item(product_number="001", quantity="10", description="Test Item")]
+    customer_number = "12345"
+    purchase_order_date = date(2023, 1, 1)
+    delivery_date = date(2023, 1, 10)
+    reference_purchase_order = "PO12345678"
+    unic_id = "UNIC001"
+
+    expected_filename = "test.xml"
+
+    # WHEN
+    creator.create_invoice(
+        customer_number,
+        purchase_order_date,
+        delivery_date,
+        reference_purchase_order,
+        unic_id,
+        items,
+        filename=expected_filename,
+    )
+
+    # THEN
+    repository_mock.upload_invoice.assert_called_once()
+    uploaded_invoice, uploaded_filename = repository_mock.upload_invoice.call_args[0]
+
+    assert uploaded_filename == expected_filename
+    assert "<CustomerNumber>12345</CustomerNumber>" in uploaded_invoice
+    assert "<ItemNumber>1</ItemNumber>" in uploaded_invoice
+    assert "<ProductNumber>001</ProductNumber>" in uploaded_invoice
+    assert "<QuantityOrdered>10</QuantityOrdered>" in uploaded_invoice
+    assert "<Text>Test Item</Text>" in uploaded_invoice