Python-Projekte/datev/export_extf.py

import calendar
import csv
import hashlib
import os
import re
import shutil
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Generator, Literal

import pyodbc

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


class DatevConfig:
    base_dir: str = str(Path(__file__).resolve().parent)
    data_path: str = base_dir + "/data"
    export_path: str = base_dir + "/export/temp"
    translation_file: str = data_path + "/uebersetzungstabelle.csv"
    csv_date: datetime = datetime.now()  # datetime(2023, 11, 20, 19, 2, 28, 714000)
    geschaeftsjahr_monat: int = 1
    periode: str = "202301"
    berater: int = 30612
    mandant: int = 10139
    konto_laenge: int = 5

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

    @property
    def datum_bis(self) -> datetime:
        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 geschaeftsjahr_beginn(self) -> datetime:
        year = int(self.periode[:4])
        if self.geschaeftsjahr_monat > datetime.now().month:
            year -= 1
        return datetime(year, self.geschaeftsjahr_monat, 1)

    @property
    def header2(self) -> str:
        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}";"";;"";;"";;;;;;"";"";"";"";"";"";"";"";"";"";"";"";"";'
        + '"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";"";;;;"";;;;"";"";;"";;'
        + ';;"";"";;"";;"";;"";"";;"";;0;;'
    )
    # '592.80;H;EUR;"15800";90900;0101;6288;Opel Bank VoST 12/22  Lagerwag;1'

    @property
    def export_file(self) -> str:
        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) -> str:
        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) -> dict[str, str]:
    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) -> Generator[list[str], Any, None]:
    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) -> Generator[list[str], Any, None]:
    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


def export_extf(period: str, import_method: Literal["csv", "db"] = "csv") -> None:
    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() -> None:
    dt = datetime.now()
    prev = str(dt.year - 1)
    periods = [f"{prev}{x:02}" for x in range(1, 13)] + [f"{dt.year}{x:02}" for x in range(1, dt.month + 1)]

    for p in periods:
        export_extf(p, "db")


def extf_get_hash(filename):
    with open(filename, "r", encoding="latin-1") as frh1:
        frh1.readline()  # ignore header
        data = frh1.read()
        return calculate_sha256(data)


def extf_files_equal_content(file1, file2):
    with open(file1, "r", encoding="latin-1") as frh1:
        frh1.readline()  # ignore header
        data1 = frh1.read()

    with open(file2, "r", encoding="latin-1") as frh2:
        frh2.readline()  # ignore header
        data2 = frh2.read()

    print(calculate_sha256(data1))
    print(calculate_sha256(data2))

    return calculate_sha256(data1) == calculate_sha256(data2)


def calculate_sha256(data) -> str:
    return hashlib.sha256(data.encode()).hexdigest()


def test_content():
    print(
        extf_files_equal_content(
            "datev/export/EXTF_Buchungsstapel_30612_10139_202312_20240514_112734.csv",
            "datev/export/EXTF_Buchungsstapel_30612_10139_202312_20240514_112734.csv",
        )
    )
    print(
        extf_files_equal_content(
            "datev/export/EXTF_Buchungsstapel_30612_10139_202312_20240222_155629.csv",
            "datev/export/EXTF_Buchungsstapel_30612_10139_202312_20240514_112734.csv",
        )
    )
    print(
        extf_files_equal_content(
            "datev/export/EXTF_Buchungsstapel_30612_10139_202312_20240514_112734.csv",
            "datev/export/EXTF_Buchungsstapel_30612_10139_202312_20240515_104021.csv",
        )
    )


def archive_files():
    last_week = (datetime.now() - timedelta(days=6)).timestamp()
    for file in Path("datev/export").glob("*.csv"):
        if file.stat().st_ctime < last_week:
            file.unlink()

    archive_path = Path("datev/export/Archiv")
    for file in Path("datev/export/temp").glob("*.csv"):
        p = re.search(r"_(\d{6})_", file.name)
        if not p:
            continue
        period = p[1]
        target = archive_path / period[:4] / period
        os.makedirs(target, exist_ok=True)
        file_hash = extf_get_hash(file)

        if has_identical_file(target, file_hash):
            file.unlink()
            continue
        shutil.copy(file, archive_path.parent / file.name)
        file.rename(target / file.name)


def has_identical_file(target: Path, file_hash: str) -> bool:
    for archived_file in Path(target).glob("*.csv"):
        if extf_get_hash(archived_file) == file_hash:
            return True
    return False


if __name__ == "__main__":
    export_all_periods()
    archive_files()
    # os.makedirs(Path(filename).parent.joinpath("info"), exist_ok=True)