diff --git a/server/vbv_lernwelt/shop/invoice/__init__.py b/server/vbv_lernwelt/shop/invoice/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/server/vbv_lernwelt/shop/invoice/abacus.py b/server/vbv_lernwelt/shop/invoice/abacus.py new file mode 100644 index 00000000..059c6e3b --- /dev/null +++ b/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) diff --git a/server/vbv_lernwelt/shop/invoice/creator.py b/server/vbv_lernwelt/shop/invoice/creator.py new file mode 100644 index 00000000..c8758e91 --- /dev/null +++ b/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 diff --git a/server/vbv_lernwelt/shop/invoice/repositories.py b/server/vbv_lernwelt/shop/invoice/repositories.py new file mode 100644 index 00000000..5f599b22 --- /dev/null +++ b/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() diff --git a/server/vbv_lernwelt/shop/tests/test_invoice.py b/server/vbv_lernwelt/shop/tests/test_invoice.py new file mode 100644 index 00000000..ffc4936f --- /dev/null +++ b/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 "12345" in invoice_xml + assert "1" in invoice_xml + assert "001" in invoice_xml + assert "10" in invoice_xml + assert "Test Item" 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 "12345" in uploaded_invoice + assert "1" in uploaded_invoice + assert "001" in uploaded_invoice + assert "10" in uploaded_invoice + assert "Test Item" in uploaded_invoice