ExtractDenado.py

import os

import jaydebeapi
import pandas as pd
from socket import gethostname
import re

from tabulate import tabulate


def extract_sql_host(uri):
    match = re.search(r"jdbc:\w+://([^:/;]+)", uri)
    return match.group(1) if match else None


# ---------------- CREATE Denodo_Objects_Final.csv ----------------

def extract_objects(df, is_source=True):
    prefix = "Source" if is_source else "Target"
    objects_df = pd.DataFrame()
    objects_df["Provider Name"] = df[f"{prefix} Provider Name"]
    objects_df["Server Name"] = df[f"{prefix} Server"]
    objects_df["Database Name"] = df[f"{prefix} Database"]
    objects_df["Schema Name"] = df[f"{prefix} Schema"]
    objects_df["Object Name"] = df[f"{prefix} Object"]
    objects_df["Object Description"] = ""  # Not available in your lineage
    objects_df["Column Name"] = df[f"{prefix} Column"]
    objects_df["Column Description"] = ""  # Not available in your lineage
    objects_df["Data Type"] = df.get(f"{prefix} Data Type", "")
    objects_df["Is Nullable"] = ""
    objects_df["Precision"] = df.get(f"{prefix} Precision", "")
    objects_df["Scale"] = df.get(f"{prefix} Scale", "")
    objects_df["Object Type"] = df.get(f"{prefix} Object Type", "")
    return objects_df


# ---------------- CONFIGURATION ----------------
denodoserver_name = "10.0.68.9"
denodoserver_jdbc_port = "9999"
denodoserver_uid = "admin"
denodoserver_pwd = "admin"
denododriver_path = "Jupiter/denodo-vdp-jdbcdriver.jar"

conn_uri = f"jdbc:vdb://{denodoserver_name}:{denodoserver_jdbc_port}/sampledb"
driver_class = "com.denodo.vdp.jdbc.Driver"

# ---------------- CONNECT TO DENODO ----------------
try:
    cnxn = jaydebeapi.connect(
        driver_class,
        conn_uri,
        {"user": denodoserver_uid, "password": denodoserver_pwd},
        jars=denododriver_path
    )
    cursor = cnxn.cursor()
except Exception as e:
    print("Connection failed:", e)
    exit()

# ---------------- GET DATABASE LIST ----------------
db_list_df = pd.read_sql("SELECT db_name FROM GET_DATABASES()", cnxn)
db_list_df = db_list_df[db_list_df["db_name"] == "sampledb"]
db_list = db_list_df['db_name'].tolist()

# ---------------- STORAGE ----------------
all_views = []
all_dependencies = []
all_datasources = []
datasource_descriptions = []
output_dir="Output"
os.makedirs(output_dir, exist_ok=True)

# Clear all files inside the Output directory
for filename in os.listdir(output_dir):
    file_path = os.path.join(output_dir, filename)
    try:
        if os.path.isfile(file_path):
            os.remove(file_path)
    except Exception as e:
        print(f"Failed to delete {file_path}: {e}")

