tools.py

from rdflib import Graph, RDF, RDFS, OWL, term, URIRef

from rdflib.exceptions import ParserError
from smolagents import tool


@tool
def validate_rdf_syntax(
    file_path: str, file_format: str = "turtle"
) -> tuple[bool, str]:
    """
    Validates the syntax of an RDF file.

    This function parses an RDF file using the specified format and checks if it
    is syntactically valid.  It handles potential parsing errors and other
    exceptions that may occur during the validation process.

    Args:
        file_path (str): The path to the RDF file to validate.
        file_format (str, optional): The RDF file format.
            Defaults to "turtle".  Other common formats include "xml" (for RDF/XML),
            "nt" (for N-Triples), and "nq" (for N-Quads).

    Returns:
        tuple: A tuple containing two elements:
            - bool: True if the RDF syntax is valid, False otherwise.
            - str: A message describing the validation result.  If the syntax is
              valid, the message is "RDF syntax is valid.".  If there is an error,
              the message provides details about the error.

    Raises:
        ParserError: If the RDF file has invalid syntax according to the specified format.
        Exception: For any other unexpected error during file processing or parsing.
    """
    g = Graph()
    try:
        g.parse(file_path, format=file_format)
        return True, "RDF syntax is valid."
    except ParserError as e:
        return False, f"RDF syntax error: {e}"
    except Exception as e:
        return False, f"An unexpected error occurred: {e}"


@tool
def write_rdf_to_file(filename: str, text: str) -> None:
    """
    Writes the given text to a file.

    Args:
        filename (str): The name of the file to write to, without extension.
        text (str): The text to write to the file.
    """
    try:
        with open(f"{filename}.ttl", "w") as file:
            file.write(text)
        print(f"Successfully wrote text to {filename}")
    except Exception as e:
        print(f"An error occurred while writing to {filename}: {e}")


@tool
def get_entities_from_kg() -> list:
    """
    Loads a knowledge graph and returns a list of identified entity URIs.
    Entities are typically resources with an rdf:type linking them to a class,
    or those explicitly declared as owl:NamedIndividual.

    Args:
        graph_path (str): The path to the knowledge graph file (local path or URL).
        format (str, optional): The format of the graph file (e.g., 'xml', 'turtle', 'json-ld').
                                 If None, rdflib will try to guess the format based on the file extension.

    Returns:
        list: A list of rdflib.URIRef or rdflib.BNode objects representing the entities.
    """
    return []


def parse_types_graph(g):
    types = set()

    # Query for classes defined using rdfs:Class
    for s, p, o in g.triples((None, RDF.type, RDFS.Class)):
        types.add(s)

    # Query for classes defined using owl:Class
    for s, p, o in g.triples((None, RDF.type, OWL.Class)):
        types.add(s)

    # You might also find classes as subjects of rdfs:subClassOf triples
    # Although this doesn't strictly define a class, it implies the subject is a class
    for s, p, o in g.triples((None, RDFS.subClassOf, None)):
        types.add(s)
        # Also add the object, as it must also be a class
        if isinstance(o, term.URIRef) or isinstance(o, term.BNode):
            types.add(o)

    return sorted(list(types))


def parse_relations_graph(g):
    predicates = set()

    # Query for resources explicitly typed as properties
    property_types = [
        RDF.Property,
        OWL.ObjectProperty,
        OWL.DatatypeProperty,
        OWL.AnnotationProperty,
    ]

    for prop_type in property_types:
        for s, p, o in g.triples((None, RDF.type, prop_type)):
            predicates.add(s)

    # Also consider any URI or BNode that is used as a predicate in any triple
    # These are implicitly properties, even if not explicitly typed
    for s, p, o in g.triples((None, None, None)):
        if isinstance(p, term.URIRef) or isinstance(p, term.BNode):
            predicates.add(p)

    return sorted(list(predicates))


@tool
def get_types_from_ontology() -> list[URIRef]:
    """
    Loads an ontology and returns a list of defined class URIs.

    Args:
        ontology_path (str): The path to the ontology file (local path or URL).
        format (str, optional): The format of the ontology file (e.g., 'xml', 'turtle', 'json-ld').
                                 If None, rdflib will try to guess the format based on the file extension.

    Returns:
        list: A list of rdflib.URIRef objects representing the defined classes.
    """
    g = Graph()

    try:
        g.parse("./sources/cacao.owl", format=None)
    except Exception as e:
        print(f"Error loading ontology: {e}")
        return []

    return parse_types_graph(g)


@tool
def get_relations_from_ontology() -> list[URIRef]:
    """
    Loads an ontology and returns a list of defined predicate (property) URIs.

    Args:
        ontology_path (str): The path to the ontology file (local path or URL).
        format (str, optional): The format of the ontology file (e.g., 'xml', 'turtle', 'json-ld').
                                 If None, rdflib will try to guess the format based on the file extension.

    Returns:
        list: A list of rdflib.URIRef objects representing the defined predicates.
    """
    g = Graph()

    try:
        g.parse("./sources/cacao.owl", format=None)
    except Exception as e:
        print(f"Error loading ontology: {e}")
        return []

    return parse_relations_graph(g)