""" * @author Xiaobo Sun * created on 29.11.2025 * @copyright (C) 2025 Sogood International GmbH - all rights reserved * @licence * Unauthorized copying of this file, via any medium is strictly prohibited * Proprietary and confidential """ import base64 import io import glob import os import logging import csv import json import re from datetime import datetime import pytz from django.http import HttpResponse from django.views.decorators.csrf import csrf_exempt import shutil from .clazz import TaxItem from .setting import CONFIG, CompanyEntry # Get an instance of a logger logger = logging.getLogger('django') def sanitize_filename(filename: str) -> str: # Define ALL invalid filename characters for Windows/macOS/Linux invalid_chars = r'[<>:"/\\|?*+]' # Replace invalid characters with a hyphen (safe) clean_name = re.sub(invalid_chars, '-', filename) # Optional: Remove extra spaces/trailing dots (extra safety) clean_name = clean_name.strip().strip('.') return clean_name @csrf_exempt def print_datev(request): """ Django View : endpoint Used by E2 x_controller :param request: httpRequest :rtype: HttpResponse """ # FIXME: using request to fetch the all parameters prefix = request.POST.get("prefix") is_sogood = bool(int(request.POST.get("is_sogood"))) order_id = request.POST.get("order_id") payment_date = request.POST.get("payment_date") payment_date = datetime.strptime(payment_date, "%d.%m.%Y %H:%M:%S") # Transaction date order_date = request.POST.get("order_date") order_date = datetime.strptime(order_date, "%d.%m.%Y %H:%M:%S") payment_full_amount = float(request.POST.get("payment_full_amount")) payment_already_paid = float(request.POST.get("payment_already_paid")) tax = int(request.POST.get("tax")) payment_id = request.POST.get("payment_id") # mandatory as TransactionId invoice_url = request.POST.get("invoice_url") # optional customer_country = request.POST.get("customer_country") selling_country = request.POST.get("selling_country") lager_country = request.POST.get("lager_country") systemId = request.POST.get("systemId") systemInvoiceId = request.POST.get("systemInvoiceId") customerTaxId = request.POST.get("customerTaxId") taxItemId = request.POST.get("taxItemId") # as database id of x_taxitem konto = request.POST.get("konto") logger.info("print_datev with prefix " + prefix + " order_id " + order_id + " systemId " + systemId + " payment_id " + payment_id + " tax " + str(tax)) # create an taxItem taxItem = TaxItem(order_id, payment_date, order_date, f"{payment_full_amount:.2f}".replace(".", ","), f"{payment_already_paid:.2f}".replace(".", ","), payment_id, invoice_url, customer_country, selling_country, lager_country, systemId, systemInvoiceId, customerTaxId, taxItemId, konto, tax) # identify the company company = CONFIG.company[0] if is_sogood else CONFIG.company[1] # export to a file source_path = os.path.dirname(os.path.abspath(__file__)) out_dir = os.path.abspath(os.path.join(source_path, "..", "datev_files")) os.makedirs(out_dir, exist_ok=True) file_path = '%s/../datev_files/%s_%s_%s_%s_%s.csv' % (source_path, prefix, systemId, sanitize_filename(payment_id), taxItemId, payment_date.strftime("%Y%m%d")) with open(file_path, "w", encoding="utf-8") as file: strD = taxItem.toDatev(company) if strD != None: file.write(strD) return HttpResponse("print_datev with file_path " + file_path) @csrf_exempt def merge_datev(request): """ Django View : endpoint Used by E2 x_controller :param request: httpRequest :rtype: HttpResponse """ prefix = request.POST.get("prefix") is_sogood = bool(int(request.POST.get("is_sogood"))) # convert str to bool company = CONFIG.company[0] if is_sogood else CONFIG.company[1] logger.info("merge_datev with prefix " + prefix) # loop all existing csv file with same prefix source_path = os.path.dirname(os.path.abspath(__file__)) paths = glob.glob(source_path + '/../datev_files/%s_*.csv' % (prefix)) paths = [f for f in paths if re.search(r'%s_[a-zA-Z0-9-]+_[a-zA-Z0-9-]+_[\d]+_[\d]+\.csv$' % re.escape(prefix), f)] paths.sort(key=os.path.getmtime) min_orderDate = "20301231" max_orderDate = "20200101" datav_objects = [] for path in paths: filename = os.path.basename(path) name, ext = os.path.splitext(filename) # 1. Read the content first with open(path) as infile: content = infile.read().strip() # .strip() handles files with only whitespace/newlines # 2. Check if content is empty if not content: logger.warning(f"Skipping empty file: {filename}") continue # This jumps to the next iteration of the 'for' loop # 3. Proceed with logic only if content exists order_id = name.split("_")[1] order_date = name.split("_")[4] logger.info(f"file: {filename} : {order_id} : {order_date}") if int(order_date) < int(min_orderDate): min_orderDate = order_date if int(order_date) > int(max_orderDate): max_orderDate = order_date logger.info(f"[" + content + "]") datav_objects.append(content) os.remove(path) logger.debug("merge_datev min_orderDate " + min_orderDate + " max_orderDate " + max_orderDate) # header lines berlin = pytz.timezone("Europe/Berlin") now = datetime.now(berlin) #FIXME: header 5 why 12? It should be 13 # JE ist Herkunft # JERA2FIBU ist Exportiert von # 5 is len(company.konto) fileHeader = ( f'"EXTF";700;21;"Buchungsstapel";13;' f'{now.strftime("%Y%m%d%H%M%S")}000;;"JE";"JERA2FIBU";"";' f'{company.beraternummer};{company.mandantennummer};"20250101";' f'{5};{min_orderDate};{max_orderDate};' f'"Belege {min_orderDate[:4]}/{min_orderDate[4:6]}";' f'JE;1;0;0;"EUR";;"";;;"";;;"";""' ) columnHeader = ("Umsatz (ohne Soll/Haben-Kz);" + # column1 "Soll/Haben-Kennzeichen;" + # column2 "WKZ Umsatz;" + # column3 "Kurs;" + # column4 "Basis-Umsatz;" + # column5 "WKZ Basis-Umsatz;" + # column6 "Konto;" + # column7 "Gegenkonto (ohne BU-Schlüssel);" + # column8 "BU-Schlüssel;" + # column9 "Belegdatum;" + # column10 "Belegfeld 1;" + # column11 "Belegfeld 2;" + # column12 "Skonto;" + # column13 "Buchungstext;" + # column14 "Postensperre;" + # column15 "Diverse Adressnummer;" + # column16 "Geschäftspartnerbank;" + # column17 "Sachverhalt;" + # column18 "Zinssperre;" + # column19 "Beleglink;" + # column20 "Beleginfo - Art 1;" + # column21 "Beleginfo - Inhalt 1;" + # column22 "Beleginfo - Art 2;" + # column23 "Beleginfo - Inhalt 2;" + # column24 "Beleginfo - Art 3;" + # column25 "Beleginfo - Inhalt 3;" + # column26 "Beleginfo - Art 4;" + # column27 "Beleginfo - Inhalt 4;" + # column28 "Beleginfo - Art 5;" + # column29 "Beleginfo - Inhalt 5;" + # column30 "Beleginfo - Art 6;" + # column31 "Beleginfo - Inhalt 6;" + # column32 "Beleginfo - Art 7;" + # column33 "Beleginfo - Inhalt 7;" + # column34 "Beleginfo - Art 8;" + # column35 "Beleginfo - Inhalt 8;" + # column36 "KOST1 - Kostenstelle;" + # column37 "KOST2 - Kostenstelle;" + # column38 "Kost-Menge;" + # column39 "EU-Land und UStID (Bestimmung);" + # column40 "EU-Steuersatz (Bestimmung);" + # column41 "Abw. Versteuerungsart;" + # column42 "Sachverhalt L+L;" + # column43 "Funktionsergänzung L+L;" + # column44 "BU 49 Hauptfunktionstyp;" + # column45 "BU 49 Hauptfunktionsnummer;" + # column46 "BU 49 Funktionsergänzung;" + # column47 "Zusatzinformation - Art 1;" + # column48 "Zusatzinformation- Inhalt 1;" + # column49 "Zusatzinformation - Art 2;" + # column50 "Zusatzinformation- Inhalt 2;" + # column51 "Zusatzinformation - Art 3;" + # column52 "Zusatzinformation- Inhalt 3;" + # column53 "Zusatzinformation - Art 4;" + # column54 "Zusatzinformation- Inhalt 4;" + # column55 "Zusatzinformation - Art 5;" + # column56 "Zusatzinformation- Inhalt 5;" + # column57 "Zusatzinformation - Art 6;" + # column58 "Zusatzinformation- Inhalt 6;" + # column59 "Zusatzinformation - Art 7;" + # column60 "Zusatzinformation- Inhalt 7;" + # column61 "Zusatzinformation - Art 8;" + # column62 "Zusatzinformation- Inhalt 8;" + # column63 "Zusatzinformation - Art 9;" + # column64 "Zusatzinformation- Inhalt 9;" + # column65 "Zusatzinformation - Art 10;" + # column66 "Zusatzinformation- Inhalt 10;" + # column67 "Zusatzinformation - Art 11;" + # column68 "Zusatzinformation- Inhalt 11;" + # column69 "Zusatzinformation - Art 12;" + # column70 "Zusatzinformation- Inhalt 12;" + # column71 "Zusatzinformation - Art 13;" + # column72 "Zusatzinformation- Inhalt 13;" + # column73 "Zusatzinformation - Art 14;" + # column74 "Zusatzinformation- Inhalt 14;" + # column75 "Zusatzinformation - Art 15;" + # column76 "Zusatzinformation- Inhalt 15;" + # column77 "Zusatzinformation - Art 16;" + # column78 "Zusatzinformation- Inhalt 16;" + # column79 "Zusatzinformation - Art 17;" + # column80 "Zusatzinformation- Inhalt 17;" + # column81 "Zusatzinformation - Art 18;" + # column82 "Zusatzinformation- Inhalt 18;" + # column83 "Zusatzinformation - Art 19;" + # column84 "Zusatzinformation- Inhalt 19;" + # column85 "Zusatzinformation - Art 20;" + # column86 "Zusatzinformation- Inhalt 20;" + # column87 "Stück;" + # column88 "Gewicht;" + # column89 "Zahlweise;" + # column90 "Forderungsart;" + # column91 "Veranlagungsjahr;" + # column92 "Zugeordnete Fälligkeit;" + # column93 "Skontotyp;" + # column94 "Auftragsnummer;" + # column95 "Buchungstyp;" + # column96 "Ust-Schlüssel (Anzahlungen);" + # column97 "EU-Land (Anzahlungen);" + # column98 "Sachverhalt L+L (Anzahlungen);" + # column99 "EU-Steuersatz (Anzahlungen);" + # column100 "Erlöskonto (Anzahlungen);" + # column101 "Herkunft-Kz;" + # column102 "Leerfeld;" + # column103 "KOST-Datum;" + # column104 "Mandatsreferenz;" + # column105 "Skontosperre;" + # column106 "Gesellschaftername;" + # column107 "Beteiligtennummer;" + # column108 "Identifikationsnummer;" + # column109 "Zeichnernummer;" + # column110 "Postensperre bis;" + # column111 "Bezeichnung SoBil-Sachverhalt;" + # column112 "Kennzeichen SoBil-Buchung;" + # column113 "Festschreibung;" + # column114 "Leistungsdatum;" + # column115 "Datum Zuord.Steuerperiode;" + # column116 "Fälligkeit;" + # column117 "Generalumkehr (GU);" + # column118 "Steuersatz;" + # column119 "Land;" + # column120 "Abrechnungsreferenz;" + # column121 "BVV-Postion;" + # column122 "EU-Land und UStID (Ursprung);" + # column123 "EU-Steuersatz (Ursprung);" # column124 ) # export to a file source_path = os.path.dirname(os.path.abspath(__file__)) file_name = ("EXTF_Buchungsstapel_Belege_" + now.strftime("%Y%m%d_%H%M%S") + ".csv") file_path = '%s/../datev_files/%s' % (source_path, file_name) try: # FIXME: why using cp1252 instead of utf-8, # Because Datev prefers to use cp1252. with open(file_path, "w", encoding="cp1252") as f: f.write(fileHeader + "\n") f.write(columnHeader + "\n") for datev_object in datav_objects: f.write(datev_object + "\n") logger.info(f"✅ write to {file_name} successfully") except IOError as e: logger.error(f"❌ write to {file_name} failed: {e}") # revert the order of rows reverse_csv_rows(file_path, file_path, 2) # =========================== # 返回 Base64 # =========================== with open(file_path, "r", encoding="cp1252") as f: csv_content = f.read() csv_base64 = base64.b64encode(csv_content.encode("cp1252")).decode("utf-8") # return response return HttpResponse(csv_base64) @csrf_exempt def revert_datev(request): """ Django View : endpoint Used by E2 x_controller :param request: httpRequest :rtype: HttpResponse """ file_name = request.POST.get("file_name") logger.info("revert_datev with file_name " + file_name) source_path = os.path.dirname(os.path.abspath(__file__)) input_file = '%s/../datev_files/%s' % (source_path, file_name) output_file = '%s/../datev_files/%s' % (source_path, file_name) reverse_csv_rows(input_file, output_file, 2) # return response return HttpResponse("revert_datev successfully ") def reverse_csv_rows(input_file, output_file, header_lines=2): """ 反转CSV数据行（保留表头不变，不修改内容格式） :param input_file: 输入CSV路径 :param output_file: 输出CSV路径 :param header_lines: 表头行数，默认2行 """ # 原样读取所有行 with open(input_file, "r", encoding="cp1252", newline="") as f: lines = f.readlines() # 拆分表头和数据 header = lines[:header_lines] data = lines[header_lines:] # 反转数据行 data_reversed = data[::-1] # 原样写入文件 with open(output_file, "w", encoding="cp1252", newline="") as f: f.writelines(header) f.writelines(data_reversed) logger.info(f"✅ 处理完成！文件已保存到：{output_file}")