diff --git a/Changelog.md b/Changelog.md index aa634d56..858f23ea 100644 --- a/Changelog.md +++ b/Changelog.md @@ -4,6 +4,24 @@ All notable changes are recorded here, newest release first. Versions follow [Semantic Versioning](https://semver.org/). Dates use America/Los_Angeles. The history before this initial versioned snapshot remains in Git. +## [1.0.8] - 2026-10-17 + +### Fixed + +- Derive stable anniversary dates from successfully sent posts instead of the + statistics collection start; allow explicit dates and handle February 29. + Skip missing history and isolate invalid channels rather than aborting all sends. +- Deliver annual introductions featuring the main feed and community, with a + deterministic paginated channel directory and direct text birthday notices. + The daily sweep no longer depends on Reddit listings or media availability. +- Persist annually unique message plans, acknowledged Telegram IDs, and run + outcomes independently of the queue cleaner. Resume rejected/unsent parts on + later sweeps; hold interrupted or ambiguous sends for review rather than risk + duplicates. Expose a read-only report/preview and keep test sends isolated. +- Propagate supply failures after reporting them so partial anniversary runs + cannot be recorded as successful tasks. Preserve the original error if + developer-chat reporting fails. No migration or historical replay is required. + ## [1.0.7] - 2026-10-17 ### Fixed diff --git a/README.md b/README.md index dbff8476..db4c7afd 100644 --- a/README.md +++ b/README.md @@ -139,3 +139,54 @@ R2T_LIVE_TESTS=1 .venv/bin/python -m unittest tests/test_live_resources.py -v ``` Leave `R2T_LIVE_TESTS` unset during offline test runs. + +Anniversary announcements +------------------------ + +`tech_anniversary` runs daily at 11:55 UTC without fetching a Reddit listing. +On a channel's anniversary, its introduction features `@reddit2telegram`, then +the `@r_channels` community, followed by the birthday greeting and a paginated +directory of other scheduled channels. A short birthday notice also goes to +the main channel, community, and developer chat. Identical destinations are +deduplicated. Announcements use text, so unavailable Reddit media cannot block +them. + +The default anniversary date is the earliest **successfully sent** post in +`urls`, never the start of statistics collection. This is an earliest-known-post +date, not a verified Telegram creation date. It is saved once in `settings` +under `_id: r2t_anniversary_date:` so history cleanup cannot +move it. A channel document's optional `anniversary_date` (`YYYY-MM-DD` or a +BSON datetime) overrides that default. Channels without either date are skipped +and reported; one missing or invalid channel does not prevent other sends. +February 29 anniversaries are observed on February 28 in non-leap years. + +Annual delivery records use `_id: r2t_anniversary_delivery::` +in `settings`. Their message text and directory order are frozen when queued; +each acknowledged Telegram message ID is saved before the next send. Daily +runs resume outstanding messages from earlier days and never repeat completed +parts. Channels removed from the schedule are not sent queued retries. Every +invocation also leaves a `kind: r2t_anniversary_run` record containing its start, +finish, outcome, due/completed/pending counts, missing history, and errors. The +next lease owner marks a crashed sweep as interrupted. The hourly task cleaner +does not remove any of these records. No historical +anniversaries are backfilled on installation. + +Inspect dates, upcoming/due channels, a complete message preview, pending parts, +and the last seven days of run history without writing state or sending messages: + +```bash +.venv/bin/python scripts/anniversary_report.py +.venv/bin/python scripts/anniversary_report.py --date 2027-02-12 --days 30 +``` + +A sweep lease prevents overlapping workers. Telegram API rejections are safe to +retry, while timeouts and interrupted requests are recorded as `uncertain`. +Telegram does not provide an idempotency key for sends, so these parts require +checking the destination before another attempt. After that check, an operator +can conditionally update the event in `settings`: select its `_id` and +`deliveries..state: uncertain`; set that part's state to `sent` and its +`message_ids` to the verified IDs if it arrived, or reset it to `pending` if it +did not. The next scheduled run resumes the remaining parts. A partial run is +reported as a supply failure, with the retained run ID for diagnosis. The +existing `supplier.py --test --sub tech_anniversary` path sends only a sample to +`@r_channels_test` and leaves production anniversary records untouched. diff --git a/VERSION b/VERSION index 238d6e88..b0f3d96f 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.0.7 +1.0.8 diff --git a/reddit2telegram/channels/tech_anniversary/app.py b/reddit2telegram/channels/tech_anniversary/app.py index a108c094..8a1a8494 100644 --- a/reddit2telegram/channels/tech_anniversary/app.py +++ b/reddit2telegram/channels/tech_anniversary/app.py @@ -1,72 +1,24 @@ -#encoding:utf-8 +"""Daily anniversary sweep; no Reddit listing is needed to send announcements.""" -import utils from utils import SupplyResult -from utils.tech import is_birthday_today, get_all_public_channels, get_dev_channel -from utils.tech import generate_list_of_channels, default_ending, chunker, get_all_public_submodules -from utils.tech import short_sleep, long_sleep -from utils.setup import get_config -from supplier import send_to_channel_from_subreddit -from channels.reddit2telegram.app import make_nice_submission +from utils.anniversaries import AnniversaryAnnouncements, announcement_messages +from utils.tech import get_dev_channel subreddit = 'all' t_channel = get_dev_channel() +requires_submission = False def send_post(submission, r2t): - channels_list = get_all_public_channels(r2t) + # supplier --test sets the initial destination. Never turn a test invocation + # into production sends or update production anniversary delivery records. + if r2t.t_channel == '@r_channels_test': + for text in announcement_messages('@r_channels_test', 1, ['@reddit2telegram', '@r_channels']): + r2t.send_text(text, parse_mode='HTML', disable_web_page_preview=True) + return SupplyResult.STOP_THIS_SUPPLY - for submodule_name in get_all_public_submodules(): - submodule = utils.channels_stuff.import_submodule(submodule_name) - channel = submodule.t_channel - bd_party, years = is_birthday_today(r2t, channel) - if bd_party and years > 0: - plural = 's' if years != 1 else '' - # To the @r_channels - long_sleep() - r2t.t_channel = '@r_channels' - cakes = 'πŸŽ‚' * years - text_to_send = '{cake}\n🎁 Today {channel} is {years_cnt} year{s} old.\nπŸŽ‰ Congratulations! 🎈'.format( - channel=channel, - years_cnt=years, - s=plural, - cake=cakes - ) - r2t.send_text(text_to_send) - # To the dev channel - long_sleep() - r2t.t_channel = get_dev_channel() - r2t.send_text(text_to_send) - # To the channels itself - long_sleep() - r2t.t_channel = channel - text1_to_send = text_to_send - list_of_channels = generate_list_of_channels(channels_list, random_permutation=True) - text3_to_send = default_ending() - r2t.send_text(text1_to_send) - short_sleep() - text2_to_send = 'Other @reddit2telegram channels powered by @r_channels:\n' - for l in chunker(list_of_channels, 100): - text2_to_send += '\n'.join(l) - r2t.send_text(text2_to_send) - text2_to_send = '' - short_sleep() - r2t.send_text(text3_to_send) - long_sleep() - if channel != '@reddit2telegram': - config = get_config() - send_to_channel_from_subreddit( - how_to_post=make_nice_submission, - channel_to_post='@reddit2telegram', - subreddit=submodule.subreddit, - submodule_name_to_promte=submodule_name, - submissions_ranking='top', - submissions_limit=1000, - config=config, - extra_args=True, - extra_ending=text_to_send - ) - long_sleep() - # It's not a proper supply, so just stop. + result = AnniversaryAnnouncements(r2t).run() + if result['status'] in ('failed', 'partial'): + raise RuntimeError('Anniversary sweep needs attention; run {}'.format(result['_id'])) return SupplyResult.STOP_THIS_SUPPLY diff --git a/reddit2telegram/reporting_stuff.py b/reddit2telegram/reporting_stuff.py index 56bbadea..8db0cdd7 100644 --- a/reddit2telegram/reporting_stuff.py +++ b/reddit2telegram/reporting_stuff.py @@ -2,6 +2,7 @@ import logging import sys import os +from functools import wraps import yaml import sentry_sdk @@ -60,13 +61,18 @@ def send_report_to_dev_chat(exc): def report_error(fn): + @wraps(fn) def wrapper(*args, **kwargs): try: - fn(*args, **kwargs) + return fn(*args, **kwargs) except Exception as e: if client: # has sentry instance client.capture_exception(e) else: - logging.exception('Exception Ignored.') - send_report_to_dev_chat(e) + logging.exception('Supply failed.') + try: + send_report_to_dev_chat(e) + except Exception: + logging.exception('Failed to report supply error to the developer chat.') + raise return wrapper diff --git a/reddit2telegram/utils/anniversaries.py b/reddit2telegram/utils/anniversaries.py new file mode 100644 index 00000000..08fb4294 --- /dev/null +++ b/reddit2telegram/utils/anniversaries.py @@ -0,0 +1,361 @@ +"""Stable anniversary dates and resumable, annually unique text announcements. + +All records live in settings, independently of the hourly task queue cleaner. +Telegram has no idempotency key: an interrupted/ambiguous send is held for review +instead of being automatically resent. Explicit API rejections can be retried. +""" + +import ast +import calendar +import csv +from datetime import date, datetime, time, timedelta, timezone +from html import escape +import logging +from pathlib import Path +import re +from uuid import uuid4 + +from pymongo import ReturnDocument +from pymongo.errors import DuplicateKeyError +from telegram.error import BadRequest, ChatMigrated, Forbidden, RetryAfter + +from utils import SupplyResult +from utils.tech import short_sleep + + +MAIN_CHANNEL = '@reddit2telegram' +COMMUNITY = '@r_channels' +DATE_KIND = 'r2t_anniversary_date' +EVENT_KIND = 'r2t_anniversary_delivery' +RUN_KIND = 'r2t_anniversary_run' +LOCK_ID = 'r2t_anniversary_sweep_lock' +LEASE = timedelta(hours=2) +HANDLE = re.compile(r'^@[a-zA-Z][a-zA-Z0-9_]{4,31}$') +EXCLUDED = {COMMUNITY, '@r_channels_test'} +CHANNELS_DIR = Path(__file__).resolve().parent.parent / 'channels' +logger = logging.getLogger(__name__) + + +def utc_now(): + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def utc_date(value): + if isinstance(value, datetime): + if value.tzinfo is not None: + value = value.astimezone(timezone.utc) + return value.date() + if isinstance(value, date): + return value + return date.fromisoformat(value) + + +def anniversary_age(birthday, today): + birthday, today = utc_date(birthday), utc_date(today) + month, day = birthday.month, birthday.day + if (month, day) == (2, 29) and not calendar.isleap(today.year): + day = 28 + years = today.year - birthday.year + return years if (today.month, today.day) == (month, day) and years > 0 else None + + +def local_channel(module): + # Legacy custom modules may have no DB document. Read their literal handle + # without executing import-time promotion selection or other side effects. + tree = ast.parse((CHANNELS_DIR / module / 'app.py').read_text()) + for statement in tree.body: + if isinstance(statement, ast.Assign) and any( + isinstance(target, ast.Name) and target.id == 't_channel' + for target in statement.targets): + return ast.literal_eval(statement.value) + raise ValueError('No literal public channel destination') + + +def announcement_messages(channel, years, channels, celebrated_on=None): + """Freeze a deterministic directory into single Telegram-sized messages.""" + plural = '' if years == 1 else 's' + introduction = ( + '🌍 Discover the Reddit2Telegram family\n\n' + 'πŸ“£ @reddit2telegram β€” our main channel for discovering posts and channels ' + 'from across the network.\n\n' + 'πŸ’¬ @r_channels β€” join the community, find new channels, ask questions, ' + 'and suggest your next favorite.\n\n' + 'πŸŽ‚ Celebrating {channel}: {years} year{plural} together!\n' + '{date_label}' + 'Thank you for reading, sharing, and being part of the community. ' + 'Here’s to another year of great discoveries!' + ).format(channel=escape(channel), years=years, plural=plural, + date_label=('πŸ“… {}\n'.format(celebrated_on.strftime('%d %B %Y')) + if celebrated_on is not None else '')) + others = sorted({name.lower() for name in channels} + - {channel.lower(), MAIN_CHANNEL, COMMUNITY, '@r_channels_test'}) + if not others: + return [introduction] + introduction += '\n\n🧭 Explore more of our channels in the directory below.' + pages, lines = [], [] + for number, name in enumerate(others, 1): + line = '{}. {}'.format(number, escape(name)) + # Count UTF-16 units and reserve room for the page heading. Keeping each + # call below 4096 avoids sender.send_text's untracked automatic splitting. + candidate = '\n'.join(lines + [line]) + if lines and (len(lines) == 100 or len(candidate.encode('utf-16-le')) // 2 > 3400): + pages.append('\n'.join(lines)) + lines = [] + lines.append(line) + if lines: + pages.append('\n'.join(lines)) + return [introduction] + [ + '🧭 More channels Β· {}/{}\n\n{}'.format(i, len(pages), page) + for i, page in enumerate(pages, 1) + ] + + +class AnniversaryAnnouncements: + def __init__(self, sender): + self.sender = sender + self.settings = sender.settings + self.owner = uuid4().hex + self.previous_owner = None + + def acquire(self): + try: + self.settings.update_one( + {'_id': LOCK_ID}, {'$setOnInsert': { + 'setting': LOCK_ID, 'expires_at': datetime.min, + }}, upsert=True, + ) + except DuplicateKeyError: + pass + now = utc_now() + previous = self.settings.find_one_and_update( + {'_id': LOCK_ID, 'expires_at': {'$lte': now}}, + {'$set': {'owner': self.owner, 'expires_at': now + LEASE}}, + return_document=ReturnDocument.BEFORE, + ) + self.previous_owner = previous.get('owner') if previous is not None else None + return previous is not None + + def renew(self): + now = utc_now() + result = self.settings.update_one( + {'_id': LOCK_ID, 'owner': self.owner, 'expires_at': {'$gt': now}}, + {'$set': {'expires_at': now + LEASE}}, + ) + if not result.matched_count: + raise RuntimeError('Anniversary sweep lease expired or changed owner') + + def release(self): + self.settings.update_one({'_id': LOCK_ID, 'owner': self.owner}, + {'$set': {'expires_at': datetime.min}}) + + def birthday(self, channel, channel_doc=None, persist=True): + selector = {'_id': '{}:{}'.format(DATE_KIND, channel)} + cached = self.settings.find_one(selector) + explicit = (channel_doc or {}).get('anniversary_date') + if explicit is not None: + birthday, source = utc_date(explicit), 'explicit' + elif cached is not None: + return utc_date(cached['date']) + else: + first_post = self.sender.urls.find_one( + {'channel': channel, 'sent': True, 'ts': {'$type': 'date'}}, + sort=[('ts', 1)], projection={'ts': 1}, + ) + if first_post is None: + return None + birthday, source = utc_date(first_post['ts']), 'first_sent_post' + if persist: + values = {'setting': selector['_id'], 'kind': DATE_KIND, 'channel': channel, + 'date': datetime.combine(birthday, time()), 'source': source} + # Automatic dates are written only once; explicit operator overrides + # deliberately replace the cached default without clearing delivery history. + update = '$set' if explicit is not None else '$setOnInsert' + self.settings.update_one(selector, {update: values}, upsert=True) + return birthday + + def inventory(self, persist=True, heartbeat=None): + channels, skipped, errors = {}, [], [] + with open(self.sender.config['cron_file']) as cron_file: + modules = sorted({row['submodule_name'] for row in csv.DictReader(cron_file, delimiter='\t') + if not row['submodule_name'].startswith('tech_')}) + for module in modules: + if heartbeat is not None: + heartbeat() + try: + # Database-backed channels need no module imports or promotion + # selection side effects simply to discover their handle/date. + doc = self.sender.urls.database['channels'].find_one({'submodule': module}) + channel = doc['channel'] if doc else local_channel(module) + if not isinstance(channel, str) or not HANDLE.fullmatch(channel): + continue + channel = channel.lower() + if channel in EXCLUDED or channel in channels: + continue + birthday = self.birthday(channel, doc, persist=persist) + if birthday is None: + skipped.append(channel) + logger.warning('Anniversary skipped: %s has no successful posting history', channel) + continue + channels[channel] = birthday + except Exception as exc: + errors.append({'submodule': module, 'error_type': type(exc).__name__}) + logger.warning('Could not inspect anniversary channel %s (%s)', module, type(exc).__name__) + return channels, skipped, errors + + def queue(self, channel, years, today, channels): + event_id = '{}:{}:{}'.format(EVENT_KIND, channel, today.year) + birthday_notice = ('πŸŽ‚ Celebrating {channel}: {years} year{plural} together!\n' + 'πŸ“… {anniversary}\n\n' + 'Discover this channel and more in our main feed @reddit2telegram.\n' + 'Join @r_channels to explore the network and meet the community.').format( + channel=escape(channel), years=years, plural='' if years == 1 else 's', + anniversary=today.strftime('%d %B %Y')) + deliveries = [{'destination': channel, 'text': text, 'state': 'pending'} + for text in announcement_messages(channel, years, channels, today)] + destinations = {str(channel).lower()} + for destination in [MAIN_CHANNEL, COMMUNITY, self.sender.config['telegram']['dev_chat']]: + if str(destination).lower() in destinations: + continue + destinations.add(str(destination).lower()) + deliveries.append({'destination': destination, 'text': birthday_notice, 'state': 'pending'}) + self.settings.update_one({'_id': event_id}, {'$setOnInsert': { + 'setting': event_id, 'kind': EVENT_KIND, 'channel': channel, 'year': today.year, + 'years': years, 'anniversary': today.isoformat(), 'created_at': utc_now(), + 'complete': False, 'deliveries': deliveries, + }}, upsert=True) + + def deliver(self, event): + blocked = set() + for index, delivery in enumerate(event['deliveries']): + self.renew() + destination = delivery['destination'] + path = 'deliveries.{}'.format(index) + selector = {'_id': event['_id']} + if delivery['state'] == 'sent' or destination in blocked: + continue + if delivery['state'] in ('sending', 'uncertain'): + # A process may have died after Telegram accepted a request. + self.settings.update_one( + {**selector, path + '.state': 'sending'}, + {'$set': {path + '.state': 'uncertain'}}, + ) + blocked.add(destination) + continue + attempt = uuid4().hex + claimed = self.settings.update_one( + {**selector, path + '.state': 'pending'}, + {'$set': {path + '.state': 'sending', path + '.attempt': attempt, + path + '.attempted_at': utc_now()}, + '$inc': {path + '.attempts': 1}}, + ) + if not claimed.matched_count: + blocked.add(destination) + continue + attempt_selector = {**selector, path + '.attempt': attempt} + acknowledged = [] + + def remember(ids): + acknowledged.extend(ids) + result = self.settings.update_one(attempt_selector, {'$set': { + path + '.state': 'sent', path + '.message_ids': list(acknowledged), + path + '.sent_at': utc_now(), + }}) + if not result.matched_count: + raise RuntimeError('Anniversary delivery attempt changed before acknowledgment') + + old_channel = self.sender.t_channel + old_callback = getattr(self.sender, 'on_message_sent', None) + try: + self.sender.t_channel = destination + self.sender.on_message_sent = remember + result = self.sender.send_text(delivery['text'], parse_mode='HTML', + disable_web_page_preview=True) + if result != SupplyResult.SUCCESSFULLY or not acknowledged: + raise RuntimeError('Anniversary message was not acknowledged') + except Exception as exc: + # Definite Telegram API rejection is safe to retry. All other + # failures may have accepted the message before losing the reply. + state = 'pending' if isinstance(exc, (BadRequest, ChatMigrated, Forbidden, RetryAfter)) else 'uncertain' + if acknowledged: + remember([]) + else: + self.settings.update_one(attempt_selector, {'$set': { + path + '.state': state, path + '.last_error_type': type(exc).__name__, + }}) + blocked.add(destination) + logger.warning('Anniversary delivery needs attention: %s to %s (%s)', + event['_id'], destination, type(exc).__name__) + finally: + self.sender.t_channel = old_channel + self.sender.on_message_sent = old_callback + short_sleep() + current = self.settings.find_one({'_id': event['_id']}) + complete = all(part['state'] == 'sent' for part in current['deliveries']) + self.settings.update_one({'_id': event['_id']}, {'$set': { + 'complete': complete, 'updated_at': utc_now(), + }}) + return complete + + def run(self, today=None): + today = utc_date(today or utc_now()) + run_id = '{}:{}'.format(RUN_KIND, uuid4().hex) + run = {'_id': run_id, 'kind': RUN_KIND, 'setting': run_id, 'owner': self.owner, + 'date': today.isoformat(), 'started_at': utc_now(), 'status': 'running', + 'channels': 0, 'due': 0, 'completed': 0, 'pending': 0, 'uncertain_messages': 0, + 'skipped_no_history': [], 'errors': []} + self.settings.insert_one(dict(run)) + acquired = False + try: + acquired = self.acquire() + if not acquired: + run['status'] = 'skipped_overlap' + return run + if self.previous_owner is not None: + self.settings.update_many( + {'kind': RUN_KIND, 'owner': self.previous_owner, 'status': 'running', + '_id': {'$ne': run_id}}, + {'$set': {'status': 'interrupted', 'finished_at': utc_now(), + 'errors': [{'error_type': 'InterruptedSweep'}]}}, + ) + channels, skipped, errors = self.inventory(heartbeat=self.renew) + run.update(channels=len(channels), skipped_no_history=skipped, errors=errors) + for channel, birthday in channels.items(): + self.renew() + years = anniversary_age(birthday, today) + if years is not None: + run['due'] += 1 + self.queue(channel, years, today, channels) + # Retry queued work even after its anniversary day has passed, but + # never post to a channel removed from the active schedule. + for event in self.settings.find({'kind': EVENT_KIND, 'complete': False}).batch_size(25): + self.renew() + if event['channel'] not in channels: + continue + try: + if self.deliver(event): + run['completed'] += 1 + except Exception as exc: + run['errors'].append({'event': event['_id'], 'error_type': type(exc).__name__}) + logger.warning('Anniversary event failed: %s (%s)', event['_id'], type(exc).__name__) + for event in self.settings.find({'kind': EVENT_KIND, 'complete': False}).batch_size(25): + run['pending'] += 1 + run['uncertain_messages'] += sum(part['state'] in ('sending', 'uncertain') + for part in event['deliveries']) + run['status'] = 'partial' if run['pending'] or run['errors'] else 'success' + return run + except Exception as exc: + run['status'] = 'failed' + run['errors'].append({'error_type': type(exc).__name__}) + raise + finally: + run['finished_at'] = utc_now() + try: + self.settings.update_one({'_id': run['_id']}, {'$set': { + key: value for key, value in run.items() if key != '_id' + }}) + finally: + if acquired: + self.release() + logger.info('Anniversary run %s: %s; due=%d complete=%d pending=%d uncertain=%d', + run['_id'], run['status'], run['due'], run['completed'], + run['pending'], run['uncertain_messages']) diff --git a/scripts/anniversary_report.py b/scripts/anniversary_report.py new file mode 100644 index 00000000..a48b660e --- /dev/null +++ b/scripts/anniversary_report.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 +"""Read-only anniversary inventory, message preview, and retained run history.""" + +import argparse +from datetime import date, datetime, timedelta +import json +import os +from pathlib import Path +import sys +from types import SimpleNamespace + +import pymongo +import yaml + + +APP_DIR = Path(__file__).resolve().parent.parent / 'reddit2telegram' +sys.path.insert(0, str(APP_DIR)) +from utils.anniversaries import AnniversaryAnnouncements, RUN_KIND, EVENT_KIND +from utils.anniversaries import anniversary_age, announcement_messages +from utils.anniversaries import utc_now + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--config', type=Path, default=APP_DIR / 'configs/prod.yml') + parser.add_argument('--date', type=date.fromisoformat, default=utc_now().date()) + parser.add_argument('--days', type=int, default=7, help='Retained run history, including today') + args = parser.parse_args() + if args.days < 1: + parser.error('--days must be positive') + with args.config.open() as config_file: + config = yaml.safe_load(config_file) + os.chdir(APP_DIR) + with pymongo.MongoClient(config['db']['host'], serverSelectionTimeoutMS=5000) as client: + db = client[config['db']['name']] + service = AnniversaryAnnouncements(SimpleNamespace( + settings=db.settings, urls=db.urls, config=config, + )) + channels, skipped, errors = service.inventory(persist=False) + due = [{'channel': channel, 'birthday': birthday.isoformat(), + 'years': anniversary_age(birthday, args.date)} + for channel, birthday in sorted(channels.items()) + if anniversary_age(birthday, args.date) is not None] + cutoff = datetime.combine(utc_now().date() - timedelta(days=args.days - 1), + datetime.min.time()) + runs = list(db.settings.find( + {'kind': RUN_KIND, 'started_at': {'$gte': cutoff}}, + ).sort('started_at', -1)) + pending = [] + for event in db.settings.find({'kind': EVENT_KIND, 'complete': False}).batch_size(25): + pending.append({'id': event['_id'], 'channel': event['channel'], + 'anniversary': event['anniversary'], + 'parts': [{'index': i, 'destination': part['destination'], + 'state': part['state'], 'message_ids': part.get('message_ids', [])} + for i, part in enumerate(event['deliveries'])]}) + example = due[0] if due else {'channel': '@example_channel', 'years': 1} + print(json.dumps({ + 'read_only': True, 'date': args.date.isoformat(), 'channels': len(channels), + 'skipped_no_history': skipped, 'errors': errors, 'due': due, 'runs': runs, + 'pending': pending, + 'preview': announcement_messages(example['channel'], example['years'], channels, args.date), + }, ensure_ascii=False, default=str, indent=2)) + + +if __name__ == '__main__': + main() diff --git a/tests/test_anniversaries.py b/tests/test_anniversaries.py new file mode 100644 index 00000000..89d0d578 --- /dev/null +++ b/tests/test_anniversaries.py @@ -0,0 +1,408 @@ +from datetime import date, datetime, timedelta, timezone +from pathlib import Path +import sys +from types import SimpleNamespace +from unittest import mock + +import mongomock +import pytest +from pymongo.errors import AutoReconnect +from telegram.error import BadRequest, Forbidden, RetryAfter, TimedOut + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'reddit2telegram')) +import supplier +import reporting_stuff +import utils +from utils import anniversaries as ann + +with mock.patch('utils.tech.get_dev_channel', return_value=-1): + from channels.tech_anniversary import app as job + + +@pytest.fixture +def sweep(monkeypatch, tmp_path): + db = mongomock.MongoClient().db + cron = tmp_path / 'own.cron' + cron.write_text('MASK\tsubmodule_name\n55 11 * * *\tr_alpha\n55 11 * * *\tr_beta\n') + sender = SimpleNamespace( + settings=db.settings, urls=db.urls, stats=db.stats, t_channel=-1, + config={'cron_file': str(cron), 'telegram': {'dev_chat': -1}}, + calls=[], errors={}, after_ack=None, on_message_sent=None, + ) + db.channels.insert_many([ + {'submodule': 'r_alpha', 'channel': '@r_alpha'}, + {'submodule': 'r_beta', 'channel': '@r_beta'}, + ]) + db.urls.insert_many([ + {'channel': '@r_alpha', 'sent': True, 'ts': datetime(2019, 10, 17)}, + {'channel': '@r_beta', 'sent': True, 'ts': datetime(2020, 10, 17)}, + ]) + + def send_text(text, **kwargs): + error = sender.errors.get(sender.t_channel) + if error is not None: + raise error + sender.calls.append((sender.t_channel, text, kwargs)) + if sender.on_message_sent is not None: + sender.on_message_sent([len(sender.calls)]) + if sender.after_ack is not None: + raise sender.after_ack + return utils.SupplyResult.SUCCESSFULLY + + sender.send_text = send_text + monkeypatch.setattr(ann, 'short_sleep', lambda: None) + return sender, db, ann.AnniversaryAnnouncements(sender) + + +def event(db, channel='@r_alpha', year=2026): + return db.settings.find_one({'kind': ann.EVENT_KIND, 'channel': channel, 'year': year}) + + +def test_date_uses_successful_posts_and_survives_history_cleanup(sweep): + sender, db, service = sweep + db.urls.insert_many([ + {'channel': '@r_alpha', 'sent': False, 'ts': datetime(2017, 1, 1)}, + {'channel': '@r_alpha', 'sent': True}, + {'channel': '@r_alpha', 'sent': True, 'ts': 'invalid'}, + ]) + db.stats.insert_one({'channel': '@r_alpha', 'ts': datetime(2026, 1, 2)}) + assert service.birthday('@r_alpha') == date(2019, 10, 17) + db.urls.delete_many({}) + assert service.birthday('@r_alpha') == date(2019, 10, 17) + assert service.birthday('@r_nohistory') is None + assert db.settings.find_one({'kind': ann.DATE_KIND})['source'] == 'first_sent_post' + + +def test_explicit_date_overrides_cached_default_and_invalid_date_is_rejected(sweep): + _, db, service = sweep + service.birthday('@r_alpha') + assert service.birthday('@r_alpha', {'anniversary_date': '2015-05-12'}) == date(2015, 5, 12) + assert service.birthday('@r_alpha') == date(2015, 5, 12) + with pytest.raises(ValueError): + service.birthday('@r_alpha', {'anniversary_date': 'not-a-date'}) + assert db.settings.find_one({'channel': '@r_alpha', 'kind': ann.DATE_KIND})['source'] == 'explicit' + + +def test_read_only_inventory_does_not_initialize_dates_or_delivery_records(sweep): + _, db, service = sweep + channels, skipped, errors = service.inventory(persist=False) + assert channels == {'@r_alpha': date(2019, 10, 17), '@r_beta': date(2020, 10, 17)} + assert skipped == errors == [] + assert db.settings.count_documents({}) == 0 + + +def test_legacy_destination_is_read_without_executing_module(sweep, monkeypatch, tmp_path): + sender, db, service = sweep + module = tmp_path / 'r_legacy' + module.mkdir() + (module / 'app.py').write_text('raise AssertionError("Do not execute me")\nt_channel = "@r_legacy"\n') + monkeypatch.setattr(ann, 'CHANNELS_DIR', tmp_path) + Path(sender.config['cron_file']).write_text('MASK\tsubmodule_name\n55 11 * * *\tr_legacy\n') + db.urls.insert_one({'channel': '@r_legacy', 'sent': True, 'ts': datetime(2018, 1, 1)}) + assert service.inventory(persist=False) == ({'@r_legacy': date(2018, 1, 1)}, [], []) + assert db.settings.count_documents({}) == 0 + + +@pytest.mark.parametrize('birthday,today,expected', [ + ('2019-10-17', '2026-10-17', 7), + ('2019-10-17', '2026-10-18', None), + ('2026-10-17', '2026-10-17', None), + ('2027-10-17', '2026-10-17', None), + ('2020-02-29', '2026-02-28', 6), + ('2020-02-29', '2026-03-01', None), + ('2020-02-29', '2028-02-28', None), + ('2020-02-29', '2028-02-29', 8), +]) +def test_anniversary_dates_including_leap_years(birthday, today, expected): + assert ann.anniversary_age(birthday, today) == expected + + +def test_anniversary_date_normalizes_aware_timestamps_to_utc(): + observed = datetime(2019, 10, 18, 1, tzinfo=timezone(timedelta(hours=3))) + assert ann.anniversary_age(observed, date(2026, 10, 17)) == 7 + + +def test_inventory_isolates_missing_configuration_history_and_invalid_dates(sweep, monkeypatch): + sender, db, service = sweep + Path(sender.config['cron_file']).write_text( + 'MASK\tsubmodule_name\n' + ''.join('55 11 * * *\t{}\n'.format(name) for name in [ + 'r_alpha', 'r_beta', 'r_missing', 'r_nourls', 'r_invalid', 'r_duplicate', + 'r_test', 'r_private', 'tech_anniversary', + ])) + db.channels.insert_many([ + {'submodule': 'r_nourls', 'channel': '@r_nourls'}, + {'submodule': 'r_invalid', 'channel': '@r_invalid', 'anniversary_date': 'bad'}, + {'submodule': 'r_duplicate', 'channel': '@R_ALPHA'}, + {'submodule': 'r_test', 'channel': '@r_channels_test'}, + {'submodule': 'r_private', 'channel': -123}, + ]) + monkeypatch.setattr(utils.channels_stuff, 'import_submodule', mock.Mock(side_effect=ValueError)) + channels, skipped, errors = service.inventory() + assert set(channels) == {'@r_alpha', '@r_beta'} + assert skipped == ['@r_nourls'] + assert {error['submodule'] for error in errors} == {'r_missing', 'r_invalid'} + + +def test_copy_promotes_featured_pair_first_and_splits_a_large_directory(): + others = ['@channel_' + str(i).zfill(23) for i in range(900)] + messages = ann.announcement_messages('@birthday_channel', 7, others + [ + '@reddit2telegram', '@r_channels', '@r_channels_test', '@birthday_channel', others[0], + ], date(2026, 10, 17)) + intro = messages[0] + assert intro.index('@reddit2telegram') < intro.index('@r_channels') < intro.index('@birthday_channel') + assert '7 years' in intro and '17 October 2026' in intro + assert all(len(text.encode('utf-16-le')) // 2 < 4096 for text in messages) + directory = '\n'.join(messages[1:]) + assert all(directory.count(handle) == 1 for handle in others) + assert '@birthday_channel' not in directory and '@r_channels' not in directory + assert all('More channels Β·' in page for page in messages[1:]) + + +def test_annual_event_is_unique_and_plan_is_frozen(sweep): + _, db, service = sweep + service.queue('@r_alpha', 7, date(2026, 10, 17), ['@r_alpha', '@r_beta']) + planned = event(db)['deliveries'] + service.queue('@r_alpha', 7, date(2026, 11, 1), ['@changed_directory']) + assert event(db)['deliveries'] == planned + service.queue('@r_alpha', 8, date(2027, 10, 17), ['@r_beta']) + assert db.settings.count_documents({'kind': ann.EVENT_KIND}) == 2 + + +def test_repeated_run_does_not_repeat_sends_and_history_survives_task_cleaner(sweep): + sender, db, service = sweep + first = service.run(date(2026, 10, 17)) + assert first['status'] == 'success' and first['due'] == first['completed'] == 2 + sends = list(sender.calls) + second = service.run(date(2026, 10, 17)) + assert second['status'] == 'success' and second['completed'] == 0 + assert sender.calls == sends + db.tasks.delete_many({}) + assert db.settings.count_documents({'kind': ann.RUN_KIND}) == 2 + assert db.settings.count_documents({'kind': ann.EVENT_KIND, 'complete': True}) == 2 + + +def test_definite_rejection_isolated_and_retries_only_unsent_parts_next_day(sweep): + sender, db, service = sweep + sender.errors['@r_alpha'] = Forbidden('not an administrator') + first = service.run(date(2026, 10, 17)) + assert first['status'] == 'partial' and first['completed'] == 1 + assert event(db, '@r_beta')['complete'] + assert not event(db)['complete'] + sent_before = list(sender.calls) + sender.errors.clear() + second = service.run(date(2026, 10, 18)) + assert second['status'] == 'success' and second['due'] == 0 and second['completed'] == 1 + retried = sender.calls[len(sent_before):] + assert retried and {destination for destination, *_ in retried} == {'@r_alpha'} + assert all('17 October 2026' in text for _, text, _ in retried if 'family' in text) + + +def test_resume_after_a_directory_page_is_rejected_does_not_repeat_intro(sweep): + sender, db, service = sweep + service.acquire() + service.queue('@r_alpha', 7, date(2026, 10, 17), ['@r_alpha', '@r_beta']) + original = sender.send_text + + def reject_directory(text, **kwargs): + if sender.t_channel == '@r_alpha' and 'More channels Β·' in text: + raise RetryAfter(30) + return original(text, **kwargs) + + sender.send_text = reject_directory + assert not service.deliver(event(db)) + sender.send_text = original + before = len(sender.calls) + assert service.deliver(event(db)) + assert len(sender.calls) == before + 1 + assert 'More channels Β·' in sender.calls[-1][1] + assert sum('family' in text for channel, text, _ in sender.calls if channel == '@r_alpha') == 1 + + +@pytest.mark.parametrize('error', [TimedOut(), RuntimeError('lost response')]) +def test_ambiguous_delivery_is_held_for_review_and_never_automatically_reposted(sweep, error): + sender, db, service = sweep + sender.errors['@r_alpha'] = error + first = service.run(date(2026, 10, 17)) + assert first['uncertain_messages'] == 1 + assert event(db)['deliveries'][0]['state'] == 'uncertain' + sender.errors.clear() + second = service.run(date(2026, 10, 18)) + assert second['status'] == 'partial' and second['uncertain_messages'] == 1 + assert not any(channel == '@r_alpha' for channel, *_ in sender.calls) + assert event(db, '@r_beta')['complete'] + + +def test_process_interruption_does_not_repeat_a_potentially_sent_message(sweep): + sender, db, service = sweep + service.queue('@r_alpha', 7, date(2026, 10, 17), ['@r_beta']) + doc = event(db) + db.settings.update_one({'_id': doc['_id']}, {'$set': {'deliveries.0.state': 'sending'}}) + run = service.run(date(2026, 10, 18)) + assert run['status'] == 'partial' and run['uncertain_messages'] == 1 + assert not any(channel == '@r_alpha' for channel, *_ in sender.calls) + assert event(db)['deliveries'][0]['state'] == 'uncertain' + + +def test_late_acknowledgment_is_not_overwritten_by_stale_recovery(sweep): + _, db, service = sweep + service.queue('@r_alpha', 7, date(2026, 10, 17), ['@r_beta']) + service.acquire() + stale = event(db) + stale['deliveries'][0]['state'] = 'sending' + db.settings.update_one({'_id': stale['_id']}, {'$set': { + 'deliveries.0.state': 'sent', 'deliveries.0.message_ids': [100], + }}) + service.deliver(stale) + assert event(db)['deliveries'][0]['state'] == 'sent' + assert event(db)['deliveries'][0]['message_ids'] == [100] + + +def test_database_outage_after_telegram_ack_does_not_cause_a_duplicate(sweep, monkeypatch): + sender, db, service = sweep + update_one = db.settings.update_one + + def lose_ack(selector, update, *args, **kwargs): + if any(key.endswith('.sent_at') for key in update.get('$set', {})): + raise AutoReconnect('DB connection lost while persisting Telegram response') + return update_one(selector, update, *args, **kwargs) + + with monkeypatch.context() as patch: + patch.setattr(db.settings, 'update_one', lose_ack) + run = service.run(date(2026, 10, 17)) + assert run['status'] == 'partial' and run['uncertain_messages'] == 2 + assert event(db)['deliveries'][0]['state'] == 'sending' + service.run(date(2026, 10, 18)) + assert event(db)['deliveries'][0]['state'] == 'uncertain' + assert sum(channel == '@r_alpha' for channel, *_ in sender.calls) == 1 + assert sum(channel == '@r_beta' for channel, *_ in sender.calls) == 1 + + +def test_acknowledgment_is_saved_even_if_a_later_step_raises(sweep): + sender, db, service = sweep + sender.after_ack = RuntimeError('failure after acknowledgment') + first = service.run(date(2026, 10, 17)) + assert first['status'] == 'partial' # Later directory parts resume next sweep. + assert first['uncertain_messages'] == 0 + before = len(sender.calls) + sender.after_ack = None + assert service.run(date(2026, 10, 18))['status'] == 'success' + assert all('More channels Β·' in text for _, text, _ in sender.calls[before:]) + assert all(part['message_ids'] for part in event(db)['deliveries']) + + +def test_non_acknowledged_success_does_not_mark_event_complete(sweep): + sender, db, service = sweep + sender.send_text = lambda *args, **kwargs: utils.SupplyResult.SUCCESSFULLY + run = service.run(date(2026, 10, 17)) + assert run['status'] == 'partial' and run['uncertain_messages'] > 0 + assert not event(db)['complete'] + + +def test_inactive_channel_is_not_sent_a_retry(sweep): + sender, db, service = sweep + service.queue('@r_inactive', 7, date(2026, 10, 17), ['@r_alpha']) + run = service.run(date(2026, 10, 18)) + assert sender.calls == [] + assert run['pending'] == 1 and run['status'] == 'partial' + + +def test_main_channel_birthday_does_not_send_two_introductions_to_same_destination(sweep): + _, db, service = sweep + service.queue(ann.MAIN_CHANNEL, 7, date(2026, 10, 17), [ann.MAIN_CHANNEL, '@r_beta']) + doc = event(db, ann.MAIN_CHANNEL) + destinations = [part['destination'] for part in doc['deliveries']] + assert destinations.count(ann.MAIN_CHANNEL) == 2 # One introduction and one directory page. + assert destinations.count(ann.COMMUNITY) == 1 + + +def test_overlap_is_recorded_and_expired_lease_cannot_be_renewed_or_released_by_old_owner(sweep): + sender, db, first = sweep + second = ann.AnniversaryAnnouncements(sender) + assert first.acquire() + assert second.run(date(2026, 10, 17))['status'] == 'skipped_overlap' + assert not second.acquire() + db.settings.update_one({'_id': ann.LOCK_ID}, {'$set': {'expires_at': datetime.min}}) + assert second.acquire() + with pytest.raises(RuntimeError, match='lease expired'): + first.renew() + first.release() + assert db.settings.find_one({'_id': ann.LOCK_ID})['owner'] == second.owner + assert db.settings.find_one({'kind': ann.RUN_KIND})['status'] == 'skipped_overlap' + + +def test_fatal_sweep_failure_is_retained_and_lock_is_released(sweep, monkeypatch): + _, db, service = sweep + monkeypatch.setattr(service, 'inventory', mock.Mock(side_effect=OSError('cron missing'))) + with pytest.raises(OSError): + service.run(date(2026, 10, 17)) + record = db.settings.find_one({'kind': ann.RUN_KIND}) + assert record['status'] == 'failed' and record['finished_at'] >= record['started_at'] + assert record['errors'] == [{'error_type': 'OSError'}] + assert db.settings.find_one({'_id': ann.LOCK_ID})['expires_at'] == datetime.min + + +def test_next_owner_marks_an_interrupted_run_in_retained_history(sweep): + sender, db, first = sweep + first.acquire() + db.settings.insert_one({'_id': 'previous-run', 'kind': ann.RUN_KIND, + 'owner': first.owner, 'status': 'running', 'started_at': ann.utc_now()}) + db.settings.update_one({'_id': ann.LOCK_ID}, {'$set': {'expires_at': datetime.min}}) + second = ann.AnniversaryAnnouncements(sender) + assert second.run(date(2026, 10, 18))['status'] == 'success' + previous = db.settings.find_one({'_id': 'previous-run'}) + assert previous['status'] == 'interrupted' + assert previous['errors'] == [{'error_type': 'InterruptedSweep'}] + + +def test_job_runs_without_reddit_and_test_mode_never_updates_production_ledger(sweep, monkeypatch): + sender, db, _ = sweep + sender.t_channel = '@r_channels_test' + monkeypatch.setattr(supplier.utils.channels_stuff, 'import_submodule', lambda name: job) + monkeypatch.setattr(supplier.utils, 'Reddit2TelegramSender', mock.MagicMock( + return_value=mock.MagicMock(__enter__=lambda _: sender))) + reddit = mock.Mock(side_effect=AssertionError('Anniversary job must not fetch Reddit')) + monkeypatch.setattr(supplier, 'send_to_channel_from_subreddit', reddit) + assert supplier.supply('tech_anniversary', sender.config, is_test=True) == utils.SupplyResult.STOP_THIS_SUPPLY + assert sender.calls and {destination for destination, *_ in sender.calls} == {'@r_channels_test'} + assert db.settings.count_documents({}) == 0 + reddit.assert_not_called() + + +def test_job_exposes_partial_delivery_as_failed_supply(sweep, monkeypatch): + sender, _, service = sweep + monkeypatch.setattr(job, 'AnniversaryAnnouncements', lambda _: service) + sender.errors['@r_alpha'] = BadRequest('chat not found') + with pytest.raises(RuntimeError, match='needs attention'): + job.send_post(None, sender) + + +def test_queue_records_partial_supply_as_failure_even_if_error_reporting_fails(sweep, monkeypatch): + import task_queue + + sender, db, service = sweep + monkeypatch.setattr(job, 'AnniversaryAnnouncements', lambda _: service) + monkeypatch.setattr(utils.channels_stuff, 'import_submodule', lambda name: job) + monkeypatch.setattr(utils, 'Reddit2TelegramSender', mock.MagicMock( + return_value=mock.MagicMock(__enter__=lambda _: sender))) + monkeypatch.setattr(supplier, 'long_sleep', lambda *args: None) + monkeypatch.setattr(reporting_stuff, 'client', None) + monkeypatch.setattr(reporting_stuff, 'send_report_to_dev_chat', mock.Mock(side_effect=OSError)) + sender.errors['@r_alpha'] = Forbidden('not an administrator') + task = db.tasks.insert_one({'status': task_queue.TaskStatus.NEW.value}) + task_queue.execute_task(db.tasks, task.inserted_id, task_queue.TASK_SUPPLY, + {'submodule_name': 'tech_anniversary', 'config': sender.config}) + assert db.tasks.find_one({'_id': task.inserted_id})['status'] == task_queue.TaskStatus.FAILED.value + assert db.settings.find_one({'kind': ann.RUN_KIND})['status'] == 'partial' + + +def test_reporting_decorator_preserves_success_value_and_original_failure(monkeypatch): + monkeypatch.setattr(reporting_stuff, 'client', None) + monkeypatch.setattr(reporting_stuff, 'send_report_to_dev_chat', mock.Mock(side_effect=OSError)) + assert reporting_stuff.report_error(lambda: 'success')() == 'success' + + @reporting_stuff.report_error + def broken_job(): + raise ValueError('original job failure') + + with pytest.raises(ValueError, match='original job failure'): + broken_job()