gctools/cognos7/mdl_convert.py

import json
from pathlib import Path
import re


id_lookup = {}


def get_field(field, block, num=1):
    if num == 2:
        res = re.search(field + r' (\d+) "([^"]*)" ', block)
        if res:
            return res[1], res[2]
        return "0", ""
    res = re.search(field + r' "([^"]*)" ', block)
    if res:
        return res[1]
    res = re.search(field + r" (\w*) ", block)
    if res:
        return res[1]
    return ""


def ignore(block):
    s = block.split(" ")
    return (s[0], block)


def model_name(block):
    return {
        "Type": "ModelName",
        "Name": get_field("Name", block),
        "ModelCodePage": get_field("ModelCodePage", block),
        "AutoAccess": get_field("AutoAccess", block),
        "UpdateCycle": get_field("UpdateCycle", block),
        "ModelStamp": get_field("ModelStamp", block),
        "Version": get_field("Version", block),
        "ModelCategoryOrderDefault": get_field("ModelCategoryOrderDefault", block),
        "ModelOrderedByDefault": get_field("ModelOrderedByDefault", block),
        "ModelNonRollupHierarchies": get_field("ModelNonRollupHierarchies", block),
    }


def cognos_source(block):
    id, name = get_field("CognosSource", block, 2)
    return {
        "Type": "CognosSource",
        "ID": id,
        "Name": name,
        "SourceType": get_field("SourceType", block),
        "SourcePath": get_field("SourcePath", block),
        "PackageTimeStamp": get_field("PackageTimeStamp", block),
    }


def cognos_package_datasource_connection(block):
    id, name = get_field("CognosPackageDatasourceConnection", block, 2)
    prs_id, prs_name = get_field("PackageReportSource", block, 2)
    return {
        "Type": "CognosPackageDatasourceConnection",
        "ID": id,
        "Name": name,
        "PackageReportSource": {
            "ID": prs_id,
            "Name": prs_name,
        },
        "CognosPackageConnection": get_field("CognosPackageConnection", block),
        "CognosPackageConnectionSignon": get_field(
            "CognosPackageConnectionSignon", block
        ),
        "CognosPackageAlwaysUseTransformerSignon": get_field(
            "CognosPackageAlwaysUseTransformerSignon", block
        ),
        "CognosPackagePowercubeSource": get_field(
            "CognosPackagePowercubeSource", block
        ),
    }


def data_source(block):
    id, name = get_field("DataSource", block, 2)
    prs_id, prs_name = get_field("PackageReportSource", block, 2)
    return {
        "Type": "DataSource",
        "ID": id,
        "Name": name,
        "Separator": get_field("Separator", block),
        "SourceType": get_field("SourceType", block),
        "CharacterSet": get_field("CharacterSet", block),
        "DecimalSep": get_field("DecimalSep", block),
        "Thousandsep": get_field("Thousandsep", block),
        "HasColumns": get_field("Columns", block),
        "Timing": get_field("Timing", block),
        "PackageReportSource": {
            "ID": prs_id,
            "Name": prs_name,
        },
        "AutoSummary": get_field("AutoSummary", block),
        "SetCurrent": get_field("SetCurrent", block),
        "ServerSource": get_field("ServerSource", block),
        "Speed": get_field("Speed", block),
        "Presummarized": get_field("Presummarized", block),
        "StreamExtractSize": get_field("StreamExtractSize", block),
        "Columns": [],
    }


def org_name(block):
    id, name = get_field("OrgName", block, 2)
    return {
        "Type": "OrgName",
        "ID": id,
        "Name": name,
        "Origin": get_field("Origin", block),
        "Offset": get_field("Offset", block),
        "Column": get_field("Column", block),
        "Storage": get_field("Storage", block),
        "Scale": get_field("Scale", block),
        "Size": get_field("Size", block),
        "Decimals": get_field("Decimals", block),
        "Class": get_field("Class", block),
        "InputScale": get_field("InputScale", block),
        "TimeArray": get_field("TimeArray", block),
        "ColSrcType": get_field("ColSrcType", block),
        "Associations": [],
    }


