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)