237 lines
8.9 KiB
Python
237 lines
8.9 KiB
Python
import datetime
|
|
from io import BytesIO
|
|
from xml.dom import minidom
|
|
from xml.etree.ElementTree import Element, SubElement, tostring
|
|
|
|
import structlog
|
|
from django.conf import settings
|
|
from paramiko.client import AutoAddPolicy, SSHClient
|
|
|
|
from vbv_lernwelt.shop.models import CheckoutInformation
|
|
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
def abacus_ssh_upload(checkout_information: CheckoutInformation):
|
|
invoice_xml_filename, invoice_xml_content = create_invoice_xml(checkout_information)
|
|
customer_xml_filename, customer_xml_content = create_customer_xml(
|
|
checkout_information
|
|
)
|
|
|
|
abacus_ssh_upload_invoice(
|
|
customer_xml_filename, customer_xml_content, folder="debitor"
|
|
)
|
|
abacus_ssh_upload_invoice(invoice_xml_filename, invoice_xml_content, folder="order")
|
|
|
|
|
|
def create_invoice_xml(checkout_information: CheckoutInformation):
|
|
# set fill abacus numbers
|
|
checkout_information = checkout_information.set_increment_abacus_order_id()
|
|
customer = checkout_information.user.set_increment_abacus_debitor_number()
|
|
|
|
invoice_xml_content = render_invoice_xml(
|
|
abacus_debitor_number=customer.abacus_debitor_number,
|
|
abacus_order_id=checkout_information.abacus_order_id,
|
|
datatrans_transaction_id=checkout_information.transaction_id,
|
|
order_date=checkout_information.created_at.date(),
|
|
# TODO was ist der korrekte text für die item_description?
|
|
item_description=f"{checkout_information.product_name} - {checkout_information.created_at.date().isoformat()} - {checkout_information.user.last_name} {checkout_information.user.first_name}",
|
|
)
|
|
|
|
# YYYYMMDDhhmmss
|
|
filename_datetime = checkout_information.created_at.strftime("%Y%m%d%H%M%S")
|
|
invoice_xml_filename = f"myVBV_orde_{customer.abacus_debitor_number}_{filename_datetime}_{checkout_information.abacus_order_id}.xml"
|
|
|
|
return invoice_xml_filename, invoice_xml_content
|
|
|
|
|
|
def create_customer_xml(checkout_information: CheckoutInformation):
|
|
customer = checkout_information.user.set_increment_abacus_debitor_number()
|
|
|
|
customer_xml_content = render_customer_xml(
|
|
abacus_debitor_number=customer.abacus_debitor_number,
|
|
last_name=checkout_information.last_name,
|
|
first_name=checkout_information.first_name,
|
|
company_name=checkout_information.organisation_detail_name
|
|
if checkout_information.invoice_address == "org"
|
|
else "",
|
|
street=(
|
|
checkout_information.organisation_street
|
|
if checkout_information.invoice_address == "org"
|
|
else checkout_information.street
|
|
),
|
|
house_number=(
|
|
checkout_information.organisation_street_number
|
|
if checkout_information.invoice_address == "org"
|
|
else checkout_information.street_number
|
|
),
|
|
zip_code=(
|
|
checkout_information.organisation_postal_code
|
|
if checkout_information.invoice_address == "org"
|
|
else checkout_information.postal_code
|
|
),
|
|
city=(
|
|
checkout_information.organisation_city
|
|
if checkout_information.invoice_address == "org"
|
|
else checkout_information.city
|
|
),
|
|
country=(
|
|
checkout_information.organisation_country_id
|
|
if checkout_information.invoice_address == "org"
|
|
else checkout_information.country_id
|
|
),
|
|
language=customer.language,
|
|
email=customer.email,
|
|
)
|
|
|
|
customer_xml_filename = f"myVBV_debi_{customer.abacus_debitor_number}.xml"
|
|
|
|
return customer_xml_filename, customer_xml_content
|
|
|
|
|
|
def render_invoice_xml(
|
|
abacus_debitor_number: int,
|
|
abacus_order_id: int,
|
|
datatrans_transaction_id: str,
|
|
order_date: datetime.date,
|
|
item_description: str,
|
|
) -> 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"
|
|
)
|
|
|
|
# Skender: Kundennummer, erste Ziffer abhängig von der Plattform (4 = LMS, 6 = myVBV, 7 = EduManager), Plattform führt ein eigenständiges hochzählendes Mapping.
|
|
SubElement(sales_order_header_fields, "CustomerNumber").text = str(
|
|
abacus_debitor_number
|
|
)
|
|
|
|
# Skender: ePayment: Ablaufnr. für ePayment Rechnung in Abacus
|
|
SubElement(sales_order_header_fields, "ProcessFlowNumber").text = "30"
|
|
|
|
# Skender: ePayment: Zahlungskondition für ePaymente in Abacus 9999 Tage Mahnungsfrist, da schon bezahlt
|
|
SubElement(sales_order_header_fields, "PaymentCode").text = "9999"
|
|
|
|
# Skender: Bestellzeitpunkt
|
|
SubElement(
|
|
sales_order_header_fields, "PurchaseOrderDate"
|
|
).text = order_date.isoformat()
|
|
|
|
# Skender: ePayment: TRANSACTION-ID von Datatrans in Bestellreferenz
|
|
SubElement(
|
|
sales_order_header_fields, "ReferencePurchaseOrder"
|
|
).text = datatrans_transaction_id
|
|
|
|
# Skender: ePayment: OrderID. max 10 Ziffern, erste Ziffer abhängig von der Plattform (4 = LMS, 6 = myVBV, 7 = EduManager)
|
|
SubElement(sales_order_header_fields, "GroupingNumberAscii1").text = str(
|
|
abacus_order_id
|
|
)
|
|
|
|
item_element = SubElement(sales_order_header, "Item", mode="SAVE")
|
|
item_fields = SubElement(item_element, "ItemFields", mode="SAVE")
|
|
SubElement(item_fields, "DeliveryDate").text = order_date.isoformat()
|
|
SubElement(item_fields, "ItemNumber").text = "1"
|
|
SubElement(item_fields, "ProductNumber").text = "30202"
|
|
SubElement(item_fields, "QuantityOrdered").text = "1"
|
|
|
|
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 create_xml_string(container)
|
|
|
|
|
|
def render_customer_xml(
|
|
abacus_debitor_number: int,
|
|
last_name: str,
|
|
first_name: str,
|
|
company_name: str,
|
|
street: str,
|
|
house_number: str,
|
|
zip_code: str,
|
|
city: str,
|
|
country: str,
|
|
language: str,
|
|
email: str,
|
|
):
|
|
container = Element("AbaConnectContainer")
|
|
task = SubElement(container, "Task")
|
|
|
|
parameter = SubElement(task, "Parameter")
|
|
SubElement(parameter, "Application").text = "DEBI"
|
|
SubElement(parameter, "ID").text = "Kunden"
|
|
SubElement(parameter, "MapID").text = "AbaDefault"
|
|
SubElement(parameter, "Version").text = "2022.00"
|
|
|
|
transaction = SubElement(task, "Transaction")
|
|
customer_element = SubElement(transaction, "Customer", mode="SAVE")
|
|
|
|
SubElement(customer_element, "CustomerNumber").text = str(abacus_debitor_number)
|
|
SubElement(customer_element, "DefaultCurrency").text = "CHF"
|
|
SubElement(customer_element, "PaymentTermNumber").text = "1"
|
|
SubElement(customer_element, "ReminderProcedure").text = "NORM"
|
|
|
|
address_data = SubElement(customer_element, "AddressData", mode="SAVE")
|
|
SubElement(address_data, "AddressNumber").text = str(abacus_debitor_number)
|
|
SubElement(address_data, "Name").text = last_name
|
|
SubElement(address_data, "FirstName").text = first_name
|
|
if company_name:
|
|
SubElement(address_data, "Text").text = company_name
|
|
SubElement(address_data, "Street").text = street
|
|
SubElement(address_data, "HouseNumber").text = house_number
|
|
SubElement(address_data, "ZIP").text = zip_code
|
|
SubElement(address_data, "City").text = city
|
|
SubElement(address_data, "Country").text = country
|
|
SubElement(address_data, "Language").text = language
|
|
SubElement(address_data, "Email").text = email
|
|
|
|
return create_xml_string(container)
|
|
|
|
|
|
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)
|
|
|
|
|
|
def abacus_ssh_upload_invoice(
|
|
filename: str, content_xml: str, folder: str = ""
|
|
) -> None:
|
|
invoice_io = BytesIO(content_xml.encode("utf-8"))
|
|
ssh_client = SSHClient()
|
|
|
|
print(invoice_io)
|
|
|
|
try:
|
|
ssh_client.set_missing_host_key_policy(AutoAddPolicy())
|
|
ssh_client.connect(
|
|
hostname=settings.ABACUS_EXPORT_SFTP_HOST,
|
|
port=settings.ABACUS_EXPORT_SFTP_PORT,
|
|
username=settings.ABACUS_EXPORT_SFTP_USERNAME,
|
|
password=settings.ABACUS_EXPORT_SFTP_PASSWORD,
|
|
)
|
|
|
|
with ssh_client.open_sftp() as sftp_client:
|
|
path = filename
|
|
if folder:
|
|
path = f"{folder}/{filename}"
|
|
|
|
sftp_client.putfo(invoice_io, path)
|
|
|
|
except Exception as e:
|
|
logger.error("Could not upload invoice", exc_info=e)
|
|
finally:
|
|
ssh_client.close()
|