Ausnahmen für Dateitypen, automatisches generieren der Batch-Dateien

c7.py

@@ -19,7 +19,11 @@ def iqd_convert():
 
 
 @app.command()
-def mdl_convert(mdl_file):
+def mdl_convert(mdl_file: str):
+    if not Path(mdl_file).exists():
+        raise typer.Exit(code=1, message=f"Datei {mdl_file} nicht gefunden")
+    if Path(mdl_file).is_dir():
+        return cognos7.convert_folder(mdl_file)
     cognos7.convert_file(mdl_file)
     source = Path(mdl_file[:-4] + ".json")
     dest = f"{cfg.cognos11.specs_dir}\\..\\DataModel\\{source.name}"
@@ -28,6 +32,11 @@ def mdl_convert(mdl_file):
     os.rename(source, dest)
 
 
+@app.command()
+def export_csv_sql():
+    cognos7.create_format_xml_from_folder(f"{cfg.system_dir}\\Export")
+
+
 @app.command()
 def move_csv():
     print("Verschiebe CSV-Dateien von IQD zu Export...")

cognos7/csv_column_types.py

@@ -9,9 +9,24 @@ data_types = {
 
 int_values = ["Menge", "Anzahl"]
 decimal_values = ["Betrag"]
+str_values = [
+    "Zipcode",
+    "PLZ_1_Stelle",
+    "PLZ_2_Stelle",
+    "PLZ_3_Stelle",
+    "PLZ_4_Stelle",
+    "PLZ",
+]
 
 
 def get_column_types(csv_file, header=None):
+    with open(csv_file, "r", encoding="latin-1") as frh:
+        header1 = frh.readline()
+        if header is not None:
+            header1 = header
+        columns = frh.readline().split(";")
+        column_types = {key: '"' in val for key, val in zip(header1, columns)}
+
     skip = 1 if header else 0
     df = pd.read_csv(
         csv_file,
@@ -23,7 +38,7 @@ def get_column_types(csv_file, header=None):
         skiprows=skip,
     )
 
-    col_types = dict([(c, get_type(df[c])) for c in df.columns])
+    col_types = dict([(c, get_type(df[c], column_types.get(c, False))) for c in df.columns])
     return col_types
 
 
@@ -35,26 +50,30 @@ def is_date(value: str):
     return re.match(r"\d{4}-\d{2}-\d{2}", value)
 
 
-def get_type(df_col: pd.Series) -> str:
-    if str(df_col.dtype) == "float64":
-        if all(df_col.isna()):
-            return "varchar(255)"
-        return "decimal(18,8)"
-
-    if str(df_col.dtype) == "int64":
-        for entry in decimal_values:
-            if entry in df_col.name:
-                return "decimal(18,8)"
-        for entry in int_values:
-            if entry in df_col.name:
-                return "int"
-        return "varchar(20)"
-
-    if all([is_datetime(str(value)) for value in df_col]):
-        return "datetime"
-
-    if all([is_date(str(value)) for value in df_col]):
-        return "date"
+def get_type(df_col: pd.Series, is_str: bool) -> str:
+    if not is_str:
+        if str(df_col.dtype) == "float64":
+            if all(df_col.isna()):
+                return "varchar(255)"
+            for entry in str_values:
+                if entry in df_col.name:
+                    return "varchar(20)"
+            return "decimal(18,8)"
+
+        if str(df_col.dtype) == "int64":
+            for entry in decimal_values:
+                if entry in df_col.name:
+                    return "decimal(18,8)"
+            for entry in int_values:
+                if entry in df_col.name:
+                    return "int"
+            return "varchar(20)"
+
+        if all([is_datetime(str(value)) for value in df_col]):
+            return "datetime"
+
+        if all([is_date(str(value)) for value in df_col]):
+            return "date"
 
     max_len = max([len(str(value)) for value in df_col])
     if max_len < 15:

cognos7/schema_ini_convert.py

@@ -3,7 +3,7 @@ import os
 from itertools import count
 from pathlib import Path
 
-from csv_column_types import get_column_types
+from cognos7.csv_column_types import get_column_types
 
 col_type_convert = {
     "varchar(20)": "to_varchar20",
@@ -64,7 +64,13 @@ def create_format_xml_files(base_dir: str, tables_with_columns: dict[str, list[s
             record.append(
                 f'    <FIELD ID="{i}" xsi:type="CharTerm" TERMINATOR=";" MAX_LENGTH="255" COLLATION="SQL_Latin1_General_CP1_CI_AS"/>'
             )
-            col_name_escape = col_name.encode("ascii", "xmlcharrefreplace").decode()
+            col_name_escape = (
+                col_name.replace("&", "&#38;")
+                .encode("ascii", "xmlcharrefreplace")
+                .decode()
+                .replace("<", "&#60;")
+                .replace(">", "&#62;")
+            )
             row.append(f'    <COLUMN SOURCE="{i}" NAME="{col_name_escape}" xsi:type="SQLVARYCHAR"/>')
         record[-1] = record[-1].replace(";", "\\r\\n")
 
@@ -80,6 +86,7 @@ def create_format_xml_files(base_dir: str, tables_with_columns: dict[str, list[s
 
 
 def create_format_xml_from_folder(base_dir: str) -> None:
+    system = Path(base_dir).parent.name
     format_folder = f"{base_dir}\\Format"
     os.makedirs(format_folder, exist_ok=True)
     sql_folder = f"{base_dir}\\SQL"
@@ -93,8 +100,9 @@ def create_format_xml_from_folder(base_dir: str) -> None:
     create_format_xml_files(format_folder, tables_with_columns)
     create_openrowset_sql_files(base_dir, format_folder, sql_folder, tables_with_columns)
     create_load_sql_files(sql_folder, tables_with_columns)
-    create_drop_create_sql_files(sql_folder, tables_with_columns)
-    create_update_sql_files(sql_folder, tables_with_columns)
+    create_drop_create_sql_files(sql_folder, tables_with_columns, system)
+    create_update_sql_files(sql_folder, tables_with_columns, system)
+    create_sql_bat_files(sql_folder)
 
 
 def create_openrowset_sql_files(
@@ -179,20 +187,45 @@ def get_tables_with_columns_from_folder(base_dir: str) -> dict[str, list[str]]:
         with open(csv_file, "r", encoding="latin-1") as frh:
             cols = frh.readline().strip("\n").split(";")
         cols_unique = []
+        cols_unique_lower = []
         for c in cols:
             c1 = c.strip('"')
-            if c1 not in cols_unique:
+            if c1.lower() not in cols_unique_lower:
                 cols_unique.append(c1)
+                cols_unique_lower.append(c1.lower())
                 continue
             for i in count(1):
                 c2 = f"{c1}_{i}"
-                if c2 not in cols and c2 not in cols_unique:
+                if c2 not in cols and c2.lower() not in cols_unique_lower:
                     cols_unique.append(c2)
+                    cols_unique_lower.append(c2.lower())
                     break
         tables_with_columns[table_name] = cols_unique
     return tables_with_columns
 
 
+def create_sql_bat_files(sql_folder: str) -> None:
+    tasks_dir = Path(sql_folder).parent.parent.parent.parent / "Tasks" / "scripts"
+    header = f'@call "{tasks_dir}\\config.bat" 0 > nul'
+
+    folder_list = [
+        "views_export",
+        "views_load",
+        "exec_drop_create",
+        "exec_update",
+    ]
+    for f in folder_list:
+        folder = f"{sql_folder}\\{f}"
+        bat_file = f"{folder}\\{f}.bat"
+
+        with open(bat_file, "w", encoding="cp850") as fwh:
+            fwh.write(header + "\n\n")
+            fwh.write(f"echo {folder}\n")
+
+            for sql_file in Path(folder).glob("*.sql"):
+                fwh.write(f"  call sqlexec2.bat {sql_file}\n")
+
+
 if __name__ == "__main__":
     # schema_convert("C:\\GlobalCube_LOCOSOFT\\GCStruct_SKR51\\Kontenrahmen")
     create_format_xml_from_folder("C:\\GlobalCube_LOCOSOFT\\System\\LOCOSOFT\\Export")

db.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 import typer
 
 import config
@@ -25,6 +27,22 @@ def run(config_file: str, increment: str = "1", max: int = 5):
     database.run(config_file, increment == "1", max)
 
 
+@app.command()
+def run_folder(folder: str):
+    folder_list = [
+        Path(folder),
+        Path(cfg.system_dir + "\\SQL\\exec\\" + folder),
+        Path(cfg.system_dir + "\\Export\\SQL\\" + folder),
+    ]
+    for f in folder_list:
+        if f.exists() and f.is_dir():
+            folder = str(f.resolve())
+            break
+    print(folder)
+    for sql_file in Path(folder).glob("*.sql"):
+        print(f"call sqlexec2.bat {sql_file}")
+
+
 @app.command()
 def schema():
     database.schema()