Volkswagen erste Version für Marco

gchr/gchr.py

@@ -7,39 +7,45 @@ from pathlib import Path
 import pandas as pd
 
 from gchr.gchr_bookings import GchrBookings
-from gchr.gchr_export import GchrExportFormat, get_export_fn
+from gchr.gchr_export import get_export_fn
 from gchr.gchr_model import ACCOUNT_INFO, GchrConfig, GchrExportConfig
 from gchr.gchr_translate import load_translation
 
 
 class GCHR:
     cfg: GchrConfig
-    booking_date: datetime
     bookings: GchrBookings
     _df_translate: pd.DataFrame = None
     df_translate2: pd.DataFrame = None
     makes: dict[str, str] = None
     sites: dict[str, str] = None
-    current_year: str
-    current_month: str
     timestamp: str
 
     def __init__(self, base_dir: str) -> None:
         self.base_dir = base_dir
-        cfg_file = self.base_dir + "\\config\\gchr2.json"
+        cfg_file = f"{self.base_dir}\\config\\gchr2.json"
         if Path(cfg_file).exists():
             with open(cfg_file, "r") as frh:
                 self.cfg = GchrConfig(**json.load(frh))
         else:
-            self.cfg = GchrConfig()
-        os.makedirs(self.base_dir + "/data", exist_ok=True)
-        os.makedirs(self.base_dir + "/export/temp", exist_ok=True)
-        os.makedirs(self.base_dir + "/logs", exist_ok=True)
+            os.makedirs(f"{self.base_dir}\\config", exist_ok=True)
+            self.cfg = GchrConfig(
+                first_month_of_financial_year="01",
+                data_dir=f"{self.base_dir}\\data",
+                gcstruct_dir=f"{self.base_dir}\\..\\GCStruct_Aufbereitung",
+                export_dir=f"{self.base_dir}\\Export",
+                export_format="SKR51",
+            )
+            with open(cfg_file, "w") as fwh:
+                json.dump(self.cfg.__dict__, fwh, indent=2)
+
+        os.makedirs(self.cfg.data_dir, exist_ok=True)
+        os.makedirs(f"{self.cfg.export_dir}\\temp", exist_ok=True)
+        os.makedirs(f"{self.base_dir}\\logs", exist_ok=True)
 
-        self.account_translation = f"{self.base_dir}/data/Kontenrahmen_uebersetzt.csv"
+        self.account_translation = f"{self.cfg.data_dir}\\Kontenrahmen_uebersetzt.csv"
 
-        self.first_month_of_financial_year = "10"
-        self.bookings = GchrBookings(self.base_dir, self.first_month_of_financial_year)
+        self.bookings = GchrBookings(self.base_dir, self.cfg.first_month_of_financial_year)
 
         self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
 
@@ -48,13 +54,13 @@ class GCHR:
 
     @property
     def debug_file(self) -> str:
-        return f"{self.logs_dir}/debug_{self.timestamp}.csv"
+        return f"{self.logs_dir}\\debug_{self.timestamp}.csv"
 
     @property
     def account_ignored(self) -> str:
-        return f"{self.export_info_dir}/ignoriert_{self.period}.csv"
+        return f"{self.export_info_dir}\\ignoriert_{self.bookings.period}.csv"
 
-    # self.account_invalid = f"{self.export_info_dir}/ungueltig_{self.period}.csv"
+    # self.account_invalid = f"{self.export_info_dir}\\ungueltig_{self.period}.csv"
 
     def export_all_periods(self, overwrite=False, today=None) -> None:
         periods = GCHR.get_all_periods(today)
@@ -77,13 +83,13 @@ class GCHR:
         return periods
 
     def export_period(self, year: str, month: str) -> str:
-        export_fn = get_export_fn(GchrExportFormat.SKR51)
+        export_fn = get_export_fn(self.cfg.export_format)
         # Kontensalden laden
         df_bookings = self.bookings.filter_bookings(year, month)
         all_periods = set(df_bookings["Bookkeep Period"].to_list())
         bookkeep_period_date = datetime(int(year), int(month), 28)
 
-        if df_bookings.shape[0] == 0 or len(all_periods) <= 1 or self.booking_date < bookkeep_period_date:
+        if df_bookings.shape[0] == 0 or len(all_periods) <= 1 or self.bookings.booking_date < bookkeep_period_date:
             logging.error("ABBRUCH!!! Keine Daten vorhanden!")
             return False
 
