import json import re import shutil import unicodedata import uuid from datetime import datetime, timedelta from pathlib import Path from linkki_poster.models import Language, PublishBinding, TextSegmentBinding, LanguageBinding from linkki_poster.scanner import KNOWN_IMAGE_EXTENSIONS POST_METADATA_FILENAME = "post.json" METADATA_SCHEMA_VERSION = 3 LANGUAGE_ORDER_DEFAULT = [Language.FI.value, Language.SV.value, Language.EN.value] IMAGE_SCOPE_PREFIXES = { "global": "Kuva", Language.FI.value: "Suomi", Language.SV.value: "Svenska", Language.EN.value: "English", } TEXT_FILENAMES = { Language.FI.value: "Suomi.md", Language.SV.value: "Svenska.md", Language.EN.value: "English.md", } LANGUAGE_FIELD_KEYS = [Language.FI.value, Language.SV.value, Language.EN.value] TELEGRAM_DELETE_WINDOW = timedelta(hours=48) def list_broadcasts(posts_dir: Path) -> list[dict]: broadcasts: list[dict] = [] for post_dir in posts_dir.iterdir(): if not post_dir.is_dir(): continue metadata = load_metadata(post_dir) broadcasts.append( { "directory": post_dir.name, "title": metadata["title"], "sort_index": metadata["sort_index"], "status": metadata["status"], "scheduled_for": metadata["scheduled_for"], "scheduled_next_attempt_at": metadata.get("scheduled_next_attempt_at"), "scheduled_retry_count": metadata.get("scheduled_retry_count", 0), "scheduled_unpin_for": metadata.get("scheduled_unpin_for"), "last_attempt_at": metadata.get("last_attempt_at"), "last_result": metadata.get("last_result"), "linked": metadata["telegram_binding"] is not None, "telegram_binding": metadata.get("telegram_binding"), "discussion_pinning_mode": metadata.get("discussion_pinning_mode"), "telegram_bot_id": metadata["telegram_bot_id"], "telegram_chat_id": metadata["telegram_chat_id"], "updated_at": metadata["updated_at"], "telegram_published_at": metadata.get("telegram_published_at"), } ) return sorted( broadcasts, key=lambda item: (item["sort_index"], item["directory"].lower()), ) def create_broadcast( posts_dir: Path, title: str, default_bot_id: str = "", default_chat_id: str = "", ) -> str: slug = slugify(title) directory_name = slug post_dir = posts_dir / directory_name if post_dir.exists(): raise ValueError(f"Directory already exists: {directory_name}") post_dir.mkdir(parents=True, exist_ok=False) for text_filename in TEXT_FILENAMES.values(): (post_dir / text_filename).write_text("", encoding="utf-8") metadata = default_metadata( slug, title, sort_index=new_broadcast_sort_index(posts_dir), telegram_bot_id=default_bot_id, telegram_chat_id=default_chat_id, ) save_metadata(post_dir, metadata) return directory_name def load_broadcast(post_dir: Path) -> dict: metadata = load_metadata(post_dir) texts = { language: (post_dir / TEXT_FILENAMES[language]).read_text(encoding="utf-8") if (post_dir / TEXT_FILENAMES[language]).exists() else "" for language in LANGUAGE_FIELD_KEYS } image_lists = metadata["image_lists"] return { "metadata": metadata, "texts": texts, "images": image_lists, "directory": post_dir.name, } def save_broadcast_changes( post_dir: Path, changes: dict, ) -> dict: metadata = load_metadata(post_dir) current_post_dir = post_dir results: dict[str, dict] = {} linked = metadata["telegram_binding"] is not None scheduled = metadata["status"] == "scheduled" requested = {key for key, value in changes.items() if value not in (None, {}, [])} if linked: linked_disallowed = { "telegram_bot_id", "telegram_chat_id", "language_order", "enabled_languages", "images", } forbidden = requested.intersection(linked_disallowed) if forbidden: for field in forbidden: results[field] = {"status": "rejected", "reason": "Field cannot be changed after publish."} if scheduled: scheduled_disallowed = {"telegram_bot_id", "telegram_chat_id"} forbidden = requested.intersection(scheduled_disallowed) if forbidden: for field in forbidden: results[field] = {"status": "rejected", "reason": "Telegram target cannot be changed after scheduling."} text_changes = changes.get("texts", {}) for language, payload in text_changes.items(): field_key = f"{language}.text" result = apply_text_change(post_dir, metadata, language, payload) results[field_key] = result preview_payload = changes.get("disable_link_previews") if preview_payload is not None: results["disable_link_previews"] = apply_disable_link_previews_change(metadata, preview_payload) pinning_payload = changes.get("discussion_pinning_mode") if pinning_payload is not None: results["discussion_pinning_mode"] = apply_discussion_pinning_mode_change(metadata, pinning_payload) delete_service_payload = changes.get("delete_pin_service_messages") if delete_service_payload is not None: results["delete_pin_service_messages"] = apply_delete_pin_service_messages_change(metadata, delete_service_payload) schedule_draft_payload = changes.get("schedule_draft_for") if schedule_draft_payload is not None: results["schedule_draft_for"] = apply_schedule_draft_for_change(metadata, schedule_draft_payload) title_payload = changes.get("title") if title_payload is not None: title_result, current_post_dir = apply_title_change(current_post_dir, metadata, title_payload) results["title"] = title_result if not linked: bot_id_payload = changes.get("telegram_bot_id") if bot_id_payload is not None and "telegram_bot_id" not in results: results["telegram_bot_id"] = apply_simple_field_change( metadata, "telegram_bot_id", { "base_revision": bot_id_payload["base_revision"], "value": str(bot_id_payload["value"]).strip(), }, ) chat_id_payload = changes.get("telegram_chat_id") if chat_id_payload is not None and "telegram_chat_id" not in results: results["telegram_chat_id"] = apply_simple_field_change( metadata, "telegram_chat_id", { "base_revision": chat_id_payload["base_revision"], "value": str(chat_id_payload["value"]).strip(), }, ) order_payload = changes.get("language_order") if order_payload is not None: results["language_order"] = apply_language_order_change(metadata, order_payload) enabled_payload = changes.get("enabled_languages") if enabled_payload is not None: results["enabled_languages"] = apply_enabled_languages_change(metadata, enabled_payload) image_changes = changes.get("images", {}) for scope, payload in image_changes.items(): field_key = f"{scope}.images" results[field_key] = apply_image_change(current_post_dir, metadata, scope, payload) if any(item.get("status") == "saved" for item in results.values()): metadata["updated_at"] = now_iso() transition_status_after_save(metadata) save_metadata(current_post_dir, metadata) return {"results": results, "metadata": metadata, "post_dir": current_post_dir} def stage_image_upload(post_dir: Path, scope: str, filename: str, content: bytes) -> dict: extension = Path(filename).suffix.lower().lstrip(".") if extension not in KNOWN_IMAGE_EXTENSIONS: raise ValueError(f"Unsupported image extension: {extension}") temp_name = f"_tmp_{scope}_{uuid.uuid4().hex}.{extension}" file_path = post_dir / temp_name file_path.write_bytes(content) return { "name": temp_name, "url": f"/announcements/{post_dir.name}/images/{temp_name}", } def mark_publish_success(post_dir: Path, binding: PublishBinding, output: str) -> None: mark_publish_success_with_result(post_dir, binding, output, "published") def mark_publish_success_with_result( post_dir: Path, binding: PublishBinding, output: str, result: str, effective_discussion_pinning_mode: str | None = None, ) -> None: metadata = load_metadata(post_dir) linked_at = now_iso() metadata["status"] = "published" metadata["scheduled_for"] = None metadata["scheduled_next_attempt_at"] = None metadata["scheduled_retry_count"] = 0 metadata["last_attempt_at"] = linked_at metadata["last_output"] = append_status_output(metadata.get("last_output", ""), output, result, linked_at) metadata["last_result"] = result metadata["telegram_binding"] = serialize_binding(binding) if result == "published" or not metadata.get("telegram_published_at"): metadata["telegram_published_at"] = linked_at if effective_discussion_pinning_mode is not None: normalized_mode = normalize_discussion_pinning_mode(effective_discussion_pinning_mode) if metadata["discussion_pinning_mode"] != normalized_mode: metadata["discussion_pinning_mode"] = normalized_mode metadata["field_revisions"]["discussion_pinning_mode"] = int(metadata["field_revisions"]["discussion_pinning_mode"]) + 1 metadata["dirty_text_languages"] = [] metadata["dirty_discussion_pinning"] = False metadata["updated_at"] = linked_at save_metadata(post_dir, metadata) def mark_publish_failure(post_dir: Path, output: str, during_update: bool) -> None: metadata = load_metadata(post_dir) attempted_at = now_iso() metadata["last_attempt_at"] = attempted_at result = "update_failed" if during_update and metadata["telegram_binding"] is not None else "publish_failed" metadata["last_output"] = append_status_output(metadata.get("last_output", ""), output, result, attempted_at) if during_update and metadata["telegram_binding"] is not None: metadata["status"] = "modified" metadata["last_result"] = "update_failed" else: metadata["status"] = "failed" metadata["scheduled_next_attempt_at"] = None metadata["scheduled_retry_count"] = 0 metadata["telegram_binding"] = None metadata["dirty_text_languages"] = [] metadata["dirty_discussion_pinning"] = False metadata["last_result"] = "publish_failed" metadata["updated_at"] = now_iso() save_metadata(post_dir, metadata) def mark_scheduled_publish_failure( post_dir: Path, output: str, partial_binding: PublishBinding | None = None, ) -> None: """Keep a scheduled post eligible for exponential-backoff retry.""" metadata = load_metadata(post_dir) attempted_at = datetime.now().astimezone() retry_count = int(metadata.get("scheduled_retry_count", 0) or 0) + 1 retry_delay = scheduled_retry_delay(retry_count) metadata["status"] = "scheduled" metadata["scheduled_retry_count"] = retry_count metadata["scheduled_next_attempt_at"] = (attempted_at + retry_delay).isoformat() metadata["last_attempt_at"] = attempted_at.isoformat() metadata["last_output"] = append_status_output( metadata.get("last_output", ""), output, "scheduled_publish_retry", metadata["last_attempt_at"], ) metadata["last_result"] = "scheduled_publish_retry" if partial_binding is not None: metadata["telegram_binding"] = serialize_binding(partial_binding) metadata["telegram_published_at"] = metadata.get("telegram_published_at") or metadata["last_attempt_at"] metadata["dirty_discussion_pinning"] = True metadata["updated_at"] = metadata["last_attempt_at"] save_metadata(post_dir, metadata) def scheduled_retry_delay(retry_count: int) -> timedelta: minutes = (1, 5, 15, 30)[min(max(retry_count, 1) - 1, 3)] return timedelta(minutes=minutes) def mark_scheduled(post_dir: Path, scheduled_for: str, schedule_draft_for: str | None = None) -> None: metadata = load_metadata(post_dir) metadata["status"] = "scheduled" metadata["scheduled_for"] = scheduled_for metadata["scheduled_next_attempt_at"] = None metadata["scheduled_retry_count"] = 0 set_schedule_draft_for(metadata, schedule_draft_for) metadata["updated_at"] = now_iso() save_metadata(post_dir, metadata) def unschedule(post_dir: Path) -> None: metadata = load_metadata(post_dir) metadata["scheduled_for"] = None metadata["scheduled_next_attempt_at"] = None metadata["scheduled_retry_count"] = 0 if metadata["telegram_binding"] is not None and ( metadata["dirty_text_languages"] or metadata.get("dirty_discussion_pinning", False) ): metadata["status"] = "modified" else: metadata["status"] = "draft" metadata["updated_at"] = now_iso() save_metadata(post_dir, metadata) def set_scheduled_unpin_for(post_dir: Path, scheduled_unpin_for: str | None) -> None: metadata = load_metadata(post_dir) metadata["scheduled_unpin_for"] = normalize_optional_string(scheduled_unpin_for) metadata["field_revisions"]["scheduled_unpin_for"] = int(metadata["field_revisions"]["scheduled_unpin_for"]) + 1 metadata["updated_at"] = now_iso() save_metadata(post_dir, metadata) def unlink_telegram(post_dir: Path) -> None: metadata = load_metadata(post_dir) metadata["status"] = "draft" metadata["scheduled_for"] = None metadata["scheduled_next_attempt_at"] = None metadata["scheduled_retry_count"] = 0 metadata["scheduled_unpin_for"] = None metadata["telegram_binding"] = None metadata["dirty_text_languages"] = [] metadata["dirty_discussion_pinning"] = False metadata["last_attempt_at"] = None metadata["last_output"] = "" metadata["last_result"] = None metadata["telegram_published_at"] = None metadata["updated_at"] = now_iso() save_metadata(post_dir, metadata) def clear_status_history(post_dir: Path) -> None: metadata = load_metadata(post_dir) metadata["last_attempt_at"] = None metadata["last_output"] = "" metadata["last_result"] = None metadata["updated_at"] = now_iso() save_metadata(post_dir, metadata) def delete_from_storage(post_dir: Path) -> None: shutil.rmtree(post_dir) def get_binding(metadata: dict) -> PublishBinding | None: binding_payload = metadata.get("telegram_binding") if binding_payload is None: return None return deserialize_binding(binding_payload) def load_metadata(post_dir: Path) -> dict: metadata_path = post_dir / POST_METADATA_FILENAME if metadata_path.exists(): try: metadata = json.loads(metadata_path.read_text(encoding="utf-8")) except json.JSONDecodeError: metadata = {} else: metadata = {} return merge_metadata_with_defaults(post_dir, metadata) def save_metadata(post_dir: Path, metadata: dict) -> None: (post_dir / POST_METADATA_FILENAME).write_text(json.dumps(metadata, indent=2, ensure_ascii=False), encoding="utf-8") def save_binding(post_dir: Path, binding: PublishBinding) -> dict: metadata = load_metadata(post_dir) metadata["telegram_binding"] = serialize_binding(binding) metadata["updated_at"] = now_iso() save_metadata(post_dir, metadata) return metadata def default_metadata( slug: str, title: str, sort_index: int, telegram_bot_id: str = "", telegram_chat_id: str = "", ) -> dict: return { "schema_version": METADATA_SCHEMA_VERSION, "slug": slug, "title": title, "sort_index": sort_index, "telegram_bot_id": telegram_bot_id, "telegram_chat_id": telegram_chat_id, "language_order": list(LANGUAGE_ORDER_DEFAULT), "discussion_pinning_mode": "first", "delete_pin_service_messages": True, "enabled_languages": {language: True for language in LANGUAGE_FIELD_KEYS}, "disable_link_previews": {language: True for language in LANGUAGE_FIELD_KEYS}, "status": "draft", "scheduled_for": None, "scheduled_next_attempt_at": None, "scheduled_retry_count": 0, "schedule_draft_for": None, "scheduled_unpin_for": None, "last_attempt_at": None, "last_output": "", "last_result": None, "telegram_published_at": None, "telegram_binding": None, "dirty_text_languages": [], "dirty_discussion_pinning": False, "field_revisions": default_field_revisions(), "image_lists": {"global": [], **{language: [] for language in LANGUAGE_FIELD_KEYS}}, "created_at": now_iso(), "updated_at": now_iso(), } def default_field_revisions() -> dict[str, int]: revisions = { "title": 0, "telegram_bot_id": 0, "telegram_chat_id": 0, "language_order": 0, "discussion_pinning_mode": 0, "delete_pin_service_messages": 0, "schedule_draft_for": 0, "scheduled_unpin_for": 0, "enabled_languages": 0, "disable_link_previews": 0, } revisions["global.images"] = 0 for language in LANGUAGE_FIELD_KEYS: revisions[f"{language}.text"] = 0 revisions[f"{language}.images"] = 0 return revisions def build_image_lists_from_disk(post_dir: Path) -> dict[str, list[str]]: image_lists = {"global": [], **{language: [] for language in LANGUAGE_FIELD_KEYS}} files = [path for path in post_dir.iterdir() if path.is_file()] for scope, prefix in IMAGE_SCOPE_PREFIXES.items(): matches: list[tuple[Path, int, str]] = [] pattern = re.compile(rf"^{re.escape(prefix)}(\d+)\.([^.]+)$", re.IGNORECASE) for file_path in files: match = pattern.match(file_path.name) if match is None: continue extension = match.group(2).lower() if extension not in KNOWN_IMAGE_EXTENSIONS: continue matches.append((file_path, int(match.group(1)), extension)) matches.sort(key=lambda item: (item[1], item[2], item[0].name.lower())) image_lists[scope] = [item[0].name for item in matches] return image_lists def apply_text_change(post_dir: Path, metadata: dict, language: str, payload: dict) -> dict: field_key = f"{language}.text" expected_revision = int(payload["base_revision"]) current_revision = int(metadata["field_revisions"][field_key]) if expected_revision != current_revision: return {"status": "conflict", "reason": "Text was changed by someone else."} new_text = normalize_saved_text(str(payload["value"])) text_path = post_dir / TEXT_FILENAMES[language] current_text = text_path.read_text(encoding="utf-8") if text_path.exists() else "" if new_text == current_text: return {"status": "unchanged"} text_path.write_text(new_text, encoding="utf-8") metadata["field_revisions"][field_key] = current_revision + 1 dirty_languages = set(metadata["dirty_text_languages"]) dirty_languages.add(language) metadata["dirty_text_languages"] = sorted(dirty_languages) return {"status": "saved", "revision": metadata["field_revisions"][field_key]} def apply_simple_field_change(metadata: dict, field_name: str, payload: dict) -> dict: expected_revision = int(payload["base_revision"]) current_revision = int(metadata["field_revisions"][field_name]) if expected_revision != current_revision: return {"status": "conflict", "reason": f"{field_name} was changed by someone else."} new_value = payload["value"] if new_value == metadata[field_name]: return {"status": "unchanged"} metadata[field_name] = new_value metadata["field_revisions"][field_name] = current_revision + 1 return {"status": "saved", "revision": metadata["field_revisions"][field_name]} def apply_title_change(post_dir: Path, metadata: dict, payload: dict) -> tuple[dict, Path]: expected_revision = int(payload["base_revision"]) current_revision = int(metadata["field_revisions"]["title"]) if expected_revision != current_revision: return {"status": "conflict", "reason": "title was changed by someone else."}, post_dir new_title = str(payload["value"]).strip() if new_title == "": return {"status": "rejected", "reason": "Title cannot be empty."}, post_dir if new_title == metadata["title"]: return {"status": "unchanged"}, post_dir metadata["title"] = new_title metadata["slug"] = slugify(new_title) metadata["field_revisions"]["title"] = current_revision + 1 target_directory_name = metadata["slug"] target_post_dir = post_dir.parent / target_directory_name if target_post_dir != post_dir: if target_post_dir.exists(): return {"status": "rejected", "reason": f"Directory already exists: {target_directory_name}"}, post_dir post_dir.rename(target_post_dir) return {"status": "saved", "revision": metadata["field_revisions"]["title"]}, target_post_dir return {"status": "saved", "revision": metadata["field_revisions"]["title"]}, post_dir def apply_language_order_change(metadata: dict, payload: dict) -> dict: values = list(payload["value"]) if sorted(values) != sorted(LANGUAGE_ORDER_DEFAULT): return {"status": "rejected", "reason": "Language order must contain fi, sv and en exactly once."} return apply_simple_field_change(metadata, "language_order", {"base_revision": payload["base_revision"], "value": values}) def apply_discussion_pinning_mode_change(metadata: dict, payload: dict) -> dict: value = str(payload["value"]).strip().lower() if value not in {"none", "first", "last", "all"}: return {"status": "rejected", "reason": "Pinning mode must be none, first, last, or all."} result = apply_simple_field_change( metadata, "discussion_pinning_mode", {"base_revision": payload["base_revision"], "value": value}, ) if result.get("status") == "saved" and metadata["telegram_binding"] is not None: metadata["dirty_discussion_pinning"] = True return result def apply_delete_pin_service_messages_change(metadata: dict, payload: dict) -> dict: result = apply_simple_field_change( metadata, "delete_pin_service_messages", { "base_revision": payload["base_revision"], "value": bool(payload["value"]), }, ) if result.get("status") == "saved" and metadata["telegram_binding"] is not None: metadata["dirty_discussion_pinning"] = True return result def set_schedule_draft_for(metadata: dict, value: str | None) -> bool: normalized = normalize_optional_string(value) if normalized == metadata["schedule_draft_for"]: return False metadata["schedule_draft_for"] = normalized metadata["field_revisions"]["schedule_draft_for"] = int(metadata["field_revisions"]["schedule_draft_for"]) + 1 return True def apply_schedule_draft_for_change(metadata: dict, payload: dict) -> dict: expected_revision = int(payload["base_revision"]) current_revision = int(metadata["field_revisions"]["schedule_draft_for"]) if expected_revision != current_revision: return {"status": "conflict", "reason": "Schedule draft time was changed by someone else."} new_value = normalize_optional_string(payload["value"]) if new_value == metadata["schedule_draft_for"]: return {"status": "unchanged"} set_schedule_draft_for(metadata, new_value) return {"status": "saved", "revision": metadata["field_revisions"]["schedule_draft_for"]} def apply_enabled_languages_change(metadata: dict, payload: dict) -> dict: values = {language: bool(enabled) for language, enabled in payload["value"].items()} merged = {**metadata["enabled_languages"], **values} return apply_simple_field_change( metadata, "enabled_languages", {"base_revision": payload["base_revision"], "value": merged}, ) def apply_disable_link_previews_change(metadata: dict, payload: dict) -> dict: expected_revision = int(payload["base_revision"]) current_revision = int(metadata["field_revisions"]["disable_link_previews"]) if expected_revision != current_revision: return {"status": "conflict", "reason": "Link preview settings were changed by someone else."} current_value = metadata["disable_link_previews"] values = normalize_disable_link_previews(payload["value"]) if values == current_value: return {"status": "unchanged"} metadata["disable_link_previews"] = values metadata["field_revisions"]["disable_link_previews"] = current_revision + 1 if metadata["telegram_binding"] is not None: dirty_languages = set(metadata["dirty_text_languages"]) for language in LANGUAGE_FIELD_KEYS: if current_value.get(language) != values.get(language): dirty_languages.add(language) metadata["dirty_text_languages"] = sorted(dirty_languages) return {"status": "saved", "revision": metadata["field_revisions"]["disable_link_previews"]} def apply_image_change(post_dir: Path, metadata: dict, scope: str, payload: dict) -> dict: field_key = f"{scope}.images" expected_revision = int(payload["base_revision"]) current_revision = int(metadata["field_revisions"][field_key]) if expected_revision != current_revision: return {"status": "conflict", "reason": "Images were changed by someone else."} requested_names = [str(name) for name in payload["value"]] current_names = list(metadata["image_lists"].get(scope, [])) if requested_names == current_names: return {"status": "unchanged"} canonical_names = canonicalize_image_scope(post_dir, scope, requested_names) metadata["image_lists"][scope] = canonical_names metadata["field_revisions"][field_key] = current_revision + 1 return {"status": "saved", "revision": metadata["field_revisions"][field_key], "value": canonical_names} def canonicalize_image_scope(post_dir: Path, scope: str, requested_names: list[str]) -> list[str]: prefix = IMAGE_SCOPE_PREFIXES[scope] reserved_names: list[tuple[Path, Path]] = [] final_names: list[str] = [] for index, name in enumerate(requested_names, start=1): source_path = post_dir / name if not source_path.exists(): raise FileNotFoundError(f"Image not found: {name}") extension = source_path.suffix.lower() target_name = f"{prefix}{index}{extension}" temp_target = post_dir / f"_rename_{uuid.uuid4().hex}{extension}" source_path.rename(temp_target) reserved_names.append((temp_target, post_dir / target_name)) final_names.append(target_name) for temp_path, target_path in reserved_names: if target_path.exists(): target_path.unlink() temp_path.rename(target_path) existing_scope_files = [ path for path in post_dir.iterdir() if path.is_file() and path.name.startswith(prefix) and path.suffix.lower().lstrip(".") in KNOWN_IMAGE_EXTENSIONS ] keep_names = set(final_names) for path in existing_scope_files: if path.name not in keep_names: path.unlink() return final_names def transition_status_after_save(metadata: dict) -> None: status = metadata["status"] dirty = bool(metadata["dirty_text_languages"]) if metadata.get("dirty_discussion_pinning", False): dirty = True if status == "published" and dirty: metadata["status"] = "modified" elif status == "modified": metadata["status"] = "modified" elif status == "failed": metadata["status"] = "draft" elif status == "scheduled": metadata["status"] = "scheduled" def serialize_binding(binding: PublishBinding) -> dict: return { "global_message_ids": binding.global_message_ids, "global_attached_to_language": binding.global_attached_to_language.value if binding.global_attached_to_language else None, "global_primary_message_id": binding.global_primary_message_id, "global_discussion_message_id": binding.global_discussion_message_id, "languages": { language.value: { "language": language.value, "image_message_ids": language_binding.image_message_ids, "text_segments": [ {"message_id": segment.message_id, "mode": segment.mode} for segment in language_binding.text_segments ], "primary_message_id": language_binding.primary_message_id, "primary_message_url": language_binding.primary_message_url, "discussion_message_id": language_binding.discussion_message_id, } for language, language_binding in binding.languages.items() }, "discussion_message_ids": {str(key): value for key, value in binding.discussion_message_ids.items()}, } def deserialize_binding(payload: dict) -> PublishBinding: languages: dict[Language, LanguageBinding] = {} for language_key, language_payload in payload.get("languages", {}).items(): language = Language(language_key) languages[language] = LanguageBinding( language=language, image_message_ids=[int(item) for item in language_payload.get("image_message_ids", [])], text_segments=[ TextSegmentBinding(message_id=int(segment["message_id"]), mode=str(segment["mode"])) for segment in language_payload.get("text_segments", []) ], primary_message_id=( int(language_payload["primary_message_id"]) if str(language_payload.get("primary_message_id", "")).lstrip("-").isdigit() else None ), primary_message_url=str(language_payload["primary_message_url"]), discussion_message_id=( int(language_payload["discussion_message_id"]) if str(language_payload.get("discussion_message_id", "")).lstrip("-").isdigit() else None ), ) attached_language = payload.get("global_attached_to_language") return PublishBinding( global_message_ids=[int(item) for item in payload.get("global_message_ids", [])], global_attached_to_language=Language(attached_language) if attached_language else None, languages=languages, discussion_message_ids={ int(key): int(value) for key, value in payload.get("discussion_message_ids", {}).items() if str(key).lstrip("-").isdigit() and str(value).lstrip("-").isdigit() }, global_primary_message_id=( int(payload["global_primary_message_id"]) if str(payload.get("global_primary_message_id", "")).lstrip("-").isdigit() else None ), global_discussion_message_id=( int(payload["global_discussion_message_id"]) if str(payload.get("global_discussion_message_id", "")).lstrip("-").isdigit() else None ), ) def slugify(title: str) -> str: normalized = unicodedata.normalize("NFKD", title.strip()) ascii_only = normalized.encode("ascii", "ignore").decode("ascii") lowered = ascii_only.lower() slug = re.sub(r"[^a-z0-9]+", "-", lowered) slug = slug.strip("-") return slug or "announcement" def deslugify_directory_name(directory_name: str) -> str: directory_name = re.sub(r"^\d+-", "", directory_name) cleaned = re.sub(r"[-_]+", " ", directory_name).strip() cleaned = re.sub(r"\s+", " ", cleaned) if not cleaned: return "Announcement" return cleaned[:1].upper() + cleaned[1:] def now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") def append_status_output(existing_output: str, new_output: str, result: str, attempted_at: str) -> str: clean_output = str(new_output or "").strip() if not clean_output: return str(existing_output or "") header = f"[{attempted_at}] {result}" entry = f"{header}\n{clean_output}" previous = str(existing_output or "").strip() if not previous: return entry return f"{previous}\n\n{entry}" def normalize_saved_text(value: str) -> str: lines = value.splitlines(keepends=True) if not lines: return value.rstrip() return "".join(_rstrip_line(line) for line in lines).rstrip() def _rstrip_line(line: str) -> str: if line.endswith("\r\n"): return line[:-2].rstrip() + "\r\n" if line.endswith("\n"): return line[:-1].rstrip() + "\n" if line.endswith("\r"): return line[:-1].rstrip() + "\r" return line.rstrip() def merge_metadata_with_defaults(post_dir: Path, raw_metadata: object) -> dict: title = derive_title_from_metadata(post_dir, raw_metadata) slug = derive_slug_from_metadata(title, raw_metadata) metadata = default_metadata(slug, title, sort_index=0) source = raw_metadata if isinstance(raw_metadata, dict) else {} metadata["schema_version"] = METADATA_SCHEMA_VERSION metadata["sort_index"] = normalize_sort_index(source.get("sort_index")) metadata["language_order"] = normalize_language_order(source.get("language_order")) metadata["discussion_pinning_mode"] = normalize_discussion_pinning_mode(source.get("discussion_pinning_mode")) metadata["delete_pin_service_messages"] = bool(source.get("delete_pin_service_messages", True)) metadata["telegram_bot_id"] = normalize_string(source.get("telegram_bot_id")).strip() metadata["telegram_chat_id"] = normalize_string(source.get("telegram_chat_id")).strip() metadata["enabled_languages"] = normalize_enabled_languages(source.get("enabled_languages")) metadata["disable_link_previews"] = normalize_disable_link_previews(source.get("disable_link_previews")) metadata["status"] = normalize_status(source.get("status")) metadata["scheduled_for"] = normalize_optional_string(source.get("scheduled_for")) metadata["scheduled_next_attempt_at"] = normalize_optional_string(source.get("scheduled_next_attempt_at")) metadata["scheduled_retry_count"] = normalize_nonnegative_int(source.get("scheduled_retry_count")) metadata["schedule_draft_for"] = normalize_optional_string(source.get("schedule_draft_for")) metadata["scheduled_unpin_for"] = normalize_optional_string(source.get("scheduled_unpin_for")) metadata["last_attempt_at"] = normalize_optional_string(source.get("last_attempt_at")) metadata["last_output"] = normalize_string(source.get("last_output")) metadata["last_result"] = normalize_optional_string(source.get("last_result")) metadata["telegram_published_at"] = normalize_optional_string(source.get("telegram_published_at")) metadata["telegram_binding"] = source.get("telegram_binding") if isinstance(source.get("telegram_binding"), dict) else None metadata["dirty_text_languages"] = normalize_dirty_languages(source.get("dirty_text_languages")) metadata["dirty_discussion_pinning"] = bool(source.get("dirty_discussion_pinning", False)) metadata["field_revisions"] = normalize_field_revisions(source.get("field_revisions")) metadata["image_lists"] = normalize_image_lists(source.get("image_lists"), post_dir) metadata["created_at"] = normalize_timestamp(source.get("created_at"), metadata["created_at"]) metadata["updated_at"] = normalize_timestamp(source.get("updated_at"), metadata["updated_at"]) return metadata def move_broadcast(posts_dir: Path, directory_name: str, direction: str) -> None: ordered_items = list_broadcasts(posts_dir) index_by_directory = {item["directory"]: index for index, item in enumerate(ordered_items)} current_index = index_by_directory.get(directory_name) if current_index is None: raise FileNotFoundError(directory_name) if direction == "up": target_index = current_index - 1 elif direction == "down": target_index = current_index + 1 else: raise ValueError(f"Unsupported direction: {direction}") if target_index < 0 or target_index >= len(ordered_items): return ordered_items[current_index], ordered_items[target_index] = ordered_items[target_index], ordered_items[current_index] for sort_index, item in enumerate(ordered_items): post_dir = posts_dir / item["directory"] metadata = load_metadata(post_dir) if metadata["sort_index"] == sort_index: continue metadata["sort_index"] = sort_index save_metadata(post_dir, metadata) def derive_title_from_metadata(post_dir: Path, raw_metadata: object) -> str: if isinstance(raw_metadata, dict): title = str(raw_metadata.get("title", "")).strip() if title: return title return deslugify_directory_name(post_dir.name) def derive_slug_from_metadata(title: str, raw_metadata: object) -> str: if isinstance(raw_metadata, dict): slug = str(raw_metadata.get("slug", "")).strip() if slug: return slugify(slug) return slugify(title) def normalize_language_order(value: object) -> list[str]: if isinstance(value, list): items = [str(item) for item in value] if sorted(items) == sorted(LANGUAGE_ORDER_DEFAULT): return items return list(LANGUAGE_ORDER_DEFAULT) def normalize_discussion_pinning_mode(value: object) -> str: text = str(value).strip().lower() return text if text in {"none", "first", "last", "all"} else "first" def normalize_enabled_languages(value: object) -> dict[str, bool]: source = value if isinstance(value, dict) else {} return {language: bool(source.get(language, True)) for language in LANGUAGE_FIELD_KEYS} def normalize_disable_link_previews(value: object) -> dict[str, bool]: source = value if isinstance(value, dict) else {} return {language: bool(source.get(language, True)) for language in LANGUAGE_FIELD_KEYS} def normalize_status(value: object) -> str: allowed = {"draft", "scheduled", "published", "modified", "failed"} text = str(value).strip() return text if text in allowed else "draft" def normalize_optional_string(value: object) -> str | None: if value is None: return None text = str(value).strip() return text or None def normalize_nonnegative_int(value: object) -> int: if isinstance(value, bool): return 0 try: number = int(value) except (TypeError, ValueError): return 0 return max(number, 0) def normalize_string(value: object) -> str: if value is None: return "" return str(value) def normalize_dirty_languages(value: object) -> list[str]: if not isinstance(value, list): return [] return sorted({str(item) for item in value if str(item) in LANGUAGE_FIELD_KEYS}) def normalize_sort_index(value: object) -> int: if isinstance(value, int): return value if isinstance(value, str) and value.lstrip("-").isdigit(): return int(value) return 0 def new_broadcast_sort_index(posts_dir: Path) -> int: lowest = 0 for post_dir in posts_dir.iterdir(): if not post_dir.is_dir(): continue metadata = load_metadata(post_dir) lowest = min(lowest, metadata["sort_index"]) return lowest - 1 def telegram_delete_deadline(metadata: dict) -> datetime | None: raw_value = metadata.get("telegram_published_at") if not raw_value: return None return datetime.fromisoformat(raw_value) + TELEGRAM_DELETE_WINDOW def is_telegram_delete_expired(metadata: dict, now: datetime | None = None) -> bool: deadline = telegram_delete_deadline(metadata) if deadline is None: return False current_time = now or datetime.now(deadline.tzinfo) return current_time >= deadline def normalize_field_revisions(value: object) -> dict[str, int]: source = value if isinstance(value, dict) else {} revisions = default_field_revisions() for key in revisions: try: revisions[key] = int(source.get(key, revisions[key])) except (TypeError, ValueError): revisions[key] = revisions[key] return revisions def normalize_image_lists(value: object, post_dir: Path) -> dict[str, list[str]]: disk_lists = build_image_lists_from_disk(post_dir) if not isinstance(value, dict): return disk_lists result: dict[str, list[str]] = {"global": [], **{language: [] for language in LANGUAGE_FIELD_KEYS}} available = {scope: set(names) for scope, names in disk_lists.items()} for scope in result: raw_items = value.get(scope) if isinstance(raw_items, list): filtered = [str(name) for name in raw_items if str(name) in available[scope]] missing = [name for name in disk_lists[scope] if name not in filtered] result[scope] = filtered + missing else: result[scope] = disk_lists[scope] return result def normalize_timestamp(value: object, fallback: str) -> str: text = normalize_optional_string(value) return text or fallback