digest.py

import os

from infrai_client import ack_digest, consume_digests, create_digest_schedule, publish_digest


def subscribers_for_digest(subscribers, week_number):
    return [subscriber for subscriber in subscribers if subscriber["active"] and subscriber["last_digest"] < week_number]


def publish_weekly_digest(subscribers, stories, week_number):
    recipients = subscribers_for_digest(subscribers, week_number)
    payload = {"week": week_number, "recipients": recipients, "stories": stories}
    return publish_digest(payload)


def main():
    stories = [{"title": "Studio notes", "url": "https://example.com/studio-notes"}]
    subscribers = [{"email": "reader@example.com", "active": True, "last_digest": 17}]
    week_number = 18
    schedule = create_digest_schedule(os.environ["DIGEST_TASK_URL"])
    published = publish_weekly_digest(subscribers, stories, week_number)
    messages = consume_digests(max_messages=10, visibility_timeout=60)
    for message in messages.get("messages", []):
        ack_digest(message["message_id"])
    print({"job_id": schedule.get("job_id"), "published": published, "consumed": len(messages.get("messages", []))})


if __name__ == "__main__":
    main()