Cognos 11 Benutzerliste und Version

c11.py

@@ -1,4 +1,5 @@
 import os
+import re
 from enum import Enum
 from pathlib import Path
 
@@ -35,7 +36,8 @@ def reportoutput(folder="", mailcsv=""):
 
     req_plan_filtered = exp.filter_request_plan(req_plan, cfg.xml_dir + "/" + mailcsv)
     exp.execute_request_plan(req_plan_filtered)
-    pdf_test.missing_data(cfg.cognos11.reportoutput_dir + "/" + folder)
+    if mailcsv != "":
+        pdf_test.missing_data(cfg.cognos11.reportoutput_dir + "/" + folder)
 
     pdf_merge.pdf_merge_files(merge_group)
     cognos11.get_converted_jobs(cfg)
@@ -91,6 +93,28 @@ def deploy():
     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()
+    version()

cognos11/c11_api.py

@@ -1,14 +1,17 @@
 import base64
-import os
-import requests
-from requests_toolbelt.multipart import decoder
-import jinja2
+import csv
 import json
+import logging
+import os
 import re
+
+import jinja2
+import requests
 from bs4 import BeautifulSoup
+from requests_toolbelt.multipart import decoder
+
 import config
 from cognos11.xml_prettify import prettify_xml
-import logging
 
 
 class c11_api:
@@ -451,6 +454,171 @@ class c11_api:
         logging.debug(error)
         return r.status_code, error
 
+    def get_users(self, export_dir):
+        self.get_licenses(export_dir)
+
+        ns = self.get_namespaces()
+        for item in ns:
+            temp_items = self.get_namespace_items(item["id"])
+            item["items"] = self.get_sub_items(temp_items)
+
+        with open(export_dir + "/namespaces.json", "w") as fwh:
+            json.dump(ns, fwh, indent=2)
+        return ns
+
+    def get_sub_items(self, items):
+        for sub_item in items:
+            if sub_item["type"] == "namespaceFolder":
+                temp_items = self.get_namespace_folder_items(sub_item["id"])
+                sub_item["items"] = self.get_sub_items(temp_items)
+
+            if sub_item["type"] == "role":
+                sub_item["members"] = self.get_role_members(sub_item["id"])
+
+            if sub_item["type"] == "group":
+                sub_item["members"] = self.get_group_members(sub_item["id"])
+        return items
+
+    def show_users(self, ns, indent=0):
+        indent_str = " " * indent
+        style = {
+            "account": "@",
+            "namespace": "#",
+            "role": "~",
+            "group": "+",
+            "namespaceFolder": "*",
+        }
+        for item in ns:
+            if item["type"] == "account":
+                print(indent_str + f"@ {item['defaultName']} (uid={item.get('userName', 'xx')})")
+            else:
+                print(indent_str + f"{style.get(item['type'], '*')} {item['defaultName']} ({item['type']})")
+            if "members" in item:
+                for user in item["members"].get("users", []):
+                    print(indent_str + f"  -> @ {user['defaultName']} (uid={user['userName']})")
+                for group in item["members"].get("groups", []):
+                    print(indent_str + f"  -> + {group['defaultName']} ({group['type']})")
+                for role in item["members"].get("roles", []):
+                    print(indent_str + f"  -> ~ {role['defaultName']} ({role['type']})")
+            if "items" in item:
+                self.show_users(item["items"], indent + 2)
+
+    def export_users(self, ns, export_dir: str):
+        data = self.export_users_recursive(ns, [])
+
+        if len(data) > 0:
+            with open(export_dir + "\\cognos_users.csv", "w", encoding="latin-1", newline="") as fwh:
+                csv_fwh = csv.DictWriter(fwh, fieldnames=data[0].keys(), delimiter=";")
+                csv_fwh.writeheader()
+                csv_fwh.writerows(data)
+
+    def export_users_recursive(self, tree, prev_folder_list):
+        res = []
+        for item in tree:
+            folder_list = prev_folder_list + [item["defaultName"]]
+            if "items" in item:
+                res.extend(self.export_users_recursive(item["items"], folder_list))
+            elif "members" in item and isinstance(item["members"], dict):
+                for e in ["users", "groups", "roles"]:
+                    res.extend(self.export_users_recursive(item["members"].get(e, []), folder_list))
+            else:
+                res.append(
+                    {
+                        "Name": item["defaultName"],
+                        "Typ": item["type"],
+                        "Pfad": "/".join(prev_folder_list),
+                        "ID": item["id"],
+                        "Suchpfad": item["searchPath"].replace('"', "'"),
+                        "UID": item.get("userName", ""),
+                        "Email": item.get("email", ""),
+                        "Erstellung": item.get("creationTime", ""),
+                    }
+                )
+        return res
+
+    def get_licenses(self, export_dir):
+        res = self.session.get(
+            f"{self.webservice}v1/licenses/details",
+            headers=self.headers,
+        )
+        data = res.text.replace(",", ";")
+        with open(export_dir + "/licenses.csv", "w", encoding="latin-1", newline="") as fwh:
+            fwh.write(data)
+        return data
+
+    def get_namespaces(self):
+        fields = ",".join(
+            [
+                "id",
+                "defaultName",
+                "searchPath",
+                "objectClass",
+                "creationTime",
+                "modificationTime",
+                "type",
+            ]
+        )
+        res = self.session.get(
+            f"{self.webservice}v1/namespaces?fields={fields}",
+            headers=self.headers,
+        )
+        return res.json()["data"]
+
+    def get_namespace_items(self, id):
+        fields = ",".join(
+            [
+                "id",
+                "defaultName",
+                "searchPath",
+                "objectClass",
+                "creationTime",
+                "modificationTime",
+                "type",
+                "email",
+                "givenName",
+                "surname",
+                "userName",
+            ]
+        )
+
+        res = self.session.get(
+            f"{self.webservice}v1/namespaces/{id}/items?fields={fields}",
+            headers=self.headers,
+        )
+        return res.json()["data"]
+
+    def get_namespace_folder_items(self, id):
+        fields = ",".join(
+            [
+                "id",
+                "defaultName",
+                "searchPath",
+                "objectClass",
+                "creationTime",
+                "modificationTime",
+                "type",
+            ]
+        )
+        res = self.session.get(
+            f"{self.webservice}v1/folders/{id}/items?fields={fields}",
+            headers=self.headers,
+        )
+        return res.json()["data"]
+
+    def get_role_members(self, id):
+        res = self.session.get(
+            f"{self.webservice}v1/roles/{id}/members",
+            headers=self.headers,
+        )
+        return res.json()
+
+    def get_group_members(self, id):
+        res = self.session.get(
+            f"{self.webservice}v1/groups/{id}/members",
+            headers=self.headers,
+        )
+        return res.json()
+
     def create_report(self, folder_id, fullpath):
         # self.session.get(self.webservice + 'v1/reports/templates?path=%2Fcontent%2Ffolder%5B%40name%3D%27Templates%27%5D/
         # *[@objectClass=%27interactiveReport%27%20or%20@objectClass=%27report%27%20or%20@objectClass=%27reportTemplate%27]&maxResults=100&locale=de')
