Skip to content
36 changes: 19 additions & 17 deletions src/palace/manager/feed/annotator/circulation.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,9 @@
FeedData,
IndirectAcquisition,
Link,
LinkContentType,
LinkKwargs,
LinkType,
PatronData,
WorkEntry,
)
Expand Down Expand Up @@ -186,15 +188,15 @@ def license_tags(
)

@classmethod
def format_types(cls, delivery_mechanism: DeliveryMechanism) -> list[str]:
def format_types(cls, delivery_mechanism: DeliveryMechanism) -> list[LinkType]:
"""Generate a set of types suitable for passing into
acquisition_link().
"""
types = []
types: list[LinkType] = []
# If this is a streaming book, you have to get an OPDS entry, then
# get a direct link to the streaming reader from that.
if delivery_mechanism.is_streaming:
types.append(OPDSFeed.ENTRY_TYPE)
types.append(LinkContentType.OPDS_ENTRY)

# If this is a DRM-encrypted book, you have to get through the DRM
# to get the goodies inside.
Expand Down Expand Up @@ -624,7 +626,7 @@ def acquisition_link(
cls,
rel: str,
href: str,
types: list[str] | None,
types: list[LinkType] | None,
active_loan: Loan | None = None,
templated: bool = False,
) -> Acquisition:
Expand All @@ -649,7 +651,7 @@ def acquisition_link(

@classmethod
def indirect_acquisition(
cls, indirect_types: list[str]
cls, indirect_types: list[LinkType]
) -> IndirectAcquisition | None:
top_level_parent: IndirectAcquisition | None = None
parent: IndirectAcquisition | None = None
Expand Down Expand Up @@ -737,16 +739,15 @@ def is_novelist_configured(self) -> bool:
def top_level_title(self) -> str:
return self._top_level_title

def permalink_for(self, identifier: Identifier) -> tuple[str, str]:
# TODO: Do not force OPDS types
def permalink_for(self, identifier: Identifier) -> tuple[str, LinkType]:
url = self.url_for(
"permalink",
identifier_type=identifier.type,
identifier=identifier.identifier,
library_short_name=self.library.short_name,
_external=True,
)
return url, OPDSFeed.ENTRY_TYPE
return url, LinkContentType.OPDS_ENTRY

def groups_url(
self, lane: WorkList | None, facets: FacetsWithEntryPoint | None = None
Expand Down Expand Up @@ -872,7 +873,6 @@ def annotate_work_entry(
identifier = entry.identifier

permalink_uri, permalink_type = self.permalink_for(identifier)
# TODO: Do not force OPDS types
if permalink_uri:
entry.computed.other_links.append(
Link(href=permalink_uri, rel="alternate", type=permalink_type)
Expand All @@ -895,7 +895,7 @@ def annotate_work_entry(
entry.computed.other_links.append(
Link(
rel="recommendations",
type=OPDSFeed.ACQUISITION_FEED_TYPE,
type=LinkContentType.OPDS_FEED,
title="Recommended Works",
href=self.url_for(
"recommendations",
Expand All @@ -912,7 +912,7 @@ def annotate_work_entry(
entry.computed.other_links.append(
Link(
rel="related",
type=OPDSFeed.ACQUISITION_FEED_TYPE,
type=LinkContentType.OPDS_FEED,
title="Recommended Works",
href=self.url_for(
"related_books",
Expand Down Expand Up @@ -1022,7 +1022,7 @@ def add_author_links(self, entry: WorkEntry) -> None:

author_entry.link = Link(
rel="contributor",
type=OPDSFeed.ACQUISITION_FEED_TYPE,
type=LinkContentType.OPDS_FEED,
title=name,
href=self.url_for(
"contributor",
Expand Down Expand Up @@ -1065,7 +1065,7 @@ def add_series_link(self, entry: WorkEntry) -> None:
)
series_entry.link = Link(
rel="series",
type=OPDSFeed.ACQUISITION_FEED_TYPE,
type=LinkContentType.OPDS_FEED,
title=series_name,
href=href,
)
Expand Down Expand Up @@ -1118,7 +1118,7 @@ def annotate_feed(self, feed: FeedData) -> None:
_external=True,
),
rel="http://opds-spec.org/shelf",
type=OPDSFeed.ACQUISITION_FEED_TYPE,
type=LinkContentType.OPDS_FEED,
)

feed.add_link(
Expand Down Expand Up @@ -1151,7 +1151,7 @@ def annotate_feed(self, feed: FeedData) -> None:
feed.add_link(
href=crawlable_url,
rel="http://opds-spec.org/crawlable",
type=OPDSFeed.ACQUISITION_FEED_TYPE,
type=LinkContentType.OPDS_FEED,
)

self.add_configuration_links(feed)
Expand Down Expand Up @@ -1358,7 +1358,7 @@ def borrow_link(
borrow_link = Acquisition(
rel=rel,
href=borrow_url,
type=OPDSFeed.ENTRY_TYPE,
type=LinkContentType.OPDS_ENTRY,
is_hold=True if active_hold else False,
)

Expand Down Expand Up @@ -1421,7 +1421,9 @@ def fulfill_link(
_external=True,
)

if template_vars := self.FULFILL_LINK_TEMPLATED_TYPES.get(format_types[0]):
first_format_type = format_types[0]
template_key = first_format_type if isinstance(first_format_type, str) else None
if template_vars := self.FULFILL_LINK_TEMPLATED_TYPES.get(template_key):
fulfill_url = fulfill_url + "{?" + ",".join(template_vars) + "}"
templated = True
else:
Expand Down
2 changes: 1 addition & 1 deletion src/palace/manager/feed/annotator/loan_and_hold.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ def user_profile_management_protocol_link(self) -> Link:
for the current patron.
"""
return Link(
rel="http://librarysimplified.org/terms/rel/user-profile",
rel="profile",
href=self.url_for(
"patron_profile",
library_short_name=self.library.short_name,
Expand Down
55 changes: 49 additions & 6 deletions src/palace/manager/feed/serializer/opds.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,10 @@
from functools import partial
from typing import Any, cast

from frozendict import frozendict
from lxml import etree

from palace.manager.core.user_profile import ProfileController
from palace.manager.feed.facets.constants import FacetConstants
from palace.manager.feed.serializer.base import SerializerInterface
from palace.manager.feed.types import (
Expand All @@ -18,6 +20,8 @@
FeedMetadata,
IndirectAcquisition,
Link,
LinkContentType,
LinkType,
PatronData,
Rating,
Series,
Expand Down Expand Up @@ -73,9 +77,36 @@ def is_sort_facet(link: Link) -> bool:


class BaseOPDS1Serializer(SerializerInterface[etree._Element], OPDSFeed, abc.ABC):
_CONTENT_TYPE_MAP: frozendict[LinkContentType, str] = frozendict(
{
LinkContentType.OPDS_FEED: OPDSFeed.ACQUISITION_FEED_TYPE,
LinkContentType.OPDS_ENTRY: OPDSFeed.ENTRY_TYPE,
}
)

# OPDS1 uses Palace-specific relation URIs for some standard rels.
# OPDS2 uses standard IANA rels directly and needs no mapping.
_REL_MAP: frozendict[str, str] = frozendict(
{
"profile": ProfileController.LINK_RELATION,
}
)

def __init__(self) -> None:
pass

def _resolve_type(self, type_value: LinkType | None) -> str | None:
"""Map semantic LinkContentType values to OPDS1-specific types."""
if isinstance(type_value, LinkContentType):
return self._CONTENT_TYPE_MAP[type_value]
return type_value

def _resolve_rel(self, rel_value: str | None) -> str | None:
"""Map standard rels to OPDS1/Palace-specific rels."""
if rel_value is not None and rel_value in self._REL_MAP:
return self._REL_MAP[rel_value]
return rel_value

def _tag(
self, tag_name: str, *args: Any, mapping: dict[str, str] | None = None
) -> etree._Element:
Expand Down Expand Up @@ -325,13 +356,15 @@ def _serialize_author_tag(self, tag: str, author: Author) -> etree._Element:
return entry

def _serialize_link(self, link: Link) -> etree._Element:
resolved_type = self._resolve_type(link.type)
resolved_rel = self._resolve_rel(link.rel)
attrs: dict[str, str] = {}
if link.href is not None:
attrs["href"] = link.href
if link.rel is not None:
attrs["rel"] = link.rel
if link.type is not None:
attrs["type"] = link.type
if resolved_rel is not None:
attrs["rel"] = resolved_rel
if resolved_type is not None:
attrs["type"] = resolved_type
if link.title is not None:
attrs["title"] = link.title
if link.role is not None:
Expand Down Expand Up @@ -408,13 +441,23 @@ def _serialize_rating(self, rating: Rating) -> etree._Element:
return entry

def _serialize_acquisition_link(self, link: Acquisition) -> etree._Element:
resolved_type = self._resolve_type(link.type)
resolved_rel = self._resolve_rel(link.rel)

attrs: dict[str, str] = {"href": link.href}
if resolved_rel is not None:
attrs["rel"] = resolved_rel
if resolved_type is not None:
attrs["type"] = resolved_type

link_func = OPDSFeed.tlink if link.templated else OPDSFeed.link
element = link_func(**link.link_attribs())
element = link_func(**attrs)

def _indirect(item: IndirectAcquisition) -> etree._Element:
tag = self._tag("indirectAcquisition")
tag.set("type", item.type)
resolved_indirect_type = self._resolve_type(item.type)
if resolved_indirect_type is not None:
tag.set("type", resolved_indirect_type)
for child in item.children:
tag.append(_indirect(child))
return tag
Expand Down
Loading
Loading