import time from dataclasses import dataclass from pathlib import Path from linkki_poster.formatting import CAPTION_LIMIT, ParsedTemplate, parse_markdown_template, render_template from linkki_poster.models import ( Config, Language, LanguageAssets, LanguageBinding, LANGUAGE_DISPLAY_NAMES, PublishBinding, ScanResult, TextSegmentBinding, ) from linkki_poster.progress import ProgressDisplay from linkki_poster.scanner import scan_directory from linkki_poster.telegram_api import ( TelegramAPI, TelegramApiError, is_message_missing_error, is_message_not_modified_error, ) from linkki_poster.workflow import ( build_edit_label, build_image_label, build_message_url, build_text_label, is_caption_too_long_error, normalize_channel_username, resolve_global_image_strategy, resolve_link_target, split_markdown_text_for_messages, ) @dataclass(frozen=True) class PublishResult: binding: PublishBinding message_urls: dict[Language, str] @dataclass(frozen=True) class DeleteBindingResult: warnings: list[str] failures: list[str] binding: PublishBinding | None @dataclass(frozen=True) class DiscussionResolutionStats: updates_seen: int discussion_group_updates: int automatic_forward_updates: int linked_channel_forward_updates: int @dataclass(frozen=True) class DiscussionPinningResult: binding: PublishBinding pinning_failed: bool @dataclass class _PublishedTextSegment: label: str binding: TextSegmentBinding parsed_template: ParsedTemplate last_rendered: str @dataclass(frozen=True) class _PublishedLanguage: binding: LanguageBinding segments: list[_PublishedTextSegment] def publish_directory( directory: Path, config: Config, ordered_languages: list[Language] | None = None, use_captions: bool = True, progress: ProgressDisplay | None = None, disable_link_previews: dict[Language, bool] | None = None, ) -> PublishResult: scan_result = scan_directory(directory) languages = order_scan_languages(scan_result, ordered_languages) reporter = progress or ProgressDisplay() api = TelegramAPI(config.bot_token) normalized_username = normalize_channel_username(config.channel_username) link_target = resolve_link_target(config.chat_id, normalized_username) language_links: dict[Language, str] = {} published_languages: dict[Language, _PublishedLanguage] = {} global_message_ids: list[int] = [] global_attached_to_language: Language | None = None posted_message_ids: list[int] = [] leading_global_images, effective_languages = resolve_global_image_strategy(scan_result.global_images, languages) if scan_result.global_images and not leading_global_images and effective_languages: global_attached_to_language = effective_languages[0].language try: if leading_global_images: global_message_ids = post_images( api=api, chat_id=config.chat_id, image_paths=leading_global_images, label=build_image_label("global", len(leading_global_images)), reporter=reporter, posted_message_ids=posted_message_ids, ) for language_assets in effective_languages: published_language = publish_language( api=api, chat_id=config.chat_id, language_assets=language_assets, language_links=language_links, disable_link_preview=bool((disable_link_previews or {}).get(language_assets.language, False)), link_target=link_target, use_captions=use_captions, reporter=reporter, posted_message_ids=posted_message_ids, ) published_languages[language_assets.language] = published_language language_links[language_assets.language] = published_language.binding.primary_message_url backfill_links( api=api, chat_id=config.chat_id, published_segments={language: item.segments for language, item in published_languages.items()}, language_links=language_links, disable_link_previews=disable_link_previews or {}, reporter=reporter, ) return PublishResult( binding=PublishBinding( global_message_ids=global_message_ids, global_attached_to_language=global_attached_to_language, languages={language: item.binding for language, item in published_languages.items()}, discussion_message_ids={}, global_primary_message_id=global_message_ids[0] if global_message_ids and global_attached_to_language is None else None, ), message_urls=language_links, ) except Exception: rollback_messages(api, config.chat_id, posted_message_ids, reporter) raise def update_texts( directory: Path, config: Config, binding: PublishBinding, changed_languages: set[Language], progress: ProgressDisplay | None = None, disable_link_previews: dict[Language, bool] | None = None, ) -> PublishBinding: reporter = progress or ProgressDisplay() api = TelegramAPI(config.bot_token) scan_result = scan_directory(directory) scan_languages = {language_assets.language: language_assets for language_assets in scan_result.languages} language_links = {language: language_binding.primary_message_url for language, language_binding in binding.languages.items()} normalized_username = normalize_channel_username(config.channel_username) link_target = resolve_link_target(config.chat_id, normalized_username) updated_languages: dict[Language, LanguageBinding] = dict(binding.languages) for language in changed_languages: language_assets = scan_languages.get(language) existing_binding = binding.languages.get(language) if language_assets is None or existing_binding is None: continue chunk_texts = split_markdown_text_for_messages(language_assets.text_raw, link_target) if len(chunk_texts) != len(existing_binding.text_segments): raise ValueError( f"Updated {LANGUAGE_DISPLAY_NAMES[language]} text changed message count. Delete or unlink and publish again." ) for index, (chunk_text, existing_segment) in enumerate(zip(chunk_texts, existing_binding.text_segments), start=1): parsed_template = parse_markdown_template(chunk_text) rendered_text = render_template(parsed_template, language_links) step_index = reporter.add_step(build_edit_label(build_text_label(language, index, len(chunk_texts)))) reporter.update(step_index, "in_progress") try: if existing_segment.mode == "message": api.edit_message_text( config.chat_id, existing_segment.message_id, rendered_text, parse_mode="HTML", disable_link_preview=bool((disable_link_previews or {}).get(language, False)), ) else: api.edit_message_caption(config.chat_id, existing_segment.message_id, rendered_text, parse_mode="HTML") except TelegramApiError as error: if is_message_missing_error(error): raise ValueError( f"Telegram message for {LANGUAGE_DISPLAY_NAMES[language]} no longer exists. " "Delete from Telegram or unlink this broadcast and publish again." ) from error if is_message_not_modified_error(error): reporter.update(step_index, "skipped", "no Telegram-visible change") continue raise reporter.update(step_index, "done") return PublishBinding( global_message_ids=binding.global_message_ids, global_attached_to_language=binding.global_attached_to_language, languages=updated_languages, discussion_message_ids=dict(binding.discussion_message_ids), global_primary_message_id=binding.global_primary_message_id, global_discussion_message_id=binding.global_discussion_message_id, ) def delete_binding_from_telegram( config: Config, binding: PublishBinding, progress: ProgressDisplay | None = None, ) -> DeleteBindingResult: reporter = progress or ProgressDisplay() api = TelegramAPI(config.bot_token) failures: list[str] = [] warnings: list[str] = [] deleted_channel_ids: set[int] = set() deleted_discussion_channel_ids: set[int] = set() deleted_discussion_message_ids: set[int] = set() if config.discussion_group_id: discussion_targets: list[tuple[int | None, int]] = [] if binding.global_discussion_message_id: discussion_targets.append((binding.global_primary_message_id, binding.global_discussion_message_id)) for language_binding in binding.languages.values(): if language_binding.discussion_message_id: discussion_targets.append((language_binding.primary_message_id, language_binding.discussion_message_id)) seen_discussion_ids: set[int] = set() for channel_message_id, discussion_message_id in reversed(discussion_targets): if discussion_message_id in seen_discussion_ids: continue seen_discussion_ids.add(discussion_message_id) step_index = reporter.add_step(f"Delete discussion message {discussion_message_id}") reporter.update(step_index, "in_progress") try: api.delete_message(config.discussion_group_id, discussion_message_id) except TelegramApiError as error: if is_message_missing_error(error): detail = f"{error.description}; it may already have been deleted manually" reporter.update(step_index, "skipped", detail) warnings.append(f"Discussion message {discussion_message_id}: {error.description}") if channel_message_id: deleted_discussion_channel_ids.add(channel_message_id) deleted_discussion_message_ids.add(discussion_message_id) continue reporter.update(step_index, "failed", error.description) failures.append(f"Discussion message {discussion_message_id}: {error.description}") continue if channel_message_id: deleted_discussion_channel_ids.add(channel_message_id) deleted_discussion_message_ids.add(discussion_message_id) reporter.update(step_index, "done") if failures: return DeleteBindingResult( warnings=warnings, failures=failures, binding=prune_binding_after_deletions( binding, deleted_channel_ids=set(), deleted_discussion_channel_ids=deleted_discussion_channel_ids, deleted_discussion_message_ids=deleted_discussion_message_ids, chat_id=config.chat_id, channel_username=config.channel_username, ), ) message_ids: list[int] = [] message_ids.extend(binding.global_message_ids) for language_binding in binding.languages.values(): message_ids.extend(language_binding.image_message_ids) message_ids.extend(segment.message_id for segment in language_binding.text_segments) for message_id in reversed(deduplicate_message_ids(message_ids)): step_index = reporter.add_step(f"Delete message {message_id}") reporter.update(step_index, "in_progress") try: api.delete_message(config.chat_id, message_id) except TelegramApiError as error: if is_message_missing_error(error): detail = f"{error.description}; it may already have been deleted manually" reporter.update(step_index, "skipped", detail) warnings.append(f"Message {message_id}: {error.description}") deleted_channel_ids.add(message_id) continue reporter.update(step_index, "failed", error.description) failures.append(f"Message {message_id}: {error.description}") continue deleted_channel_ids.add(message_id) reporter.update(step_index, "done") return DeleteBindingResult( warnings=warnings, failures=failures, binding=prune_binding_after_deletions( binding, deleted_channel_ids=deleted_channel_ids, deleted_discussion_channel_ids=deleted_discussion_channel_ids, deleted_discussion_message_ids=deleted_discussion_message_ids, chat_id=config.chat_id, channel_username=config.channel_username, ), ) def prune_binding_after_deletions( binding: PublishBinding, deleted_channel_ids: set[int], deleted_discussion_channel_ids: set[int], deleted_discussion_message_ids: set[int], chat_id: str, channel_username: str, ) -> PublishBinding | None: link_target = resolve_link_target(chat_id, normalize_channel_username(channel_username)) global_message_ids = [message_id for message_id in binding.global_message_ids if message_id not in deleted_channel_ids] languages: dict[Language, LanguageBinding] = {} for language, language_binding in binding.languages.items(): image_message_ids = [message_id for message_id in language_binding.image_message_ids if message_id not in deleted_channel_ids] text_segments = [ TextSegmentBinding(message_id=segment.message_id, mode=segment.mode) for segment in language_binding.text_segments if segment.message_id not in deleted_channel_ids ] if not image_message_ids and not text_segments: continue primary_message_id = text_segments[0].message_id if text_segments else image_message_ids[0] languages[language] = LanguageBinding( language=language, image_message_ids=image_message_ids, text_segments=text_segments, primary_message_id=primary_message_id, primary_message_url=build_message_url(link_target, primary_message_id), discussion_message_id=None, ) discussion_message_ids = { channel_message_id: discussion_message_id for channel_message_id, discussion_message_id in binding.discussion_message_ids.items() if channel_message_id not in deleted_channel_ids and channel_message_id not in deleted_discussion_channel_ids } global_primary_message_id = ( binding.global_primary_message_id if binding.global_primary_message_id and binding.global_primary_message_id not in deleted_channel_ids else (global_message_ids[0] if global_message_ids and binding.global_attached_to_language is None else None) ) global_discussion_message_id = binding.global_discussion_message_id if global_primary_message_id and global_primary_message_id in deleted_discussion_channel_ids: global_discussion_message_id = None elif global_primary_message_id and global_primary_message_id in discussion_message_ids: global_discussion_message_id = discussion_message_ids[global_primary_message_id] elif not global_primary_message_id: global_discussion_message_id = None elif global_discussion_message_id in deleted_discussion_message_ids: global_discussion_message_id = None languages = { language: LanguageBinding( language=language_binding.language, image_message_ids=list(language_binding.image_message_ids), text_segments=list(language_binding.text_segments), primary_message_id=language_binding.primary_message_id, primary_message_url=language_binding.primary_message_url, discussion_message_id=( None if language_binding.discussion_message_id in deleted_discussion_message_ids else ( discussion_message_ids.get(language_binding.primary_message_id) if language_binding.primary_message_id else language_binding.discussion_message_id ) ), ) for language, language_binding in languages.items() } if not global_message_ids and not languages: return None return PublishBinding( global_message_ids=global_message_ids, global_attached_to_language=binding.global_attached_to_language if global_message_ids else None, languages=languages, discussion_message_ids=discussion_message_ids, global_primary_message_id=global_primary_message_id, global_discussion_message_id=global_discussion_message_id, ) def snapshot_next_update_offset(config: Config) -> int: api = TelegramAPI(config.bot_token) offset: int | None = None latest_offset = 0 while True: updates = api.get_updates(offset=offset, limit=100, timeout=0) if not updates: return latest_offset latest_offset = max(int(update["update_id"]) for update in updates) + 1 if len(updates) < 100: return latest_offset offset = latest_offset def apply_discussion_group_pinning( config: Config, binding: PublishBinding, ordered_languages: list[Language], pinning_mode: str, delete_service_messages: bool = False, update_offset: int = 0, progress: ProgressDisplay | None = None, ) -> DiscussionPinningResult: if not config.discussion_group_id: return DiscussionPinningResult(binding=binding, pinning_failed=False) reporter = progress or ProgressDisplay() binding, update_offset = ensure_discussion_group_message_ids( config=config, binding=binding, ordered_languages=ordered_languages, update_offset=update_offset, progress=progress, required=True, ) primary_channel_message_ids = build_primary_channel_message_ids(binding, ordered_languages) ordered_discussion_ids = [ binding.discussion_message_ids[channel_message_id] for channel_message_id in primary_channel_message_ids if channel_message_id in binding.discussion_message_ids ] if not ordered_discussion_ids: return DiscussionPinningResult(binding=binding, pinning_failed=False) api = TelegramAPI(config.bot_token) keep_ids = select_discussion_keep_ids(ordered_discussion_ids, pinning_mode) pinning_failed = False for discussion_message_id in ordered_discussion_ids: if discussion_message_id in keep_ids: pin_step = reporter.add_step(f"Pin discussion message {discussion_message_id}") reporter.update(pin_step, "in_progress") try: api.pin_chat_message(config.discussion_group_id, discussion_message_id, disable_notification=True) except TelegramApiError as error: pinning_failed = True reporter.update(pin_step, "failed", error.description) continue if delete_service_messages: update_offset = delete_pin_service_message( api=api, discussion_group_id=config.discussion_group_id, pinned_message_id=discussion_message_id, update_offset=update_offset, reporter=reporter, ) reporter.update(pin_step, "done") elif discussion_message_id not in keep_ids: unpin_step = reporter.add_step(f"Unpin discussion message {discussion_message_id}") reporter.update(unpin_step, "in_progress") try: api.unpin_chat_message(config.discussion_group_id, discussion_message_id) except TelegramApiError as error: pinning_failed = True reporter.update(unpin_step, "failed", error.description) continue reporter.update(unpin_step, "done") return DiscussionPinningResult( binding=PublishBinding( global_message_ids=binding.global_message_ids, global_attached_to_language=binding.global_attached_to_language, languages=binding.languages, discussion_message_ids=dict(binding.discussion_message_ids), global_primary_message_id=binding.global_primary_message_id, global_discussion_message_id=binding.global_discussion_message_id, ), pinning_failed=pinning_failed, ) def select_discussion_keep_ids(ordered_discussion_ids: list[int], pinning_mode: str | None) -> set[int]: if not ordered_discussion_ids: return set() if pinning_mode == "all": return set(ordered_discussion_ids) if pinning_mode == "first": return {ordered_discussion_ids[0]} if pinning_mode == "last": return {ordered_discussion_ids[-1]} return set() def order_scan_languages(scan_result: ScanResult, ordered_languages: list[Language] | None) -> list[LanguageAssets]: if ordered_languages is None: return scan_result.languages by_language = {language_assets.language: language_assets for language_assets in scan_result.languages} ordered: list[LanguageAssets] = [] for language in ordered_languages: match = by_language.get(language) if match is None: continue ordered.append(match) return ordered def build_primary_channel_message_ids(binding: PublishBinding, ordered_languages: list[Language]) -> list[int]: message_ids: list[int] = [] if binding.global_primary_message_id and binding.global_attached_to_language is None: message_ids.append(binding.global_primary_message_id) elif binding.global_message_ids and binding.global_attached_to_language is None: message_ids.append(binding.global_message_ids[0]) for language in ordered_languages: language_binding = binding.languages.get(language) if language_binding is None: continue if language_binding.primary_message_id: message_ids.append(language_binding.primary_message_id) elif language_binding.text_segments: message_ids.append(language_binding.text_segments[0].message_id) elif language_binding.image_message_ids: message_ids.append(language_binding.image_message_ids[0]) return message_ids def ensure_discussion_group_message_ids( config: Config, binding: PublishBinding, ordered_languages: list[Language], update_offset: int = 0, progress: ProgressDisplay | None = None, required: bool = True, ) -> tuple[PublishBinding, int]: if not config.discussion_group_id: return binding, update_offset reporter = progress or ProgressDisplay() primary_channel_message_ids = build_primary_channel_message_ids(binding, ordered_languages) discussion_message_ids = dict(binding.discussion_message_ids) missing_message_ids = [ message_id for message_id in primary_channel_message_ids if message_id not in discussion_message_ids ] if not missing_message_ids: return binding, update_offset step_index = reporter.add_step("Resolve discussion-group messages") reporter.update(step_index, "in_progress") matched_ids, next_offset, stats = collect_discussion_message_ids( config=config, channel_message_ids=primary_channel_message_ids, update_offset=update_offset, ) discussion_message_ids.update(matched_ids) ordered_discussion_ids = [ discussion_message_ids[channel_message_id] for channel_message_id in primary_channel_message_ids if channel_message_id in discussion_message_ids ] if not ordered_discussion_ids: detail = build_discussion_resolution_detail( stats=stats, expected_count=len(primary_channel_message_ids), matched_count=0, ) reporter.update(step_index, "failed", detail) if required: raise ValueError( "Could not find forwarded discussion-group messages for this announcement. " f"{detail}" ) return binding, next_offset if len(ordered_discussion_ids) < len(primary_channel_message_ids): reporter.update( step_index, "done", build_discussion_resolution_detail( stats=stats, expected_count=len(primary_channel_message_ids), matched_count=len(ordered_discussion_ids), ), ) else: reporter.update(step_index, "done") updated_languages = { language: LanguageBinding( language=language_binding.language, image_message_ids=list(language_binding.image_message_ids), text_segments=list(language_binding.text_segments), primary_message_id=language_binding.primary_message_id, primary_message_url=language_binding.primary_message_url, discussion_message_id=( discussion_message_ids.get(language_binding.primary_message_id) if language_binding.primary_message_id else language_binding.discussion_message_id ), ) for language, language_binding in binding.languages.items() } return ( PublishBinding( global_message_ids=binding.global_message_ids, global_attached_to_language=binding.global_attached_to_language, languages=updated_languages, discussion_message_ids=discussion_message_ids, global_primary_message_id=binding.global_primary_message_id, global_discussion_message_id=( discussion_message_ids.get(binding.global_primary_message_id) if binding.global_primary_message_id else binding.global_discussion_message_id ), ), next_offset, ) def collect_discussion_message_ids( config: Config, channel_message_ids: list[int], update_offset: int, timeout_seconds: int = 12, ) -> tuple[dict[int, int], int, DiscussionResolutionStats]: api = TelegramAPI(config.bot_token) expected = set(channel_message_ids) matched: dict[int, int] = {} offset = update_offset deadline = time.monotonic() + timeout_seconds updates_seen = 0 discussion_group_updates = 0 automatic_forward_updates = 0 linked_channel_forward_updates = 0 while time.monotonic() < deadline and len(matched) < len(expected): updates = api.get_updates(offset=offset, limit=100, timeout=1) updates_seen += len(updates) if updates: offset = max(int(update["update_id"]) for update in updates) + 1 for update in updates: message = extract_message_from_update(update) if message is None: continue if str((message.get("chat") or {}).get("id", "")).strip() != config.discussion_group_id: continue discussion_group_updates += 1 if not bool(message.get("is_automatic_forward", False)): continue automatic_forward_updates += 1 origin = message.get("forward_origin") if not isinstance(origin, dict): continue origin_chat = origin.get("chat") if not isinstance(origin_chat, dict): continue if str(origin_chat.get("id", "")).strip() != config.chat_id: continue linked_channel_forward_updates += 1 channel_message_id = int(origin.get("message_id", 0) or 0) if channel_message_id not in expected or channel_message_id in matched: continue matched[channel_message_id] = int(message["message_id"]) return ( matched, offset, DiscussionResolutionStats( updates_seen=updates_seen, discussion_group_updates=discussion_group_updates, automatic_forward_updates=automatic_forward_updates, linked_channel_forward_updates=linked_channel_forward_updates, ), ) def build_discussion_resolution_detail( stats: DiscussionResolutionStats, expected_count: int, matched_count: int, ) -> str: detail = [f"matched {matched_count}/{expected_count} discussion-group messages"] if stats.updates_seen == 0: detail.append("saw no new bot updates") elif stats.discussion_group_updates == 0: detail.append( f"saw {stats.updates_seen} bot updates, but none were from the configured discussion group" ) elif stats.automatic_forward_updates == 0: detail.append( f"saw {stats.discussion_group_updates} discussion-group updates, but none were automatic forwards" ) elif stats.linked_channel_forward_updates == 0: detail.append( f"saw {stats.automatic_forward_updates} automatic forwards in the discussion group, but none came from the linked channel" ) else: detail.append( f"saw {stats.linked_channel_forward_updates} automatic forwards from the linked channel" ) detail.append( "check bot message access in the discussion group and make sure no other process is consuming getUpdates" ) detail.append( "if discussion-group links were missed, they must be linked manually" ) return "; ".join(detail) def delete_pin_service_message( api: TelegramAPI, discussion_group_id: str, pinned_message_id: int, update_offset: int, reporter: ProgressDisplay, timeout_seconds: float = 2.0, ) -> int: deadline = time.monotonic() + timeout_seconds offset = update_offset while time.monotonic() < deadline: updates = api.get_updates(offset=offset, limit=100, timeout=1) if updates: offset = max(int(update["update_id"]) for update in updates) + 1 for update in updates: message = extract_message_from_update(update) if message is None: continue if str((message.get("chat") or {}).get("id", "")).strip() != discussion_group_id: continue pinned_payload = message.get("pinned_message") if not isinstance(pinned_payload, dict): continue if int(pinned_payload.get("message_id", 0) or 0) != pinned_message_id: continue service_message_id = int(message.get("message_id", 0) or 0) if service_message_id <= 0: return offset cleanup_step = reporter.add_step(f"Delete pin service message {service_message_id}") reporter.update(cleanup_step, "in_progress") try: api.delete_message(discussion_group_id, service_message_id) except TelegramApiError as error: if is_message_missing_error(error): reporter.update(cleanup_step, "skipped", "already gone") else: reporter.update(cleanup_step, "failed", error.description) else: reporter.update(cleanup_step, "done") return offset return offset def extract_message_from_update(update: dict) -> dict | None: for key in ("message", "edited_message", "channel_post", "edited_channel_post"): payload = update.get(key) if isinstance(payload, dict): return payload return None def publish_language( api: TelegramAPI, chat_id: str, language_assets: LanguageAssets, language_links: dict[Language, str], disable_link_preview: bool, link_target: str, use_captions: bool, reporter: ProgressDisplay, posted_message_ids: list[int], ) -> _PublishedLanguage: full_template = parse_markdown_template(language_assets.text_raw) full_rendered_text = render_template(full_template, language_links) chunk_texts = split_markdown_text_for_messages(language_assets.text_raw, link_target) image_message_ids: list[int] = [] published_segments: list[_PublishedTextSegment] = [] if language_assets.images: can_use_caption = use_captions and len(chunk_texts) == 1 and len(full_rendered_text) <= CAPTION_LIMIT image_label = build_image_label(LANGUAGE_DISPLAY_NAMES[language_assets.language], len(language_assets.images)) if can_use_caption: try: image_message_ids = post_images( api=api, chat_id=chat_id, image_paths=language_assets.images, label=image_label, reporter=reporter, posted_message_ids=posted_message_ids, caption=full_rendered_text, ) text_label = build_text_label(language_assets.language, 1, 1) reporter.add_step(text_label, status="done", detail="included as caption") published_segments.append( _PublishedTextSegment( label=text_label, binding=TextSegmentBinding(message_id=image_message_ids[0], mode="caption"), parsed_template=full_template, last_rendered=full_rendered_text, ) ) except TelegramApiError as error: if not is_caption_too_long_error(error): raise reporter.add_step(image_label, status="failed", detail="caption too long, retrying without caption") image_message_ids = post_images( api=api, chat_id=chat_id, image_paths=language_assets.images, label=image_label, reporter=reporter, posted_message_ids=posted_message_ids, ) published_segments = post_text_segments( api=api, chat_id=chat_id, language=language_assets.language, chunk_texts=chunk_texts, language_links=language_links, disable_link_preview=disable_link_preview, reporter=reporter, posted_message_ids=posted_message_ids, ) else: image_message_ids = post_images( api=api, chat_id=chat_id, image_paths=language_assets.images, label=image_label, reporter=reporter, posted_message_ids=posted_message_ids, ) published_segments = post_text_segments( api=api, chat_id=chat_id, language=language_assets.language, chunk_texts=chunk_texts, language_links=language_links, disable_link_preview=disable_link_preview, reporter=reporter, posted_message_ids=posted_message_ids, ) else: published_segments = post_text_segments( api=api, chat_id=chat_id, language=language_assets.language, chunk_texts=chunk_texts, language_links=language_links, disable_link_preview=disable_link_preview, reporter=reporter, posted_message_ids=posted_message_ids, ) if not published_segments: raise ValueError(f"Language '{language_assets.language.value}' produced no text segments.") binding = LanguageBinding( language=language_assets.language, image_message_ids=image_message_ids, text_segments=[segment.binding for segment in published_segments], primary_message_id=published_segments[0].binding.message_id, primary_message_url=build_message_url(link_target, published_segments[0].binding.message_id), ) return _PublishedLanguage(binding=binding, segments=published_segments) def post_images( api: TelegramAPI, chat_id: str, image_paths: list[Path], label: str, reporter: ProgressDisplay, posted_message_ids: list[int], caption: str | None = None, ) -> list[int]: step_index = reporter.add_step(label) reporter.update(step_index, "in_progress") result = api.send_media_group(chat_id, image_paths, caption=caption, parse_mode="HTML" if caption else None) message_ids = [int(item["message_id"]) for item in result] posted_message_ids.extend(message_ids) reporter.update(step_index, "done", "with caption" if caption else "") return message_ids def post_text_segments( api: TelegramAPI, chat_id: str, language: Language, chunk_texts: list[str], language_links: dict[Language, str], disable_link_preview: bool, reporter: ProgressDisplay, posted_message_ids: list[int], ) -> list[_PublishedTextSegment]: segments: list[_PublishedTextSegment] = [] for index, chunk_text in enumerate(chunk_texts, start=1): label = build_text_label(language, index, len(chunk_texts)) step_index = reporter.add_step(label) reporter.update(step_index, "in_progress") parsed_template = parse_markdown_template(chunk_text) rendered_text = render_template(parsed_template, language_links) result = api.send_message(chat_id, rendered_text, parse_mode="HTML", disable_link_preview=disable_link_preview) message_id = int(result["message_id"]) posted_message_ids.append(message_id) detail = f"split part {index}/{len(chunk_texts)}" if len(chunk_texts) > 1 else "" reporter.update(step_index, "done", detail) segments.append( _PublishedTextSegment( label=label, binding=TextSegmentBinding(message_id=message_id, mode="message"), parsed_template=parsed_template, last_rendered=rendered_text, ) ) return segments def backfill_links( api: TelegramAPI, chat_id: str, published_segments: dict[Language, list[_PublishedTextSegment]], language_links: dict[Language, str], disable_link_previews: dict[Language, bool], reporter: ProgressDisplay, ) -> None: for language, segments in published_segments.items(): for segment in segments: updated_text = render_template(segment.parsed_template, language_links) if updated_text == segment.last_rendered: continue step_index = reporter.add_step(build_edit_label(segment.label)) reporter.update(step_index, "in_progress") if segment.binding.mode == "message": api.edit_message_text( chat_id, segment.binding.message_id, updated_text, parse_mode="HTML", disable_link_preview=bool(disable_link_previews.get(language, False)), ) else: api.edit_message_caption(chat_id, segment.binding.message_id, updated_text, parse_mode="HTML") reporter.update(step_index, "done") segment.last_rendered = updated_text def rollback_messages(api: TelegramAPI, chat_id: str, posted_message_ids: list[int], reporter: ProgressDisplay) -> None: for message_id in reversed(deduplicate_message_ids(posted_message_ids)): step_index = reporter.add_step(f"Delete message {message_id}") reporter.update(step_index, "in_progress") try: api.delete_message(chat_id, message_id) except TelegramApiError as error: reporter.update(step_index, "failed", error.description) continue reporter.update(step_index, "done") def deduplicate_message_ids(message_ids: list[int]) -> list[int]: seen: set[int] = set() ordered: list[int] = [] for message_id in message_ids: if message_id in seen: continue seen.add(message_id) ordered.append(message_id) return ordered