feat: implement offline page package

simontaurus · simontaurus · commit 349b1bd7eb82 · 2026-01-03T04:27:31.000+01:00
diff --git a/examples/offline_content_package.py b/examples/offline_content_package.py
@@ -0,0 +1,105 @@
+import os
+from pathlib import Path
+
+from osw import wiki_tools
+from osw.auth import CredentialManager
+from osw.core import OSW
+from osw.model.page_package import PagePackage, PagePackageBundle, PagePackageConfig
+from osw.params import CreatePagePackageParam, PageDumpConfig
+from osw.wtsite import WtSite
+
+# Create/update the password file under examples/accounts.pwd.yaml
+pwd_file_path = os.path.join(
+    os.path.dirname(os.path.abspath(__file__)), "accounts.pwd.yaml"
+)
+
+# login to demo.open-semantic-lab.org
+osw_obj = OSW(
+    site=WtSite(
+        WtSite.WtSiteConfig(
+            iri="demo.open-semantic-lab.org",
+            cred_mngr=CredentialManager(cred_filepath=pwd_file_path),
+        )
+    )
+)
+
+# download all pages in a PagePackage
+# pages = wiki_tools.semantic_search(
+# osw_obj.site._site,
+# "[[Category:Entity]] OR [[:Category:+]]
+# )
+# only classes
+pages = wiki_tools.semantic_search(osw_obj.site._site, "[[HasType::Category:Category]]")
+print(f"Found {len(pages)} Entity pages.")
+target_dir = Path("osw_files") / "packages" / "local.offline"
+package = osw_obj.site.create_page_package(
+    CreatePagePackageParam(
+        config=PagePackageConfig(
+            name="local.offline",
+            config_path=target_dir / "package.json",
+            content_path=target_dir,  # / "content",
+            titles=pages,
+            bundle=PagePackageBundle(
+                packages={
+                    "local.offline": PagePackage(
+                        globalID="local.offline",
+                        description="Offline OSW package example",
+                        version="0.1.0",
+                        baseURL="http://local.offline",
+                    )
+                }
+            ),
+        ),
+        dump_config=PageDumpConfig(target_dir=target_dir),
+    )
+)
+
+# load pages from local offline package
+offline_pages = {}
+result = osw_obj.site.read_page_package(
+    WtSite.ReadPagePackageParam(
+        storage_path=target_dir,
+    )
+)
+offline_pages = {page.title: page for page in result.pages}
+print(offline_pages["Category:Category"])
+res = osw_obj.load_entity(
+    OSW.LoadEntityParam(titles=pages, offline_pages=offline_pages)
+)
+
+result = osw_obj.export_jsonld(
+    params=OSW.ExportJsonLdParams(
+        entities=res.entities,
+        mode=OSW.JsonLdMode.expand,
+        build_rdf_graph=True,
+        context_loader_config=WtSite.JsonLdContextLoaderParams(
+            prefer_external_vocal=False, offline_pages=offline_pages
+        ),
+    )
+)
+
+graph = result.graph
+# all triples in the graph
+qres = graph.query(
+    """
+    SELECT ?s ?p ?o
+    WHERE {
+    ?s ?p ?o .
+    }
+    """
+)
+# Count triples
+print(f"\nTotal triples in the graph: {len(qres)}")
+
+# query all properties of Category:Item
+qres = graph.query(
+    """
+    SELECT ?property ?object
+    WHERE {
+    Category:Item ?property ?object .
+    }
+    """
+)
+print("\nProperties of Category:Item:")
+for row in qres:
+    print(row.property, row.object)
diff --git a/src/osw/core.py b/src/osw/core.py
@@ -348,6 +348,14 @@ class FetchSchemaParam(BaseModel):
         )
         legacy_generator: Optional[bool] = False
         """uses legacy command line for code generation if true"""
+        offline_pages: Optional[Dict[str, WtPage]] = None
+        """pages to be used offline instead of fetching them from the OSW instance"""
+        result_model_path: Optional[Union[str, pathlib.Path]] = None
+        """path to the generated model file, if None,
+        the default path ./model/entity.py is used"""
+
+        class Config:
+            arbitrary_types_allowed = True
 
     def fetch_schema(self, fetchSchemaParam: FetchSchemaParam = None) -> None:
         """Loads the given schemas from the OSW instance and auto-generates python
@@ -370,6 +378,8 @@ def fetch_schema(self, fetchSchemaParam: FetchSchemaParam = None) -> None:
                     schema_title=schema_title,
                     mode=mode,
                     legacy_generator=fetchSchemaParam.legacy_generator,
+                    offline_pages=fetchSchemaParam.offline_pages,
+                    result_model_path=fetchSchemaParam.result_model_path,
                 )
             )
             first = False
@@ -396,6 +406,14 @@ class _FetchSchemaParam(BaseModel):
         )
         legacy_generator: Optional[bool] = False
         """uses legacy command line for code generation if true"""
+        offline_pages: Optional[Dict[str, WtPage]] = None
+        """pages to be used offline instead of fetching them from the OSW instance"""
+        result_model_path: Optional[Union[str, pathlib.Path]] = None
+        """path to the generated model file, if None,
+        the default path ./model/entity.py is used"""
+
+        class Config:
+            arbitrary_types_allowed = True
 
     def _fetch_schema(self, fetchSchemaParam: _FetchSchemaParam = None) -> None:
         """Loads the given schema from the OSW instance and autogenerates python
