"""Caption discovery, UTF-8/whitespace cleanup and Telegram-safe splitting.""" from __future__ import annotations import html as html_module import json import logging import re from collections.abc import Iterator from dataclasses import dataclass from typing import Any, Protocol from bs4 import BeautifulSoup, Tag logger = logging.getLogger(__name__) _MAX_RAW_TEXT = 50_000 _BLOCK_TAGS = { "p", "div", "section", "article", "li", "blockquote", "br", "h1", "h2", "h3", "h4", "tr", } _WHITESPACE_RE = re.compile(r"[ \t\f\v]+") _BLANK_LINES_RE = re.compile(r"\n{3,}") class SelectorAdapter(Protocol): title_selectors: tuple[str, ...] caption_selectors: tuple[str, ...] @dataclass(frozen=True, slots=True) class CaptionResult: title: str caption: str source: str def telegram_text_length(text: str) -> int: """Telegram limits are effectively UTF-16 code units; emoji count as two.""" return len(text.encode("utf-16-le")) // 2 def normalize_text(value: str | None, *, from_html: bool = False) -> str: """Clean text while preserving paragraph boundaries and Persian ZWNJ.""" if not value: return "" text = value[:_MAX_RAW_TEXT] if from_html: soup = BeautifulSoup(text, "html.parser") for tag in soup(["script", "style", "noscript", "template", "svg"]): tag.decompose() for br in soup.find_all("br"): br.replace_with("\n") for block in soup.find_all(_BLOCK_TAGS): if block.name != "br": block.append("\n") text = soup.get_text(" ", strip=False) text = html_module.unescape(text) text = text.replace("\r\n", "\n").replace("\r", "\n") text = text.replace("\u00a0", " ").replace("\u2007", " ").replace("\u202f", " ") text = "".join( char for char in text if char in "\n\t" or (ord(char) >= 0x20 and ord(char) != 0x7F) ) text = _WHITESPACE_RE.sub(" ", text) lines = [line.strip() for line in text.split("\n")] text = "\n".join(lines) text = _BLANK_LINES_RE.sub("\n\n", text) return text.strip() def truncate_utf16(text: str, limit: int, *, ellipsis: str = "…") -> str: if limit <= 0: return "" if telegram_text_length(text) <= limit: return text budget = max(0, limit - telegram_text_length(ellipsis)) used = 0 chars: list[str] = [] for char in text: units = 2 if ord(char) > 0xFFFF else 1 if used + units > budget: break chars.append(char) used += units prefix = "".join(chars).rstrip() if len(prefix) > 80: cut = max(prefix.rfind(" "), prefix.rfind("\n")) if cut > len(prefix) * 0.6: prefix = prefix[:cut].rstrip() return (prefix + ellipsis).strip() def split_caption(text: str, limit: int, extra_limit: int = 3800) -> tuple[str, list[str]]: """Split long captions into a media caption and supplementary messages. No content is discarded merely because Telegram's caption field is short; the remainder is split into bounded extra messages. """ normalized = normalize_text(text) if not normalized: return "", [] if telegram_text_length(normalized) <= limit: return normalized, [] def consume(value: str, budget: int, ellipsis: bool) -> tuple[str, str]: reserved = telegram_text_length("…") if ellipsis else 0 usable = max(1, budget - reserved) used = 0 cut = len(value) for index, char in enumerate(value): units = 2 if ord(char) > 0xFFFF else 1 if used + units > usable: cut = index break used += units prefix = value[:cut] remainder = value[cut:] if remainder and ellipsis: trimmed = prefix.rstrip() if len(trimmed) > 80: boundary = max(trimmed.rfind(" "), trimmed.rfind("\n")) if boundary > len(trimmed) * 0.7: trimmed = trimmed[:boundary] consumed = len(trimmed) prefix = trimmed + "…" remainder = value[consumed:] if not prefix.endswith("…"): prefix = prefix.rstrip() + "…" return prefix, remainder first, remainder = consume(normalized, max(2, limit), ellipsis=True) extras: list[str] = [] remainder = remainder.lstrip() while remainder and len(extras) < 20: chunk, remainder = consume(remainder, max(2, extra_limit), ellipsis=True) extras.append(chunk) return first, extras def _meta_content(soup: BeautifulSoup, *, prop: str | None = None, name: str | None = None) -> str: if prop: tag = soup.find("meta", attrs={"property": prop}) or soup.find( "meta", attrs={"property": prop.lower()} ) if isinstance(tag, Tag) and tag.get("content"): return normalize_text(str(tag.get("content"))) if name: tag = soup.find("meta", attrs={"name": name}) or soup.find( "meta", attrs={"name": name.lower()} ) if isinstance(tag, Tag) and tag.get("content"): return normalize_text(str(tag.get("content"))) return "" def iter_jsonld_objects(soup: BeautifulSoup) -> Iterator[dict[str, Any]]: for script in soup.find_all("script", attrs={"type": re.compile(r"application/ld\+json", re.I)}): raw = script.string or script.get_text() if not raw or not raw.strip(): continue try: data = json.loads(raw.strip()) except (ValueError, TypeError): continue stack: list[Any] = [data] seen = 0 while stack and seen < 500: current = stack.pop() seen += 1 if isinstance(current, list): stack.extend(current) elif isinstance(current, dict): yield current graph = current.get("@graph") if isinstance(graph, (list, dict)): stack.append(graph) for key in ("hasPart", "mainEntity", "itemListElement", "item"): nested = current.get(key) if isinstance(nested, (list, dict)): stack.append(nested) def _types_of(obj: dict[str, Any]) -> set[str]: raw = obj.get("@type") or obj.get("type") or "" values = raw if isinstance(raw, list) else [raw] return {str(value).lower().rsplit("/", 1)[-1] for value in values} def first_video_object(soup: BeautifulSoup) -> dict[str, Any] | None: for obj in iter_jsonld_objects(soup): if _types_of(obj) & {"videoobject", "clip", "movie"}: return obj return None def _selector_text(soup: BeautifulSoup, selectors: tuple[str, ...]) -> str: for selector in selectors: try: tag = soup.select_one(selector) except Exception: # invalid third-party selector should not crash a job logger.warning("adapter selector is invalid: %s", selector) continue if isinstance(tag, Tag): for bad in tag.find_all(["script", "style", "noscript"]): bad.decompose() value = normalize_text(tag.get_text("\n", strip=True)) if value: return value return "" def _article_text(soup: BeautifulSoup) -> str: article = soup.select_one("article") or soup.select_one("[itemprop='articleBody']") if not isinstance(article, Tag): return "" paragraphs = [ normalize_text(p.get_text(" ", strip=True)) for p in article.find_all("p") if isinstance(p, Tag) ] paragraphs = [p for p in paragraphs if p and len(p) >= 30] if not paragraphs: return "" joined = "\n\n".join(paragraphs[:12]) return normalize_text(joined)[:4000] class CaptionParser: """Resolve the most specific available title/caption for a page.""" def parse( self, soup: BeautifulSoup, *, adapter: SelectorAdapter | None = None, fallback_title: str = "", ) -> CaptionResult: video_object = first_video_object(soup) title = "" if adapter is not None: title = _selector_text(soup, adapter.title_selectors) if not title and video_object: title = normalize_text(str(video_object.get("name") or video_object.get("headline") or "")) if not title: heading = soup.find("h1") if isinstance(heading, Tag): title = normalize_text(heading.get_text(" ", strip=True)) if not title: title = _meta_content(soup, prop="og:title") or _meta_content( soup, name="twitter:title" ) if not title and soup.title: title = normalize_text(soup.title.get_text(" ", strip=True)) title = normalize_text(title) or normalize_text(fallback_title) or "ویدئو" title = truncate_utf16(title, 300, ellipsis="…") caption = "" caption_source = "" if adapter is not None: caption = _selector_text(soup, adapter.caption_selectors) if caption: caption_source = "adapter_caption" if not caption: itemprop = soup.select_one("[itemprop='description']") if isinstance(itemprop, Tag): content = itemprop.get("content") if itemprop.name == "meta" else None caption = normalize_text( str(content) if content else itemprop.get_text(" ", strip=True) ) if caption: caption_source = "itemprop" if not caption: caption = _meta_content(soup, prop="og:description") or _meta_content( soup, name="twitter:description" ) if caption: caption_source = "open_graph" if not caption: caption = _meta_content(soup, name="description") if caption: caption_source = "meta_description" if not caption and video_object: caption = normalize_text(str(video_object.get("description") or "")) if caption: caption_source = "jsonld" if not caption: caption = _article_text(soup) if caption: caption_source = "article" if not caption: caption = title caption_source = "title_fallback" if telegram_text_length(caption) < telegram_text_length(title) and caption != title: # A very short site tagline is less useful than the actual title. if len(caption) < 20: caption = f"{title}\n\n{caption}" if caption else title caption_source = "title_plus_short_description" caption = normalize_text(caption) if not caption: caption = title caption_source = "title_fallback" return CaptionResult(title=title, caption=caption, source=caption_source)