generaty_sdmx_glossary.py

from pysdmx.io.xml.sdmx21.writer.structure import write

from pysdmx.model.organisation import AgencyScheme, Agency
from pysdmx.model.code import Codelist, Code
from pysdmx.model.concept import ConceptScheme, Concept, DataType

from xml.sax.saxutils import escape

import pyodbc

EBA_ACCESS_DB_PATH = 'DPM2.0_release_4.0_2024-12-10.accdb'

def create_agency_scheme():
    agency_scheme = AgencyScheme(
        agency="SDMX",
        id="EBA",
        name="EBA",
        description="EBA DPM glossary SDMX import",
        items = [Agency(
            id="EBA",
            name="EBA",
            description="EBA DPM glossary SDMX import"
        )])
    
    return [agency_scheme]


def get_items(conn, category_id):
    query = """
        SELECT 
            Item.Name, Item.Description, ItemCategory.CategoryID, ItemCategory.StartReleaseID, ItemCategory.EndReleaseID, ItemCategory.Code
        FROM 
            Item 
            INNER JOIN ItemCategory ON Item.ItemID = ItemCategory.ItemID
        where CategoryID= ? and EndReleaseID is null
        ORDER BY ItemCategory.StartReleaseID DESC;
    """
    
    cursor = conn.cursor()
    cursor.execute(query, category_id)
    codes = []
    code_ids = set()
    for item in cursor.fetchall():
        code_id = item[5].replace('.', '_')
        code = Code(
            id=code_id,
            name=escape(item[0]),
            description=escape(item[1]) if item[1] else None
        )
        if code_id not in code_ids:
            code_ids.add(code_id)
            codes.append(code)
        else:
            print(f"Code {code_id} - {item[0]} duplicated")
    
    return codes


def get_categories(conn):
    query = """
        SELECT DISTINCT
            Category.Code, Category.Name, Category.CategoryID
        FROM 
            Category
            INNER JOIN ItemCategory ON Category.CategoryID = ItemCategory.CategoryID
        where 
            EndReleaseID is null 
            and Category.Code not in ('_PR', '_TE');
        """
    
    cursor = conn.cursor()
    cursor.execute(query)

    codelists = []
    for category_version in cursor.fetchall():
        codes = get_items(conn, category_version[2])

        category_code = category_version[0]
        if category_code[0] == '_':
            category_code = category_code.replace('_', 'ZZZ_')

        codelist = Codelist(
            agency="EBA",
            id=category_code,
            name=category_version[1],
            items=codes
        )

        codelists.append(codelist)

    return codelists


def get_properties(conn, codelists):
    concepts = []

    query = """
        SELECT
            Item.Name, Item.Description, ItemCategory.Code, Category.Code, ItemCategory.EndReleaseID, DataType.Code
        FROM 
            DataType INNER JOIN (Category INNER JOIN (((Item INNER JOIN Property ON Item.ItemID = Property.PropertyID) INNER JOIN ItemCategory ON Item.ItemID = ItemCategory.ItemID) INNER JOIN PropertyCategory ON Property.PropertyID = PropertyCategory.PropertyID) ON Category.CategoryID = PropertyCategory.CategoryID) ON DataType.DataTypeID = Property.DataTypeID
        WHERE ItemCategory.EndReleaseID Is Null
        ORDER BY ItemCategory.StartReleaseID DESC
    """
    
    DATA_TYPES_MAPPING = {
        'i': DataType.INTEGER,
        'r': DataType.DECIMAL,
        's': DataType.STRING,
        'b': DataType.BOOLEAN,
        'd': DataType.DATE,
        'm': DataType.DECIMAL,
        'p': DataType.DECIMAL,
        'es': DataType.STRING,
        'e': DataType.STRING,
        't': DataType.BOOLEAN
    }


    cursor = conn.cursor()
    cursor.execute(query)
    
    concept_ids = set()

    for property in cursor.fetchall():
        codelist = None
        dtype = None
        for codelist_enum in codelists:
            if codelist_enum.id == property[3]:
                codelist = codelist_enum

        if codelist is None:
            if property[5] in DATA_TYPES_MAPPING:
                dtype = DATA_TYPES_MAPPING[property[5]]
            else:
                print(f"Data type {property[5]} used for property {property[2]} using category {property[3]}")

        concept = Concept(
            id=property[2],
            name=escape(property[0]),
            description=escape(property[1]) if property[1] else None,
            codes=codelist,
            dtype=dtype
        )
        if property[2] not in concept_ids:
            concept_ids.add(property[2])
            concepts.append(concept)
        else:
            print(f"Concept {property[2]} duplicated")

        concept_scheme = ConceptScheme(
            agency="EBA",
            id="EBA",
            name="EBA",
            description="EBA DPM glossary SDMX import",
            items=concepts
        )

    return concept_scheme


if __name__ == "__main__":
    conn_str = ('DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};'
                f'DBQ={EBA_ACCESS_DB_PATH};')
    conn = pyodbc.connect(conn_str)

    agency_scheme = create_agency_scheme()

    with open("output/agency_scheme.xml", "w", encoding="utf-8") as f:
        f.write(write(agency_scheme))
    
    codelists = get_categories(conn)

    with open("output/codelists.xml", "w", encoding="utf-8") as f:
        f.write(write(codelists))

    properties = get_properties(conn, codelists)

    with open("output/properties.xml", "w", encoding="utf-8") as f:
        f.write(write([properties]))