@@ -413,11 +431,21 @@ def _fetch_schema(self, fetchSchemaParam: _FetchSchemaParam = None) -> None:
         schema_title = fetchSchemaParam.schema_title
         root = fetchSchemaParam.root
         schema_name = schema_title.split(":")[-1]
-        page = self.site.get_page(WtSite.GetPageParam(titles=[schema_title])).pages[0]
-        if not page.exists:
-            print(f"Error: Page {schema_title} does not exist")
-            return
-        # not only in the JsonSchema namespace the schema is located in the main sot
+        if (
+            fetchSchemaParam.offline_pages is not None
+            and schema_title in fetchSchemaParam.offline_pages
+        ):
+            print(f"Fetch {schema_title} from offline pages")
+            page = fetchSchemaParam.offline_pages[schema_title]
+        else:
+            print(f"Fetch {schema_title} from online pages")
+            page = self.site.get_page(WtSite.GetPageParam(titles=[schema_title])).pages[
+                0
+            ]
+            if not page.exists:
+                print(f"Error: Page {schema_title} does not exist")
+                return
+        # not only in the JsonSchema namespace the schema is located in the main slot
         # in all other namespaces, the json_schema slot is used
         if schema_title.startswith("JsonSchema:"):
             schema_str = ""
@@ -441,7 +469,6 @@ def _fetch_schema(self, fetchSchemaParam: _FetchSchemaParam = None) -> None:
             )
             # fix https://github.com/koxudaxi/datamodel-code-generator/issues/1910
         )
-        print(f"Fetch {schema_title}")
 
         jsonpath_expr = parse("$..dollarref")
         for match in jsonpath_expr.find(schema):
@@ -462,9 +489,10 @@ def _fetch_schema(self, fetchSchemaParam: _FetchSchemaParam = None) -> None:
             if (
                 ref_schema_title != schema_title
             ):  # prevent recursion in case of self references
