V2 cataloger

backend/.env.example

@@ -1,4 +1,7 @@
 OPENPECHA_ENDPOINT=
+# OpenPecha API v2 (X-API-Key) and default X-Application for /v2 routes
+OPENPECHA_API_KEY=
+OPENPECHA_X_APPLICATION=
 API_ENDPOINT=
 BDRC_SEARCH_ENDPOINT=
 PORT=8000

backend/cataloger/controller/openpecha_api/__init__.py

@@ -0,0 +1 @@
+"""HTTP client helpers for the external OpenPecha API."""

backend/cataloger/controller/openpecha_api/annotations.py

@@ -0,0 +1,115 @@
+from typing import Any, Callable, Dict, List, Literal, Optional, Tuple
+
+import requests
+from fastapi import HTTPException
+
+from cataloger.controller.openpecha_api.base import json_headers, openpecha_headers, openpecha_url
+from cataloger.controller.openpecha_api.v2_adapters import (
+    alignment_output_to_legacy_data,
+    segmentation_output_to_legacy_data,
+)
+
+
+def _try_get(url: str, *, timeout: Optional[int] = None) -> Optional[requests.Response]:
+    kw: Dict[str, Any] = {}
+    if timeout is not None:
+        kw["timeout"] = timeout
+    r = requests.get(url, headers=openpecha_headers(), **kw)
+    if r.status_code == 200:
+        return r
+    return None
+
+
+def get_annotation(annotation_id: str, *, timeout: Optional[int] = None) -> Any:
+    bases: List[Tuple[str, Callable[[Dict[str, Any]], Dict[str, Any]]]] = [
+        (
+            "annotations/alignment",
+            lambda b: {"data": alignment_output_to_legacy_data(b)},
+        ),
+        ("annotations/segmentation", lambda b: {"data": segmentation_output_to_legacy_data(b)}),
+        ("annotations/pagination", lambda b: {"data": b}),
+        ("annotations/bibliographic", lambda b: {"data": b}),
+        ("annotations/durchen", lambda b: {"data": b}),
+    ]
+    for path_prefix, wrap in bases:
+        r = _try_get(
+            openpecha_url(path_prefix, annotation_id),
+            timeout=timeout,
+        )
+        if r:
+            body = r.json()
+            merged = wrap(body) if isinstance(body, dict) else {"data": body}
+            eid = body.get("id") if isinstance(body, dict) else None
+            merged.setdefault("id", eid or annotation_id)
+            return merged
+    raise HTTPException(status_code=404, detail="Annotation not found for v2 typed paths")
+
+
+def update_annotation_body(annotation_id: str, payload: Dict[str, Any]) -> Any:
+    del annotation_id, payload
+    raise HTTPException(
+        status_code=501,
+        detail="OpenPecha v2 has no generic annotation update; delete and recreate via edition annotations.",
+    )
+
+
+def create_annotation_for_instance(instance_id: str, payload: Dict[str, Any]) -> Any:
+    """instance_id is an edition id; maps legacy union payload to AnnotationRequestInput."""
+    ptype = payload.get("type")
+    ann: Dict[str, Any] = {}
+    if ptype == "segmentation" and payload.get("annotation"):
+        segs = []
+        for item in payload["annotation"]:
+            if isinstance(item, dict) and item.get("span"):
+                segs.append({"lines": [item["span"]]})
+        if segs:
+            ann["segmentation"] = {"segments": segs}
+    elif ptype == "alignment":
+        ta = payload.get("target_annotation") or []
+        aa = payload.get("alignment_annotation") or []
+        target_segments = [
+            {"lines": [x["span"]]} for x in ta if isinstance(x, dict) and x.get("span")
+        ]
+        aligned_segments = []
+        for x in aa:
+            if isinstance(x, dict) and x.get("span"):
+                aligned_segments.append(
+                    {
+                        "lines": [x["span"]],
+                        "alignment_indices": x.get("alignment_index") or [],
+                    }
+                )
+        tid = payload.get("target_manifestation_id") or payload.get("target_id")
+        if not tid:
+            raise HTTPException(status_code=400, detail="alignment requires target_manifestation_id")
+        ann["alignment"] = {
+            "target_id": tid,
+            "target_segments": target_segments,
+            "aligned_segments": aligned_segments,
+        }
+    elif ptype == "table_of_contents":
+        ann["pagination"] = {"volume": {"pages": payload.get("annotation") or []}}
+    if not ann:
+        raise HTTPException(
+            status_code=400,
+            detail="Unsupported or empty annotation payload for v2",
+        )
+    r = requests.post(
+        openpecha_url("editions", instance_id, "annotations"),
+        json=ann,
+        headers=json_headers(),
+        timeout=120,
+    )
+    if r.status_code not in (200, 201):
+        raise HTTPException(status_code=r.status_code, detail=r.text)
+    return r.json()
+
+
+
+annotationTypes = Literal["alignments", "segmentations", "paginations", "bibliographic", "durchens"]
+def delete_annotation(type: annotationTypes, annotation_id: str) -> Any:
+    url = openpecha_url(type, annotation_id)
+    r = requests.delete(url, headers=openpecha_headers(), timeout=60)
+    if r.status_code not in (200, 204):
+        raise HTTPException(status_code=r.status_code, detail=r.text)
+    return {"status": "deleted", "annotation_id": annotation_id}

