import os import re import time from datetime import datetime from enum import Enum from itertools import chain from pathlib import Path import typer import cognos7.mdl_convert import cognos11 import config from pdf import pdf_merge, pdf_test def flatten_chain(list_of_lists): return list(chain.from_iterable(list_of_lists)) class ExportFormat(Enum): PDF = "PDF" XML = "XML" app = typer.Typer() cfg = config.Config() @app.command() def export(folder="", overwrite="0"): exp = cognos11.c11_export(cfg) exp.export_folder(folder, overwrite=overwrite != "0") @app.command() def reportoutput(folder="", mailcsv="", exp=None): if re.match(r"^[\d\.\-\/]+$", folder): return reportoutput_by_date(folder, mailcsv) if exp is None: exp = cognos11.c11_export(cfg) # folder2 = exp.get_folder(folder) req_plan = exp.get_folder_pdf_request_plan(folder) merge_group = exp.get_merge_group(req_plan) exp.template(req_plan, merge_group) req_plan_filtered = exp.filter_request_plan(req_plan, f"{cfg.xml_dir}\\{mailcsv}") req_plan_flat = flatten_chain(req_plan_filtered) print("Exportiere Reports...") for i in range(1, 6): print(f"Durchlauf {i}...") req_plan_flat = exp.execute_request_plan(req_plan_flat) if not req_plan_flat: break time.sleep(20) exp.api.login() if req_plan_flat: print("!! Fehler: Einige Reports konnten nicht exportiert werden !!") for report_req in req_plan_flat: print(Path(report_req.filename).relative_to(cfg.cognos11.reportoutput_dir)) if mailcsv == "": pdf_test.missing_data(f"{cfg.cognos11.reportoutput_dir}\\{folder}") pdf_merge.pdf_merge_files(merge_group) cognos11.get_converted_jobs(cfg) # pdf_merge.merge_reports_in_folder(self.cfg, folder2) def reportoutput_by_date(date_str: str, mailcsv: str = "") -> None: current_date = None if re.match(r"^\d{4}-\d{2}-\d{2}$", date_str): current_date = datetime.strptime(date_str, "%Y-%m-%d") elif re.match(r"^\d{2}\.\d{2}\.\d{4}$", date_str): current_date = datetime.strptime(date_str, "%d.%m.%Y") elif re.match(r"^\d{2}\/\d{2}\/\d{4}$", date_str): current_date = datetime.strptime(date_str, "%m/%d/%Y") if current_date is None: print(f"Ungültiges Datumsformat: {date_str}") return exp = cognos11.c11_export(cfg) folders = [f["name"] for f in exp.api.get_folders()] # for f in folders: # print(f) weekdays = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"] current_weekday = weekdays[current_date.weekday()] print(f"Berichte für {current_weekday}, {current_date.strftime('%d.%m.%Y')}...") # taeglich daily_interval = ["Mo-Fr", "Mo-Sa", "Mo-So", "Di-So"] daily_interval_special = { "Mo": ["Mo-Fr", "Mo-Sa", "Mo-So"], "Sa": ["Mo-Sa", "Di-So"], "So": ["Mo-So", "Di-So"] } current_interval = daily_interval_special.get(current_weekday, daily_interval) for c in current_interval: search = f"Team Content/ReportOutput/taeglich/{c}" for f in folders: if search in f: print(f) # woechentlich search = f"Team Content/ReportOutput/woechentlich/{current_weekday}" for f in folders: if search in f: print(f) # 14-taeglich week = (int(current_date.strftime("%W")) % 2) + 1 search = f"Team Content/ReportOutput/14-taeglich/{current_weekday}{week}" for f in folders: if search in f: print(f) # monatlich day_of_month = current_date.strftime("%d") search = f"Team Content/ReportOutput/monatlich/{day_of_month}" for f in folders: if search in f: print(f) @app.command() def merge(folder="", config=""): folder2 = cognos11.c11_export.get_folder(folder) config_json = pdf_merge.get_config_json(config, cfg.cognos11.config_dir) pdf_merge.merge_reports_in_folder(cfg, folder2, config_json) @app.command() def errors(): exp = cognos11.c11_export(cfg) exp.export_errors() @app.command() def create(path: str): cognos11.c11_create(cfg).create_path(path) @app.command() def template(): exp = cognos11.c11_export(cfg) cognos11.get_converted_jobs(cfg) req_plan = exp.get_folder_pdf_request_plan("") merge_group = exp.get_merge_group(req_plan) exp.template(req_plan, merge_group) @app.command() def mailtemplate(): return template() @app.command() def mdlconvert(mdl_file): cognos7.mdl_convert.convert_file(mdl_file) source = mdl_file[:-4] + ".json" target = cfg.cognos11.specs_dir + "/../DataModel/" + Path(source).name os.makedirs(os.path.dirname(target), exist_ok=True) Path(target).unlink(missing_ok=True) os.rename(source, target) @app.command() def deploy(): cube_out = cfg.system_dir + "\\Cube_out" cubes_dir = cfg.portal_dir + "\\Cubes" cognos11.deploy(cube_out, cubes_dir) @app.command() def users(): api = cognos11.c11_api(cfg).login() ns = api.get_users(cfg.cognos11.config_dir) api.export_users(ns, cfg.cognos11.config_dir) api.show_users(ns) @app.command() def version(): with open(cfg.cognos11.server_dir + "\\cmplst.txt", "r") as frh: for line in frh.readlines(): match = re.search(r"Manifest=casrv-manifest-(.*).json", line) if match: ver = match.group(1) print("Cognos Version: " + ver) if not ver.startswith("11.1.7") and not ver.startswith("12."): print("!! Fehler: Cognos Version ist veraltet !!") return if __name__ == "__main__": app() # reportoutput() # reportoutput("2023-10-02")