Black Formatter, isort und VSCode

.gitignore

@@ -1,3 +1,4 @@
 __pycache__
 supervisord.pid
 app/temp/*
+.venv/

.vscode/launch.json

@@ -6,7 +6,7 @@
     "configurations": [
         {
             "name": "Python: Aktuelle Datei",
-            "type": "python",
+            "type": "debugpy",
             "request": "launch",
             "program": "${file}",
             "console": "integratedTerminal",

.vscode/settings.json

@@ -0,0 +1,16 @@
+{  
+    "[python]": {
+        "editor.defaultFormatter": "ms-python.black-formatter",
+        "editor.formatOnSave": true,
+        "editor.codeActionsOnSave": {
+            "source.organizeImports": "explicit",
+        },
+    },
+    "isort.args":["--profile", "black"],
+    "black-formatter.args": ["--line-length", "120"],
+    "python.testing.pytestEnabled": false,
+    "python.testing.unittestEnabled": true,
+    "files.associations": {
+        "*.mac": "vbs"
+    }
+}

app/api.py

@@ -1,7 +1,8 @@
 from flask import Blueprint
 
-bp = Blueprint('api', __name__)
+bp = Blueprint("api", __name__)
 
-@bp.route('/')
+
+@bp.route("/")
 def main():
-    return 'Test'
+    return "Test"

app/tools/config.py

@@ -1,6 +1,6 @@
 import os
-from pathlib import Path
 from dataclasses import dataclass
+from pathlib import Path
 
 
 @dataclass
@@ -29,85 +29,87 @@ def joinpath(path, *other):
 
 
 class Config:
-    kunde = 'Test'
-    version = 'C11'
+    kunde = "Test"
+    version = "C11"
     versand_separat = False
     cognos7: Cognos7Config
     _cfg = {}
 
-    def __init__(self, ini='GAPS.ini'):
+    def __init__(self, ini="GAPS.ini"):
         self._cfg = {}
 
         self.tools_dir = Path(os.curdir).absolute()
         self.tasks_dir = str(self.tools_dir.parent).lower()
-        if self.tools_dir.parent.name.lower() != 'tasks':
+        if self.tools_dir.parent.name.lower() != "tasks":
             # development
-            self.tools_dir = self.tools_dir.joinpath('gctools')
-            self.portal_dir = 'c:\\globalcube'
+            self.tools_dir = self.tools_dir.joinpath("gctools")
+            self.portal_dir = "c:\\globalcube"
             if not Path(self.portal_dir).exists():
-                self.portal_dir = joinpath(self.tools_dir, 'Portal')
+                self.portal_dir = joinpath(self.tools_dir, "Portal")
         else:
             base_dir = self.tools_dir.parent.parent
-            if base_dir.name.lower() != 'globalcube':
-                self.version = 'C7'
+            if base_dir.name.lower() != "globalcube":
+                self.version = "C7"
             self.portal_dir = str(base_dir)
 
-        self.system_dir = joinpath(self.portal_dir, 'System')
-        self.tasks_dir = joinpath(self.portal_dir, 'Tasks')
-        self.xml_dir = joinpath(self.tasks_dir, 'config')
-        self.log_dir = joinpath(self.tasks_dir, 'logs')
+        self.system_dir = joinpath(self.portal_dir, "System")
+        self.tasks_dir = joinpath(self.portal_dir, "Tasks")
+        self.xml_dir = joinpath(self.tasks_dir, "config")
+        self.log_dir = joinpath(self.tasks_dir, "logs")
 
-        if ':' not in ini:
+        if ":" not in ini:
             ini = joinpath(self.tasks_dir, ini)
-        with open(ini, 'r') as stream:
+        with open(ini, "r") as stream:
             for line in stream.readlines():
-                if '=' in line:
-                    key, value = line.split('=')
-                    self._cfg[key] = value.replace('"', '').replace('\n', '')
-        self.kunde = self._cfg.get('KUNDE', 'Test')
-        self.kunde_safe = self.kunde.replace(' ', '-').lower()
-
-        self.system = self._cfg.get('SYSTEM', '')
-        if self.system != '':
+                if "=" in line:
+                    key, value = line.split("=")
+                    self._cfg[key] = value.replace('"', "").replace("\n", "")
+        self.kunde = self._cfg.get("KUNDE", "Test")
+        self.kunde_safe = self.kunde.replace(" ", "-").lower()
+
+        self.system = self._cfg.get("SYSTEM", "")
+        if self.system != "":
             self.system_dir = joinpath(self.system_dir, self.system)
         self.mail_config()
         self.cognos7_config()
         self.cognos11_config()
 
     def mail_config(self):
-        self.smtp = MailConfig(**{
-            'server': 'mail.global-cube.com',
-            'port': '465',
-            'secure': 'ssl',
-            'username': 'versand',
-            'password': 'y6!avXX3tQvr',
-            'email': 'versand+neuer-kunde@global-cube.com'
-        })
-        if self._cfg.get('SMTP_HOST', '') != '':
-            secure = {'': '', 'N': '', 'J': 'ssl', 'SSL': 'ssl', 'TLS': 'tls'}
+        self.smtp = MailConfig(
+            **{
+                "server": "mail.global-cube.com",
+                "port": "465",
+                "secure": "ssl",
+                "username": "versand",
+                "password": "y6!avXX3tQvr",
+                "email": "versand+neuer-kunde@global-cube.com",
+            }
+        )
+        if self._cfg.get("SMTP_HOST", "") != "":
+            secure = {"": "", "N": "", "J": "ssl", "SSL": "ssl", "TLS": "tls"}
 
             self.smtp = MailConfig(
-                server=self._cfg.get('SMTP_HOST'),
-                port=self._cfg.get('SMTP_PORT'),
-                secure=secure[self._cfg.get('SMTP_SSL', '')],
-                username=self._cfg.get('SMTP_USER'),
-                password=self._cfg.get('SMTP_PW'),
-                email=self._cfg.get('SMTP_FROM'),
+                server=self._cfg.get("SMTP_HOST"),
+                port=self._cfg.get("SMTP_PORT"),
+                secure=secure[self._cfg.get("SMTP_SSL", "")],
+                username=self._cfg.get("SMTP_USER"),
+                password=self._cfg.get("SMTP_PW"),
+                email=self._cfg.get("SMTP_FROM"),
             )
-        self.versand_separat = self._cfg.get('VERSAND_SEPARAT') == 'J'
+        self.versand_separat = self._cfg.get("VERSAND_SEPARAT") == "J"
 
     def cognos7_config(self):
         self.cognos7 = Cognos7Config(
-            program_dir=self._cfg.get('COGNOS', 'c:\\program files (x86)\\cognos\\cer5\\bin'),
-            iqd_dir=joinpath(self.system_dir, 'IQD'),
-            models_dir=joinpath(self.system_dir, 'Models'),
-            cube_dir=joinpath(self.system_dir, 'Cube_out'),
-            report_dir=joinpath(self.system_dir, 'Report'),
-            publish_dir=joinpath(self.portal_dir, 'Publish', 'daten'),
-            prot_dir=joinpath(self.system_dir, 'Prot')
+            program_dir=self._cfg.get("COGNOS", "c:\\program files (x86)\\cognos\\cer5\\bin"),
+            iqd_dir=joinpath(self.system_dir, "IQD"),
+            models_dir=joinpath(self.system_dir, "Models"),
+            cube_dir=joinpath(self.system_dir, "Cube_out"),
+            report_dir=joinpath(self.system_dir, "Report"),
+            publish_dir=joinpath(self.portal_dir, "Publish", "daten"),
+            prot_dir=joinpath(self.system_dir, "Prot"),
         )
         if not Path(self.cognos7.publish_dir).exists():
-            self.cognos7.publish_dir = joinpath(self.portal_dir, 'daten'),
+            self.cognos7.publish_dir = (joinpath(self.portal_dir, "daten"),)
 
     def cognos11_config(self):
         pass

app/tools/inspect_files.py

@@ -1,51 +1,52 @@
 # from itertools import chain
 import json
+import re
+from collections import defaultdict
 from pathlib import Path
+
 import pandas as pd
 from bs4 import BeautifulSoup
-from collections import defaultdict
-import re
-import sys
 from config import Config
 
 
 def get_path_info(base_dir):
-    path_info_df = pd.read_csv(base_dir + '/logs/status/path_info.csv', sep=';',
-                               encoding='latin-1', converters={'process': str})
-    path_info_df.rename(columns={'name': 'filename'}, inplace=True)
-    path_info_df['filename'] = path_info_df['filename'].str.lower()
-    path_info_df.set_index('filename', inplace=True, drop=False)
-    return path_info_df.to_dict(orient='index')
+    path_info_df = pd.read_csv(
+        base_dir + "/logs/status/path_info.csv", sep=";", encoding="latin-1", converters={"process": str}
+    )
+    path_info_df.rename(columns={"name": "filename"}, inplace=True)
+    path_info_df["filename"] = path_info_df["filename"].str.lower()
+    path_info_df.set_index("filename", inplace=True, drop=False)
+    return path_info_df.to_dict(orient="index")
 
 
 def get_cubes(base_dir, cfg: Config, path_info, cubes_models):
-    ver_files = [k for k in path_info.keys() if re.search(r'\\cubes\\.*\.ver', k)]
+    ver_files = [k for k in path_info.keys() if re.search(r"\\cubes\\.*\.ver", k)]
     cubes = {}
     for file in ver_files:
-        match = re.search(r'\\cubes\\(.*)__\d+\.ver$', file)
+        match = re.search(r"\\cubes\\(.*)__\d+\.ver$", file)
         cube_name = match[1]
         cube_subdir = f"{file[:-4]}\\{cube_name}.mdc"
         cube_out = f"{cfg.system_dir}\\cube_out\\{cube_name}.mdc"
 
         cubes[cube_name] = {
-            'deployed_mdc': path_info[cube_subdir],
-            'cube_out_mdc': path_info[cube_out],
-            'model': cubes_models[cube_name],
-            'errors': []
+            "deployed_mdc": path_info[cube_subdir],
+            "cube_out_mdc": path_info[cube_out],
+            "model": cubes_models[cube_name],
+            "errors": [],
         }
     return cubes
 
 
 def get_models(base_dir, cfg: Config, path_info, fm_sources):
     models = {}
-    for file in Path(base_dir + '\\config\\models').glob('*.log'):
-        with open(file, 'r') as frh:
-            model_infos = frh.read().lower().replace('"', '').replace(',', '').split('--')
-        datasources = model_infos[1].split('\n')
-        datasources = [d for d in datasources if d != '']
+    for file in Path(base_dir + "\\config\\models").glob("*.log"):
+        with open(file, "r") as frh:
+            model_infos = frh.read().lower().replace('"', "").replace(",", "").split("--")
+        datasources = model_infos[1].split("\n")
+        datasources = [d for d in datasources if d != ""]
         fm_src = []
         fm_package = None
-        if '[' in datasources[0]:
+        if "[" in datasources[0]:
             fm_package = datasources.pop(0).upper()
             fm_src = datasources
             datasources = []
@@ -54,43 +55,43 @@ def get_models(base_dir, cfg: Config, path_info, fm_sources):
                     datasources.extend(fm_sources[src])
             datasources = sorted(set(datasources))
 
-        cube = re.search(r'\\cube_out\\(.*)\.mdc', model_infos[0])[1]
+        cube = re.search(r"\\cube_out\\(.*)\.mdc", model_infos[0])[1]
         models[file.name[:-8].lower()] = {
-            'framework_manager': 'J' if fm_package else 'N',
-            'fm_package': fm_package,
-            'fm_sources': fm_src,
-            'datasources': datasources,
-            'cube': cube,
-            'logfile_tasks': path_info.get(cfg.portal_dir + '\\tasks\\logs\\' + file.name[:-8] + '.log', {'mtime': '0'}),
-            'logfile_system': path_info.get(cfg.system_dir + '\\logs\\' + file.name[:-8] + '.log', {'mtime': '0'}),
-            'model_file': path_info.get(cfg.system_dir + '\\models\\' + file.name[:-4], {'mtime': '0'}),
-            'model_file_filled': path_info.get(cfg.system_dir + '\\models\\gefuellt\\' + file.name[:-4], {'mtime': '0'}),
+            "framework_manager": "J" if fm_package else "N",
+            "fm_package": fm_package,
+            "fm_sources": fm_src,
+            "datasources": datasources,
+            "cube": cube,
+            "logfile_tasks": path_info.get(
+                cfg.portal_dir + "\\tasks\\logs\\" + file.name[:-8] + ".log", {"mtime": "0"}
+            ),
+            "logfile_system": path_info.get(cfg.system_dir + "\\logs\\" + file.name[:-8] + ".log", {"mtime": "0"}),
+            "model_file": path_info.get(cfg.system_dir + "\\models\\" + file.name[:-4], {"mtime": "0"}),
+            "model_file_filled": path_info.get(
+                cfg.system_dir + "\\models\\gefuellt\\" + file.name[:-4], {"mtime": "0"}
+            ),
         }
     return models
 
 
 def get_database_info(base_dir, cfg: Config):
-    db_info_df = pd.read_csv(base_dir + '/logs/status/db_info.csv', sep=';',
-                             encoding='latin-1')
-    db_info_df = db_info_df[db_info_df['DatabaseName'] == 'GC']
-    db_info_df['table'] = db_info_df['TableName'].str.lower()
-    db_info_df.set_index('table', inplace=True)
-    return db_info_df.to_dict(orient='index')
+    db_info_df = pd.read_csv(base_dir + "/logs/status/db_info.csv", sep=";", encoding="latin-1")
+    db_info_df = db_info_df[db_info_df["DatabaseName"] == "GC"]
+    db_info_df["table"] = db_info_df["TableName"].str.lower()
+    db_info_df.set_index("table", inplace=True)
+    return db_info_df.to_dict(orient="index")
 
 
 def get_fm_sources(base_dir, cfg):
-    bs = BeautifulSoup(open(base_dir + '\\config\\fm\\model.xml', 'r'), 'xml')
+    bs = BeautifulSoup(open(base_dir + "\\config\\fm\\model.xml", "r"), "xml")
     sources = defaultdict(list)
-    for item in bs.find_all('queryItem'):
-        p = item.parent.parent.find('name').string
-        if item.parent.parent.name == 'folder':
-            p = item.parent.parent.parent.find('name').string
-        parent = "[{0}].[{1}]".format(
-            p,
-            item.parent.find('name').string
-        )
-        src = ''
-        exp = ''
+    for item in bs.find_all("queryItem"):
+        p = item.parent.parent.find("name").string
+        if item.parent.parent.name == "folder":
+            p = item.parent.parent.parent.find("name").string
+        parent = "[{0}].[{1}]".format(p, item.parent.find("name").string)
+        src = ""
+        exp = ""
         if item.expression:
             if item.expression.refobj:
                 src = item.expression.refobj.string
@@ -98,26 +99,26 @@ def get_fm_sources(base_dir, cfg):
                 exp = item.expression.string
         elif item.externalName:
             exp = item.externalName.string
-        sources[parent].append((item.find('name').string, src, exp))
+        sources[parent].append((item.find("name").string, src, exp))
     interface = {}
     for k, fields in sources.items():
-        if '[Schnittstelle]' not in k:
+        if "[Schnittstelle]" not in k:
             continue
-        key = k.split('.')[-1][1:-1].lower()
+        key = k.split(".")[-1][1:-1].lower()
         links = []
         for field in fields:
-            links.append(follow_links(sources, field, ''))
-        interface[key] = sorted(list(set([re.search(r'\.\[(.*)\]$', e)[1].lower() for e in links if '[Import]' in e])))
+            links.append(follow_links(sources, field, ""))
+        interface[key] = sorted(list(set([re.search(r"\.\[(.*)\]$", e)[1].lower() for e in links if "[Import]" in e])))
 
     return interface
 
 
 def follow_links(sources, field, value):
-    if field[1] == '':
+    if field[1] == "":
         if field[2] == field[0]:
             return value
         return field[2]
-    match = re.search(r'(\[.*\]\.\[.*\])\.\[(.*)\]', field[1])
+    match = re.search(r"(\[.*\]\.\[.*\])\.\[(.*)\]", field[1])
     key1 = match[1]
     val1 = match[2]
     if key1 in sources:
@@ -129,35 +130,43 @@ def follow_links(sources, field, value):
 
 
 def get_datasources(base_dir, cfg, path_info):
-    all_datasources = set([re.search(r'\\iqd\\.*\\(.*)\.imr', k)[1]
-                           for k in path_info.keys()
-                           if re.search(r'\\iqd\\.*\\.*\.imr', k)])
+    all_datasources = set(
+        [re.search(r"\\iqd\\.*\\(.*)\.imr", k)[1] for k in path_info.keys() if re.search(r"\\iqd\\.*\\.*\.imr", k)]
+    )
     datasources = {}
     for ds in all_datasources:
-        ds_search = f'\\{ds}.imr'
-        imr_files = [k for k in path_info.keys()
-                     if re.search(r'\\iqd\\.*\.imr', k) and ds_search in k
-                     and 'austausch' not in k]
+        ds_search = f"\\{ds}.imr"
+        imr_files = [
+            k for k in path_info.keys() if re.search(r"\\iqd\\.*\.imr", k) and ds_search in k and "austausch" not in k
+        ]
         if len(imr_files) == 0:
-            imr_file = '0.imr'
+            imr_file = "0.imr"
         else:
             imr_file = imr_files.pop(0)
         datasources[ds] = {
-            'imr_file': path_info.get(imr_file, {'mtime': '0'}),
-            'iqd_file': path_info.get(imr_file[:-4] + '.iqd', {'mtime': '0'}),
-            'csv_file': path_info.get(cfg.system_dir + '\\export\\' + ds + '.csv', {'mtime': '0'}),
-            'csv_file_iqd_folder': path_info.get(imr_file[:-4] + '.csv', {'mtime': '0'}),
-            'duplicates': imr_files
+            "imr_file": path_info.get(imr_file, {"mtime": "0"}),
+            "iqd_file": path_info.get(imr_file[:-4] + ".iqd", {"mtime": "0"}),
+            "csv_file": path_info.get(cfg.system_dir + "\\export\\" + ds + ".csv", {"mtime": "0"}),
+            "csv_file_iqd_folder": path_info.get(imr_file[:-4] + ".csv", {"mtime": "0"}),
+            "duplicates": imr_files,
         }
     return datasources
 
 
 def cubes_to_models(models):
-    models_sort = sorted([(v.get('logfile_tasks')['mtime'],
-                           v.get('logfile_system')['mtime'],
-                           v.get('model_file_filled')['mtime'],
-                           v.get('model_file')['mtime'],
-                           k, v['cube']) for k, v in models.items()])
+    models_sort = sorted(
+        [
+            (
+                v.get("logfile_tasks")["mtime"],
+                v.get("logfile_system")["mtime"],
+                v.get("model_file_filled")["mtime"],
+                v.get("model_file")["mtime"],
+                k,
+                v["cube"],
+            )
+            for k, v in models.items()
+        ]
+    )
     result = {}
     for m in models_sort:
         result[m[5]] = m[4]
@@ -165,8 +174,8 @@ def cubes_to_models(models):
 
 
 def main():
-    base_dir = 'app/temp/unzipped/loeffler-c11_2022-07-08_112547'
-    cfg = Config(str(Path(base_dir + '\\gaps.ini').absolute()))
+    base_dir = "app/temp/unzipped/loeffler-c11_2022-07-08_112547"
+    cfg = Config(str(Path(base_dir + "\\gaps.ini").absolute()))
     # Dateiliste
     path_info = get_path_info(base_dir)
     # random_bat_file = [k for k in path_info.keys() if re.search(r'\\Tasks\\.*\.bat', k)][0]
@@ -176,16 +185,16 @@ def main():
     result = {}
 
     # Modelle und Datenquellen
-    result['fm_sources'] = get_fm_sources(base_dir, cfg)
-    result['models'] = get_models(base_dir, cfg, path_info, result['fm_sources'])
-    result['database'] = get_database_info(base_dir, cfg)
-    result['datasources'] = get_datasources(base_dir, cfg, path_info)
-    cubes_models = cubes_to_models(result['models'])
-    result['cubes'] = get_cubes(base_dir, cfg, path_info, cubes_models)
+    result["fm_sources"] = get_fm_sources(base_dir, cfg)
+    result["models"] = get_models(base_dir, cfg, path_info, result["fm_sources"])
+    result["database"] = get_database_info(base_dir, cfg)
+    result["datasources"] = get_datasources(base_dir, cfg, path_info)
+    cubes_models = cubes_to_models(result["models"])
+    result["cubes"] = get_cubes(base_dir, cfg, path_info, cubes_models)
     # Cubes aktuell?
     # Rest aktuell?
-    json.dump(result, open('app/temp/logs/export.json', 'w'), indent=2)
+    json.dump(result, open("app/temp/logs/export.json", "w"), indent=2)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()

app/tools/status_server.py

@@ -1,20 +1,21 @@
+from datetime import datetime
+
 from flask import Flask, request
 from sqlalchemy import create_engine
-from datetime import datetime
 
 app = Flask(__name__)
-filename = 'access.log'
-engine = create_engine('mysql+pymysql://gaps:Gcbs12ma@192.168.2.41/status')
+filename = "access.log"
+engine = create_engine("mysql+pymysql://gaps:Gcbs12ma@192.168.2.41/status")
 
 
 @app.route("/ping/<kunde>")
 def ping(kunde):
-    uptime = int(request.args.get('uptime', '0'))
+    uptime = int(request.args.get("uptime", "0"))
     current_date = datetime.now().timestamp()
-    print(kunde + ' hat sich um ' + str(current_date) + ' gemeldet')
-    with open(filename, 'a') as f:
-        f.write(str(current_date) + ';' + kunde + '\n')
+    print(kunde + " hat sich um " + str(current_date) + " gemeldet")
+    with open(filename, "a") as f:
+        f.write(str(current_date) + ";" + kunde + "\n")
     with engine.connect() as con:
         query = f"INSERT INTO ping (kunde, pingtime, uptime) VALUES ('{kunde}', '{int(current_date)}', '{uptime}')"
         con.execute(query)
-    return 'Hallo ' + kunde + '!'
+    return "Hallo " + kunde + "!"

app/tools/unzip.py

@@ -1,15 +1,14 @@
 import zipfile
 from pathlib import Path
-import os
 
 
 def unzip_file(filename, target):
     subdir = Path(filename).name[:-4]
-    target_subdir = target + '/' + subdir
+    target_subdir = target + "/" + subdir
     Path(target_subdir).mkdir()
-    with zipfile.ZipFile(filename, 'r', compression=zipfile.ZIP_DEFLATED, compresslevel=9) as zip:
+    with zipfile.ZipFile(filename, "r", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as zip:
         zip.extractall(target_subdir)
 
 
-if __name__ == '__main__':
-    unzip_file('app/temp/zip_files/loeffler-c11_2022-07-08_112547.zip', 'app/temp/unzipped')
+if __name__ == "__main__":
+    unzip_file("app/temp/zip_files/loeffler-c11_2022-07-08_112547.zip", "app/temp/unzipped")

app/views.py

@@ -1,12 +1,13 @@
 from flask import Blueprint, render_template
 
-bp = Blueprint('views', __name__)
+bp = Blueprint("views", __name__)
 
-@bp.route('/')
+
+@bp.route("/")
 def home():
-   return 'hello world!!'
+    return "hello world!!"
 
 
-@bp.route('/template')
+@bp.route("/template")
 def template():
-    return render_template('home.html')
+    return render_template("home.html")

main.py

@@ -1,6 +1,11 @@
 from flask import Flask
-from app import views, api
 
-app = Flask(__name__, template_folder='app/templates')
+from app import api, views
+
+app = Flask(__name__, template_folder="app/templates")
 app.register_blueprint(views.bp)
-app.register_blueprint(api.bp, url_prefix='/api')
+app.register_blueprint(api.bp, url_prefix="/api")
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port="8097")

requirements.txt

@@ -1,5 +1,4 @@
-Flask>=2.2.0
+flask>=2.2.0
 beautifulsoup4
 pandas
-sqlalchemy
-zipfile
+sqlalchemy

tox.ini

@@ -0,0 +1,5 @@
+[flake8]
+ignore = E203, E712, W503, W504
+max-line-length = 140
+# exclude = tests/*
+# max-complexity = 10