Python-Projekte/gctools/c11_admin.py

import base64
import os
import time
import requests
from requests_toolbelt.multipart import decoder
import pandas as pd
import jinja2
import json
import re
from bs4 import BeautifulSoup


class ca_webscraper:
    webservice = "http://localhost:9300/bi/"
    templates_dir = "C:/Projekte/Python/cognos11/templates"
    export_dir = "C:/Projekte/Python/export"
    log_dir = "C:/Projekte/Python/logs"
    credentials = '{"parameters":[{"name":"CAMNamespace","value":"CognosEx"},{"name":"h_CAM_action","value":"logonAs"},{"name":"CAMUsername","value":"Global1"},{"name":"CAMPassword","value":"Cognos#11"}]}'
    headers = {}
    caf = ""
    cam = ""

    def __init__(self):
        self._env = jinja2.Environment(
            loader=jinja2.FileSystemLoader(self.templates_dir),
            autoescape=jinja2.select_autoescape(['html', 'xml'])
        )
        self.template = self._env.get_template('get_report.xml')

    def generate_token(self, message_base64):
        version = "V1".encode("utf-8")
        header_len = 4
        msg = base64.b64decode(message_base64)[1:]
        chunks = []

        while len(msg) >= header_len:
            chunk_len = int.from_bytes(msg[:header_len], byteorder="little")
            msg = msg[header_len:]
            chunks.append(msg[:chunk_len])
            msg = msg[chunk_len:]

        return base64.b64encode(version + chunks[-1]).decode("utf-8")

    def login(self):
        self.session = requests.Session()
        r = self.session.get(self.webservice)
        self.headers = {'Content-Type': "application/json; charset=UTF-8", 'X-XSRF-TOKEN': self.session.cookies.get('XSRF-TOKEN')}

        r = self.session.post(self.webservice + "v1/login", data=self.credentials, headers=self.headers)

        self.caf = r.json()['cafContextId']
        self.cam = self.generate_token(r.cookies["usersessionid"])
        return r.status_code

    def report_list(self):
        # "v1/objects/_dot_public_folders/items"
        filter = "type|analysis|interactiveReport|powerPlayReport|powerPlay8Report|powerPlay8ReportView|query|report|reportTemplate"
        res = self.session.get(self.webservice + "v1/search/cm?fields=defaultName|id|ancestors&results=1000&query=.&hide_internal=all&filter=" + filter, headers=self.headers)
        self.reports = res.json()['results']
        for r in self.reports:
            r['path'] = "/".join([a['defaultName'].replace('/', '_') for a in r['ancestors']])
            r['name'] = r['defaultName']
            del(r['ancestors'])
            del(r['defaultName'])

        filename = self.log_dir + '/config/reports.json'
        os.makedirs(os.path.dirname(filename), exist_ok=True)
        json.dump(self.reports, open(filename, 'w'), indent=2)
        return self.reports

    def export_folder(self, folder):
        reports = [r for r in self.reports if r['path'].startswith(folder)]
        for r in reports:
            print(r['name'])
            # continue
            path = r['path'].replace(folder, '')
            r['filename'] = f"{path}/{r['name']}.pdf"
            r['params'] = list(re.findall(r'\[([^\]]+)\]', r['filename']))
            for i, p in enumerate(r['params']):
                r['filename'] = r['filename'].replace('[' + p + ']', '{' + str(i) + '}')
            self.export(r, 'PDF')

    def export(self, report, format='XML'):
        headers = {
            'Content-Type': 'text/xml; charset=UTF-8', 
            'X-XSRF-TOKEN': self.headers['X-XSRF-TOKEN'],
            'X-RsCMStoreID': report['id'],
            'X-UseRsConsumerMode': 'true',
            'SOAPAction': 'http://www.ibm.com/xmlns/prod/cognos/reportService/202004/'
        }
        soap = self.template.render({"caf": self.caf, "cam": self.cam,
                                "report": report, "format": 'XHTML',
                                "prompt": 'true', "tracking": "", "params": {}})

        r = self.session.post(self.webservice + 'v1/reports', data=soap, headers=headers)
        parts = decoder.MultipartDecoder.from_response(r).parts
        # for i, p in enumerate(parts):
        #     with open(f"export/{report['report']}_{i}.xml", "w") as f:
        #         f.write(p.text.replace('\x81', ''))

        meta = {'required': {}, 'optional': {}}
        bs = BeautifulSoup(parts[1].content, 'lxml')
        # print(bs.prettify())
        for sv in bs.find_all('selectvalue'):
            k = sv['parameter']
            req = 'required' if sv['required'] == 'true' else 'optional'
            v = dict([(opt['usevalue'], opt['displayvalue']) for opt in sv.find_all('selectoption')])
            meta[req][k] = v

            # <selectValue specname="selectValue" parameter="p_Auswahl_Monate" range="false" required="false" hideAdornments="false" multiSelect="true" prePopulateIfParentOptional="false" selectValueUI="checkboxGroup" rowsPerPage="5000" autoSubmit="false" name="_P3591935746" style="width:100%" columnName="Alle Autoh�user" roid="i9241">
            #     <selectChoices>
            #         <selectOption useValue="letzte 13 Monate" displayValue="letzte 13 Monate"/>
            #     </selectChoices>
            #     <selectOptions>
            #         <selectOption useValue="letzte 13 Monate" displayValue="letzte 13 Monate"/>
            #     </selectOptions>
            # </selectValue>
        filename = self.log_dir + f"/config/{report['name']}.json"
        os.makedirs(os.path.dirname(filename), exist_ok=True)
        json.dump(meta, open(filename, 'w'), indent=2)
        
        if format == 'PDF':
            return self.export_pdf(report, meta)
        
        payload = json.dumps({'reportspec_stubbed': parts[2].text, 'storeid': report['id']})
        token = re.findall(r'<bus:authenticityToken xsi:type="xs:base64Binary">(.*)</bus:authenticityToken>', parts[0].text)[0]
        headers = {
            'Content-Type': 'application/json; charset=UTF-8',
            'X-XSRF-TOKEN': self.session.cookies.get('XSRF-TOKEN'),
            'authenticityToken': token,
            'X-UseRsConsumerMode': 'true',
            'cafContextId': self.caf
        }

        r = self.session.post(self.webservice + 'v1/reports/unstubreport', data=payload, headers=headers)
        unstubbed = json.loads(r.content.decode('latin-1'))['reportspec_full']
        unstubbed = re.sub(r' iid="[^"]*"', '', unstubbed).replace('\x9f', '').replace('\x96', '')

        bs = BeautifulSoup(unstubbed, 'xml')
        for xa in bs.find_all('XMLAttributes'):
            xa.decompose()

        filename = self.export_dir + f"unstubbed/{report['name']}.xml"
        os.makedirs(os.path.dirname(filename), exist_ok=True)
        with open(filename, "w") as f:
            f.write(bs.prettify())
        return soap

    def export_pdf(self, report, meta):
        headers = {
            'Content-Type': 'text/xml; charset=UTF-8', 
            'X-XSRF-TOKEN': self.headers['X-XSRF-TOKEN'],
            'X-RsCMStoreID': report['id'],
            'X-UseRsConsumerMode': 'true',
            'SOAPAction': 'http://www.ibm.com/xmlns/prod/cognos/reportService/202004/'
        }
        
        if len(report['params']) == 0:
            params = {}
            filename = report['filename']
            self.request_file(report, headers, params, filename)
            return True
        if len(report['params']) == 1:
            key1 = report['params'][0]
            for k1, v1 in meta['optional'][key1].items():
                filename = report['filename'].format(v1)
                params = {key1: {k1: v1}}
                self.request_file(report, headers, params, filename)
            return True
        if len(report['params']) == 2:
            key1, key2 = report['params']
            for k1, v1 in meta['optional'][key1].items():
                for k2, v2 in meta['optional'][key2].items():
                    filename = report['filename'].format(v1, v2)
                    params = {key1: {k1: v1}, key2: {k2: v2}}
                    self.request_file(report, headers, params, filename)
            return True            

    def request_file(self, report, headers, params, filename):
        soap = self.template.render({"caf": self.caf, "cam": self.cam,
                                     "report": report, "format": 'PDF',
                                     "prompt": 'false', "tracking": "", "params": params})
        try:
            r = self.session.post(self.webservice + 'v1/reports', data=soap, headers=headers)
        except UnicodeEncodeError:
            filename = self.log_dir + '/' + os.path.basename(filename) + '.log'
            os.makedirs(os.path.dirname(filename), exist_ok=True)
            with open(filename, "w") as f:
                f.write(soap)
            return False

        if r.status_code == 200:
            parts = decoder.MultipartDecoder.from_response(r).parts
            filename = self.export_dir + filename
            os.makedirs(os.path.dirname(filename), exist_ok=True)
            with open(filename, "wb") as f:
                f.write(parts[1].content)
        else:
            filename = self.log_dir + '/' + os.path.basename(filename) + '.log'
            os.makedirs(os.path.dirname(filename), exist_ok=True)
            with open(filename, "wb") as f:
                f.write(r.content)
        return True

    def admin(self):
        r = self.session.get(self.webservice + "v1/disp?m_p_owner=&changed_m_p_owner=0&changed_genprop=0&so.select=&so.return.m=portal%2Fproperties_general.xts&so.defaultLocation=&so.defaultObject=&ro_name=false&origDefLang=de&m_email=&m_defaultName=&m_t_default_name_de=CARLO_F_Belege&m_t_default_description_de=&m_t_default_screenTip_de=&ifrmcmd=save&m_p_disabled=false&m_p_hidden=false&icon_radio=standard&m_transloc=de&pty_activeLang=de&pty_deactivLang=&pty_del=&pty_add=&pty_name=CARLO_F_Belege&pty_scrt=&pty_desc=&m_new_class=&b_action=xts.run&sharedPagesChanged=&from_tool=true&backURL=%2Fbi%2Fv1%2Fdisp%3Fb_action%3Dxts.run%26m%3Dportal%2Flegacy_tools%2Ftools_directory.xts%26m_pathID%3Di339AF66BADEC411E943590402582B75B%26m_path%3DCAMID%28%2522%253a%2522%29%252fdataSource%255b%2540name%253d%2527CARLO_F_Belege%2527%255d%26tool_tab%3Dd&m_selectedPage=&m_classSubtype=&m_obj=CAMID%28%22%3A%22%29%2FdataSource%5B%40name%3D%27CARLO_F_Belege%27%5D%2FdataSourceConnection%5B%40name%3D%27CARLO_F_Belege%27%5D&b_report_type=&encoding=UTF-8&m=portal%2Fproperties_connection.xts&m_class=dataSourceConnection&m_name=CARLO_F_Belege&ui.cafcontextid=CAFW000000a0Q0FGQTYwMDAwMDAwMDlBaFFBQUFERWpZV1g4bEExbmlJd29ualF1cEgwWVVTeGtnY0FBQUJUU0VFdE1qVTJJQUFBQUxha3gqeHQ5TXN3Ukw2dGhjMTJVRzN1NVhaMWVzNU5FLXRvWXI1VzlwYTE0NDI0NzN8cHM_&m_path=CAMID%28%22%3A%22%29%2FdataSource%5B%40name%3D%27CARLO_F_Belege%27%5D&cmd=&m_location=&reportLocation=&ps_nav_op=maintain&ps_nav_stack=&ps_nav_source=portal%2Fproperties_general.xts")
        print(r.cookies.keys())