backend/cataloger/controller/openpecha_api/base.py

@@ -0,0 +1,44 @@
+import os
+from typing import Any, Dict, Optional
+
+from dotenv import load_dotenv
+from fastapi import HTTPException
+
+load_dotenv(override=True)
+
+OPENPECHA_ENDPOINT_ENV = "OPENPECHA_ENDPOINT"
+OPENPECHA_API_KEY_ENV = "OPENPECHA_API_KEY"
+OPENPECHA_X_APPLICATION_ENV = "OPENPECHA_X_APPLICATION"
+MISSING_ENDPOINT_DETAIL = "OPENPECHA_ENDPOINT environment variable is not set"
+
+API_V2_PREFIX = "/v2"
+
+
+def require_openpecha_base_url() -> str:
+    base = os.getenv(OPENPECHA_ENDPOINT_ENV)
+    if not base:
+        raise HTTPException(status_code=500, detail=MISSING_ENDPOINT_DETAIL)
+    return base.rstrip("/")
+
+
+def openpecha_url(*path_parts: str) -> str:
+    base = require_openpecha_base_url()
+    tail = "/".join(p.strip("/") for p in path_parts if p)
+    return f"{base}{API_V2_PREFIX}/{tail}"
+
+
+def openpecha_headers(*, x_application: Optional[str] = None) -> Dict[str, str]:
+    headers: Dict[str, str] = {"Accept": "application/json"}
+    api_key = os.getenv(OPENPECHA_API_KEY_ENV)
+    if api_key:
+        headers["X-API-Key"] = api_key
+    app = x_application if x_application is not None else os.getenv(OPENPECHA_X_APPLICATION_ENV)
+    if app:
+        headers["X-Application"] = app
+    return headers
+
+
+def json_headers(*, x_application: Optional[str] = None) -> Dict[str, str]:
+    h = openpecha_headers(x_application=x_application)
+    h["Content-Type"] = "application/json"
+    return h

backend/cataloger/controller/openpecha_api/categories.py