def dimension(block):
    id, name = get_field("Dimension", block, 2)
    return {
        "Type": "Dimension",
        "ID": id,
        "Name": name,
        "DimType": get_field("DimType", block),
        "EarliestDate": get_field("EarliestDate", block),
        "LatestDate": get_field("LatestDate", block),
        "ManualPeriods": get_field("ManualPeriods", block),
        "DaysInWeek": get_field("DaysInWeek", block),
        "NewCatsLock": get_field("NewCatsLock", block),
        "ExcludeAutoPartitioning": get_field("ExcludeAutoPartitioning", block),
        "DimDefaultCategory": get_field("DimDefaultCategory", block),
        "Root": {},
        "Levels": [],
        "Categories": [],
        "SpecialCategories": [],
    }


def root(block):
    id, name = get_field("Root", block, 2)
    return {
        "Type": "Root",
        "ID": id,
        "Name": name,
        "Inclusion": get_field("Inclusion", block),
        "Lastuse": get_field("Lastuse", block),
        "Date": get_field("Date", block),
        "Filtered": get_field("Filtered", block),
        "Suppressed": get_field("Suppressed", block),
        "Sign": get_field("Sign", block),
        "HideValue": get_field("HideValue", block),
        "IsKeyOrphanage": get_field("IsKeyOrphanage", block),
        "IsTruncated": get_field("IsTruncated", block),
        "Blanks": get_field("Blanks", block),
        "Drill": {},
    }


def levels(block):
    id, name = get_field("Levels", block, 2)
    return {
        "Type": "Levels",
        "ID": id,
        "Name": name,
        "Blanks": get_field("Blanks", block),
        "Inclusion": get_field("Inclusion", block),
        "DateFunction": get_field("DateFunction", block),
        "Generate": get_field("Generate", block),
        "RefreshLabel": get_field("RefreshLabel", block),
        "RefreshDescription": get_field("RefreshDescription", block),
        "RefreshShortName": get_field("RefreshShortName", block),
        "NewCatsLock": get_field("NewCatsLock", block),
        "CatLabFormat": get_field("CatLabFormat", block),
        "Timerank": get_field("Timerank", block),
        "UniqueCategories": get_field("UniqueCategories", block),
        "UniqueMove": get_field("UniqueMove", block),
        "Associations": [],
    }


def category(block):
    id, name = get_field("Category", block, 2)
    return {
        "Type": "Category",
        "ID": id,
        "Name": name,
        "Parent": get_field("Parent", block),
        "Levels": get_field("Levels", block),
        "OrderBy": get_field("OrderBy", block),
        "Value": get_field("Value", block),
        "Label": get_field("Label", block),
        "Lastuse": get_field("Lastuse", block),
        "SourceValue": get_field("SourceValue", block),
        "Date": get_field("Date", block),
        "Filtered": get_field("Filtered", block),
        "Suppressed": get_field("Suppressed", block),
        "Sign": get_field("Sign", block),
        "HideValue": get_field("HideValue", block),
        "IsKeyOrphanage": get_field("IsKeyOrphanage", block),
        "IsTruncated": get_field("IsTruncated", block),
        "Blanks": get_field("Blanks", block),
    }


def special_category(block):
    id, name = get_field("SpecialCategory", block, 2)
    return {
        "Type": "SpecialCategory",
        "ID": id,
        "Name": name,
        "Parent": get_field("Parent", block),
        "Levels": get_field("Levels", block),
        "Lastuse": get_field("Lastuse", block),
        "Rollup": get_field("Rollup", block),
        "TimeAggregate": get_field("TimeAggregate", block),
        "RunningPeriods": get_field("RunningPeriods", block),
        "TargetOffset": get_field("TargetOffset", block),
        "TargetLevel": get_field("TargetLevel", block),
        "ContextOffset": get_field("ContextOffset", block),
        "DateDrill": get_field("DateDrill", block),
        "Primary": get_field("Primary", block),
        "Sign": get_field("Sign", block),
    }


def map_drills(block):
    return {
        "Type": "MapDrills",
        # "MapDrills MapDrill 1469 "
    }


def view_name(block):
    id, name = get_field("ViewName", block, 2)
    return {
        "Type": "ViewName",
        "ID": id,
        "Name": name,
        "ViewType": get_field("Type", block),
        "ViewCustomView": get_field("ViewCustomView", block),
    }


