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