# CRN=http%3A%2F%2Fdeveloper.cognos.com%2Fceba%2Fconstants%2FbiDirectionalOptionEnum%23biDirectionalFeaturesEnabled%3Dfalse%26http%3A%2F%2Fdeveloper.cognos.com%2Fceba%2Fconstants%2FsystemOptionEnum%23accessibilityFeatures%3Dfalse%26skin%3Dcorporate%26contentLocale%3Dde-de%26showHiddenObjects%3Dfalse%26showWelcomePage%3Dtrue%26backgroundSessionLogging%3D1970-01-01%2B00%253A00%253A00%26showOptionSummary%3Dtrue%26productLocale%3Dde%26listViewSeparator%3Dnone%26showHints%3DhideAll%26timeZoneID%3DEurope%252FBerlin%26linesPerPage%3D15%26displayMode%3Dlist%26automaticPageRefresh%3D30%26format%3DHTML%26columnsPerPage%3D3%26;
# cea-ssa=false;
# userCapabilities=f%3Bfdbffc6d%3Bf07c1faf%3Bff27defa%26AwcAAABTSEEtMjU2FAAAAMSNhZfyUDWeIjCieNC6kfRhRLGSDPui1gb2UKZPMWVW5x3QHrHM%2BIVk5gFTzsNq1oOMioU%3D;
# userCapabilitiesEx=603%3Bf%3Bfdbffc6d%3Bf07c1faf%3Bff27defa%26AhQAAADEjYWX8lA1niIwonjQupH0YUSxkgcAAABTSEEtMjU2IAAAAMI%2BRz7opwjhYGXonwHJpD3Ya1agg0lFOa1JK%2FkZiDTS;
# caf=CAFW000000e0Q0FGQTYwMDAwMDAwM2FBaFFBQUFERWpZV1g4bEExbmlJd29ualF1cEgwWVVTeGtnY0FBQUJUU0VFdE1qVTJJQUFBQUY3d3JJLW1zREhXcnQtT0VWLUhhKjF0c01TZFVrWlB0aXNYdEVJOUcqZm40NDI0NzN8MTAxOjc5ZTJiNDdhLTIyNjgtZDcxYS1jMGIxLWYwMDYyY2QxYWQwZTozNDAxNTQ5MzI1;
# cc_session=s_cc:|s_conf:na|s_sch:td|s_hd:sa|s_serv:na|s_disp:na|s_set:|s_dep:na|s_dir:na|s_sms:dd|s_ct:sa|s_cs:sa|s_so:sa|e_hp:CAMID(*22CognosEx*3au*3auid*3dglobal1*22)|e_proot:Team*20Content|prootid:i1F610DE4196544319A27C5709282EF95|e_mroot:Eigene*20Ordner|mrootid:iBAA3C16EC2D743B7B221BBFCDBC625D7|e_mrootpath:CAMID(*22CognosEx*3au*3auid*3dglobal1*22)*2ffolder*5b*40name*3d*27Eigene*20Ordner*27*5d|e_user:Global*20Cube|e_tenantID:|e_tenantDisplayName:|e_showTenantInfo:false|e_isSysAdmin:true|e_isTenantAdmin:false|e_isImpersonating:false|cl:de-de|dcid:i1F610DE4196544319A27C5709282EF95|show_logon:false|uig:|ui:|rsuiprofile:all|lch:f|lca:f|ci:f|write:true|eom:0|pp:3401549325;
# up=H4sIAAAAAAAAAFWQ204CMRCGX4X0GuIuF5rlDgXEBBQUgxIT0sOwW2k7mx4UNL67sxgRk0ma+Tr5+nc+WdQWVujgZsB6bJg81vBydgneaMfaLGzp6LFHF8E7qCw0cIPe8kh4vJhOqBdcbkuPyakHCEGjm2BZalfSRF5cZJ0sp2plWe9QjbXC97F2MdBEpRX0jSGaAvhbboHgtUHBTU4w7usGcCnJHwlUaGHGywZSR7x5Umij434EPCYPZN1wE6DNdOg7dHuL6Y/VHlWScYKSm0ai4BhIKXB34hVkPFEomtnNu4UTtvi4fyrP1aLO+XJerLq7t2c7CtP5ek0Kae0MfeTmJ5jQA+3JRNvg5jfY0HFhQB3lCjY8mfjv062rJJpIEmnn7jRn5xBV8AAL2MWjnu76KSL7+gaz/bFgzAEAAA==;
# MRUStorage=%7B%22xQ29nbm9zRXg6dTp1aWQ9Z2xvYmFsMQ__%22%3Atrue%7D;
# usersessionid=AggAAADOAvteAAAAAAoAAAAPMXuCsi7eLj2GFAAAAMSNhZfyUDWeIjCieNC6kfRhRLGSBwAAAFNIQS0yNTYgAAAAxhTjzCpYnRF3ryJ/z/gpU9G5UeFYyp1vKc/PjuCT/r4=;
# cam_passport=MTsxMDE6NzllMmI0N2EtMjI2OC1kNzFhLWMwYjEtZjAwNjJjZDFhZDBlOjM0MDE1NDkzMjU7MDszOzA7

if __name__ == '__main__':
    caws = ca_webscraper()
    caws.login()
    caws.report_list()
    caws.export_folder('Team Content/Export/Test')