@@ -92,7 +98,7 @@ class GCHR:
 
         logging.info("df_bookings: " + str(df_bookings.shape))
         # Join auf Übersetzung
-        df_combined = df_bookings.merge(self._df_translate, how="inner", on="Konto_Nr_Händler")
+        df_combined = df_bookings.merge(self.df_translate, how="inner", on="Konto_Nr_Händler")
         logging.info(f"df_combined: {df_combined.shape}")
 
         df_pivot = df_combined.pivot_table(
@@ -141,16 +147,17 @@ class GCHR:
             if i > 0:
                 filename = f"{filename[:-4]}_{main_site}.xml"
             export_cfg = GchrExportConfig(
-                main_site,
-                year,
-                month,
-                makes_used,
-                sites_used,
-                self.first_month_of_financial_year,
-                period_no,
-                self.bookings.bookkeep_filter,
-                filename,
-                df.to_dict(orient="records"),
+                main_site=main_site,
+                current_year=year,
+                current_month=month,
+                makes_used=makes_used,
+                sites_used=sites_used,
+                first_month=self.cfg.first_month_of_financial_year,
+                period_no=period_no,
+                bookkeep_filter=self.bookings.bookkeep_filter,
+                extraction_date=self.bookings.booking_date,
+                export_file=filename,
+                bookkeep_records=df.to_dict(orient="records"),
             )
 
             export_fn(export_cfg)
@@ -180,25 +187,15 @@ class GCHR:
 
     @property
     def export_info_dir(self) -> str:
-        return f"{self.base_dir}/Export/{self.current_year}/info/"
+        return f"{self.cfg.export_dir}\\{self.bookings.current_year}\\info\\"
 
     @property
     def logs_dir(self) -> str:
-        return f"{self.base_dir}/Logs/"
+        return f"{self.base_dir}\\Logs\\"
 
     @property
     def export_invalid_filename(self) -> str:
-        return f"{self.base_dir}/Export/ungueltig.csv"
+        return f"{self.cfg.export_dir}\\ungueltig.csv"
 
     def export_filename_for_period(self, year: str, month: str) -> str:
-        return f"{self.base_dir}/Export/{year}/export_{year}-{month}.xml"
-
-
-def gchr_local() -> None:
-    base_dir = os.getcwd() + "/../GCHR2_Testdaten/Kunden"
-    for path in Path(base_dir).glob("*"):
-        if not path.is_dir():
-            continue
-        print(path.name)
-        gchr = GCHR(str(path))
-        gchr.export_all_periods()
+        return f"{self.cfg.export_dir}\\{year}\\export_{year}-{month}.xml"

gchr/gchr_bookings.py

@@ -1,4 +1,5 @@
 import logging
+import os
 from datetime import datetime
 from pathlib import Path
 
@@ -42,7 +43,7 @@ class GchrBookings:
         self.current_year = year
         self.current_month = month
         self.period = f"{year}-{month}"
-        prot_file = f"{self.export_info_dir}/protokoll_{self.period}.log"
+        prot_file = f"{self.export_info_dir}\\protokoll_{self.period}.log"
         logging.basicConfig(
             filename=prot_file,
             filemode="w",
@@ -139,4 +140,6 @@ class GchrBookings:
 
     @property
     def export_info_dir(self) -> str:
-        return f"{self.base_dir}/Export/{self.current_year}/info/"
+        info_dir = f"{self.base_dir}\\Export\\{self.current_year}\\info\\"
+        os.makedirs(info_dir, exist_ok=True)
+        return info_dir

gchr/gchr_export.py

@@ -27,5 +27,6 @@ def export_dummy(export_cfg: GchrExportConfig) -> None:
     pass
 
 
-def get_export_fn(export_format: GchrExportFormat) -> GchrExportFn:
-    return EXPORT_FN.get(export_format, export_dummy)
+def get_export_fn(export_format: str) -> GchrExportFn:
+    export_format_enum = GchrExportFormat[export_format]
+    return EXPORT_FN.get(export_format_enum, export_dummy)

gchr/gchr_export_volkswagen.py

@@ -1,53 +1,90 @@
+import csv
 import xml.etree.ElementTree as ET
+from pathlib import Path
 from xml.dom import minidom
 
 from gchr.gchr_model import GchrExportConfig
 
+# <summary>
+#  V = Volkswagen,
+#  A = Audi,
+#  S = Seat,
+#  C = Skoda,
+#  E = Bentley,
+#  L = Lamborghini
+#  </summary>
+# internal enum Hauptmarke { V, A, S, C, E, L };
+
+MODEL_CODE = {
+    "01": "0600",
+    "02": "0603",
+}
+
 
 def export_volkswagen_xml(export_cfg: GchrExportConfig):
     header = {
-        "PartnerKey": {
-            "Country": "DEU",
-            "Brand": "V",
-            "PartnerNumber": "21996",
+        "tns:PartnerKey": {
+            "tns:Country": "DEU",
+            "tns:Brand": "V",
+            "tns:PartnerNumber": "21996",
         },
-        "IsCumulative": "true",
-        "AccountingDate": {
-            "AccountingMonth": "09",
-            "AccountingYear": "2024",
+        "tns:IsCumulative": "true",
+        "tns:AccountingDate": {
+            "tns:AccountingMonth": export_cfg.current_month,
+            "tns:AccountingYear": export_cfg.current_year,
         },
-        "Currency": "EUR",
-        "Level": "1",
+        "tns:Currency": "EUR",
+        "tns:Level": "1",
     }
-    record_elements = {"ProfitCenter": "00", "AccountKey": "010600002700000007000", "AccountValue": "+11.00"}
 
-    ET.register_namespace("tns", "http://xmldefs.volkswagenag.com/Retail/AccountBalanceDTS/V1")
-    root = ET.Element("tns:ShowAccountBalance")
+    records = []
+    for row in sorted(export_cfg.bookkeep_records, key=lambda x: x["Account"] + x["CostAccountingString"]):
+        val = "{0:.2f}".format(row["CumulatedYear"] * -1)
+        if val[0] != "-":
+            val = "+" + val
+        records.append(
+            {
+                "tns:ProfitCenter": "00",
+                "tns:AccountKey": account_number(row),
+                "tns:AccountValue": val,
+            }
+        )
+
+    add_values = (
+        Path(export_cfg.export_file).parent.parent.parent
+        / "data"
+        / f"Manuelle_Eingabe_{export_cfg.current_year}-{export_cfg.current_month}.csv"
+    )
+    if add_values.exists():
+        with add_values.open("r", encoding="latin-1") as frh:
+            csv_frh = csv.DictReader(frh, delimiter=";")
+            for row in csv_frh:
+                records.append(
+                    {
+                        "tns:ProfitCenter": "00",
+                        "tns:AccountKey": row["Kontonummer"][1:],
+                        "tns:AccountValue": "+{0:.2f}".format(float(row["Akt.Monat"].replace(",", "."))),
+                    }
+                )
+
+    nsmap = {"xmlns:tns": "http://xmldefs.volkswagenag.com/Retail/AccountBalanceDTS/V1"}
+    root = ET.Element("tns:ShowAccountBalance", nsmap)
     root = dict_to_xml(root, header)
 
-    record_list = ET.SubElement(root, "Accounts")
-    for row in export_cfg.bookkeep_records:
-        record = ET.SubElement(record_list, "Account")
-        for e, v in record_elements.items():
-            child = ET.SubElement(record, e)
-            field = row.get(e, v)
-            if str(field) == "nan":
-                field = "0"
-            elif type(field) is float:
-                field = "{:.2f}".format(field * -1)
-            child.text = str(field)
+    record_list = ET.SubElement(root, "tns:Accounts")
+    for r in records:
+        dict_to_xml(ET.SubElement(record_list, "tns:Account"), r)
 
+    raw_xml = ET.tostring(root, encoding="latin-1")
     with open(export_cfg.export_file, "w", encoding="latin-1") as fwh:
-        fwh.write(
-            minidom.parseString(ET.tostring(root, encoding="ISO-8859-1")).toprettyxml(indent="  ", encoding="latin-1")
-        )
+        fwh.write(minidom.parseString(raw_xml).toprettyxml(indent="  "))
 
 
 def dict_to_xml(root: ET.Element, subtree: dict):
-    # if isinstance(subtree, list):
-    #     for item in subtree:
-    #         dict_to_xml(root, item)
-    #     return root
+    if isinstance(subtree, list):
+        for item in subtree:
+            dict_to_xml(root, item)
+        return root
 
     for key, value in subtree.items():
         e = ET.SubElement(root, key)
@@ -56,3 +93,17 @@ def dict_to_xml(root: ET.Element, subtree: dict):
         else:
             e.text = str(value)
     return root
+
+
+def account_number(row: dict[str, str]) -> str:
+    res = {
+        "Brand": row["Make"],
+        "ModelCode": MODEL_CODE.get(row["Make"], "0000"),
+        "Account": row["Account"],
+        "CostCentre": row["Origin"],
+        "TradeChannel": row["SalesChannel"],
+        "CostUnit": row["CostCarrier"],
+        "Location": row["Site"],
+        "TaxCode": "000",
+    }
+    return "".join(res.values())

gchr/gchr_model.py

@@ -25,13 +25,13 @@ class GchrExportConfig:
     bookkeep_filter: dict[str, str]
     extraction_date: datetime
     export_file: str
-    bookkeep_records = dict[str, list[str]]
+    bookkeep_records: dict[str, list[str]]
 
 
 @dataclass
 class GchrConfig:
     first_month_of_financial_year: str = "01"
-    data_dir: str = os.curdir + "\\data"
-    gcstruct_dir: str = os.curdir + "\\..\\GCStruct_Aufbereitung"
-    export_dir: str = os.curdir + "\\export"
+    data_dir: str = os.getcwd() + "\\data"
+    gcstruct_dir: str = os.getcwd() + "\\..\\GCStruct_Aufbereitung"
+    export_dir: str = os.getcwd() + "\\Export"
     export_format: str = "SKR51"

gchr2.spec

@@ -1,9 +1,6 @@
 # -*- mode: python ; coding: utf-8 -*-
 
 
-block_cipher = None
-
-
 a = Analysis(
     ['gchr2.py'],
     pathex=['C:\\Projekte\\Python\\'],
@@ -14,18 +11,15 @@ a = Analysis(
     hooksconfig={},
     runtime_hooks=[],
     excludes=[],
-    win_no_prefer_redirects=False,
-    win_private_assemblies=False,
-    cipher=block_cipher,
     noarchive=False,
+    optimize=0,
 )
-pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
+pyz = PYZ(a.pure)
 
 exe = EXE(
     pyz,
     a.scripts,
     a.binaries,
-    a.zipfiles,
     a.datas,
     [],
     name='gchr2',

gchr2_local.py

@@ -0,0 +1,18 @@
+import os
+from pathlib import Path
+
+from gchr.gchr import GCHR
+
+
+def gchr_local() -> None:
+    base_dir = os.getcwd() + "\\..\\GCHR2_Testdaten\\Kunden"
+    for path in Path(base_dir).glob("*"):
+        if not path.is_dir():
+            continue
+        print(path.name)
+        gchr = GCHR(str(path))
+        gchr.export_all_periods()
+
+
+if __name__ == "__main__":
+    gchr_local()

sandbox/access.py

@@ -0,0 +1,30 @@
+import codecs
+import sys
+
+
+def crack_access_mdb(file):
+    no_pass_62 = "0C"
+    no_pass_42 = "BE68EC3765D79CFAFECD28E62B258A606C077B36CDE1DFB14F671343F73C"
+
+    with open(file, "rb") as f:
+        f.seek(66, 0)  # x42 == 66
+        myfile_42 = f.read(30)
+        f.seek(98)  # x62 == 98
+        myfile_62 = f.read(1)
+
+    salt = ord(codecs.decode(no_pass_62, "hex")) ^ ord(myfile_62)
+
+    add_salt = True
+    word = ""
+    for i in range(0, 52, 4):
+        xored = ord(codecs.decode(no_pass_42[i : i + 2], "hex")) ^ myfile_42[i // 2]
+        if add_salt:
+            xored = xored ^ salt
+        word = word + chr(xored)
+        add_salt = not add_salt
+    print(word)
+
+
+if __name__ == "__main__":
+    file = sys.argv[1]
+    crack_access_mdb(file)