"""Tolerant normalization of a IIIF document into the :mod:`iiisight.model`.
:func:`normalize` accepts a parsed JSON document (a ``dict``) and returns a
:class:`~iiisight.model.Manifest` or :class:`~iiisight.model.Collection`. It never
raises on spec-imperfect input: missing or wrong-typed fields are repaired and
each repair is recorded as a :class:`~iiisight.diagnostics.Diagnostic` on the
returned document.
A v2 document is detected and upgraded to a v3-shaped dict by
:func:`iiisight.upgrade.upgrade` before the v3 parse runs, so callers always work
against one model.
"""
from collections.abc import Mapping
from typing import Any
from .diagnostics import Diagnostic, Severity
from .language import LanguageMap
from .model import (
Agent,
Annotation,
AnnotationPage,
Canvas,
Collection,
ContentResource,
ImageService,
Link,
Manifest,
MetadataEntry,
Range,
Reference,
Selector,
Service,
SpecificResource,
TextualBody,
)
from .upgrade import upgrade, upgrade_annotation_list
_IMAGE_BODY_TYPES = {"Image", "Video", "Sound", "Text", "Dataset", "Model"}
[docs]
def normalize(document: Mapping[str, Any]) -> Manifest | Collection:
"""Normalize a IIIF document into a :class:`Manifest` or :class:`Collection`.
Args:
document: A parsed IIIF JSON document.
Returns:
The normalized document, with any repairs recorded on ``.diagnostics``.
"""
diags: list[Diagnostic] = []
version = _detect_version(document, diags)
if version == 2:
document = upgrade(dict(document), diags)
doc_type = document.get("type")
if doc_type == "Collection":
return _collection(document, diags)
if doc_type != "Manifest":
diags.append(
Diagnostic(
"unknown-document-type",
Severity.WARNING,
"/type",
f"Unexpected top-level type {doc_type!r}; parsed as a Manifest.",
recovered="assumed Manifest",
)
)
return _manifest(document, diags)
[docs]
def parse_annotation_page(document: Mapping[str, Any]) -> AnnotationPage:
"""Parse a standalone AnnotationPage (e.g. a Content Search response).
Accepts a v3 ``AnnotationPage`` (``items``) or a v2 ``sc:AnnotationList``
(``resources``), which is upgraded first.
"""
diags: list[Diagnostic] = []
if document.get("@type") == "sc:AnnotationList" or (
"resources" in document and "items" not in document
):
document = upgrade_annotation_list(dict(document), diags)
return _annotation_page(document, diags, "")
# --- Version detection ------------------------------------------------------
def _detect_version(document: Mapping[str, Any], diags: list[Diagnostic]) -> int:
"""Return 2 or 3 for the document's Presentation API version."""
context = document.get("@context")
context_text = " ".join(context if isinstance(context, list) else [context or ""])
if "presentation/2" in context_text or "@type" in document or "sequences" in document:
return 2
if "presentation/3" in context_text or "type" in document:
return 3
diags.append(
Diagnostic(
"version-unknown",
Severity.WARNING,
"",
"Could not determine Presentation API version; assumed 3.0.",
recovered="assumed v3",
)
)
return 3
# --- Structural parsers -----------------------------------------------------
def _manifest(document: Mapping[str, Any], diags: list[Diagnostic]) -> Manifest:
base = _described(document, diags, "")
return Manifest(
**base,
canvases=tuple(
_canvas(item, diags, f"/items/{i}")
for i, item in enumerate(_as_list(document.get("items")))
),
structures=tuple(
_range(item, diags, f"/structures/{i}")
for i, item in enumerate(_as_list(document.get("structures")))
),
annotations=tuple(
_annotation_page(item, diags, f"/annotations/{i}")
for i, item in enumerate(_as_list(document.get("annotations")))
),
viewing_direction=document.get("viewingDirection"),
diagnostics=tuple(diags),
)
def _collection(document: Mapping[str, Any], diags: list[Diagnostic]) -> Collection:
base = _described(document, diags, "")
return Collection(
**base,
items=tuple(
_collection_item(item, diags, f"/items/{i}")
for i, item in enumerate(_as_list(document.get("items")))
),
viewing_direction=document.get("viewingDirection"),
diagnostics=tuple(diags),
)
def _collection_item(
item: Mapping[str, Any], diags: list[Diagnostic], path: str
) -> Manifest | Collection | Reference:
"""A collection's child is usually a bare reference; a full item is parsed."""
item_type = item.get("type")
if item_type == "Manifest" and "items" in item:
return _manifest(item, diags)
if item_type == "Collection" and "items" in item:
return _collection(item, diags)
return Reference(
id=_require_id(item, diags, path),
type=str(item_type or "Manifest"),
label=_language_map(item.get("label"), diags, f"{path}/label"),
)
def _canvas(document: Mapping[str, Any], diags: list[Diagnostic], path: str) -> Canvas:
base = _described(document, diags, path)
return Canvas(
**base,
height=_int(document.get("height"), diags, f"{path}/height"),
width=_int(document.get("width"), diags, f"{path}/width"),
duration=_float(document.get("duration"), diags, f"{path}/duration"),
items=tuple(
_annotation_page(page, diags, f"{path}/items/{i}")
for i, page in enumerate(_as_list(document.get("items")))
),
annotations=tuple(
_annotation_page(page, diags, f"{path}/annotations/{i}")
for i, page in enumerate(_as_list(document.get("annotations")))
),
)
def _range(document: Mapping[str, Any], diags: list[Diagnostic], path: str) -> Range:
base = _described(document, diags, path)
items: list[Any] = []
for i, item in enumerate(_as_list(document.get("items"))):
item_path = f"{path}/items/{i}"
item_type = item.get("type")
if item_type == "Range":
items.append(_range(item, diags, item_path))
elif item_type == "Canvas" and ("items" in item or "height" in item or "width" in item):
items.append(_canvas(item, diags, item_path))
else:
items.append(
Reference(
id=_require_id(item, diags, item_path),
type=str(item_type or "Canvas"),
label=_language_map(item.get("label"), diags, f"{item_path}/label"),
)
)
return Range(**base, items=tuple(items))
def _annotation_page(
document: Mapping[str, Any], diags: list[Diagnostic], path: str
) -> AnnotationPage:
return AnnotationPage(
id=str(document.get("id", "")),
type=str(document.get("type", "AnnotationPage")),
items=tuple(
_annotation(item, diags, f"{path}/items/{i}")
for i, item in enumerate(_as_list(document.get("items")))
),
)
def _annotation(document: Mapping[str, Any], diags: list[Diagnostic], path: str) -> Annotation:
motivation = document.get("motivation")
if motivation is None:
motivations: tuple[str, ...] = ()
elif isinstance(motivation, str):
motivations = (motivation,)
else:
motivations = tuple(str(m) for m in motivation)
return Annotation(
id=str(document.get("id", "")),
type=str(document.get("type", "Annotation")),
motivation=motivations,
body=_body(document.get("body"), diags, f"{path}/body"),
target=_target(document.get("target")),
)
def _body(value: Any, diags: list[Diagnostic], path: str):
if value is None:
return None
if isinstance(value, list):
return tuple(_body(item, diags, f"{path}/{i}") for i, item in enumerate(value))
if not isinstance(value, Mapping):
diags.append(
Diagnostic(
"unexpected-body",
Severity.ERROR,
path,
f"Annotation body was {type(value).__name__}, not an object; dropped.",
recovered="dropped body",
)
)
return None
if value.get("type") == "TextualBody":
return TextualBody(
value=str(value.get("value", "")),
format=value.get("format"),
language=_string_tuple(value.get("language")),
)
return _content_resource(value, diags, path)
def _content_resource(
document: Mapping[str, Any], diags: list[Diagnostic], path: str
) -> ContentResource:
body_type = document.get("type")
if body_type not in _IMAGE_BODY_TYPES:
diags.append(
Diagnostic(
"unknown-body-type",
Severity.INFO,
f"{path}/type",
f"Unrecognized content resource type {body_type!r}; kept as-is.",
)
)
image_services: list[ImageService] = []
other_services: list[Service] = []
for i, entry in enumerate(_as_list(document.get("service"))):
if not isinstance(entry, Mapping):
continue
if _is_image_service(entry):
image_services.append(_image_service(entry, diags, f"{path}/service/{i}"))
else:
service = _service_from(entry, diags, f"{path}/service/{i}")
if service is not None:
other_services.append(service)
return ContentResource(
id=_require_id(document, diags, path),
type=str(body_type or "Image"),
format=document.get("format"),
height=_int(document.get("height"), diags, f"{path}/height"),
width=_int(document.get("width"), diags, f"{path}/width"),
duration=_float(document.get("duration"), diags, f"{path}/duration"),
label=_language_map(document.get("label"), diags, f"{path}/label"),
language=_string_tuple(document.get("language")),
service=tuple(image_services),
services=tuple(other_services),
)
def _image_service(document: Mapping[str, Any], diags: list[Diagnostic], path: str) -> ImageService:
service_id = document.get("id") or document.get("@id")
if not service_id:
diags.append(
Diagnostic(
"image-service-missing-id",
Severity.ERROR,
path,
"Image service has no id; image URLs cannot be built.",
recovered="empty id",
)
)
service_id = ""
profile = document.get("profile")
service_type = document.get("type") or document.get("@type")
api_version, level = _image_api_version(service_type, profile)
return ImageService(id=str(service_id), api_version=api_version, compliance_level=level)
def _image_api_version(service_type: Any, profile: Any) -> tuple[int, str | None]:
"""Infer (api_version, compliance_level) from a service's type/profile."""
profile_is_level = isinstance(profile, str) and profile.startswith("level")
profile_is_uri = isinstance(profile, str) and "/image/" in profile
level: str | None = None
if profile_is_level:
level = profile
elif profile_is_uri:
level = _level_from_uri(profile)
if service_type == "ImageService3" or profile_is_level:
return 3, level
if service_type in {"ImageService2", "ImageService1"} or profile_is_uri:
return 2, level
# No strong signal: default to v3 (the normalization target).
return 3, level
def _level_from_uri(profile: str) -> str | None:
"""Pull ``levelN`` out of a v2 profile URI like ``.../image/2/level2.json``."""
tail = profile.rstrip("/").rsplit("/", 1)[-1]
name = tail.split(".", 1)[0]
return name if name.startswith("level") else None
# --- Shared descriptive parsing ---------------------------------------------
def _described(document: Mapping[str, Any], diags: list[Diagnostic], path: str) -> dict[str, Any]:
"""Parse the shared descriptive properties into kwargs for a model type."""
return {
"id": _require_id(document, diags, path),
"type": str(document.get("type", "")),
"label": _language_map(document.get("label"), diags, f"{path}/label"),
"metadata": _metadata(document.get("metadata"), diags, f"{path}/metadata"),
"summary": _language_map(document.get("summary"), diags, f"{path}/summary"),
"required_statement": _metadata_entry(
document.get("requiredStatement"), diags, f"{path}/requiredStatement"
),
"rights": document.get("rights"),
"provider": _agents(document.get("provider"), diags, f"{path}/provider"),
"thumbnail": _content_resources(document.get("thumbnail"), diags, f"{path}/thumbnail"),
"homepage": _links(document.get("homepage"), diags, f"{path}/homepage"),
"rendering": _links(document.get("rendering"), diags, f"{path}/rendering"),
"see_also": _links(document.get("seeAlso"), diags, f"{path}/seeAlso"),
"part_of": _references(document.get("partOf"), diags, f"{path}/partOf"),
"services": _services(document.get("service"), diags, f"{path}/service"),
"behavior": _string_tuple(document.get("behavior")),
"nav_date": document.get("navDate"),
"raw": dict(document),
}
_NON_IMAGE_SERVICE_TYPE_PREFIXES = ("Auth", "SearchService", "AutoComplete")
def _services(value: Any, diags: list[Diagnostic], path: str) -> tuple[Service, ...]:
services: list[Service] = []
for i, entry in enumerate(_as_list(value)):
service = _service_from(entry, diags, f"{path}/{i}")
if service is not None:
services.append(service)
return tuple(services)
def _service_from(entry: Any, diags: list[Diagnostic], path: str) -> Service | None:
"""Parse a single service entry, preserving its raw JSON for consumers."""
if not isinstance(entry, Mapping):
return None
service_id = entry.get("id") or entry.get("@id")
if not service_id:
return None
return Service(
id=str(service_id),
type=entry.get("type") or entry.get("@type"),
profile=_profile_text(entry.get("profile")),
label=_language_map(entry.get("label"), diags, f"{path}/label"),
context=_profile_text(entry.get("@context")),
raw=dict(entry),
)
def _is_image_service(entry: Mapping[str, Any]) -> bool:
"""Whether a service on a content resource is an Image API service.
Auth/search/autocomplete services are explicitly typed; everything else on a
content resource is treated as an image service (v2 image services often carry
only an ``@id`` and a profile URI, with no type).
"""
service_type = entry.get("type") or entry.get("@type")
if isinstance(service_type, str) and service_type.startswith(_NON_IMAGE_SERVICE_TYPE_PREFIXES):
return False
return True
def _profile_text(value: Any) -> str | None:
"""Return a profile/context as a string, taking the first entry of a list."""
if isinstance(value, str):
return value
if isinstance(value, list):
for item in value:
if isinstance(item, str):
return item
return None
def _metadata(value: Any, diags: list[Diagnostic], path: str) -> tuple[MetadataEntry, ...]:
entries: list[MetadataEntry] = []
for i, entry in enumerate(_as_list(value)):
parsed = _metadata_entry(entry, diags, f"{path}/{i}")
if parsed is not None:
entries.append(parsed)
return tuple(entries)
def _metadata_entry(value: Any, diags: list[Diagnostic], path: str) -> MetadataEntry | None:
"""Parse a single ``{label, value}`` pair (a metadata row or requiredStatement)."""
if not isinstance(value, Mapping):
return None
return MetadataEntry(
label=_language_map(value.get("label"), diags, f"{path}/label") or LanguageMap(),
value=_language_map(value.get("value"), diags, f"{path}/value") or LanguageMap(),
)
def _agents(value: Any, diags: list[Diagnostic], path: str) -> tuple[Agent, ...]:
"""Parse ``provider`` into a tuple of :class:`Agent`, skipping id-less entries."""
agents: list[Agent] = []
for i, entry in enumerate(_as_list(value)):
if not isinstance(entry, Mapping):
continue
agent_id = entry.get("id") or entry.get("@id")
if not agent_id:
continue
agents.append(
Agent(
id=str(agent_id),
type=str(entry.get("type", "Agent")),
label=_language_map(entry.get("label"), diags, f"{path}/{i}/label"),
homepage=_links(entry.get("homepage"), diags, f"{path}/{i}/homepage"),
logo=_content_resources(entry.get("logo"), diags, f"{path}/{i}/logo"),
see_also=_links(entry.get("seeAlso"), diags, f"{path}/{i}/seeAlso"),
)
)
return tuple(agents)
def _links(value: Any, diags: list[Diagnostic], path: str) -> tuple[Link, ...]:
"""Parse a ``homepage``/``rendering``/``seeAlso`` list into :class:`Link`s."""
links: list[Link] = []
for i, entry in enumerate(_as_list(value)):
if not isinstance(entry, Mapping):
continue
link_id = entry.get("id") or entry.get("@id")
if not link_id:
continue
links.append(
Link(
id=str(link_id),
type=entry.get("type") or entry.get("@type"),
label=_language_map(entry.get("label"), diags, f"{path}/{i}/label"),
format=entry.get("format"),
profile=_profile_text(entry.get("profile")),
language=_string_tuple(entry.get("language")),
)
)
return tuple(links)
def _references(value: Any, diags: list[Diagnostic], path: str) -> tuple[Reference, ...]:
"""Parse ``partOf`` into a tuple of :class:`Reference` pointers."""
references: list[Reference] = []
for i, entry in enumerate(_as_list(value)):
if not isinstance(entry, Mapping):
continue
ref_id = entry.get("id") or entry.get("@id")
if not ref_id:
continue
references.append(
Reference(
id=str(ref_id),
type=str(entry.get("type") or entry.get("@type") or "Collection"),
label=_language_map(entry.get("label"), diags, f"{path}/{i}/label"),
)
)
return tuple(references)
def _content_resources(
value: Any, diags: list[Diagnostic], path: str
) -> tuple[ContentResource, ...]:
"""Parse a ``thumbnail``/``logo`` list into :class:`ContentResource`s."""
resources: list[ContentResource] = []
for i, entry in enumerate(_as_list(value)):
if not isinstance(entry, Mapping):
continue
resources.append(_content_resource(entry, diags, f"{path}/{i}"))
return tuple(resources)
def _language_map(value: Any, diags: list[Diagnostic], path: str) -> LanguageMap | None:
"""Coerce a value into a :class:`LanguageMap`, recording repairs."""
if value is None:
return None
if isinstance(value, str):
diags.append(
Diagnostic(
"coerced-language-map",
Severity.WARNING,
path,
"Expected a language map object but found a bare string.",
recovered='wrapped under "none"',
)
)
return LanguageMap({"none": [value]})
if isinstance(value, Mapping):
coerced: dict[str, list[str]] = {}
for language, strings in value.items():
if isinstance(strings, str):
coerced[language] = [strings]
diags.append(
Diagnostic(
"coerced-language-map",
Severity.WARNING,
f"{path}/{language}",
"Language map value was a bare string, not a list.",
recovered="wrapped in a list",
)
)
else:
coerced[language] = [str(s) for s in strings]
return LanguageMap(coerced)
diags.append(
Diagnostic(
"coerced-language-map",
Severity.ERROR,
path,
f"Could not interpret {type(value).__name__} as a language map; dropped.",
recovered="dropped",
)
)
return None
def _target(value: Any) -> str | SpecificResource | None:
if value is None or isinstance(value, str):
return value
if isinstance(value, Mapping):
source = value.get("source") or value.get("id") or value.get("@id") or ""
selector = value.get("selector")
parsed_selector = None
if isinstance(selector, Mapping):
parsed_selector = Selector(
type=str(selector.get("type", "")),
value=selector.get("value"),
conforms_to=selector.get("conformsTo"),
)
return SpecificResource(source=str(source), selector=parsed_selector)
return None
# --- Coercion helpers -------------------------------------------------------
def _require_id(document: Mapping[str, Any], diags: list[Diagnostic], path: str) -> str:
value = document.get("id") or document.get("@id")
if not value:
diags.append(
Diagnostic(
"missing-id",
Severity.WARNING,
f"{path}/id",
"Resource has no id.",
recovered="empty string",
)
)
return ""
return str(value)
def _int(value: Any, diags: list[Diagnostic], path: str) -> int | None:
if value is None or isinstance(value, bool):
return None
if isinstance(value, int):
return value
try:
coerced = int(value)
except (TypeError, ValueError):
diags.append(
Diagnostic(
"coerced-int",
Severity.ERROR,
path,
f"Expected an integer but found {value!r}; dropped.",
recovered="dropped",
)
)
return None
diags.append(
Diagnostic(
"coerced-int",
Severity.WARNING,
path,
f"Expected an integer but found {value!r}; coerced.",
recovered=f"coerced to {coerced}",
)
)
return coerced
def _float(value: Any, diags: list[Diagnostic], path: str) -> float | None:
if value is None or isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return float(value)
try:
coerced = float(value)
except (TypeError, ValueError):
diags.append(
Diagnostic(
"coerced-float",
Severity.ERROR,
path,
f"Expected a number but found {value!r}; dropped.",
recovered="dropped",
)
)
return None
diags.append(
Diagnostic(
"coerced-float",
Severity.WARNING,
path,
f"Expected a number but found {value!r}; coerced.",
recovered=f"coerced to {coerced}",
)
)
return coerced
def _string_tuple(value: Any) -> tuple[str, ...]:
if value is None:
return ()
if isinstance(value, str):
return (value,)
return tuple(str(item) for item in value)
def _as_list(value: Any) -> list[Any]:
if value is None:
return []
if isinstance(value, list):
return value
return [value]