Python-Projekte/datev/export_extf.py

from datetime import datetime
import calendar
import csv
import pyodbc
from pathlib import Path


DSN = "dsn=GC_OPTIMA_64;uid=gaps;pwd=Gcbs12ma"


class DatevConfig:
    data_path: str = "datev/data"
    export_path: str = "datev/export"
    translation_file: str = "datev/data/uebersetzungstabelle.csv"
    csv_date: datetime = datetime.now()  # datetime(2023, 11, 20, 19, 2, 28, 714000)
    geschaeftsjahr_beginn: datetime = datetime(2024, 1, 1)
    periode: str = "202301"
    berater: int = 30612
    mandant: int = 10139
    konto_laenge: int = 5

    @property
    def datum_von(self):
        return datetime(int(self.periode[:4]), int(self.periode[4:]), 1)

    @property
    def datum_bis(self):
        year = int(self.periode[:4])
        month = int(self.periode[4:])
        end_of_month = calendar.monthrange(year, month)[1]
        return datetime(year, month, end_of_month)

    @property
    def header2(self):
        res = [
            "Umsatz (ohne Soll/Haben-Kz)",
            "Soll/Haben-Kennzeichen",
            "WKZ Umsatz",
            "Kurs",
            "Basis-Umsatz",
            "WKZ Basis-Umsatz",
            "Konto",
            "Gegenkonto (ohne BU-Schlüssel)",
            "BU-Schlüssel",
            "Belegdatum",
            "Belegfeld 1",
            "Belegfeld 2",
            "Skonto",
            "Buchungstext",
            "Postensperre",
            "Diverse Adressnummer",
            "Geschäftspartnerbank",
            "Sachverhalt",
            "Zinssperre",
            "Beleglink",
            "Beleginfo - Art 1",
            "Beleginfo - Inhalt 1",
            "Beleginfo - Art 2",
            "Beleginfo - Inhalt 2",
            "Beleginfo - Art 3",
            "Beleginfo - Inhalt 3",
            "Beleginfo - Art 4",
            "Beleginfo - Inhalt 4",
            "Beleginfo - Art 5",
            "Beleginfo - Inhalt 5",
            "Beleginfo - Art 6",
            "Beleginfo - Inhalt 6",
            "Beleginfo - Art 7",
            "Beleginfo - Inhalt 7",
            "Beleginfo - Art 8",
            "Beleginfo - Inhalt 8",
            "KOST1 - Kostenstelle",
            "KOST2 - Kostenstelle",
            "Kost-Menge",
            "EU-Land u. UStID",
            "EU-Steuersatz",
            "Abw. Versteuerungsart",
            "Sachverhalt L+L",
            "Funktionsergänzung L+L",
            "BU 49 Hauptfunktionstyp",
            "BU 49 Hauptfunktionsnummer",
            "BU 49 Funktionsergänzung",
            "Zusatzinformation - Art 1",
            "Zusatzinformation- Inhalt 1",
            "Zusatzinformation - Art 2",
            "Zusatzinformation- Inhalt 2",
            "Zusatzinformation - Art 3",
            "Zusatzinformation- Inhalt 3",
            "Zusatzinformation - Art 4",
            "Zusatzinformation- Inhalt 4",
            "Zusatzinformation - Art 5",
            "Zusatzinformation- Inhalt 5",
            "Zusatzinformation - Art 6",
            "Zusatzinformation- Inhalt 6",
            "Zusatzinformation - Art 7",
            "Zusatzinformation- Inhalt 7",
            "Zusatzinformation - Art 8",
            "Zusatzinformation- Inhalt 8",
            "Zusatzinformation - Art 9",
            "Zusatzinformation- Inhalt 9",
            "Zusatzinformation - Art 10",
            "Zusatzinformation- Inhalt 10",
            "Zusatzinformation - Art 11",
            "Zusatzinformation- Inhalt 11",
            "Zusatzinformation - Art 12",
            "Zusatzinformation- Inhalt 12",
            "Zusatzinformation - Art 13",
            "Zusatzinformation- Inhalt 13",
            "Zusatzinformation - Art 14",
            "Zusatzinformation- Inhalt 14",
            "Zusatzinformation - Art 15",
            "Zusatzinformation- Inhalt 15",
            "Zusatzinformation - Art 16",
            "Zusatzinformation- Inhalt 16",
            "Zusatzinformation - Art 17",
            "Zusatzinformation- Inhalt 17",
            "Zusatzinformation - Art 18",
            "Zusatzinformation- Inhalt 18",
            "Zusatzinformation - Art 19",
            "Zusatzinformation- Inhalt 19",
            "Zusatzinformation - Art 20",
            "Zusatzinformation- Inhalt 20",
            "Stück",
            "Gewicht",
            "Zahlweise",
            "Forderungsart",
            "Veranlagungsjahr",
            "Zugeordnete Fälligkeit",
            "Skontotyp",
            "Auftragsnummer",
            "Buchungstyp",
            "Ust-Schlüssel (Anzahlungen)",
            "EU-Land (Anzahlungen)",
            "Sachverhalt L+L (Anzahlungen)",
            "EU-Steuersatz (Anzahlungen)",
            "Erlöskonto (Anzahlungen)",
            "Herkunft-Kz",
            "Leerfeld",
            "KOST-Datum",
            "Mandatsreferenz",
            "Skontosperre",
            "Gesellschaftername",
            "Beteiligtennummer",
            "Identifikationsnummer",
            "Zeichnernummer",
            "Postensperre bis",
            "Bezeichnung SoBil-Sachverhalt",
            "Kennzeichen SoBil-Buchung",
            "Festschreibung",
            "Leistungsdatum",
            "Datum Zuord.Steuerperiode",
        ]
        return ";".join(res)

    row_template = (
        '{0};"{1}";"{2}";;;"";"{9}";"{4}";"";{5};"{6}";"";;"{7}";;"";;;;"";"";'
        + '"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"{10}";"";;"";;"";;;;;;"";"";"";"";"";"";"";"";"";"";"";"";"";'
        + '"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";;;;"";;;;"";"";;"";;'
        + ';;"";"";;"";;"";;"";"";;"";;{8};;'
    )
    # '592.80;H;EUR;"15800";90900;0101;6288;Opel Bank VoST 12/22  Lagerwag;1'

    @property
    def export_file(self):
        timestamp = self.csv_date.strftime("%Y%m%d_%H%M%S")
        period = self.datum_von.strftime("%Y%m")
        return f"{self.export_path}/EXTF_Buchungsstapel_30612_10139_{period}_{timestamp}.csv"

    @property
    def header(self):
        datev_header = {
            "Datev-Format-KZ": "EXTF",
            "Versionsnummer": 510,
            "Datenkategorie": 21,
            "Formatname": "Buchungsstapel",
            "Formatversion": 7,
            "Erzeugt_am": self.csv_date.strftime("%Y%m%d%H%M%S%f")[:-3],
            "Importiert_am": "",
            "Herkunftskennzeichen": "SV",
            "Exportiert_von": "dracar",
            "Importiert_von": "",
            "Berater": self.berater,
            "Mandant": self.mandant,
            "WJ-Beginn": self.geschaeftsjahr_beginn.strftime("%Y%m%d"),
            "Sachkontenlänge": self.konto_laenge,
            "Datum_von": self.datum_von.strftime("%Y%m%d"),
            "Datum_bis": self.datum_bis.strftime("%Y%m%d"),
            "Bezeichnung": "",
            "Diktatkürzel": "HE",
            "Buchungstyp": 1,
            "Rechnungslegungszweck": "",
            "Festschreibeinformation": 1,
            "WKZ": "",
            "reserviert_1": "",
            "Derivatskennzeichen": "",
            "reserviert_2": "",
            "reserviert_3": "",
            "SKR": "",
            "Branchenlösung-Id": "",
            "reserviert_4": "",
            "reserviert_5": "",
            "Anwendungsinformation": "",
        }
        template = (
            '"EXTF";{Versionsnummer};{Datenkategorie};"Buchungsstapel";{Formatversion};{Erzeugt_am};'
            + ';"SV";"dracar";"";{Berater};{Mandant};{WJ-Beginn};{Sachkontenlänge};{Datum_von};{Datum_bis};"";"HE";1;;1;"";;"";;;"";;;"";""'
        )
        return template.format(**datev_header)


