gctools/cognos11/c11_export.py

from pathlib import Path
from cognos11.c11_api import c11_api
import config
import json
import logging
import os
from datetime import datetime
from dataclasses import dataclass


@dataclass
class ReportRequest:
    report_id: str
    params: dict
    filename: str
    report_format: str


class c11_export:
    api: c11_api
    cfg: config.Config = None

    def __init__(self, cfg=None, api=None):
        self.cfg = cfg
        if cfg is None:
            self.cfg = config.Config()
        self.api = api
        if api is None:
            # if self.cfg.cognos11.
            self.api = c11_api(cfg).login()

        now = datetime.now().strftime("%Y%m%d_%H%M%S")
        prot_file = f"{self.cfg.cognos11.logs_dir}/error_{now}.log"
        os.makedirs(self.cfg.cognos11.logs_dir, exist_ok=True)
        logging.basicConfig(
            filename=prot_file,
            filemode="w",
            encoding="utf-8",
            level=logging.DEBUG,
            force=True,
        )

    @staticmethod
    def get_folder(folder, format="PDF"):
        if folder == "":
            folder = "Team Content" if format == "XML" else "Team Content/ReportOutput"
        elif not folder.startswith("Team Content"):
            folder = "Team Content/ReportOutput/" + folder
        return folder

    def export_folder(self, folder="") -> None:
        folder = self.get_folder(folder, "XML")
        reports = self.api.get_reports_in_folder(folder, True)
        for r in reports:
            print(r["name"])
            self.export_unstubbed(r["id"])

    def export_unstubbed(self, report_id):
        report = self.api.get_report(report_id)
        unstubbed_report = self.api.request_unstubbed(report_id)
        if unstubbed_report:
            filename = (
                f"{self.cfg.cognos11.specs_dir}/{report['path']}/{report['name']}.xml"
            )
            os.makedirs(os.path.dirname(filename), exist_ok=True)
            with open(filename, "w") as f:
                f.write(unstubbed_report)

    def get_folder_pdf_request_plan(self, folder=""):
        folder = self.get_folder(folder, "PDF")
        reports = self.api.get_reports_in_folder(folder, True)
        return [self.export_pdf(r) for r in reports]

    def export_pdf(self, report_id, folder=None):
        report = self.api.get_report(report_id)
        if report["type"] == "shortcut":
            report_link = report
            report = self.api.get_report(report["target_id"]).copy()
            for k in ("name", "description", "path", "type", "filename"):
                report[k] = report_link[k]
            json.dump(report, open("dump.json", "w"), indent=2)
        if "meta" not in report:
            logging.warning(report["name"] + " is not accessible!")
            return False
        params = {}
        if len(report["meta"]["required"]) > 0:
            if set(report["meta"]["required"].keys()) != {"p_Von", "p_Bis"}:
                return False
            params["p_Von"] = {"2022-10-12": "12.10.2022"}
            params["p_Bis"] = {"2022-10-12": "12.10.2022"}
            # params["p_Zeitraum"] = {"Einzelne Monate": "Einzelne Monate"}

            for k, v in report["meta"]["optional"].items():
                if k in ["p_Zeit", "p_Auswahl_Monate", "p_12_Monate"]:
                    for k1, v1 in reversed(v.items()):
                        if v1 != "Invalid Dates":
                            params[k] = {k1: v1}
                            continue  # use last element only

        if len(report["params"]) == 0:
            filename = report["filename"]
            return [ReportRequest(report["id"], params, filename, report["format"])]
        result = []
        if len(report["params"]) == 1:
            filename = report["filename"].format("_Summe")
            result.append(
                ReportRequest(report["id"], params, filename, report["format"])
            )

            key1 = report["params"][0]
            for k1, v1 in report["meta"]["optional"][key1].items():
                filename = report["filename"].format(v1)
                params[key1] = {k1: v1}
                result.append(
                    ReportRequest(report["id"], params, filename, report["format"])
                )
            return result
        if len(report["params"]) == 2:
            filename = report["filename"].format("_Summe", "").replace("_.", ".")
            result.append(
                ReportRequest(report["id"], params, filename, report["format"])
            )

            key1, key2 = report["params"]
            for k1, v1 in report["meta"]["optional"][key1].items():
                filename = report["filename"].format(v1, "_Summe")
                params = {}
                params[key1] = {k1: v1}
                result.append(
                    ReportRequest(report["id"], params, filename, report["format"])
                )

                for k2, v2 in report["meta"]["optional"][key2].items():
                    filename = report["filename"].format(v1, v2)
                    params[key2] = {k2: v2}
                    result.append(
                        ReportRequest(report["id"], params, filename, report["format"])
                    )
            return result

    def execute_request_plan(self, req_plan):
        for req_group in req_plan:
            for report_req in req_group:
                self.request_and_save_file(report_req)

    def request_and_save_file(self, report_request: ReportRequest):
        logging.debug(report_request.filename)
        logging.debug(report_request.params)
        status_code, content = self.api.request_file(
            report_request.report_id,
            report_request.params,
            report_request.report_format,
        )
        if status_code == 200:
            os.makedirs(os.path.dirname(report_request.filename), exist_ok=True)
            with open(report_request.filename, "wb") as f:
                f.write(content)
        else:
            logging.warning(content)

    def get_merge_group(self, req_plan):
        res = {}
        for req_group in req_plan:
            if len(req_group) == 1:
                continue
            files = [r.filename for r in req_group]
            filename = files[0].replace("_Summe", "_Summe_und_Schichten")
            res[filename] = files

            files2 = [r for r in files if "_Summe" not in r]
            filename2 = filename.replace("_Summe_und_Schichten", "_nur_Schichten")
            res[filename2] = files2
        return res

    def export_errors(self):
        reports = self.api.get_reports_in_folder(
            "Team Content", recursive=True, specs=True
        )
        errors = [r for r in reports if "error" in r]
        filename = self.cfg.cognos11.logs_dir + "/c11_report_errors.json"
        json.dump(errors, open(filename, "w"), indent=2)

    def mail_template(self, folder: str):
        current_dir = Path(self.cfg.cognos11.reportoutput_dir + "\\" + folder)

        subfolders = set(
            [f.name for f in current_dir.glob("*") if f.is_dir()]
        ).difference([".", "..", "leer"])

        folder2 = folder + "\\" if folder != "" else ""

        export_files = [
            f"{folder2}{f.name};"
            for f in current_dir.glob("*.*")
            if f.suffix in [".pdf", ".xls"]
        ]

        if len(export_files) > 0:
            folder2 = folder.replace("\\", "_") if folder != "" else "Versand"
            template_file = Path(
                self.cfg.cognos11.config_dir + "\\" + folder2 + ".template.csv"
            )
            print(template_file)
            with template_file.open("w") as fwh:
                fwh.write("Report;Empfaenger\n")
                fwh.write("\n".join(export_files))

        for f in subfolders:
            subfolder = folder + "\\" + f if folder != "" else f
            self.mail_template(subfolder)


if __name__ == "__main__":
    api = c11_api()
    api.login()
    pdf = c11_export(None, api)
    # pdf.export_folder('Team Content/Verkauf/1. Gesamtverkauf', 'PDF')
    # pdf.export_folder('Team Content/Aftersales/1. Service')
    pdf.export_errors()