def associations(block):
    id, name = get_field("Associations", block, 2)
    return {
        "Type": "Associations",
        "ID": id,
        "Name": name,
        "AssociationType": get_field("AssociationType", block),
        "AssociationRole": get_field("AssociationRole", block),
        "AssociationReferenced": get_field("AssociationReferenced", block),
        "SortOrder": get_field("SortOrder", block),
        "SortAs": get_field("SortAs", block),
        "Parent": "0",
    }


def drill(block):
    id, name = get_field("Drill", block, 2)
    return {
        "Type": "Drill",
        "ID": id,
        "Name": name,
        "Label": get_field("Label", block),
        "Inclusion": get_field("Inclusion", block),
        "Filtered": get_field("Filtered", block),
        "Suppressed": get_field("Suppressed", block),
        "PrimaryDrill": get_field("PrimaryDrill", block),
        "HideValue": get_field("HideValue", block),
        "YearBegins": get_field("YearBegins", block),
        "PartialWeek": get_field("PartialWeek", block),
        "ExtraWeek": get_field("ExtraWeek", block),
        "WeekBegins": get_field("WeekBegins", block),
    }


def measure(block):
    id, name = get_field("Measure", block, 2)
    return {
        "Type": "Measure",
        "ID": id,
        "Name": name,
        "Missing": get_field("Missing", block),
        "IgnoreMissingValue": get_field("IgnoreMissingValue", block),
        "Storage": get_field("Storage", block),
        "OutPutScale": get_field("OutPutScale", block),
        "Decimals": get_field("Decimals", block),
        "ReverseSign": get_field("ReverseSign", block),
        "IsCurrency": get_field("IsCurrency", block),
        "IsFolder": get_field("IsFolder", block),
        "Format": get_field("Format", block),
        "DrillThrough": get_field("DrillThrough", block),
        "Associations": [],
    }


def signon(block):
    id, name = get_field("Signon", block, 2)
    return {
        "Type": "Signon",
        "ID": id,
        "Name": name,
        "UserId": get_field("UserId", block),
        "PromptForPassword": get_field("PromptForPassword", block),
        "EncryptedPW": get_field("EncryptedPW", block),
        "AutoLogon": get_field("AutoLogon", block),
        "SignonType": get_field("SignonType", block),
    }


def dimension_view(block):
    id, name = get_field("DimensionView", block, 2)
    return {
        "Type": "DimensionView",
        "ID": id,
        "Name": name
        # "DimensionView 1463 \"All Categories\" DimensionView 1521 \"All Categories\" DimensionView 1551 \"All Categories\"
        # DimensionView 1575 \"All Categories\" DimensionView 1591 \"All Categories\" DimensionView 1651 \"All Categories\"
        # DimensionView 1665 \"All Categories\" DimensionView 1693 \"All Categories\" DimensionView 15741 \"All Categories\"
        # MeasureInclude 9829 Yes MeasureInclude 10053 Yes MeasureInclude 10309 Yes MeasureInclude 10313 Yes
        # MeasureInclude 10317 Yes MeasureInclude 15761 Yes "
    }


def allocation_add(block):
    # only in DimensionView
    return {
        "Type": "AllocationAdd",
        # "AllocationAdd Measure 9829 Type Default AllocationAdd Measure 10053 Type Default AllocationAdd Measure 10309 Type Default
        # AllocationAdd Measure 10313 Type Default AllocationAdd Measure 10317 Type Default AllocationAdd Measure 15761 Type Default "
    }


