Admin and Verbose annotator as well as the AdminFeed implementation

core/feed_protocol/admin.py

@@ -0,0 +1,58 @@
+from sqlalchemy import and_
+
+from core.feed_protocol.acquisition import OPDSAcquisitionFeed
+from core.lane import Pagination
+from core.model.licensing import LicensePool
+
+
+class AdminFeed(OPDSAcquisitionFeed):
+    @classmethod
+    def suppressed(cls, _db, title, url, annotator, pagination=None):
+        pagination = pagination or Pagination.default()
+
+        q = (
+            _db.query(LicensePool)
+            .filter(
+                and_(
+                    LicensePool.suppressed == True,
+                    LicensePool.superceded == False,
+                )
+            )
+            .order_by(LicensePool.id)
+        )
+        pools = pagination.modify_database_query(_db, q).all()
+
+        works = [pool.work for pool in pools]
+        feed = cls(title, url, works, None, pagination, annotator)
+        feed.generate_feed()
+
+        # Render a 'start' link
+        top_level_title = annotator.top_level_title()
+        start_uri = annotator.groups_url(None)
+
+        feed.add_link(start_uri, rel="start", title=top_level_title)
+
+        # Render an 'up' link, same as the 'start' link to indicate top-level feed
+        feed.add_link(start_uri, rel="up", title=top_level_title)
+
+        if len(works) > 0:
+            # There are works in this list. Add a 'next' link.
+            feed.add_link(
+                href=annotator.suppressed_url(pagination.next_page),
+                rel="next",
+            )
+
+        if pagination.offset > 0:
+            feed.add_link(
+                annotator.suppressed_url(pagination.first_page),
+                rel="first",
+            )
+
+        previous_page = pagination.previous_page
+        if previous_page:
+            feed.add_link(
+                annotator.suppressed_url(previous_page),
+                rel="previous",
+            )
+
+        return feed

core/feed_protocol/annotator/admin.py

@@ -0,0 +1,93 @@
+from core.feed_protocol.annotator.circulation import LibraryAnnotator
+from core.feed_protocol.annotator.verbose import VerboseAnnotator
+from core.feed_protocol.types import FeedData, Link, WorkEntry
+from core.mirror import MirrorUploader
+from core.model import DataSource
+from core.model.configuration import ExternalIntegrationLink
+
+
+class AdminAnnotator(LibraryAnnotator):
+    def __init__(self, circulation, library, test_mode=False):
+        super().__init__(circulation, None, library, test_mode=test_mode)
+
+    def annotate_work_entry(self, entry: WorkEntry):
+        super().annotate_work_entry(entry)
+        VerboseAnnotator.add_ratings(entry)
+
+        identifier = entry.identifier
+        active_license_pool = entry.license_pool
+
+        # Find staff rating and add a tag for it.
+        for measurement in identifier.measurements:
+            if (
+                measurement.data_source.name == DataSource.LIBRARY_STAFF
+                and measurement.is_most_recent
+            ):
+                entry.computed.ratings.append(
+                    self.rating(measurement.quantity_measured, measurement.value)
+                )
+
+        if active_license_pool and active_license_pool.suppressed:
+            entry.computed.other_links.append(
+                Link(
+                    href=self.url_for(
+                        "unsuppress",
+                        identifier_type=identifier.type,
+                        identifier=identifier.identifier,
+                        _external=True,
+                    ),
+                    rel="http://librarysimplified.org/terms/rel/restore",
+                )
+            )
+        else:
+            entry.computed.other_links.append(
+                Link(
+                    href=self.url_for(
+                        "suppress",
+                        identifier_type=identifier.type,
+                        identifier=identifier.identifier,
+                        _external=True,
+                    ),
+                    rel="http://librarysimplified.org/terms/rel/hide",
+                )
+            )
+
+        entry.computed.other_links.append(
+            Link(
+                href=self.url_for(
+                    "edit",
+                    identifier_type=identifier.type,
+                    identifier=identifier.identifier,
+                    _external=True,
+                ),
+                rel="edit",
+            )
+        )
+
+        # If there is a storage integration for the collection, changing the cover is allowed.
+        mirror = MirrorUploader.for_collection(
+            active_license_pool.collection, ExternalIntegrationLink.COVERS
+        )
+        if mirror:
+            entry.computed.other_links.append(
+                Link(
+                    href=self.url_for(
+                        "work_change_book_cover",
+                        identifier_type=identifier.type,
+                        identifier=identifier.identifier,
+                        _external=True,
+                    ),
+                    rel="http://librarysimplified.org/terms/rel/change_cover",
+                )
+            )
+
+    def suppressed_url(self, pagination):
+        kwargs = dict(list(pagination.items()))
+        return self.url_for("suppressed", _external=True, **kwargs)
+
+    def annotate_feed(self, feed: FeedData):
+        # Add a 'search' link.
+        search_url = self.url_for("lane_search", languages=None, _external=True)
+        feed.add_link(
+            search_url, rel="search", type="application/opensearchdescription+xml"
+        )