-                self._fetch_schema(
-                    OSW._FetchSchemaParam(schema_title=ref_schema_title, root=False)
-                )  # resolve references recursive
+                _param = fetchSchemaParam.copy()
+                _param.root = False
+                _param.schema_title = ref_schema_title
+                self._fetch_schema(_param)  # resolve references recursive
 
         model_dir_path = os.path.join(
             os.path.dirname(os.path.abspath(__file__)), "model"
@@ -730,6 +758,11 @@ class LoadEntityParam(BaseModel):
         from the jsondata."""
         disable_cache: bool = False
         """If true, disable the cache for the loading process"""
+        offline_pages: Optional[Dict[str, WtPage]] = None
+        """pages to be used offline instead of fetching them from the OSW instance"""
+
+        class Config:
+            arbitrary_types_allowed = True  # allow any class as type
 
         def __init__(self, **data):
             super().__init__(**data)
@@ -795,7 +828,9 @@ def load_entity(
             self.site.enable_cache()
 
         entities = []
-        pages = self.site.get_page(WtSite.GetPageParam(titles=param.titles)).pages
+        pages = self.site.get_page(
+            WtSite.GetPageParam(titles=param.titles, offline_pages=param.offline_pages)
+        ).pages
         for page in pages:
             entity = None
             schemas = []
@@ -806,7 +841,11 @@ def load_entity(
             if jsondata:
                 for category in jsondata["type"]:
                     schema = (
-                        self.site.get_page(WtSite.GetPageParam(titles=[category]))
+                        self.site.get_page(
+                            WtSite.GetPageParam(
+                                titles=[category], offline_pages=param.offline_pages
+                            )
+                        )
                         .pages[0]
                         .get_slot_content("jsonschema")
                     )
@@ -820,7 +859,9 @@ def load_entity(
                             if param.autofetch_schema:
                                 self.fetch_schema(
                                     OSW.FetchSchemaParam(
-                                        schema_title=category, mode="append"
+                                        schema_title=category,
+                                        mode="append",
+                                        offline_pages=param.offline_pages,
                                     )
                                 )
                         if not hasattr(model, cls_name):
@@ -1586,6 +1627,10 @@ class ExportJsonLdParams(OswBaseModel):
         build_rdf_graph: Optional[bool] = False
         """If True, the output is a graph."""
         debug: Optional[bool] = False
+        """If True, debug information is printed."""
+
+        class Config:
+            arbitrary_types_allowed = True
 
         def __init__(self, **data):
             super().__init__(**data)
diff --git a/src/osw/wtsite.py b/src/osw/wtsite.py
@@ -251,6 +251,11 @@ class GetPageParam(OswBaseModel):
         """Whether to raise an exception if an error occurs"""
         raise_warning: Optional[bool] = True
         """Whether to raise a warning if a page does not exist occurs"""
+        offline_pages: Optional[Dict[str, "WtPage"]] = None
+        """pages to be used offline instead of fetching them from the OSW instance"""
+
+        class Config:
+            arbitrary_types_allowed = True  # allows to use WtPage in type hints
 
         def __init__(self, **data):
             super().__init__(**data)
@@ -292,7 +297,11 @@ def get_page_(title: str, index: int = None):
                 if index is not None:
                     msg = f"({index + 1}/{max_index}) "
                 try:
-                    if self._cache_enabled and title in self._page_cache:
+                    if param.offline_pages and title in param.offline_pages:
+                        wtpage = param.offline_pages[title]
+                        wtpage.exists = True
+                        msg += "Page loaded from offline pages. "
+                    elif self._cache_enabled and title in self._page_cache:
                         wtpage = self._page_cache[title]
                         msg += "Page loaded from cache. "
                     else:
@@ -954,18 +963,25 @@ def read_page_package(self, param: ReadPagePackageParam) -> ReadPagePackageResul
             search_path=storage_path, recursive=True
         )
         sub_dirs = top_level_content["directories"]
-        if len(top_level_content["directories"]) == 0:
+        if len(sub_dirs) == 0:
             # No subdirectories found, assume that the pages files are located in the
             #  top level
             sub_dirs = [storage_path]
 
+        # if subdirs are namespaces, skip them as well
+        if any(
+            d.name in ["Category", "Item", "File", "Module"]
+            for d in storage_path_content["directories"]
+        ):
+            sub_dirs = [storage_path]
+
         def get_slot_content(
             parent_dir: List[Union[str, Path]],
             url_path: str,
             files_in_storage_path: List[Path],
         ) -> Union[str, Dict]:
             for pdir in parent_dir:
-                slot_path = storage_path / pdir / url_path
+                slot_path = pdir / url_path
                 if slot_path in files_in_storage_path:
                     with open(slot_path, "r", encoding="utf-8") as f:
                         file_content = f.read()
@@ -986,8 +1002,18 @@ def get_slot_content(
                 namespace = page["namespace"].split("_")[-1].capitalize()
                 name = page["name"]
                 # Create the WtPage object
-                page_obj = WtPage(wtSite=self, title=f"{namespace}:{name}")
-                if "main" in selected_slots:
+                page_obj = WtPage(
+                    wtSite=self, title=f"{namespace}:{name}", do_init=False
+                )
+                if selected_slots is None:
+                    _selected_slots = page["slots"]
+                else:
+                    _selected_slots = {
+                        slot_name: slot_dict
+                        for slot_name, slot_dict in page["slots"].items()
+                        if slot_name in selected_slots
+                    }
+                if "main" in _selected_slots:
                     # Main slot is special
                     slot_content = get_slot_content(
                         parent_dir=sub_dirs,
@@ -999,14 +1025,6 @@ def get_slot_content(
                             slot_key="main",
                             content=slot_content,
                         )
-                if selected_slots is None:
-                    _selected_slots = page["slots"]
-                else:
-                    _selected_slots = {
-                        slot_name: slot_dict
-                        for slot_name, slot_dict in page["slots"].items()
-                        if slot_name in selected_slots
-                    }
                 for slot_name, slot_dict in _selected_slots.items():
                     slot_content = get_slot_content(
                         parent_dir=sub_dirs,
@@ -1125,6 +1143,12 @@ class JsonLdContextLoaderParams(OswBaseModel):
         prefer_external_vocal: Optional[bool] = True
         """Whether to prefer external vocabularies (e.g. skos, schema, etc.)
         over the local properties (Namespace 'Property:')"""
+        offline_pages: Optional[Dict[str, "WtPage"]] = None
+        """A dictionary of pages that are already loaded. Pages in this dictionary
+        will not be fetched again."""
+
+        class Config:
+            arbitrary_types_allowed = True
 
     def _replace_jsonld_context_mapping(
         self, context: Union[str, list, dict], config: JsonLdContextLoaderParams
@@ -1202,7 +1226,11 @@ def loader(url, options=None):
             # print("Requesting", url)
             if "/wiki/" in url:
                 title = url.split("/wiki/")[-1].split("?")[0]
-                page = self.get_page(WtSite.GetPageParam(titles=[title])).pages[0]
+                page = self.get_page(
+                    WtSite.GetPageParam(
+                        titles=[title], offline_pages=params.offline_pages
+                    )
+                ).pages[0]
                 if "JsonSchema:" in title:
                     schema = page.get_slot_content("main")
                 else:
@@ -2177,8 +2205,10 @@ def import_xml(self, config: ImportConfig) -> ImportResult:
 WtPage.PageCopyResult.update_forward_refs()
 WtSite.CopyPagesParam.update_forward_refs()
 WtSite.UploadPageParam.update_forward_refs()
+WtSite.GetPageParam.update_forward_refs()
 WtSite.GetPageResult.update_forward_refs()
 WtSite.CreatePagePackageParam.update_forward_refs()
 WtSite.UploadPagePackageParam.update_forward_refs()
 WtSite.ReadPagePackageResult.update_forward_refs()
 WtSite.DeletePageParam.update_forward_refs()
+WtSite.JsonLdContextLoaderParams.update_forward_refs()
