gctools/database/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")