| 1 | """Telegram public channel posts via ScrapeCreators API for /last30days. |
| 2 | |
| 3 | Uses ScrapeCreators REST API to fetch recent posts from named public Telegram |
| 4 | channels. No keyword search - channel handles only. |
| 5 | |
| 6 | Requires SCRAPECREATORS_API_KEY in config plus a channel list (TELEGRAM_SOURCES |
| 7 | env var or --telegram-sources CLI flag). |
| 8 | |
| 9 | API docs: https://docs.scrapecreators.com/v1/telegram/channel/posts |
| 10 | """ |
| 11 | |
| 12 | import math |
| 13 | import os |
| 14 | import re |
| 15 | from typing import Any |
| 16 | |
| 17 | from . import dates, http, log |
| 18 | from .relevance import token_overlap_relevance as _compute_relevance |
| 19 | |
| 20 | SCRAPECREATORS_BASE = "https://api.scrapecreators.com/v1/telegram" |
| 21 | |
| 22 | DEPTH_PAGE_CAPS = { |
| 23 | "quick": 1, |
| 24 | "default": 3, |
| 25 | "deep": 6, |
| 26 | } |
| 27 | |
| 28 | |
| 29 | def _log(msg: str): |
| 30 | log.source_log("Telegram", msg, tty_only=False) |
| 31 | |
| 32 | |
| 33 | class InvalidChannelHandle(ValueError): |
| 34 | """Raised when a channel handle is rejected (joinchat, numeric -100 ID).""" |
| 35 | |
| 36 | |
| 37 | def parse_channel_handle(raw: str) -> str: |
| 38 | """Normalize a Telegram channel identifier to a bare handle. |
| 39 | |
| 40 | Accepts: |
| 41 | - bare username: aipost |
| 42 | - @handle: @aipost |
| 43 | - t.me URL: https://t.me/aipost |
| 44 | - t.me/s preview URL: https://t.me/s/aipost |
| 45 | |
| 46 | Rejects (raises InvalidChannelHandle): |
| 47 | - joinchat links: https://t.me/joinchat/xxxxx |
| 48 | - numeric -100 supergroup IDs: -1001234567890 |
| 49 | |
| 50 | Returns: |
| 51 | Bare handle string (no @ prefix). |
| 52 | """ |
| 53 | handle = raw.strip() |
| 54 | if not handle: |
| 55 | raise InvalidChannelHandle("Empty channel handle") |
| 56 | |
| 57 | if handle.lstrip("-").isdigit() and handle.startswith("-100"): |
| 58 | raise InvalidChannelHandle( |
| 59 | f"Numeric supergroup IDs are not supported: {handle}" |
| 60 | ) |
| 61 | |
| 62 | if handle.startswith("@"): |
| 63 | handle = handle[1:] |
| 64 | if not handle: |
| 65 | raise InvalidChannelHandle("Empty handle after @ prefix") |
| 66 | return handle |
| 67 | |
| 68 | url_match = re.match( |
| 69 | r"(?:https?://)?(?:www\.)?t\.me/(?:s/)?([^/?#]+)", |
| 70 | handle, |
| 71 | re.IGNORECASE, |
| 72 | ) |
| 73 | if url_match: |
| 74 | extracted = url_match.group(1) |
| 75 | if extracted.lower() == "joinchat": |
| 76 | raise InvalidChannelHandle( |
| 77 | f"Private joinchat links are not supported: {raw}" |
| 78 | ) |
| 79 | return extracted |
| 80 | |
| 81 | if "joinchat" in handle.lower(): |
| 82 | raise InvalidChannelHandle( |
| 83 | f"Private joinchat links are not supported: {raw}" |
| 84 | ) |
| 85 | |
| 86 | return handle |
| 87 | |
| 88 | |
| 89 | def parse_channel_sources(raw: str) -> list[str]: |
| 90 | """Parse a comma-separated list of channel handles. |
| 91 | |
| 92 | Filters out invalid handles (logs a warning) and returns valid ones. |
| 93 | """ |
| 94 | handles: list[str] = [] |
| 95 | for part in raw.split(","): |
| 96 | part = part.strip() |
| 97 | if not part: |
| 98 | continue |
| 99 | try: |
| 100 | handle = parse_channel_handle(part) |
| 101 | if handle.lower() not in {h.lower() for h in handles}: |
| 102 | handles.append(handle) |
| 103 | except InvalidChannelHandle as exc: |
| 104 | _log(f"Skipping invalid channel: {exc}") |
| 105 | return handles |
| 106 | |
| 107 | |
| 108 | def _get_channel_sources(config: dict[str, Any]) -> list[str]: |
| 109 | """Get configured Telegram channel sources from config or env.""" |
| 110 | raw = config.get("TELEGRAM_SOURCES") or os.environ.get("TELEGRAM_SOURCES") or "" |
| 111 | return parse_channel_sources(raw) |
| 112 | |
| 113 | |
| 114 | def is_telegram_configured(config: dict[str, Any]) -> bool: |
| 115 | """True when Telegram has both API key and at least one channel.""" |
| 116 | return bool( |
| 117 | config.get("SCRAPECREATORS_API_KEY") |
| 118 | and _get_channel_sources(config) |
| 119 | ) |
| 120 | |
| 121 | |
| 122 | def _parse_date(item: dict[str, Any]) -> str | None: |
| 123 | """Parse date from Telegram post to YYYY-MM-DD.""" |
| 124 | for key in ("published_at", "date", "created_at"): |
| 125 | val = item.get(key) |
| 126 | if val is None: |
| 127 | continue |
| 128 | dt = dates.parse_date(str(val)) |
| 129 | if dt: |
| 130 | return dt.strftime("%Y-%m-%d") |
| 131 | return None |
| 132 | |
| 133 | |
| 134 | def _parse_post( |
| 135 | raw: dict[str, Any], |
| 136 | channel: dict[str, Any], |
| 137 | topic: str, |
| 138 | index: int, |
| 139 | ) -> dict[str, Any]: |
| 140 | """Parse a single Telegram post into normalized dict.""" |
| 141 | post_id = str(raw.get("id") or f"TG{index + 1}") |
| 142 | text = str(raw.get("text") or "").strip() |
| 143 | url = str(raw.get("url") or "") |
| 144 | date_str = _parse_date(raw) |
| 145 | |
| 146 | handle = str(raw.get("channel_handle") or channel.get("handle") or "") |
| 147 | author_name = str(raw.get("author_name") or channel.get("name") or handle) |
| 148 | |
| 149 | view_count = raw.get("view_count") or 0 |
| 150 | reaction_count = raw.get("reaction_count") or 0 |
| 151 | subscriber_count = channel.get("subscriber_count") or 0 |
| 152 | |
| 153 | text_relevance = _compute_relevance(topic, text) |
| 154 | rank_score = max(0.3, 1.0 - (index * 0.02)) |
| 155 | engagement_boost = min(0.2, math.log1p(view_count + reaction_count * 10) / 50) |
| 156 | relevance = min(1.0, text_relevance * 0.5 + rank_score * 0.3 + engagement_boost + 0.1) |
| 157 | |
| 158 | return { |
| 159 | "id": post_id, |
| 160 | "handle": handle, |
| 161 | "display_name": author_name, |
| 162 | "text": text, |
| 163 | "url": url, |
| 164 | "date": date_str, |
| 165 | "engagement": { |
| 166 | "views": view_count, |
| 167 | "reactions": reaction_count, |
| 168 | "subscribers": subscriber_count, |
| 169 | }, |
| 170 | "relevance": round(relevance, 2), |
| 171 | "why_relevant": f"Telegram @{handle}: {text[:60]}" if text else f"Telegram: @{handle}", |
| 172 | } |
| 173 | |
| 174 | |
| 175 | def _fetch_channel_posts( |
| 176 | handle: str, |
| 177 | token: str, |
| 178 | *, |
| 179 | from_date: str, |
| 180 | topic: str, |
| 181 | max_pages: int, |
| 182 | ) -> list[dict[str, Any]]: |
| 183 | """Fetch posts from a single channel, paginating until date cutoff.""" |
| 184 | items: list[dict[str, Any]] = [] |
| 185 | cursor: str | None = None |
| 186 | pages_fetched = 0 |
| 187 | |
| 188 | while pages_fetched < max_pages: |
| 189 | _log(f"Fetching @{handle} (page {pages_fetched + 1}/{max_pages})") |
| 190 | |
| 191 | params: dict[str, Any] = {"handle": handle} |
| 192 | if cursor: |
| 193 | params["cursor"] = cursor |
| 194 | |
| 195 | try: |
| 196 | data = http.get( |
| 197 | f"{SCRAPECREATORS_BASE}/channel/posts", |
| 198 | params=params, |
| 199 | headers=http.scrapecreators_headers(token), |
| 200 | timeout=30, |
| 201 | retries=2, |
| 202 | ) |
| 203 | except http.HTTPError as exc: |
| 204 | _log(f"HTTP error fetching @{handle}: {exc}") |
| 205 | break |
| 206 | |
| 207 | if not data.get("success"): |
| 208 | error_msg = data.get("error") or data.get("message") or "Unknown error" |
| 209 | _log(f"API error for @{handle}: {error_msg}") |
| 210 | break |
| 211 | |
| 212 | channel = data.get("channel") or {} |
| 213 | posts = data.get("posts") or [] |
| 214 | |
| 215 | if not posts: |
| 216 | _log(f"No posts returned for @{handle}") |
| 217 | break |
| 218 | |
| 219 | page_all_old = True |
| 220 | for idx, raw_post in enumerate(posts): |
| 221 | parsed = _parse_post(raw_post, channel, topic, len(items) + idx) |
| 222 | items.append(parsed) |
| 223 | if parsed["date"] and parsed["date"] >= from_date: |
| 224 | page_all_old = False |
| 225 | |
| 226 | pages_fetched += 1 |
| 227 | |
| 228 | if page_all_old: |
| 229 | _log(f"All posts on page older than {from_date}, stopping pagination") |
| 230 | break |
| 231 | |
| 232 | cursor = data.get("cursor") |
| 233 | if not data.get("has_more") or not cursor: |
| 234 | break |
| 235 | |
| 236 | return items |
| 237 | |
| 238 | |
| 239 | def search_telegram( |
| 240 | topic: str, |
| 241 | from_date: str, |
| 242 | to_date: str, |
| 243 | depth: str = "default", |
| 244 | token: str | None = None, |
| 245 | config: dict[str, Any] | None = None, |
| 246 | ) -> dict[str, Any]: |
| 247 | """Fetch recent posts from configured Telegram channels. |
| 248 | |
| 249 | Args: |
| 250 | topic: Search topic (for relevance scoring) |
| 251 | from_date: Start date (YYYY-MM-DD) |
| 252 | to_date: End date (YYYY-MM-DD) |
| 253 | depth: 'quick', 'default', or 'deep' |
| 254 | token: ScrapeCreators API key |
| 255 | config: Config dict (for TELEGRAM_SOURCES) |
| 256 | |
| 257 | Returns: |
| 258 | Dict with 'items' list and optional 'error'. |
| 259 | """ |
| 260 | config = config or {} |
| 261 | |
| 262 | if not token: |
| 263 | return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"} |
| 264 | |
| 265 | channels = _get_channel_sources(config) |
| 266 | if not channels: |
| 267 | return {"items": [], "error": "No TELEGRAM_SOURCES configured (channel list required)"} |
| 268 | |
| 269 | base_cap = DEPTH_PAGE_CAPS.get(depth, DEPTH_PAGE_CAPS["default"]) |
| 270 | override = config.get("TELEGRAM_MAX_PAGES") |
| 271 | if override: |
| 272 | try: |
| 273 | max_pages = max(base_cap, int(override)) |
| 274 | except (ValueError, TypeError): |
| 275 | max_pages = base_cap |
| 276 | else: |
| 277 | max_pages = base_cap |
| 278 | |
| 279 | _log(f"Searching {len(channels)} channel(s) for '{topic}' (depth={depth}, max_pages={max_pages})") |
| 280 | |
| 281 | all_items: list[dict[str, Any]] = [] |
| 282 | for handle in channels: |
| 283 | channel_items = _fetch_channel_posts( |
| 284 | handle, |
| 285 | token, |
| 286 | from_date=from_date, |
| 287 | topic=topic, |
| 288 | max_pages=max_pages, |
| 289 | ) |
| 290 | all_items.extend(channel_items) |
| 291 | |
| 292 | in_range = [ |
| 293 | item for item in all_items |
| 294 | if item["date"] and from_date <= item["date"] <= to_date |
| 295 | ] |
| 296 | out_of_range = len(all_items) - len(in_range) |
| 297 | if in_range: |
| 298 | items = in_range |
| 299 | if out_of_range: |
| 300 | _log(f"Filtered {out_of_range} posts outside date range") |
| 301 | else: |
| 302 | items = all_items |
| 303 | _log(f"No posts within date range, keeping all {len(items)}") |
| 304 | |
| 305 | items.sort(key=lambda x: x.get("relevance", 0), reverse=True) |
| 306 | |
| 307 | _log(f"Found {len(items)} Telegram posts") |
| 308 | return {"items": items} |
| 309 | |
| 310 | |
| 311 | def parse_telegram_response(response: dict[str, Any]) -> list[dict[str, Any]]: |
| 312 | """Parse Telegram search response to normalized format. |
| 313 | |
| 314 | Returns: |
| 315 | List of item dicts ready for normalization. |
| 316 | """ |
| 317 | return response.get("items", []) |
| 318 |