Python-Projekte/dbtools/csv_import.py

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="utf-8", 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")