midpoint - go

python_services/owl_imports_combiner/main.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import logging
+import os
+
+from owl_imports_combiner import (
+    build_combined_graph,
+    output_location_to_path,
+    resolve_output_location,
+    serialize_graph_to_ttl,
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+def _env_bool(name: str, *, default: bool = False) -> bool:
+    val = os.getenv(name)
+    if val is None:
+        return default
+    return val.strip().lower() in {"1", "true", "yes", "y", "on"}
+
+
+def main() -> None:
+    logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO").upper())
+
+    if not _env_bool("COMBINE_OWL_IMPORTS_ON_START", default=False):
+        logger.info("Skipping combine step (COMBINE_OWL_IMPORTS_ON_START=false)")
+        return
+
+    entry_location = os.getenv("COMBINE_ENTRY_LOCATION") or os.getenv("TTL_PATH")
+    if not entry_location:
+        raise SystemExit("Set COMBINE_ENTRY_LOCATION (or TTL_PATH) to the ontology file/URL to load.")
+
+    output_name = os.getenv("COMBINE_OUTPUT_NAME", "combined_ontology.ttl")
+    output_location = resolve_output_location(
+        entry_location,
+        output_location=os.getenv("COMBINE_OUTPUT_LOCATION"),
+        output_name=output_name,
+    )
+
+    output_path = output_location_to_path(output_location)
+    force = _env_bool("COMBINE_FORCE", default=False)
+    if output_path.exists() and not force:
+        logger.info("Skipping combine step (output exists): %s", output_location)
+        return
+
+    graph = build_combined_graph(entry_location)
+    logger.info("Finished combining imports; serializing to: %s", output_location)
+    serialize_graph_to_ttl(graph, output_location)
+
+
+if __name__ == "__main__":
+    main()