# ---------------- PROCESS EACH DATABASE ----------------
for db_name in db_list:
    print(f"\nProcessing database: {db_name}")

    # Get list of datasources
    try:
        ds_query = "SELECT database_name, name as datasourceName, subtype FROM GET_ELEMENTS() WHERE type='datasource'"
        ds_df = pd.read_sql(ds_query, cnxn)
        ds_filtered = ds_df[ds_df['database_name'] == db_name]
        all_datasources.append(ds_filtered)

        for _, row in ds_filtered.iterrows():
            subtype = row['subtype'].upper()
            name = row['datasourcename']
            conn_uri = f"jdbc:vdb://{denodoserver_name}:{denodoserver_jdbc_port}/{db_name}"
            cnxn = jaydebeapi.connect(
                driver_class,
                conn_uri,
                {"user": denodoserver_uid, "password": denodoserver_pwd},
                jars=denododriver_path
            )
            desc_stmt = f'DESC DATASOURCE {subtype} "{name}"'
            try:
                db_details = pd.read_sql(desc_stmt, cnxn)
                print("db_details")
                # print(tabulate(db_details, headers='keys', tablefmt='psql'))

                for _, r in db_details.iterrows():
                    db_uri = r.get("database uri") or r.get("data route")
                    database_type = database_type = "route" if r.get("data route") else r.get("database name")
                    ds_source = r["name"]

            except Exception as e:
                print(f"  DESC failed for {name}: {e}")

    except Exception as e:
        print(f"  Skipping datasources in {db_name}: {e}")

    # Get list of views
    try:
        views_df = pd.read_sql(f"SELECT folder,description,database_name,name as view_name,type as view_type FROM GET_VIEWS('{db_name}')", cnxn)
        print("List of views:")
        # print(tabulate(views_df, headers='keys', tablefmt='psql'))
        for _, row in views_df.iterrows():
            viewName = row['view_name']
            # if viewName=="address_j_addresstype":
            all_views.append({'database': db_name, 'view_name': row['view_name'], 'view_type': row['view_type']})


                # Get dependencies for each view
            try:
                    dep_query = f"""
                    SELECT 
                        input_view_database_name as TargetDB,
                        input_view_name as TargetLayerName,
                        column_name as source_column,
                        column_name as Target_column,
                        dependency_type, 
                        dependency_name as source_sql,
                        view_type as TargetobjectType
                    FROM COLUMN_DEPENDENCIES('{db_name}', '{row['view_name']}') 
                    where depth=1
                    """
                    deps_df = pd.read_sql(dep_query, cnxn)
                    deps_df['database'] = db_name
                    folder=row['folder']
                    description=row ['description']
                    view_type=row['view_type']
                    viewName=row['view_name']

                    print(f"folder: { row['folder']}")
                    print(f"viewName: {row['view_name']}")
                    print(f"view_type: {row['view_type']}")
                    print(f"DbName: {db_name}")
                    viewName=row['view_name']

                    desc_stmt = f'DESC VIEW "{viewName}"'
                    try:
                        view_def = pd.read_sql(desc_stmt, cnxn)
                        view_def["databaseName"]=db_name
                        view_def["viewName"]=viewName
                        print("view_def:")
                        # print(tabulate(view_def, headers='keys', tablefmt='psql'))
                        query=f"CALL GET_SOURCE_TABLE('{db_name}','{viewName}')"
                        physical_data=pd.read_sql(  query,cnxn)
                        print("physical_data:")
                        # print(tabulate(physical_data, headers='keys', tablefmt='psql'))

                    except Exception as e:
                        physical_data=pd.DataFrame()
                        print(e)

                    print("deps_df:")
                    # print(tabulate(deps_df, headers='keys', tablefmt='psql'))

                    columns = [
                        "Process Name", "Process Path", "Process Type", "Process Description",
                        "Task Name", "Task Path", "Source Provider Name", "Source Component",
                        "Source Server", "Source Database", "Source Schema", "Source Data Type",
                        "Source Object", "Source Precision", "Source Column", "Source Scale",
                        "Source Object Type", "Target Provider Name", "Target Component",
                        "Target Server", "Target Database", "Target Schema", "Target Object",
                        "Target Column", "Target Data Type", "Target Precision", "Target Scale",
                        "Target Object Type", "Expression", "Link Type", "Link Description"
                    ]

                    # Create an empty DataFrame with the specified columns

                    # Step 1: Define the shared metadata
                    shared_metadata = {
                        "Process Name": viewName,
                        "Process Path": f"{folder}/{viewName}",
                        "Process Type": view_type,
                        "Process Description": description,
                        "Task Name": viewName,
                        "Task Path": f"{db_name}.{viewName}",
                        "ViewName":viewName
                    }

                    # Step 2: Create a DataFrame with the same number of rows as deps_df, and fill it with metadata
                    metadata_df = pd.DataFrame([shared_metadata] * len(deps_df))

                    # Step 3: Concatenate metadata_df and deps_df horizontally
                    df_final = pd.concat([metadata_df.reset_index(drop=True), deps_df.reset_index(drop=True)], axis=1)

                    if not physical_data.empty:
                        physical_data = physical_data.rename(columns={
                            "view_name": "ViewName",
                            "database_name": "database"
                        })

                        merged_df = pd.merge(
                            df_final,
                            physical_data[
                                ["database", "ViewName", "source_catalog_name", "source_schema_name", "source_table_name","view_type"]],
                            on=["database", "ViewName"],
                            how="left"
                        )

                        # Rename merged columns to match your target column names
                        merged_df = merged_df.rename(columns={
                            "source_catalog_name": "Source Database",
                            "source_schema_name": "Source Schema",
                            "source_table_name": "Source Object",
                            "view_type":"Source Object Type"
                        })

                        # Perform the join
                        joined_df = merged_df.merge(db_details, how="left", left_on="source_sql", right_on="name")

                        # Rename 'database uri' to 'Source Server'
                        joined_df = joined_df.rename(columns={"database uri": "Source Server"})
                        joined_df["Source Server"] = joined_df["Source Server"].apply(extract_sql_host)
                        joined_df['Source Provider Name']=joined_df["database name"]



                        final_df = joined_df.merge(
                            view_def,
                            how="left",
                            left_on=["database", "targetlayername", "source_column"],
                            right_on=["databaseName", "viewName", "fieldname"]
                        )

                        # Rename selected columns as specified
                        final_df = final_df.rename(columns={
                            "fieldtype": "Source Data Type",
                            "fieldPrecision": "Source Precision",
                            "fieldDecimals": "Source Scale"
                        })

                        # print(tabulate(final_df, headers='keys', tablefmt='psql'))

                        new_df = pd.DataFrame()
                        new_df["Process Name"] = final_df["Process Name"]
                        new_df["Process Path"] = final_df["Process Path"]
                        new_df["Process Type"] = final_df["Process Type"]
                        new_df["Process Description"] = final_df["Process Description"]
                        new_df["Task Name"] = final_df["Task Name"]
                        new_df["Task Path"] = final_df["Task Path"]
                        new_df["Source Provider Name"] = final_df["Source Provider Name"]
                        new_df["Source Component"] = final_df["Source Object"]
                        new_df["Source Server"] = final_df["Source Server"]
                        new_df["Source Database"] = final_df["Source Database"]
                        new_df["Source Schema"] = final_df["Source Schema"]
                        new_df["Source Data Type"] = final_df["Source Data Type"]
                        new_df["Source Object"] = final_df["Source Object"]
                        new_df["Source Precision"] = final_df["Source Precision"]
                        new_df["Source Column"] = final_df["source_column"]
                        new_df["Source Scale"] = final_df["Source Scale"]
                        new_df["Source Object Type"] = final_df["Source Object Type"]
                        new_df["Target Provider Name"] = "Denodo"
                        new_df["Target Component"] = final_df["targetlayername"]
                        new_df["Target Server"] = ""
                        new_df["Target Database"] = final_df["targetdb"]
                        new_df["Target Schema"] = final_df["targetdb"]
                        new_df["Target Object"] = final_df["targetlayername"]
                        new_df["Target Column"] = final_df["source_column"]
                        new_df["Target Data Type"] = final_df["Source Data Type"]
                        new_df["Target Precision"] = final_df["Source Precision"]
                        new_df["Target Scale"] = final_df["Source Scale"]
                        new_df["Target Object Type"] = final_df["targetobjecttype"]
                        new_df["Expression"] = ""
                        new_df["Link Type"] = "DataFlow"
                        new_df["Link Description"] = ""

                        new_df["Target Precision"] = pd.to_numeric(new_df["Target Precision"], errors="coerce")

                        # Replace values > 100 with -1
                        new_df["Target Precision"] = new_df["Target Precision"].apply(
                            lambda x: -1 if pd.notnull(x) and x > 100 else x
                        )

                        new_df["Source Precision"] = pd.to_numeric(new_df["Source Precision"], errors="coerce")

                        # Replace values > 100 with -1
                        new_df["Source Precision"] = new_df["Source Precision"].apply(
                            lambda x: -1 if pd.notnull(x) and x > 100 else x
                        )


                        print(tabulate(new_df, headers='keys', tablefmt='psql'))

                        filename = f"Output\{db_name}_{viewName}.csv".replace(" ", "_")
                        new_df.to_csv(filename, index=False)
                        print(f"Saved: {filename}")



                    else:

                        print(tabulate(df_final, headers='keys', tablefmt='psql'))
                        df_final = df_final[df_final["targetobjecttype"] != "Base View"]

                        df_final_n = pd.DataFrame()
                        df_final_n["Process Name"] = df_final["Process Name"]
                        df_final_n["Process Path"] = df_final["Process Path"]
                        df_final_n["Process Type"] = df_final["Process Type"]
                        df_final_n["Process Description"] = df_final["Process Description"]
                        df_final_n["Task Name"] = df_final["Task Name"]
                        df_final_n["Task Path"] = df_final["Task Path"]
                        df_final_n["Source Provider Name"] = "Denodo"
                        df_final_n["Source Component"] = df_final["source_sql"]
                        df_final_n["Source Server"] = ""
                        df_final_n["Source Database"] = df_final["targetdb"]
                        df_final_n["Source Schema"] = df_final["targetdb"]
                        df_final_n["Source Data Type"] = ""
                        df_final_n["Source Object"] = df_final["source_sql"]
                        df_final_n["Source Precision"] = ""
                        df_final_n["Source Column"] = df_final["source_column"]
                        df_final_n["Source Scale"] = ""
                        df_final_n["Source Object Type"] = df_final["Process Type"]
                        df_final_n["Target Provider Name"] = "Denodo"
                        df_final_n["Target Component"] = df_final["targetlayername"]
                        df_final_n["Target Server"] = ""
                        df_final_n["Target Database"] = df_final["targetdb"]
                        df_final_n["Target Schema"] = df_final["targetdb"]
                        df_final_n["Target Object"] = df_final["targetlayername"]
                        df_final_n["Target Column"] = df_final["source_column"]
                        df_final_n["Target Data Type"] = ""
                        df_final_n["Target Precision"] = ""
                        df_final_n["Target Scale"] = ""
                        df_final_n["Target Object Type"] = df_final["targetobjecttype"]
                        df_final_n["Expression"] = ""
                        df_final_n["Link Type"] = "DataFlow"
                        df_final_n["Link Description"] = ""

                        df_final_joined = df_final_n.merge(
                            view_def,
                            how="left",
                            left_on=["Target Component", "Target Database", "Target Column"],
                            right_on=["viewName", "databaseName", "fieldname"]
                        )

                        # Update Source and Target columns from view_def
                        df_final_joined["Source Data Type"] = df_final_joined["fieldtype"]

                        df_final_joined["Source Precision"] = pd.to_numeric(df_final_joined["Source Precision"], errors="coerce")

                        # Replace values > 100 with -1
                        df_final_joined["Source Precision"] = df_final_joined["Source Precision"].apply(
                            lambda x: -1 if pd.notnull(x) and x > 100 else x
                        )


                        df_final_joined["Source Scale"] = df_final_joined["fieldDecimals"]
                        df_final_joined["Target Data Type"] = df_final_joined["fieldtype"]
                        df_final_joined["Target Precision"] = df_final_joined["fieldPrecision"]

                        df_final_joined["Target Precision"] = pd.to_numeric(df_final_joined["Target Precision"],
                                                                            errors="coerce")

                        # Replace values > 100 with -1
                        df_final_joined["Target Precision"] = df_final_joined["Target Precision"].apply(
                            lambda x: -1 if pd.notnull(x) and x > 100 else x
                        )


                        df_final_joined["Target Scale"] = df_final_joined["fieldDecimals"]



                        df_final_joined["Target Object Type"]=df_final_joined["Process Type"]
                        columns_to_keep = df_final_n.columns.tolist()
                        df_final_joined_cleaned = df_final_joined[columns_to_keep]
                        filename = f"Output\{db_name}_{viewName}.csv".replace(" ", "_")
                        df_final_joined_cleaned.to_csv(filename, index=False)


            except Exception as e:
                    print(f"  Skipping dependency fetch for {row['view_name']} due to error: {e}")

    except Exception as e:
        print(f"  Skipping views in {db_name}: {e}")


# ---------------- COMBINE ALL CSV FILES ----------------
import glob

csv_files = glob.glob(os.path.join(output_dir, "*.csv"))

combined_df = pd.concat((pd.read_csv(f) for f in csv_files), ignore_index=True)

# Save the final combined file
combined_output_path = os.path.join(output_dir, "Denodo_Links_all.csv")
combined_df.to_csv(combined_output_path, index=False)
print(f"\nUnified CSV saved to: {combined_output_path}")

filtered_df_target = combined_df[combined_df["Target Object Type"].str.lower() == "view"]

# Reuse the combined_df created earlier
source_objects = extract_objects(filtered_df_target, is_source=True)
target_objects = extract_objects(filtered_df_target, is_source=False)

# Combine and drop duplicates
denodo_objects_df = pd.concat([source_objects, target_objects], ignore_index=True).drop_duplicates()

# Save the final objects file
objects_output_path = os.path.join(output_dir, "Denodo_Objects_Final.csv")
denodo_objects_df.to_csv(objects_output_path, index=False)
print(f"Unified object metadata saved to: {objects_output_path}")