import plac from pathlib import Path import csv import re import pandas as pd from sqlalchemy import create_engine, inspect import json csv_dir = Path('C:\\GlobalCube\\System\\AUTOLINE\\Datenbank\\Full_zusammengesetllt') target_dsn = {'user': 'sa', 'pass': 'Mffu3011#', 'server': 'GC-SERVER1\\GLOBALCUBE', 'database': 'AUTOLINE'} temp_schema = 'temp' target_schema = 'import' transform = [] def get_dtype(db_type): if db_type == "DATETIME": return "datetime64" if db_type == "DECIMAL(28, 8)" or db_type == "DECIMAL(18, 0)" or db_type == "NUMERIC(18, 0)": return "float64" return "object" def conn_string(dsn): return f"mssql+pyodbc://{dsn['user']}:{dsn['pass']}@{dsn['server']}/{dsn['database']}?driver=SQL+Server+Native+Client+11.0" def conn_params(dsn): return f"-S {dsn['server']} -d {dsn['database']} -U {dsn['user']} -P {dsn['pass']}" def columns_from_csv(source_csv): df = pd.read_csv(source_csv, sep=",", encoding="utf8", decimal=".", nrows=1) source_columns_list = list(df.columns) for i, col in enumerate([col for col in source_columns_list if col[-2:] == ".1"]): source_columns_list[i] = col[:-2] + "2" source_columns_list = [col.replace(".", " ") for col in source_columns_list] return source_columns_list def transform_template(target_insp, source_csv, table, target_schema): target_insp_cols = target_insp.get_columns(table, schema=target_schema) target_columns_list = [col['name'] for col in target_insp_cols] source_columns_list = columns_from_csv(source_csv) target_columns = set(target_columns_list) source_columns = set(source_columns_list) # intersect = source_columns.intersection(target_columns) # print("Auf beiden Seiten: " + ";".join(intersect)) diff1 = source_columns.difference(target_columns) if len(diff1) > 0: print("rem Nur in Quelle: " + ";".join(diff1)) diff2 = target_columns.difference(source_columns) if len(diff2) > 0: print("rem Nur in Ziel: " + ";".join(diff2)) template = [] for i, col in enumerate(target_columns_list): if col in source_columns_list: pos = source_columns_list.index(col) else: pos = -1 template.append((pos, get_dtype(str(target_insp_cols[i]['type'])))) return template def transform_line(line): pattern = re.compile(r"\d{4}-\d\d-\d\d \d\d:\d\d") pattern2 = re.compile(r"(\d{2})[/\.](\d{2})[/\.](\d{4})") result = [] for pos, f in transform: e = "" if pos > -1 and pos < len(line): e = line[pos] if f == "float64": e = e.replace(",", ".") if f == "datetime64": m = pattern2.match(e) if m: e = f"{m[3]}-{m[2]}-{m[1]}" elif not pattern.match(e): e = "" # e += ":00.000" result.append(e) return result def fix_nulls(s): for line in s: yield line.replace('\0', ' ') def transform_file(source_csv, template): global transform transform = json.loads(template) stage_csv = Path(f"{source_csv.parent}\\stage\\{source_csv.name}") if stage_csv.exists() and stage_csv.stat().st_ctime > source_csv.stat().st_ctime: print(f"Stage-CSV '{stage_csv.name}' ist bereits aktuell.") return False print(f"Importiere {source_csv.name}...") with open(source_csv, "r", encoding="utf-8", errors="ignore", newline="") as source_file, open(stage_csv, "w", encoding="utf-8", newline="") as target_file: csv_read = csv.reader(fix_nulls(source_file), delimiter=",") csv_write = csv.writer(target_file, delimiter="\t") next(csv_read) # ignore header i = 0 for cols in csv_read: csv_write.writerow(transform_line(cols)) i += 1 print(f"...{i} Zeilen konvertiert.") def csv_tables(csv_dir, target_tables_ci): p = re.compile(r"_\d+$") result = [] if not csv_dir.is_dir(): print(f"Verzeichnis {csv_dir} existiert nicht!") return result for source_csv in csv_dir.glob("*.csv"): if source_csv.is_dir(): continue table = source_csv.name[:-4].lower() if table not in target_tables_ci: table = p.sub("", table) if table not in target_tables_ci: print(f"rem Ziel-Tabelle '{table}' existiert nicht!") continue result.append((table, source_csv)) return result def target_tables(target_dsn, target_schema): engine = create_engine(conn_string(target_dsn)) target_insp = inspect(engine) target_tables = target_insp.get_table_names(schema=target_schema) return (target_insp, list(map(str.lower, target_tables))) def batch(csv_dir, action): target_insp, target_tables_ci = target_tables(target_dsn, target_schema) stage_schema = target_schema if action == "overwrite" else temp_schema print("@echo off") print("cd /d %~dp0") print("set PYTHON=\"C:\\dev\\Python\\Python38-32\"") for (table, source_csv) in csv_tables(csv_dir, target_tables_ci): print(f"echo =={table}==") stage_csv = Path(f"{source_csv.parent}\\stage\\{source_csv.name}") try: tf_template = transform_template(target_insp, source_csv, table, target_schema) template_json = json.dumps(tf_template).replace("\"", "\\\"") print(f"sqlcmd.exe {conn_params(target_dsn)} -p -Q \"TRUNCATE TABLE [{stage_schema}].[{table}]\" ") print(f"%PYTHON%\\python.exe csv_import.py transform \"{source_csv}\" -t \"{template_json}\" ") print(f"bcp.exe [{stage_schema}].[{table}] in \"{stage_csv}\" {conn_params(target_dsn)} -c -C 65001 -e \"{stage_csv}.log\" ") pkeys = target_insp.get_pk_constraint(table, schema=target_schema) if len(pkeys['constrained_columns']) > 0: delete_sql = f"DELETE T1 FROM [{target_schema}].[{table}] T1 INNER JOIN [{temp_schema}].[{table}] T2 ON " + " AND ".join([f"T1.[{col}] = T2.[{col}]" for col in pkeys['constrained_columns']]) print(f"sqlcmd.exe {conn_params(target_dsn)} -p -Q \"{delete_sql}\" ") insert_sql = f"INSERT INTO [{target_schema}].[{table}] SELECT * FROM [{temp_schema}].[{table}]" print(f"sqlcmd.exe {conn_params(target_dsn)} -p -Q \"{insert_sql}\" ") print("") except Exception: print(f"rem {source_csv} fehlerhaft!") @plac.pos('action', "", choices=['batch', 'transform']) @plac.pos('csv_dir', "", type=Path) @plac.opt('mode', "", choices=['overwrite', 'append', 'update']) @plac.opt('template', "") def main(action, csv_dir, mode="overwrite", template="[]"): if action == "transform": transform_file(csv_dir, template) else: batch(csv_dir, mode) if __name__ == '__main__': plac.call(main) # main("batch", csv_dir, "append")