@@ -523,15 +691,3 @@ class c11_api:
     def request_search_path(self, id):
         res = self.session.get(f"{self.webservice}v1/objects/{id}?fields=searchPath", headers=self.headers)
         return res.json()["data"][0]["searchPath"]
-
-
-if __name__ == "__main__":
-    api = c11_api()
-    api.login()
-    folders = api.get_folders()
-    # filename = "C:/GlobalCube/Tasks/gctools/logs/config/folders.json"
-    # json.dump(folders, open(filename, "w"), indent=2)
-    # filename = "C:/GlobalCube/Tasks/gctools/logs/config/reports.json"
-    # json.dump(api.reports, open(filename, "w"), indent=2)
-    # filename = "C:/GlobalCube/Tasks/gctools/logs/config/jobs.json"
-    # json.dump(api.jobs, open(filename, "w"), indent=2)

cognos11/c11_export.py

@@ -159,7 +159,7 @@ class c11_export:
             for row in rfh.readlines():
                 if ";" not in row:
                     continue
-                filename, mailto = row.split(";")
+                filename, mailto = row.split(";", 2)
                 if "@" not in mailto:
                     continue
                 required_files.append(str(Path(self.cfg.cognos11.reportoutput_dir + "/" + filename).resolve()))

config/config.py

@@ -1,6 +1,6 @@
 import os
-from pathlib import Path
 from dataclasses import dataclass
+from pathlib import Path
 
 
 @dataclass
@@ -41,6 +41,7 @@ class Cognos11Config:
     folders_file: str
     reports_file: str
     jobs_file: str
+    server_dir: str
     credentials: Cognos11Credentials
 
 
@@ -147,4 +148,5 @@ class Config:
                 username=self._cfg.get("C11_U", "Global1"),
                 password=self._cfg.get("C11_P", "Cognos#11"),
             ),
+            server_dir="C:\\Program Files\\ibm\\cognos\\analytics",
         )