core/feed_protocol/annotator/base.py

@@ -11,6 +11,7 @@
 
 from core.classifier import Classifier
 from core.feed_protocol.types import (
+    Author,
     FeedData,
     FeedEntryType,
     Link,
@@ -40,7 +41,7 @@ def authors(cls, edition):
             for bibliographic information, including the list of
             Contributions.
         """
-        authors = {"author": [], "contributor": []}
+        authors = {"authors": [], "contributors": []}
         state = defaultdict(set)
         for contribution in edition.contributions:
             info = cls.contributor_tag(contribution, state)
@@ -49,15 +50,15 @@ def authors(cls, edition):
                 # need a tag.
                 continue
             key, tag = info
-            authors[key].append(tag)
+            authors[f"{key}s"].append(tag)
 
-        if authors["author"]:
+        if authors["authors"]:
             return authors
 
         # We have no author information, so we add empty <author> tag
         # to avoid the implication (per RFC 4287 4.2.1) that this book
         # was written by whoever wrote the OPDS feed.
-        authors["author"].append(FeedEntryType(name=""))
+        authors["authors"].append(Author(name=""))
         return authors
 
     @classmethod
@@ -98,7 +99,7 @@ def contributor_tag(cls, contribution, state):
         properties = dict()
         if marc_role:
             properties["role"] = marc_role
-        entry = FeedEntryType(name=name, **properties)
+        entry = Author(name=name, **properties)
 
         # Record the fact that we credited this person with this role,
         # so that we don't do it again on a subsequent call.
@@ -117,6 +118,11 @@ def series(cls, series_name, series_position):
             series_details["position"] = str(series_position)
         return FeedEntryType(**series_details)
 
+    @classmethod
+    def rating(cls, type_uri, value):
+        """Generate a schema:Rating tag for the given type and value."""
+        return FeedEntryType(ratingValue="%.4f" % value, additionalType=type_uri)
+
     @classmethod
     def samples(cls, edition: Edition) -> list[Hyperlink]:
         if not edition:
@@ -231,7 +237,7 @@ def __init__(self, library) -> None:
         self.library = library
 
 
-class Annotator(OPDSAnnotator):
+class Annotator(OPDSAnnotator, ToFeedEntry):
     def annotate_work_entry(self, entry: WorkEntry, updated=None) -> None:
         if entry.computed:
             return
@@ -258,7 +264,7 @@ def annotate_work_entry(self, entry: WorkEntry, updated=None) -> None:
                 image_type = "image/gif"
             image_links.append(Link(rel=rel, href=url, type=image_type))
 
-        samples = ToFeedEntry.samples(edition)
+        samples = self.samples(edition)
         for sample in samples:
             other_links.append(
                 Link(
@@ -268,7 +274,7 @@ def annotate_work_entry(self, entry: WorkEntry, updated=None) -> None:
                 )
             )
 
-        content = ToFeedEntry.content(work)
+        content = self.content(work)
         if isinstance(content, bytes):
             content = content.decode("utf8")
 
@@ -285,21 +291,19 @@ def annotate_work_entry(self, entry: WorkEntry, updated=None) -> None:
 
         # TODO: Is VerboseAnnotator used anywhere?
 
-        author_entries = ToFeedEntry.authors(edition)
-        computed.contributors = author_entries.get("contributor")
-        computed.authors = author_entries.get("author")
+        author_entries = self.authors(edition)
+        computed.contributors = author_entries.get("contributors")
+        computed.authors = author_entries.get("authors")
 
         if edition.series:
-            computed.series = ToFeedEntry.series(
-                edition.series, edition.series_position
-            )
+            computed.series = self.series(edition.series, edition.series_position)
 
         if content:
             computed.summary = FeedEntryType(text=content, type="html")
 
         computed.pwid = edition.permanent_work_id
 
-        categories_by_scheme = ToFeedEntry.categories(work)
+        categories_by_scheme = self.categories(work)
         category_tags = []
         for scheme, categories in list(categories_by_scheme.items()):
             for category in categories:

core/feed_protocol/annotator/verbose.py

@@ -0,0 +1,102 @@
+from collections import defaultdict
+
+from sqlalchemy.orm import Session
+
+from core.feed_protocol.annotator.base import Annotator
+from core.feed_protocol.types import Author, WorkEntry
+from core.model import PresentationCalculationPolicy
+from core.model.classification import Subject
+from core.model.identifier import Identifier
+from core.model.measurement import Measurement
+from core.model.work import Work
+
+
+class VerboseAnnotator(Annotator):
+    """The default Annotator for machine-to-machine integration.
+
+    This Annotator describes all categories and authors for the book
+    in great detail.
+    """
+
+    opds_cache_field = Work.verbose_opds_entry.name
+
+    def annotate_work_entry(self, entry, updated=None):
+        super().annotate_work_entry(entry, updated=updated)
+        self.add_ratings(entry)
+
+    @classmethod
+    def add_ratings(cls, entry: WorkEntry):
+        """Add a quality rating to the work."""
+        work = entry.work
+        for type_uri, value in [
+            (Measurement.QUALITY, work.quality),
+            (None, work.rating),
+            (Measurement.POPULARITY, work.popularity),
+        ]:
+            if value:
+                entry.computed.ratings.append(cls.rating(type_uri, value))
+
+    @classmethod
+    def categories(cls, work, policy=None):
+        """Send out _all_ categories for the work.
+
+        (So long as the category type has a URI associated with it in
+        Subject.uri_lookup.)
+
+        :param policy: A PresentationCalculationPolicy to
+            use when deciding how deep to go when finding equivalent
+            identifiers for the work.
+        """
+        policy = policy or PresentationCalculationPolicy(
+            equivalent_identifier_cutoff=100
+        )
+        _db = Session.object_session(work)
+        by_scheme_and_term = dict()
+        identifier_ids = work.all_identifier_ids(policy=policy)
+        classifications = Identifier.classifications_for_identifier_ids(
+            _db, identifier_ids
+        )
+        for c in classifications:
+            subject = c.subject
+            if subject.type in Subject.uri_lookup:
+                scheme = Subject.uri_lookup[subject.type]
+                term = subject.identifier
+                weight_field = "ratingValue"
+                key = (scheme, term)
+                if not key in by_scheme_and_term:
+                    value = dict(term=subject.identifier)
+                    if subject.name:
+                        value["label"] = subject.name
+                    value[weight_field] = 0
+                    by_scheme_and_term[key] = value
+                by_scheme_and_term[key][weight_field] += c.weight
+
+        # Collapse by_scheme_and_term to by_scheme
+        by_scheme = defaultdict(list)
+        for (scheme, term), value in list(by_scheme_and_term.items()):
+            by_scheme[scheme].append(value)
+        by_scheme.update(super().categories(work))
+        return by_scheme
+
+    @classmethod
+    def authors(cls, edition):
+        """Create a detailed <author> tag for each author."""
+        return {
+            "authors": [
+                cls.detailed_author(author) for author in edition.author_contributors
+            ],
+            "contributors": [],
+        }
+
+    @classmethod
+    def detailed_author(cls, contributor):
+        """Turn a Contributor into a detailed <author> tag."""
+        author = Author()
+        author.name = contributor.display_name
+        author.sort_name = contributor.sort_name
+        author.family_name = contributor.family_name
+        author.wikipedia_name = contributor.wikipedia_name
+        author.viaf = f"http://viaf.org/viaf/{contributor.viaf}"
+        author.lc = f"http://id.loc.gov/authorities/names/{contributor.lc}"
+
+        return author