def get_translation(cfg: DatevConfig):
    translation = {}
    with Path(cfg.translation_file).open("r", encoding="latin-1") as frh:
        for line in csv.reader(frh, delimiter=";"):
            acct_no = line[0][:4] + "0"
            acct_details = "11" + line[0][11:].replace("-", "")
            translation[line[2]] = (acct_no, acct_details)
    return translation


def from_database(period):
    with pyodbc.connect(DSN) as conn:
        cursor = conn.cursor()
        query = (
            "SELECT * FROM [import].[DATEV_Buchungsstapel] "
            + f"WHERE [BOOKKEEP_PERIOD] = '{period}' ORDER BY [BOOKKEEP_DATE], [UNIQUE_IDENT]"
        )
        cursor.execute(query)
        for row in cursor.fetchall():
            yield list(map(str, row[:9]))


def from_csv(import_file):
    with import_file.open("r", encoding="latin-1") as frh:
        csv_reader = csv.reader(frh, delimiter=";")
        next(csv_reader)  # ignore header
        for row in csv_reader:
            yield row[:9]


def export_extf(period, import_method="csv"):
    cfg = DatevConfig()
    cfg.periode = period
    translation = get_translation(cfg)

    if import_method == "csv":
        import_file = Path(f"datev/data/{period}.csv")
        cfg.csv_date = datetime.fromtimestamp(import_file.stat().st_mtime)
        get_row = from_csv(import_file)
    else:
        get_row = from_database(cfg.periode)

    missing = []

    with Path(cfg.export_file).open("w", encoding="latin-1", newline="") as fwh:
        fwh.write(cfg.header + "\r\n")
        fwh.write(cfg.header2 + "\r\n")

        for row in get_row:
            row[0] = row[0].replace(".", ",")
            row.extend(translation.get(row[3], (row[3], "11000000")))
            if row[9] == row[3]:
                missing.append(row[3])
            fwh.write(cfg.row_template.format(*row) + "\r\n")
    print(set(missing))


def export_all_periods():
    for period in range(202301, 202313):
        export_extf(str(period), "db")


if __name__ == "__main__":
    # export_all_periods()
    export_extf("202401", "csv")