def cube(block):
    id, name = get_field("Cube", block, 2)
    return {
        "Type": "Cube",
        "ID": id,
        "Name": name,
        "MdcFile": get_field("MdcFile", block),
        "EncryptedPW": get_field("EncryptedPW", block),
        "Status": get_field("Status", block),
        "CubeCreation": get_field("CubeCreation", block),
        "Optimize": get_field("Optimize", block),
        "ConsolidatedRecords": get_field("ConsolidatedRecords", block),
        "PartitionSize": get_field("PartitionSize", block),
        "PassesNumber": get_field("PassesNumber", block),
        "Compress": get_field("Compress", block),
        "IncrementalUpdate": get_field("IncrementalUpdate", block),
        "ServerCube": get_field("ServerCube", block),
        "CubeStamp": get_field("CubeStamp", block),
        "CubeCycle": get_field("CubeCycle", block),
        "BlockParentTotals": get_field("BlockParentTotals", block),
        "Caching": get_field("Caching", block),
        "UseAlternateFileName": get_field("UseAlternateFileName", block),
        "DeployType": get_field("DeployType", block),
        "DeployLocations": get_field("DeployLocations", block),
        "DeployToAvailableLocationsAutomatic": get_field(
            "DeployToAvailableLocationsAutomatic", block
        ),
        "DeployCleanupEnabled": get_field("DeployCleanupEnabled", block),
        "DeployCleanupNumberOfCubes": get_field("DeployCleanupNumberOfCubes", block),
        "DrillThrough": get_field("DrillThrough", block),
        "DataSourceSignon": get_field("DataSourceSignon", block),
        "PublishEnable": get_field("PublishEnable", block),
        "PublishStatus": get_field("PublishStatus", block),
        "PublishAllowNullSuppression": get_field("PublishAllowNullSuppression", block),
        "PublishAllowMultiEdgeSuppression": get_field(
            "PublishAllowMultiEdgeSuppression", block
        ),
        "PublishAllowAccessToSuppressionOptions": get_field(
            "PublishAllowAccessToSuppressionOptions", block
        ),
    }


def custom_view(block):
    id, name = get_field("CustomView", block, 2)
    return {
        "Type": "CustomView",
        "ID": id,
        "Name": name,
        "DimensionView": [
            {"ID": b[0], "Name": b[1]}
            for b in re.findall(r'DimensionView (\d+) "([^"]+)"', block)
        ],
        "MeasureInclude": [
            {"ID": b[0], "Name": b[1]}
            for b in re.findall(r"MeasureInclude (\d+) (\w+) ", block)
        ],
        "ChildList": {},
    }


def custom_view_child_list(block):
    id, name = get_field("CustomViewChildList", block, 2)
    return {
        "Type": "CustomViewChildList",
        "ID": id,
        "Name": name,
    }


def security_namespace(block):
    id, name = get_field("SecurityNameSpace", block, 2)
    return {
        "Type": "SecurityNameSpace",
        "ID": id,
        "Name": name,
        "SecurityNameSpaceCAMID": get_field("SecurityNameSpaceCAMID", block),
        "Objects": [],
    }


def security_object(block):
    id, name = get_field("SecurityObject", block, 2)
    return {
        "Type": "SecurityObject",
        "ID": id,
        "Name": name,
        "SecurityObjectDisplayName": get_field("SecurityObjectDisplayName", block),
        "SecurityObjectType": get_field("SecurityObjectType", block),
        "CustomViewList": get_field("CustomViewList", block),
    }


CONVERSION = {
    "Name": model_name,
    "CognosSource": cognos_source,
    "CognosPackageDatasourceConnection": cognos_package_datasource_connection,
    "DataSource": data_source,
    "OrgName": org_name,
    "Dimension": dimension,
    "Root": root,
    "Drill": drill,
    "Levels": levels,
    "Associations": associations,
    "Category": category,
    "SpecialCategory": special_category,
    "MapDrills": map_drills,
    "ViewName": view_name,
    "Measure": measure,
    "Signon": signon,
    "Cube": cube,
    "CustomView": custom_view,
    "CustomViewChildList": custom_view_child_list,
    "SecurityNameSpace": security_namespace,
    "SecurityObject": security_object,
}


def convert_block(block):
    block = block.replace("\n", "")
    block_type = block.split(" ")[0]
    # block_pair = re.findall(r'("[^"]+"|\w+) ', block)
    # return (block_type, list(zip((block_pair[::2], block_pair[1::2]))))
    if block_type in CONVERSION:
        return CONVERSION[block_type](block)
    return {"Type": block_type}


def remove_ids(nested):
    nested.pop("ID", "")
    nested.pop("DateDrill", "")
    nested.pop("Primary", "")
    nested.pop("Lastuse", "")
    for col in ["Parent", "Levels", "CustomViewList"]:
        if col not in nested:
            continue
        if col == "Levels" and (
            isinstance(nested["Levels"], list) or nested["Levels"] == "0"
        ):
            continue
        nested[col] = id_lookup.get(nested[col], {}).get("Name", "undefined")

    for child in nested.values():
        if isinstance(child, dict):
            remove_ids(child)
        if isinstance(child, list):
            for entry in child:
                remove_ids(entry)
    return nested


