diff --git a/server/pesacheck/ingest/ghost_feeding_service.py b/server/pesacheck/ingest/ghost_feeding_service.py index 00c9051dc..2f2467fd6 100644 --- a/server/pesacheck/ingest/ghost_feeding_service.py +++ b/server/pesacheck/ingest/ghost_feeding_service.py @@ -2,10 +2,15 @@ import logging from itertools import islice +from celery.result import allow_join_result + +from superdesk import get_resource_service from superdesk.errors import ParserError from superdesk.io.feeding_services.file_service import FileFeedingService from superdesk.io.registry import register_feeding_service +from superdesk.metadata.item import CONTENT_STATE, GUID_FIELD from superdesk.notification import push_notification +from superdesk.resource_fields import ID_FIELD from superdesk.utils import get_sorted_files, FileSortAttributes @@ -13,6 +18,12 @@ BATCH_SIZE = 10 +# Desk the ingested fact-checks are fetched onto on their way to being +# published. Publisher assigns the route from the article's language, not the +# desk, so the desk is only a required waypoint in Superdesk's publish workflow; +# override it with the provider config ``publish_desk``. +DEFAULT_PUBLISH_DESK = "Newsdesk" + class GhostFeedingService(FileFeedingService): """ @@ -87,16 +98,25 @@ async def _update(self, provider, update): items_gen = parser.iter_items(file_path, provider) + file_guids = [] while True: batch = list(islice(items_gen, BATCH_SIZE)) if not batch: break + file_guids.extend( + item[GUID_FIELD] for item in batch if item.get(GUID_FIELD) + ) yield batch await self.move_file( self.path, filename, provider=provider, success=True ) + # The consumer stores each batch before asking for the next, so + # once the file is moved every item above is in the ingest + # collection and can be fetched onto a desk and published. + await self._auto_publish(provider, file_guids) + except Exception as ex: if last_updated and self.is_old_content(last_updated): await self.move_file( @@ -108,5 +128,82 @@ async def _update(self, provider, update): push_notification("ingest:update") + async def _auto_publish(self, provider, guids): + """Fetch each freshly ingested item onto a desk and publish it. + + Publishing is what carries an item through the http_push destination to + Superdesk Publisher, where the language rules drop it onto the matching + route — so with this in place ingest is hands-off end to end. Failures + are logged per item and never abort the ingest: a story that will not + validate simply stays unpublished for an editor to finish by hand. + """ + if not guids: + return + if not provider.get("config", {}).get("auto_publish", True): + return + + desk_id, stage_id = await self._resolve_publish_target(provider) + if desk_id is None: + logger.warning( + "Ghost auto-publish skipped: no desk to fetch onto " + "(set the provider config 'publish_desk')" + ) + return + + ingest_service = get_resource_service("ingest") + fetch_service = get_resource_service("fetch") + publish_service = get_resource_service("archive_publish") + + for guid in guids: + try: + ingest_item = await ingest_service.find_one_async(req=None, guid=guid) + if not ingest_item: + continue + # ``archived`` is stamped by the fetch service, so a set value + # means this item was already fetched — don't publish it twice. + if ingest_item.get("archived"): + continue + + fetch_doc = { + ID_FIELD: ingest_item[ID_FIELD], + "desk": str(desk_id), + "state": CONTENT_STATE.ROUTED, + } + if stage_id: + fetch_doc["stage"] = str(stage_id) + + archive_id = (await fetch_service.fetch([fetch_doc]))[0] + # ``allow_join_result`` is required because we publish from + # inside the ``update_provider`` celery task. Publishing an item + # with an image cascades to publishing the picture association, + # which matches a ``polling=True`` publish channel; that path + # calls ``enqueue_published.apply_async()`` and then ``.get()`` + # on the (eager) result. Celery forbids ``.get()`` inside a + # running task, but here the subtask runs eagerly in-process, so + # there is no remote worker to block on — this scopes the join + # guard off just for the publish. (Text-only items take a + # ``polling=False`` path and never hit ``.get()``, which is why + # image-free items published fine without this.) + with allow_join_result(): + await publish_service.patch_async( + archive_id, {"auto_publish": True} + ) + logger.info("Ghost auto-publish: published %s", guid) + except Exception: + logger.exception("Ghost auto-publish failed for %s", guid) + + async def _resolve_publish_target(self, provider): + """Return the ``(desk_id, stage_id)`` to fetch onto, or ``(None, None)``.""" + desks_service = get_resource_service("desks") + desk_name = ( + provider.get("config", {}).get("publish_desk") or DEFAULT_PUBLISH_DESK + ) + desk = await desks_service.find_one_async(req=None, name=desk_name) + if not desk: + desk = await desks_service.find_one_async(req=None) + if not desk: + return None, None + return desk[ID_FIELD], desk.get("incoming_stage") + register_feeding_service(GhostFeedingService)