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(2023, 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 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()