Generate the `kg2c_v1.0_KG2.10.3.sqlite` file **KG2.10.3 DEV ROADMAP SUB-ISSUE**

[hodgesf]

[hodgesf]

The script mentioned unfortunately did not work for our purposes, so I wrote a new one, as seen below. The script will be located in RTXTeam/RTX-KG2/process/ and is called generate_sqlite.py. The SQLite file has been generated, but needs to be validated. The file is located on the buildkg2103.rtx.ai instance, and is located in the kg2-build directory.

#!/usr/bin/env python3

"""
Rebuild KG2c-style SQLite database from conflated JSONL files.

This script reproduces the original KG2c SQLite structure, including:

    • Delimiter-encoded list fields
    • Deterministic triple key
    • node_pair column
    • Unique triple index
    • node_id index

The resulting SQLite database matches the legacy KG2c schema.
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sqlite3
from pathlib import Path
from typing import Any, Dict, Iterable, Mapping, Optional, Set


KG2C_ARRAY_DELIMITER = "ǂ"


# ---------------------------------------------------------------------
# Serialization helpers (legacy behavior)
# ---------------------------------------------------------------------

def _convert_list_to_string_encoded_format(
    value: Optional[Iterable[str]]
) -> str:
    if not value:
        return ""

    cleaned: list[str] = []
    for item in value:
        if not item:
            continue
        if isinstance(item, str):
            cleaned.append(item)
        else:
            logging.warning("List contains non-str items; excluding them")

    return KG2C_ARRAY_DELIMITER.join(cleaned)


def _prep_for_sqlite(value: Any) -> str:
    if not value:
        return ""

    if isinstance(value, str):
        return value

    if isinstance(value, dict):
        return json.dumps(value)

    if isinstance(value, Iterable) and not isinstance(value, (str, bytes)):
        return _convert_list_to_string_encoded_format(value)

    return str(value)


def _get_edge_key(
    subject: str,
    object: str,
    predicate: str,
    primary_knowledge_source: str,
    qualified_predicate: Optional[str],
    qualified_object_direction: Optional[str],
    qualified_object_aspect: Optional[str],
) -> str:

    qualified_portion = "--".join(
        [
            qualified_predicate or "",
            qualified_object_direction or "",
            qualified_object_aspect or "",
        ]
    )

    return "--".join(
        [
            subject,
            predicate,
            qualified_portion,
            object,
            primary_knowledge_source,
        ]
    )


# ---------------------------------------------------------------------
# Explicit property schema (legacy KG2c)
# ---------------------------------------------------------------------

NODE_PROPERTIES: Set[str] = {
    "id",
    "publications",
    "description",
    "name",
    "category",
    "synonym",
    "taxon",
    "all_categories",
    "iri",
    "same_as",
}

EDGE_PROPERTIES: Set[str] = {
    "subject",
    "object",
    "id",
    "kg2_ids",
    "predicate",
    "agent_type",
    "knowledge_level",
    "primary_knowledge_source",
    "domain_range_exclusion",
    "qualified_predicate",
    "object_direction_qualifier",
    "object_aspect_qualifier",
    "publications",
    "publications_info",
}


# ---------------------------------------------------------------------
# Core builder
# ---------------------------------------------------------------------

def create_kg2c_sqlite_db(
    nodes_dict: Dict[str, Mapping[str, Any]],
    edges_dict: Dict[str, Mapping[str, Any]],
    output_db: Path,
) -> None:

    if output_db.exists():
        output_db.unlink()

    connection = sqlite3.connect(output_db)

    # -------------------------
    # Nodes
    # -------------------------

    sqlite_node_properties = sorted(NODE_PROPERTIES)

    cols_with_types = ", ".join(f"{col} TEXT" for col in sqlite_node_properties)
    connection.execute(f"CREATE TABLE nodes ({cols_with_types})")

    question_marks = ", ".join("?" for _ in sqlite_node_properties)

    node_rows = [
        [
            _prep_for_sqlite(node.get(prop))
            for prop in sqlite_node_properties
        ]
        for node in nodes_dict.values()
    ]

    connection.executemany(
        f"INSERT INTO nodes VALUES ({question_marks})",
        node_rows,
    )

    connection.execute("CREATE UNIQUE INDEX node_id_index ON nodes (id)")
    connection.commit()

    # -------------------------
    # Edges
    # -------------------------

    sqlite_edge_properties = sorted(EDGE_PROPERTIES)

    cols_with_types = ", ".join(f"{col} TEXT" for col in sqlite_edge_properties)

    connection.execute(
        f"CREATE TABLE edges (triple TEXT, node_pair TEXT, {cols_with_types})"
    )

    question_marks = ", ".join("?" for _ in sqlite_edge_properties)

    edge_rows = []

    for edge in edges_dict.values():

        triple = _get_edge_key(
            subject=edge["subject"],
            object=edge["object"],
            predicate=edge["predicate"],
            primary_knowledge_source=edge.get("primary_knowledge_source", ""),
            qualified_predicate=edge.get("qualified_predicate"),
            qualified_object_direction=edge.get("qualified_object_direction"),
            qualified_object_aspect=edge.get("qualified_object_aspect"),
        )

        node_pair = f"{edge['subject']}--{edge['object']}"

        serialized_props = [
            _prep_for_sqlite(edge.get(prop))
            for prop in sqlite_edge_properties
        ]

        edge_rows.append([triple, node_pair] + serialized_props)

    connection.executemany(
        f"INSERT INTO edges (triple, node_pair, {', '.join(sqlite_edge_properties)}) "
        f"VALUES (?, ?, {question_marks})",
        edge_rows,
    )

    connection.execute("CREATE UNIQUE INDEX triple_index ON edges (triple)")
    connection.execute("CREATE INDEX node_pair_index ON edges (node_pair)")
    connection.commit()

    connection.close()


# ---------------------------------------------------------------------
# JSONL loader
# ---------------------------------------------------------------------

def load_jsonl_as_dict(path: Path, key_field: str) -> Dict[str, Dict[str, Any]]:
    result: Dict[str, Dict[str, Any]] = {}
    with path.open() as f:
        for line in f:
            obj = json.loads(line)
            result[obj[key_field]] = obj
    return result


# ---------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Rebuild legacy KG2c SQLite from conflated JSONL files."
    )
    parser.add_argument("--nodes", type=Path, required=True)
    parser.add_argument("--edges", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)

    args = parser.parse_args()

    logging.basicConfig(level=logging.INFO)

    logging.info("Loading nodes...")
    nodes = load_jsonl_as_dict(args.nodes, "id")

    logging.info("Loading edges...")
    edges = load_jsonl_as_dict(args.edges, "id")

    logging.info("Building SQLite...")
    create_kg2c_sqlite_db(nodes, edges, args.output)

    logging.info("Done.")


if __name__ == "__main__":
    main()