Datev mit Datei-Archivierung

datev/export_extf.py

@@ -1,11 +1,14 @@
-from datetime import datetime
 import calendar
 import csv
 import hashlib
-from typing import Any, Generator, Literal
-import pyodbc
+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"
 
@@ -13,7 +16,7 @@ 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"
+    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
@@ -280,9 +283,15 @@ def export_all_periods() -> None:
     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 year, month in periods:
-        period = f"{year}{month}"
-        export_extf(period, "db")
+    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):
@@ -304,8 +313,7 @@ def calculate_sha256(data) -> str:
     return hashlib.sha256(data.encode()).hexdigest()
 
 
-if __name__ == "__main__":
-    # export_all_periods()
+def test_content():
     print(
         extf_files_equal_content(
             "datev/export/EXTF_Buchungsstapel_30612_10139_202312_20240514_112734.csv",
@@ -325,4 +333,38 @@ if __name__ == "__main__":
         )
     )
 
+
+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)

datev/onedrive.py

@@ -1,28 +1,80 @@
+from pathlib import Path
+
 from O365 import Account
+from O365.drive import Folder
+
+CLIENT_ID = "925f74dc-f96a-4718-9ca7-d6cc3fa43e1e"
+CLIENT_SECRET = "SS~8Q~QpBZV9toZuwkzW1XGGen2Hn833spNMtdq5"
+FLOW_TYPE = "credentials"
+TENANT_ID = "2ad0dff5-07ce-4cc2-a852-99ce8b91c218"
+
+USER_ID = "bedner@global-cube.net"
+# password = "Platinum512!!"
+
+
+def get_remote_files(folder: Folder, prefix=""):
+    res = {}
+    for item in folder.get_items():
+        name = item.name
+        if prefix != "":
+            name = prefix + "\\" + name
+
+        if item.is_file:
+            res[name] = item.object_id
+        if item.is_folder:
+            res.update(get_remote_files(item, name))
+    return res
+
+
+def get_or_create_subfolder(remote_folder: Folder, folder: str):
+    if folder == "":
+        return remote_folder
+
+    if "\\" in folder:
+        folder, subfolder = folder.split("\\", 1)
+    else:
+        subfolder = ""
+
+    for f in remote_folder.get_child_folders():
+        if f.name == folder:
+            return get_or_create_subfolder(f, subfolder)
+    new_folder = remote_folder.create_child_folder(folder)
+    return get_or_create_subfolder(new_folder, subfolder)
+
+
+def main():
+    account = Account((CLIENT_ID, CLIENT_SECRET), auth_flow_type=FLOW_TYPE, tenant_id=TENANT_ID)
+    account.authenticate()
+
+    storage = account.storage(resource=USER_ID)
+    my_drive = storage.get_default_drive()
+    remote_folder = my_drive.get_item_by_path("/Keyloop_Datentransfer")
 
+    # folder.create_child_folder
+    local_folder = Path("datev/export")
 
-client_id = "925f74dc-f96a-4718-9ca7-d6cc3fa43e1e"
-client_secret = "SS~8Q~QpBZV9toZuwkzW1XGGen2Hn833spNMtdq5"
-flow_type = "credentials"
+    local_files = set([str(p.relative_to(local_folder)) for p in local_folder.rglob("*") if p.is_file()])
+    # uploaded_file = remote_folder.upload_file(item="tox.ini")
 
-user_id = "bedner@global-cube.net"
-password = "Platinum512!!"
+    remote_files_dict = get_remote_files(remote_folder)
+    remote_files = set(remote_files_dict.keys())
 
-account = Account(
-    (client_id, client_secret), auth_flow_type="credentials", tenant_id="2ad0dff5-07ce-4cc2-a852-99ce8b91c218"
-)
+    delete_candidates = remote_files.difference(local_files)
+    create_candidates = local_files.difference(remote_files)
 
-account.authenticate()
+    for filename in delete_candidates:
+        print(filename)
+        my_drive.get_item(remote_files_dict[filename]).delete()
 
-storage = account.storage(resource=user_id)
-my_drive = storage.get_default_drive()
-folder = my_drive.get_item_by_path("/Keyloop_Datentransfer")
-archive = my_drive.get_item_by_path("/Keyloop_Datentransfer/Archiv/2023")
-print(archive.name)
+    for filename in sorted(create_candidates):
+        print(filename)
+        full_filename = local_folder / filename
+        subfolder = ""
+        if "\\" in filename:
+            subfolder, filename = filename.rsplit("\\", 1)
+        remote_subfolder = get_or_create_subfolder(remote_folder, subfolder)
+        remote_subfolder.upload_file(item=full_filename)
 
-# folder.create_child_folder
-uploaded_file = folder.upload_file(item="tox.ini")
 
-for item in folder.get_items():
-    print(item.name)
-    print("--> ", item.object_id)
+if __name__ == "__main__":
+    main()