def convert_file(filename):
    with open(filename, "r", encoding="latin-1") as frh:
        mdl_str = frh.read()

    mdl_str = re.sub(r"\n+", "\n", mdl_str)
    mdl_str = re.sub(r'\nLevels (\d+ [^"])', r"Levels \1", mdl_str)
    mdl_str = re.sub(r" Associations ", " \nAssociations ", mdl_str)
    mdl_str = re.sub(r'([^ ])""', r"\1'", mdl_str)
    mdl_str = re.sub(r'""([^ ])', r"'\1", mdl_str)
    tags = "|".join(list(CONVERSION.keys()))
    mdl_str = re.sub(r"\n(" + tags + r") ", r"\n\n\1 ", mdl_str)
    mdl_blocks = mdl_str.split("\n\n")

    converted = [convert_block(b) for b in mdl_blocks]
    result = {
        "Model": {},
        "Connections": [],
        "DataSources": [],
        "Dimensions": [],
        "Measures": [],
        "Signons": [],
        "CustomViews": [],
        "Security": [],
        "Cubes": [],
    }
    types = [c["Type"] for c in converted]
    ids = [c.get("ID", "0") for c in converted]

    id_lookup.update(dict(zip(ids, converted)))

    current = None
    level_ids = []

    for c, t in zip(converted, types):
        if t in [""]:
            continue
        if (
            t in ["Category", "SpecialCategory"]
            and result["Dimensions"][-1]["Name"] == "Zeit"
        ):
            if t == "Category" or c["Name"][0].isnumeric():
                continue

        if t in ["ModelName"]:
            result["Model"] = c
        elif t in ["CognosSource", "CognosPackageDatasourceConnection"]:
            result["Connections"].append(c)
        elif t in ["DataSource"]:
            result["DataSources"].append(c)
        elif t in ["OrgName"]:
            result["DataSources"][-1]["Columns"].append(c)
        elif t in ["Dimension"]:
            level_ids = []
            result["Dimensions"].append(c)
        elif t in ["Root"]:
            result["Dimensions"][-1]["Root"] = c
        elif t in ["Drill"]:
            result["Dimensions"][-1]["Root"]["Drill"] = c
        elif t in ["Levels"]:
            current = c
            level_ids.append(c["ID"])
            result["Dimensions"][-1]["Levels"].append(c)
        elif t in ["Category"]:
            if c["Levels"] in level_ids[0:2]:
                result["Dimensions"][-1]["Categories"].append(c)
        elif t in ["SpecialCategory"]:
            result["Dimensions"][-1]["SpecialCategories"].append(c)
        elif t in ["Measure"]:
            current = c
            result["Measures"].append(c)
        elif t in ["Associations"]:
            c["Parent"] = current["ID"]
            current["Associations"].append(c)
            for ds in result["DataSources"]:
                for col in ds["Columns"]:
                    if col["Column"] == c["AssociationReferenced"]:
                        col["Associations"].append(c)
        elif t in ["Signon"]:
            result["Signons"].append(c)
        elif t in ["Cube"]:
            result["Cubes"].append(c)
        elif t in ["CustomView"]:
            result["CustomViews"].append(c)
        elif t in ["CustomViewChildList"]:
            for cv in result["CustomViews"]:
                if cv["ID"] == c["ID"]:
                    cv["ChildList"] = c
        elif t in ["SecurityNameSpace"]:
            result["Security"].append(c)
        elif t in ["SecurityObject"]:
            result["Security"][-1]["Objects"].append(c)
        # else:
        #     print(t, c)

    json.dump(result, open(filename[:-4] + "_ori.json", "w"), indent=2)
    # yaml.safe_dump(result, open(filename[:-4] + ".yaml", "w"))

    result = remove_ids(result)
    json.dump(result, open(filename[:-4] + ".json", "w"), indent=2)


def convert_folder(base_dir):
    files = sorted([(f.stat().st_mtime, f) for f in Path(base_dir).rglob("*.mdl")])
    for timestamp, filename in files:
        convert_file(str(filename))


if __name__ == "__main__":
    # convert_file("data/S_Offene_Auftraege.mdl")
    # convert_file("data/F_Belege_SKR_SKR_Boettche.mdl")
    convert_folder("data/")