coulomb/kontextual-engine
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

projection-only multifiling

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-07 02:52:49 +02:00
parent 801d2f4851
commit 81e132b33b
9 changed files with 250 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
src/kontextual_engine/core/cmis.py
View File

@@ -48,6 +48,7 @@ class CMISAction(str, Enum):
GET_REPOSITORY_INFO = "get_repository_info"
GET_TYPE_DEFINITION = "get_type_definition"
GET_CHILDREN = "get_children"
GET_OBJECT_PARENTS = "get_object_parents"
GET_OBJECT = "get_object"
GET_CONTENT_STREAM = "get_content_stream"
GET_ACL = "get_acl"
@@ -76,6 +77,7 @@ ACTION_CAPABILITIES: dict[CMISAction, CMISCapability] = {
CMISAction.GET_REPOSITORY_INFO: CMISCapability.REPOSITORY,
CMISAction.GET_TYPE_DEFINITION: CMISCapability.TYPE_DEFINITIONS,
CMISAction.GET_CHILDREN: CMISCapability.NAVIGATION,
CMISAction.GET_OBJECT_PARENTS: CMISCapability.NAVIGATION,
CMISAction.GET_OBJECT: CMISCapability.OBJECT_READ,
CMISAction.GET_CONTENT_STREAM: CMISCapability.CONTENT_STREAM_READ,
CMISAction.GET_ACL: CMISCapability.ACL,
@@ -486,7 +488,7 @@ class CMISDomainMapper:
"capability_renditions": "read" if profile.has_capability(CMISCapability.RENDITIONS) else "none",
"capability_get_descendants": profile.has_capability(CMISCapability.NAVIGATION),
"capability_get_folder_tree": profile.has_capability(CMISCapability.NAVIGATION),
"capability_multifiling": False,
"capability_multifiling": profile.has_capability(CMISCapability.NAVIGATION),
"capability_unfiling": False,
"capability_version_specific_filing": False,
"capability_pwc_searchable": False,
@@ -613,21 +615,67 @@ class CMISDomainMapper:
def asset_object_id(self, asset_id: str) -> str:
return f"cmis:asset:{asset_id}"
def folder_object_id(self, path: str) -> str:
return "cmis:folder:" + _normalize_path(path).strip("/").replace("/", "::")
def asset_path(self, asset: KnowledgeAsset) -> str:
return self.asset_paths(asset)[0]
def asset_paths(self, asset: KnowledgeAsset) -> tuple[str, ...]:
paths: list[str] = []
explicit = asset.metadata.get("cmis_path")
if explicit:
return _normalize_path(str(explicit))
paths.append(_normalize_path(str(explicit)))
for value in asset.metadata.get("cmis_paths", ()):
paths.append(_normalize_path(str(value)))
if asset.source_refs:
source_ref = asset.source_refs[0]
source_root = _safe_path_segment(source_ref.source_system)
if source_ref.path:
return _normalize_path(f"/sources/{source_root}/{source_ref.path}")
paths.append(_normalize_path(f"/sources/{source_root}/{source_ref.path}"))
if source_ref.external_id:
return _normalize_path(f"/sources/{source_root}/{source_ref.external_id}")
topics = asset.classification.topics
if topics:
return _normalize_path(f"/topics/{topics[0]}/{asset.id}")
return _normalize_path(f"/assets/{asset.classification.asset_type}/{asset.id}")
paths.append(_normalize_path(f"/sources/{source_root}/{source_ref.external_id}"))
for topic in asset.classification.topics:
paths.append(_normalize_path(f"/topics/{topic}/{asset.id}"))
if asset.classification.owner:
paths.append(_normalize_path(f"/owners/{asset.classification.owner}/{asset.id}"))
paths.append(_normalize_path(f"/lifecycle/{_enum_value(asset.lifecycle)}/{asset.id}"))
paths.append(_normalize_path(f"/assets/{asset.classification.asset_type}/{asset.id}"))
return tuple(dict.fromkeys(paths))
def parent_folders_for_asset(self, asset: KnowledgeAsset) -> tuple[dict[str, Any], ...]:
folders = []
for path in self.asset_paths(asset):
parent = _parent_path(path)
folders.append(
{
"object_id": self.folder_object_id(parent),
"path": parent,
"name": _path_name(parent),
"base_type_id": CMISBaseType.FOLDER.value,
"type_id": "kontextual:folder",
"filing_source": _filing_source(parent),
}
)
return tuple({folder["path"]: folder for folder in folders}.values())
def folder_projection(self, path: str) -> dict[str, Any]:
normalized = _normalize_path(path)
return {
"object_id": self.folder_object_id(normalized),
"base_type_id": CMISBaseType.FOLDER.value,
"type_id": "kontextual:folder",
"name": _path_name(normalized),
"path": normalized,
"properties": {
"cmis:objectId": self.folder_object_id(normalized),
"cmis:name": _path_name(normalized),
"cmis:baseTypeId": CMISBaseType.FOLDER.value,
"cmis:objectTypeId": "kontextual:folder",
"kontextual:filingSource": _filing_source(normalized),
},
"allowable_actions": [CMISAction.GET_CHILDREN.value],
}
def asset_properties(
self,
@@ -716,6 +764,7 @@ class CMISDomainMapper:
CMISAction.GET_CONTENT_STREAM,
CMISAction.GET_ACL,
CMISAction.GET_RELATIONSHIPS,
CMISAction.GET_OBJECT_PARENTS,
CMISAction.UPDATE_PROPERTIES,
CMISAction.DELETE_OBJECT,
CMISAction.SET_CONTENT_STREAM,
@@ -819,3 +868,22 @@ def _normalize_path(path: str) -> str:
def _safe_path_segment(value: str) -> str:
return str(value).strip().strip("/") or "_"
def _parent_path(path: str) -> str:
parts = _normalize_path(path).strip("/").split("/")
if len(parts) <= 1:
return "/"
return "/" + "/".join(parts[:-1])
def _path_name(path: str) -> str:
normalized = _normalize_path(path)
if normalized == "/":
return "root"
return normalized.rsplit("/", 1)[-1]
def _filing_source(path: str) -> str:
parts = _normalize_path(path).strip("/").split("/")
return parts[0] if parts and parts[0] else "root"