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 previous in Path(target).glob("*.csv"):
if extf_get_hash(previous) == file_hash:
return True
return False
if __name__ == "__main__":
export_all_periods()
archive_files()
# os.makedirs(Path(filename).parent.joinpath("info"), exist_ok=True)