datev_status.py 2.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970
  1. import json
  2. from datetime import datetime
  3. from pathlib import Path
  4. import pyodbc
  5. pyodbc.lowercase = False
  6. def run_query(cur: pyodbc.Cursor, query: str) -> list[dict[str, str]]:
  7. cur.execute(query)
  8. col_names = [x[0] for x in cur.description]
  9. res = []
  10. while True:
  11. row = cur.fetchone()
  12. if row is None:
  13. return res
  14. r = {}
  15. for i, col in enumerate(row):
  16. r[col_names[i]] = str(col).strip()
  17. res.append(r)
  18. def print_mdb_details(mdb_file: str):
  19. print("--")
  20. print("[" + mdb_file + "]")
  21. ts = datetime.fromtimestamp(Path(mdb_file).stat().st_mtime)
  22. res = {"Datum_MDB_Datei": ts.strftime("%Y-%m-%d %H:%M:%S")}
  23. try:
  24. conn = pyodbc.connect("Driver={Microsoft Access Driver (*.mdb, *.accdb)};Dbq=" + mdb_file + ";")
  25. cur = conn.cursor()
  26. except pyodbc.Error:
  27. print("Error connecting to file")
  28. return
  29. date_range = "SELECT min(dtDatum) as Datum_min, max(dtDatum) as Datum_max, min(dtWJ) as Wirtschaftsjahr FROM tblKNW"
  30. res1 = run_query(cur, date_range)
  31. res.update(res1[0])
  32. main_site = (
  33. "SELECT FIRMAI as Firma_Nr, FIRMAS as Firma_Name, BERATER as Berater_Nr, "
  34. "MANDANT as Mandant_Nr, dtWJAnzeige as WJ_Anzeige FROM ORDDAT"
  35. )
  36. res2 = run_query(cur, main_site)
  37. res.update(res2[0])
  38. sites = "SELECT KSTESA_NUMMER as Standort_Nr, KSTESA_BEZEICHNUNG as Standort_Name FROM tblKS_KSTESA WHERE KSTESA_OWN_ID = 2"
  39. res3 = run_query(cur, sites)
  40. res["Standorte"] = res3
  41. print(json.dumps(res, indent=2))
  42. client_db = res.get("Mandant_Nr", "1")
  43. year_id = res.get("Wirtschaftsjahr", "00000101")
  44. new_filename = str(Path(mdb_file).parent) + f"\\KostSt_{client_db}_{year_id}.mdb"
  45. with open(new_filename + ".json", "w") as fwh:
  46. json.dump(res, fwh, indent=2)
  47. cur.close()
  48. conn.close()
  49. def print_all_mdb_files(folder: str):
  50. for mdb_file in Path(folder).resolve().rglob("KostSt.mdb"):
  51. print_mdb_details(str(mdb_file))
  52. if __name__ == "__main__":
  53. print_all_mdb_files("C:\\Projekte\\Datev")