@@ -0,0 +1,111 @@
+from typing import Any, Dict, Optional
+
+import requests
+from fastapi import HTTPException
+
+from cataloger.controller.openpecha_api.base import json_headers, openpecha_headers, openpecha_url
+
+
+def _category_v2_to_legacy_row(c: Dict[str, Any]) -> Dict[str, Any]:
+    title_val = c.get("title")
+    if isinstance(title_val, dict) and title_val:
+        title_str = next(iter(title_val.values()), "")
+    else:
+        title_str = str(title_val or "")
+    children = c.get("children") or []
+    return {
+        "id": c.get("id", ""),
+        "parent": c.get("parent_id"),
+        "title": title_str,
+        "has_child": bool(children),
+    }
+
+
+def _require_application_header(application: Optional[str]) -> str:
+    if not application:
+        raise HTTPException(
+            status_code=400,
+            detail="OpenPecha v2 categories require `application` query/body or OPENPECHA_X_APPLICATION",
+        )
+    return application
+
+
+def list_categories(
+    *,
+    application: Optional[str] = None,
+    language: Optional[str] = None,
+    parent_id: Optional[str] = None,
+    timeout: int = 30,
+) -> Any:
+    del language
+    headers = openpecha_headers(x_application=application)
+    if "X-Application" not in headers:
+        raise HTTPException(
+            status_code=400,
+            detail="OpenPecha v2 categories require `application` query param or OPENPECHA_X_APPLICATION",
+        )
+    params: Dict[str, str] = {}
+    if parent_id:
+        params["parent_id"] = parent_id
+    try:
+        response = requests.get(
+            openpecha_url("categories"),
+            params=params,
+            headers=headers,
+            timeout=timeout,
+        )
+        if response.status_code != 200:
+            raise HTTPException(status_code=response.status_code, detail=response.text)
+        raw = response.json()
+        return raw
+    except requests.exceptions.Timeout:
+        raise HTTPException(
+            status_code=504,
+            detail="Request to OpenPecha API timed out after 30 seconds",
+        )
+    except requests.exceptions.RequestException as e:
+        raise HTTPException(
+            status_code=502,
+            detail=f"Error connecting to OpenPecha API: {str(e)}",
+        )
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
+
+
+def create_category(
+    *,
+    application: str,
+    title: dict,
+    parent: Optional[str],
+    timeout: int = 30,
+) -> Any:
+    headers = json_headers(x_application=_require_application_header(application))
+    payload: Dict[str, Any] = {"title": title}
+    if parent is not None:
+        payload["parent_id"] = parent
+    try:
+        response = requests.post(
+            openpecha_url("categories"),
+            json=payload,
+            headers=headers,
+            timeout=timeout,
+        )
+        if response.status_code not in (200, 201):
+            raise HTTPException(status_code=response.status_code, detail=response.text)
+        return response.json()
+    except requests.exceptions.Timeout:
+        raise HTTPException(
+            status_code=504,
+            detail="Request to OpenPecha API timed out after 30 seconds",
+        )
+    except requests.exceptions.RequestException as e:
+        raise HTTPException(
+            status_code=502,
+            detail=f"Error connecting to OpenPecha API: {str(e)}",
+        )
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

backend/cataloger/controller/openpecha_api/enums.py

@@ -0,0 +1,59 @@
+from typing import Any
+
+import requests
+from fastapi import HTTPException
+
+from cataloger.controller.openpecha_api.base import (
+    openpecha_headers,
+    openpecha_url,
+    require_openpecha_base_url,
+)
+
+
+def get_enum(enum_type: str, *, timeout: int = 30) -> Any:
+    if enum_type == "language":
+        try:
+            response = requests.get(
+                openpecha_url("languages"),
+                headers=openpecha_headers(),
+                timeout=timeout,
+            )
+            if response.status_code != 200:
+                raise HTTPException(status_code=response.status_code, detail=response.text)
+            return response.json()
+        except requests.exceptions.Timeout:
+            raise HTTPException(
+                status_code=504,
+                detail="Request to OpenPecha API timed out after 30 seconds",
+            )
+        except requests.exceptions.RequestException as e:
+            raise HTTPException(
+                status_code=502,
+                detail=f"Error connecting to OpenPecha API: {str(e)}",
+            )
+        except HTTPException:
+            raise
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
+
+    base = require_openpecha_base_url()
+    try:
+        params = {"type": enum_type}
+        response = requests.get(f"{base}/enum", params=params, timeout=timeout)
+        if response.status_code != 200:
+            raise HTTPException(status_code=response.status_code, detail=response.text)
+        return response.json()
+    except requests.exceptions.Timeout:
+        raise HTTPException(
+            status_code=504,
+            detail="Request to OpenPecha API timed out after 30 seconds",
+        )
+    except requests.exceptions.RequestException as e:
+        raise HTTPException(
+            status_code=502,
+            detail=f"Error connecting to OpenPecha API: {str(e)}",
+        )
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")