github-featured/mvanhorn--last30days-skill
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat(telegram): add opt-in Telegram public channel source (#1035)

Browse Source 
Co-authored-by: Cursor Agent <cursoragent@cursor.com>
This commit is contained in:
Matt Van Horn
2026-08-21 22:59:01 +01:00
committed by GitHub
parent 86a8825103
commit d05389d39b
10 changed files with 838 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CONFIGURATION.md
+1
View File
@@ -156,6 +156,7 @@ python3 skills/last30days/scripts/last30days.py "MCP servers" \
| Threads | `SCRAPECREATORS_API_KEY` + `INCLUDE_SOURCES` contains `threads` | Threads items | 10K free calls |
| Pinterest | `SCRAPECREATORS_API_KEY` + `INCLUDE_SOURCES` contains `pinterest` | Pinterest items | 10K free calls |
| LinkedIn | `SCRAPECREATORS_API_KEY` + `INCLUDE_SOURCES` contains `linkedin` | LinkedIn posts + articles (articles rank as high signal on person topics) | 10K free calls; power-user opt-in, not offered during first-run onboarding |
| Telegram | `SCRAPECREATORS_API_KEY` + (`--telegram-sources=<handles>` **or** `TELEGRAM_SOURCES=<handles>` + `INCLUDE_SOURCES` contains `telegram`) | **opt-in, off by default**; public channel posts only (no keyword discovery). `--telegram-sources=aipost,durov` (or `TELEGRAM_SOURCES` env) auto-activates for that run like `--trustpilot-domain`. Accepts bare handle, `@handle`, `t.me/URL`, or `t.me/s/URL`; rejects joinchat links and numeric -100 IDs. `INCLUDE_SOURCES=telegram` or `--search telegram` without a channel list does not fetch. `EXCLUDE_SOURCES=telegram` wins. `TELEGRAM_MAX_PAGES` overrides page cap (quick=1, default=3, deep=6). Never on Recommended onboarding tier. | 1 credit per live posts page; 10K free calls |
| Xiaohongshu (RED) | logged-in x-mcp browser plugin or `xiaohongshu-mcp` service; optional `XIAOHONGSHU_API_BASE` for custom URLs | requested-only via `--search xhs` or `--search xiaohongshu`; auto-probes `http://localhost:18060` then `http://host.docker.internal:18060` | no last30days API key; depends on your local browser-session service |
| Bluesky | `BSKY_HANDLE` + `BSKY_APP_PASSWORD` | Bluesky items | yes (app password at bsky.app) |
| TruthSocial | `TRUTHSOCIAL_TOKEN` | TruthSocial items | yes |
changelog.d/990.added.md
+1
View File
@@ -0,0 +1 @@
Telegram public channel source: opt-in via `--telegram-sources=handle1,handle2` or `TELEGRAM_SOURCES` + `INCLUDE_SOURCES=telegram`. Named public channels only (no keyword discovery); fetches recent posts via ScrapeCreators API and scores by views, reactions, and topic relevance.
skills/last30days/SKILL.md
+1
View File
@@ -705,6 +705,7 @@ The magic of /last30days is Reddit comments + X posts together - and both are fr
- `PERPLEXITY_API_KEY=xxx` - preferred Agent/Search API path with citations; set `INCLUDE_SOURCES=perplexity`. Existing `OPENROUTER_API_KEY` installs keep the synchronous Sonar fallback.
- `XIAOHONGSHU_API_BASE=http://localhost:18060` - Xiaohongshu/RED via a logged-in x-mcp browser plugin or `xiaohongshu-mcp` service; optional unless the local service runs on a custom URL. Opt in per run with `--search xhs`, or persistently via `INCLUDE_SOURCES=xiaohongshu`.
- DripStack (premium financial newsletter search) is opt-in only: per run with `--search dripstack`, or persistently via `INCLUDE_SOURCES=dripstack`. Free public search API, no key; never active without the opt-in.
- Telegram (public channels) is opt-in via `--telegram-sources=handle1,handle2` (auto-activates for that run) or persistently via `TELEGRAM_SOURCES=handles` + `INCLUDE_SOURCES=telegram`. Requires `SCRAPECREATORS_API_KEY`. Named public channels only; no keyword discovery.
- `BSKY_HANDLE=you.bsky.social` + `BSKY_APP_PASSWORD=xxx` - Bluesky (free app password).
- `BRAVE_API_KEY=xxx` or `EXA_API_KEY=xxx` - web search backends.
skills/last30days/scripts/last30days.py
+65
View File
@@ -210,6 +210,49 @@ def activate_trustpilot_for_explicit_domain(
return requested_sources
def activate_telegram_for_explicit_sources(
config: dict,
requested_sources: list[str] | None,
*,
channels: str,
) -> list[str] | None:
"""Activate the opt-in Telegram source when the user pinned channel(s).
Passing ``--telegram-sources`` is unambiguous intent — silently ignoring it
when Telegram is not in ``INCLUDE_SOURCES`` / ``--search`` is the same
failure mode as #873 (Trustpilot). Auto-activate the source.
``EXCLUDE_SOURCES=telegram`` still wins. Mutates ``config`` in place and
returns the (possibly extended) ``requested_sources`` list.
"""
excluded = {
token.strip().lower()
for token in str(config.get("EXCLUDE_SOURCES") or "").split(",")
if token.strip()
}
if "telegram" in excluded:
sys.stderr.write(
f"[Telegram] --telegram-sources={channels} ignored: telegram is in EXCLUDE_SOURCES\n"
)
return requested_sources
config["TELEGRAM_SOURCES"] = channels
include = str(config.get("INCLUDE_SOURCES") or "")
tokens = [token.strip() for token in include.split(",") if token.strip()]
if "telegram" not in {token.lower() for token in tokens}:
tokens.append("telegram")
config["INCLUDE_SOURCES"] = ",".join(tokens)
sys.stderr.write(
f"[Telegram] --telegram-sources={channels} activated telegram source "
"(add to INCLUDE_SOURCES permanently to skip this auto-enable)\n"
)
if requested_sources is not None and "telegram" not in requested_sources:
requested_sources = [*requested_sources, "telegram"]
return requested_sources
def slugify(value: str, max_length: int = 180) -> str:
slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
if len(slug) > max_length:
@@ -780,6 +823,16 @@ def build_parser() -> argparse.ArgumentParser:
"Requires the brightdata CLI on PATH and logged in."
),
)
parser.add_argument(
"--telegram-sources",
help=(
"Comma-separated list of public Telegram channel handles or t.me URLs. "
"Auto-activates the opt-in Telegram source for this run. "
"Accepts: bare handle (aipost), @handle (@aipost), "
"t.me URL (https://t.me/aipost), or preview URL (https://t.me/s/aipost). "
"Rejects joinchat links and numeric -100 supergroup IDs."
),
)
parser.add_argument(
"--competitors",
nargs="?",
@@ -3241,6 +3294,18 @@ def _main(
requested_sources,
reason=f"--trustpilot-domain={cli_trustpilot_domain}",
)
# Explicit --telegram-sources is user intent: activate the opt-in source
# before diagnose/run so the flag cannot silently no-op (same pattern as
# Trustpilot #873). Sets TELEGRAM_SOURCES in config for pipeline.
cli_telegram_sources = (
args.telegram_sources.strip() if args.telegram_sources else ""
)
if cli_telegram_sources:
requested_sources = activate_telegram_for_explicit_sources(
config,
requested_sources,
channels=cli_telegram_sources,
)
diag = pipeline.diagnose(config, requested_sources, safe=args.diagnose)
if args.diagnose:
skills/last30days/scripts/lib/doctor.py
+32
View File
@@ -161,6 +161,7 @@ SOURCE_ORDER = (
"tiktok",
"instagram",
"threads",
"telegram",
"bluesky",
"truthsocial",
"perplexity",
@@ -666,6 +667,36 @@ def _threads_record(config):
return _sc_optin_record(config, "threads", "threads")
def _telegram_record(config):
# Telegram needs the key AND an INCLUDE_SOURCES=telegram opt-in AND a
# channel list (TELEGRAM_SOURCES). Without named channels there is no
# discovery endpoint to call.
requires = "SCRAPECREATORS_API_KEY + INCLUDE_SOURCES=telegram + TELEGRAM_SOURCES"
if not config.get("SCRAPECREATORS_API_KEY"):
return _record(status="unconfigured", requires=requires, fix=_sc_fix())
from . import telegram
channels = telegram._get_channel_sources(config)
if "telegram" in env.include_sources(config):
if channels:
return _record(
status=health.OK,
requires=requires,
detail=f"SCRAPECREATORS_API_KEY present, {len(channels)} channel(s) configured",
)
return _record(
status="unconfigured",
requires=requires,
fix="set TELEGRAM_SOURCES to a comma-separated list of public channel handles",
note="key present and opt-in active, but no channels configured",
)
return _record(
status="opt-in",
requires=requires,
fix="add telegram to INCLUDE_SOURCES and set TELEGRAM_SOURCES to channel handles",
note="key present; opt-in only, channels required",
)
def _bluesky_record(config):
if env.is_bluesky_available(config):
return _record(status=health.OK, requires="BSKY_HANDLE + BSKY_APP_PASSWORD")
@@ -845,6 +876,7 @@ _SOURCE_BUILDERS: Dict[str, Callable[[Dict[str, Any]], Dict[str, Any]]] = {
"tiktok": _tiktok_record,
"instagram": _instagram_record,
"threads": _threads_record,
"telegram": _telegram_record,
"bluesky": _bluesky_record,
"truthsocial": _truthsocial_record,
"perplexity": _perplexity_record,
skills/last30days/scripts/lib/normalize.py
+3
View File
@@ -58,6 +58,9 @@ def normalize_source_items(
"threads": lambda s, i, idx, fd, td: _normalize_microblog(
s, i, idx, fd, td, "TH", "Threads post"
),
"telegram": lambda s, i, idx, fd, td: _normalize_microblog(
s, i, idx, fd, td, "TG", "Telegram post"
),
"xquik": _normalize_x,
"pinterest": _normalize_pinterest,
"polymarket": _normalize_polymarket,
skills/last30days/scripts/lib/pipeline.py
+19
View File
@@ -63,6 +63,7 @@ from . import (
snippet,
stocktwits,
techmeme,
telegram,
threads,
tiktok,
topic_shape,
@@ -106,6 +107,7 @@ SEARCH_ALIAS = {
# product search. Extra streams would be pure redundancy at one credit each.
MAX_SOURCE_FETCHES: dict[str, int] = {
"x": 2, "jobs": 1, "linkedin": 1, "stocktwits": 1, "trustpilot": 1, "amazon": 1,
"telegram": 1,
}
_FAILURE_SPECIFICITY = {
@@ -214,6 +216,7 @@ MOCK_AVAILABLE_SOURCES = [
"linkedin",
"corpus",
"dripstack",
"telegram",
]
@@ -362,6 +365,14 @@ def available_sources(
"pinterest" in include_sources or (requested_sources and "pinterest" in requested_sources)
):
available.append("pinterest")
# Telegram: opt-in via INCLUDE_SOURCES AND requires a channel list. The
# channel list (TELEGRAM_SOURCES env or --telegram-sources CLI) is the gate:
# without named channels there is no discovery endpoint to call.
if config.get("SCRAPECREATORS_API_KEY") and (
"telegram" in include_sources or (requested_sources and "telegram" in requested_sources)
):
if telegram.is_telegram_configured(config):
available.append("telegram")
# xquik is a backend of the single "x" source (see env.x_backend_chain),
# not a separate parallel source — registered via the "x" entry above.
exclude = {s.strip().lower() for s in (config.get("EXCLUDE_SOURCES") or "").split(",") if s.strip()}
@@ -4694,6 +4705,14 @@ def _retrieve_stream_impl(
token=config.get("SCRAPECREATORS_API_KEY"),
)
return threads.parse_threads_response(result), _result_outcome_artifact(source, result)
if source == "telegram":
result = telegram.search_telegram(
subquery.search_query, from_date, to_date,
depth=depth,
token=config.get("SCRAPECREATORS_API_KEY"),
config=config,
)
return telegram.parse_telegram_response(result), _result_outcome_artifact(source, result)
if source == "truthsocial":
result = truthsocial.search_truthsocial(subquery.search_query, from_date, to_date, depth=depth, config=config)
return truthsocial.parse_truthsocial_response(result), _result_outcome_artifact(source, result)
skills/last30days/scripts/lib/planner.py
+1
View File
@@ -152,6 +152,7 @@ SOURCE_CAPABILITIES = {
"trustpilot": {"reference", "company_signal", "social"},
"amazon": {"reference", "company_signal", "product_signal"},
"xiaohongshu": {"video", "video_shortform", "social"},
"telegram": {"discussion", "social"},
"github": {"discussion", "link"},
"grounding": {"web", "reference", "link"},
"perplexity": {"web", "reference", "analysis"},
skills/last30days/scripts/lib/telegram.py
+317
View File
@@ -0,0 +1,317 @@
"""Telegram public channel posts via ScrapeCreators API for /last30days.
Uses ScrapeCreators REST API to fetch recent posts from named public Telegram
channels. No keyword search - channel handles only.
Requires SCRAPECREATORS_API_KEY in config plus a channel list (TELEGRAM_SOURCES
env var or --telegram-sources CLI flag).
API docs: https://docs.scrapecreators.com/v1/telegram/channel/posts
"""
import math
import os
import re
from typing import Any
from . import dates, http, log
from .relevance import token_overlap_relevance as _compute_relevance
SCRAPECREATORS_BASE = "https://api.scrapecreators.com/v1/telegram"
DEPTH_PAGE_CAPS = {
"quick": 1,
"default": 3,
"deep": 6,
}
def _log(msg: str):
log.source_log("Telegram", msg, tty_only=False)
class InvalidChannelHandle(ValueError):
"""Raised when a channel handle is rejected (joinchat, numeric -100 ID)."""
def parse_channel_handle(raw: str) -> str:
"""Normalize a Telegram channel identifier to a bare handle.
Accepts:
- bare username: aipost
- @handle: @aipost
- t.me URL: https://t.me/aipost
- t.me/s preview URL: https://t.me/s/aipost
Rejects (raises InvalidChannelHandle):
- joinchat links: https://t.me/joinchat/xxxxx
- numeric -100 supergroup IDs: -1001234567890
Returns:
Bare handle string (no @ prefix).
"""
handle = raw.strip()
if not handle:
raise InvalidChannelHandle("Empty channel handle")
if handle.lstrip("-").isdigit() and handle.startswith("-100"):
raise InvalidChannelHandle(
f"Numeric supergroup IDs are not supported: {handle}"
)
if handle.startswith("@"):
handle = handle[1:]
if not handle:
raise InvalidChannelHandle("Empty handle after @ prefix")
return handle
url_match = re.match(
r"(?:https?://)?(?:www\.)?t\.me/(?:s/)?([^/?#]+)",
handle,
re.IGNORECASE,
)
if url_match:
extracted = url_match.group(1)
if extracted.lower() == "joinchat":
raise InvalidChannelHandle(
f"Private joinchat links are not supported: {raw}"
)
return extracted
if "joinchat" in handle.lower():
raise InvalidChannelHandle(
f"Private joinchat links are not supported: {raw}"
)
return handle
def parse_channel_sources(raw: str) -> list[str]:
"""Parse a comma-separated list of channel handles.
Filters out invalid handles (logs a warning) and returns valid ones.
"""
handles: list[str] = []
for part in raw.split(","):
part = part.strip()
if not part:
continue
try:
handle = parse_channel_handle(part)
if handle.lower() not in {h.lower() for h in handles}:
handles.append(handle)
except InvalidChannelHandle as exc:
_log(f"Skipping invalid channel: {exc}")
return handles
def _get_channel_sources(config: dict[str, Any]) -> list[str]:
"""Get configured Telegram channel sources from config or env."""
raw = config.get("TELEGRAM_SOURCES") or os.environ.get("TELEGRAM_SOURCES") or ""
return parse_channel_sources(raw)
def is_telegram_configured(config: dict[str, Any]) -> bool:
"""True when Telegram has both API key and at least one channel."""
return bool(
config.get("SCRAPECREATORS_API_KEY")
and _get_channel_sources(config)
)
def _parse_date(item: dict[str, Any]) -> str | None:
"""Parse date from Telegram post to YYYY-MM-DD."""
for key in ("published_at", "date", "created_at"):
val = item.get(key)
if val is None:
continue
dt = dates.parse_date(str(val))
if dt:
return dt.strftime("%Y-%m-%d")
return None
def _parse_post(
raw: dict[str, Any],
channel: dict[str, Any],
topic: str,
index: int,
) -> dict[str, Any]:
"""Parse a single Telegram post into normalized dict."""
post_id = str(raw.get("id") or f"TG{index + 1}")
text = str(raw.get("text") or "").strip()
url = str(raw.get("url") or "")
date_str = _parse_date(raw)
handle = str(raw.get("channel_handle") or channel.get("handle") or "")
author_name = str(raw.get("author_name") or channel.get("name") or handle)
view_count = raw.get("view_count") or 0
reaction_count = raw.get("reaction_count") or 0
subscriber_count = channel.get("subscriber_count") or 0
text_relevance = _compute_relevance(topic, text)
rank_score = max(0.3, 1.0 - (index * 0.02))
engagement_boost = min(0.2, math.log1p(view_count + reaction_count * 10) / 50)
relevance = min(1.0, text_relevance * 0.5 + rank_score * 0.3 + engagement_boost + 0.1)
return {
"id": post_id,
"handle": handle,
"display_name": author_name,
"text": text,
"url": url,
"date": date_str,
"engagement": {
"views": view_count,
"reactions": reaction_count,
"subscribers": subscriber_count,
},
"relevance": round(relevance, 2),
"why_relevant": f"Telegram @{handle}: {text[:60]}" if text else f"Telegram: @{handle}",
}
def _fetch_channel_posts(
handle: str,
token: str,
*,
from_date: str,
topic: str,
max_pages: int,
) -> list[dict[str, Any]]:
"""Fetch posts from a single channel, paginating until date cutoff."""
items: list[dict[str, Any]] = []
cursor: str | None = None
pages_fetched = 0
while pages_fetched < max_pages:
_log(f"Fetching @{handle} (page {pages_fetched + 1}/{max_pages})")
params: dict[str, Any] = {"handle": handle}
if cursor:
params["cursor"] = cursor
try:
data = http.get(
f"{SCRAPECREATORS_BASE}/channel/posts",
params=params,
headers=http.scrapecreators_headers(token),
timeout=30,
retries=2,
)
except http.HTTPError as exc:
_log(f"HTTP error fetching @{handle}: {exc}")
break
if not data.get("success"):
error_msg = data.get("error") or data.get("message") or "Unknown error"
_log(f"API error for @{handle}: {error_msg}")
break
channel = data.get("channel") or {}
posts = data.get("posts") or []
if not posts:
_log(f"No posts returned for @{handle}")
break
page_all_old = True
for idx, raw_post in enumerate(posts):
parsed = _parse_post(raw_post, channel, topic, len(items) + idx)
items.append(parsed)
if parsed["date"] and parsed["date"] >= from_date:
page_all_old = False
pages_fetched += 1
if page_all_old:
_log(f"All posts on page older than {from_date}, stopping pagination")
break
cursor = data.get("cursor")
if not data.get("has_more") or not cursor:
break
return items
def search_telegram(
topic: str,
from_date: str,
to_date: str,
depth: str = "default",
token: str | None = None,
config: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Fetch recent posts from configured Telegram channels.
Args:
topic: Search topic (for relevance scoring)
from_date: Start date (YYYY-MM-DD)
to_date: End date (YYYY-MM-DD)
depth: 'quick', 'default', or 'deep'
token: ScrapeCreators API key
config: Config dict (for TELEGRAM_SOURCES)
Returns:
Dict with 'items' list and optional 'error'.
"""
config = config or {}
if not token:
return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"}
channels = _get_channel_sources(config)
if not channels:
return {"items": [], "error": "No TELEGRAM_SOURCES configured (channel list required)"}
base_cap = DEPTH_PAGE_CAPS.get(depth, DEPTH_PAGE_CAPS["default"])
override = config.get("TELEGRAM_MAX_PAGES")
if override:
try:
max_pages = max(base_cap, int(override))
except (ValueError, TypeError):
max_pages = base_cap
else:
max_pages = base_cap
_log(f"Searching {len(channels)} channel(s) for '{topic}' (depth={depth}, max_pages={max_pages})")
all_items: list[dict[str, Any]] = []
for handle in channels:
channel_items = _fetch_channel_posts(
handle,
token,
from_date=from_date,
topic=topic,
max_pages=max_pages,
)
all_items.extend(channel_items)
in_range = [
item for item in all_items
if item["date"] and from_date <= item["date"] <= to_date
]
out_of_range = len(all_items) - len(in_range)
if in_range:
items = in_range
if out_of_range:
_log(f"Filtered {out_of_range} posts outside date range")
else:
items = all_items
_log(f"No posts within date range, keeping all {len(items)}")
items.sort(key=lambda x: x.get("relevance", 0), reverse=True)
_log(f"Found {len(items)} Telegram posts")
return {"items": items}
def parse_telegram_response(response: dict[str, Any]) -> list[dict[str, Any]]:
"""Parse Telegram search response to normalized format.
Returns:
List of item dicts ready for normalization.
"""
return response.get("items", [])
tests/test_telegram_source.py
+398
View File
@@ -0,0 +1,398 @@
"""Tests for the Telegram source adapter (lib/telegram.py).
Covers handle parsing, availability gating, pagination, date filtering,
and the channels-required constraint.
"""
from __future__ import annotations
import pytest
from lib import pipeline, telegram
# ---- handle parsing ----
@pytest.mark.parametrize("raw,expected", [
("aipost", "aipost"),
("@aipost", "aipost"),
("https://t.me/aipost", "aipost"),
("https://t.me/s/aipost", "aipost"),
("http://t.me/durov", "durov"),
("t.me/channel", "channel"),
(" @spaced ", "spaced"),
])
def test_parse_channel_handle_valid(raw, expected):
assert telegram.parse_channel_handle(raw) == expected
@pytest.mark.parametrize("raw", [
"",
" ",
"-1001234567890",
"-100999888777",
"https://t.me/joinchat/abcdef",
"joinchat/xyz",
])
def test_parse_channel_handle_invalid(raw):
with pytest.raises(telegram.InvalidChannelHandle):
telegram.parse_channel_handle(raw)
def test_parse_channel_sources_filters_invalid():
raw = "aipost, @durov, https://t.me/joinchat/secret, -1001234, valid"
handles = telegram.parse_channel_sources(raw)
assert handles == ["aipost", "durov", "valid"]
def test_parse_channel_sources_dedupes():
raw = "aipost, @aipost, https://t.me/aipost, AIPOST"
handles = telegram.parse_channel_sources(raw)
assert handles == ["aipost"]
# ---- opt-in availability gating ----
def test_not_available_by_default():
config = {"SCRAPECREATORS_API_KEY": "test-key"}
avail = pipeline.available_sources(config)
assert "telegram" not in avail
def test_not_available_without_channels():
config = {
"SCRAPECREATORS_API_KEY": "test-key",
"INCLUDE_SOURCES": "telegram",
}
avail = pipeline.available_sources(config)
assert "telegram" not in avail
def test_available_when_fully_configured():
config = {
"SCRAPECREATORS_API_KEY": "test-key",
"INCLUDE_SOURCES": "telegram",
"TELEGRAM_SOURCES": "aipost,durov",
}
avail = pipeline.available_sources(config)
assert "telegram" in avail
def test_available_when_requested():
config = {
"SCRAPECREATORS_API_KEY": "test-key",
"TELEGRAM_SOURCES": "aipost",
}
avail = pipeline.available_sources(config, requested_sources=["telegram"])
assert "telegram" in avail
def test_excluded_when_in_exclude_sources():
config = {
"SCRAPECREATORS_API_KEY": "test-key",
"INCLUDE_SOURCES": "telegram",
"TELEGRAM_SOURCES": "aipost",
"EXCLUDE_SOURCES": "telegram",
}
avail = pipeline.available_sources(config)
assert "telegram" not in avail
# ---- is_telegram_configured ----
def test_is_telegram_configured_true():
config = {
"SCRAPECREATORS_API_KEY": "test-key",
"TELEGRAM_SOURCES": "aipost",
}
assert telegram.is_telegram_configured(config) is True
def test_is_telegram_configured_false_no_key():
config = {"TELEGRAM_SOURCES": "aipost"}
assert telegram.is_telegram_configured(config) is False
def test_is_telegram_configured_false_no_channels():
config = {"SCRAPECREATORS_API_KEY": "test-key"}
assert telegram.is_telegram_configured(config) is False
# ---- search without channels returns error ----
def test_search_returns_error_without_channels():
result = telegram.search_telegram(
"test topic",
"2026-08-01",
"2026-08-21",
token="test-key",
config={},
)
assert result["items"] == []
assert "channel list required" in result.get("error", "").lower()
def test_search_returns_error_without_token():
result = telegram.search_telegram(
"test topic",
"2026-08-01",
"2026-08-21",
token=None,
config={"TELEGRAM_SOURCES": "aipost"},
)
assert result["items"] == []
assert "SCRAPECREATORS_API_KEY" in result.get("error", "")
# ---- pagination and date filtering (mocked HTTP) ----
MOCK_CHANNEL = {
"handle": "testchannel",
"name": "Test Channel",
"subscriber_count": 10000,
}
MOCK_POSTS_PAGE1 = [
{
"id": "100",
"channel_handle": "testchannel",
"url": "https://t.me/testchannel/100",
"author_name": "Test Channel",
"text": "Recent post about AI agents",
"published_at": "2026-08-20T10:00:00+00:00",
"view_count": 5000,
"reaction_count": 100,
},
{
"id": "99",
"channel_handle": "testchannel",
"url": "https://t.me/testchannel/99",
"author_name": "Test Channel",
"text": "Another recent post",
"published_at": "2026-08-19T10:00:00+00:00",
"view_count": 3000,
"reaction_count": 50,
},
]
MOCK_POSTS_PAGE2 = [
{
"id": "98",
"channel_handle": "testchannel",
"url": "https://t.me/testchannel/98",
"author_name": "Test Channel",
"text": "Old post",
"published_at": "2026-07-01T10:00:00+00:00",
"view_count": 1000,
"reaction_count": 10,
},
]
def test_pagination_stops_on_old_page(monkeypatch):
pages_fetched = []
def mock_get(url, **kwargs):
params = kwargs.get("params", {})
cursor = params.get("cursor")
pages_fetched.append(cursor)
if cursor is None:
return {
"success": True,
"channel": MOCK_CHANNEL,
"posts": MOCK_POSTS_PAGE1,
"cursor": "98",
"has_more": True,
}
else:
return {
"success": True,
"channel": MOCK_CHANNEL,
"posts": MOCK_POSTS_PAGE2,
"cursor": "50",
"has_more": True,
}
monkeypatch.setattr(telegram.http, "get", mock_get)
result = telegram.search_telegram(
"AI agents",
"2026-08-01",
"2026-08-21",
depth="default",
token="test-key",
config={"TELEGRAM_SOURCES": "testchannel"},
)
assert len(pages_fetched) == 2
assert len(result["items"]) == 2
def test_date_filtering(monkeypatch):
def mock_get(url, **kwargs):
return {
"success": True,
"channel": MOCK_CHANNEL,
"posts": MOCK_POSTS_PAGE1 + MOCK_POSTS_PAGE2,
"has_more": False,
}
monkeypatch.setattr(telegram.http, "get", mock_get)
result = telegram.search_telegram(
"AI agents",
"2026-08-15",
"2026-08-21",
token="test-key",
config={"TELEGRAM_SOURCES": "testchannel"},
)
dates = [item["date"] for item in result["items"]]
assert all(d >= "2026-08-15" for d in dates)
assert "2026-07-01" not in dates
def test_topic_relevance_scoring(monkeypatch):
def mock_get(url, **kwargs):
return {
"success": True,
"channel": MOCK_CHANNEL,
"posts": [
{
"id": "1",
"text": "AI agents are transforming software development",
"published_at": "2026-08-20T10:00:00+00:00",
"view_count": 100,
"reaction_count": 10,
},
{
"id": "2",
"text": "Unrelated post about cooking recipes",
"published_at": "2026-08-20T10:00:00+00:00",
"view_count": 100,
"reaction_count": 10,
},
],
"has_more": False,
}
monkeypatch.setattr(telegram.http, "get", mock_get)
result = telegram.search_telegram(
"AI agents",
"2026-08-01",
"2026-08-21",
token="test-key",
config={"TELEGRAM_SOURCES": "testchannel"},
)
items = result["items"]
assert len(items) == 2
ai_item = next(i for i in items if "AI agents" in i["text"])
cook_item = next(i for i in items if "cooking" in i["text"])
assert ai_item["relevance"] > cook_item["relevance"]
def test_depth_page_caps():
assert telegram.DEPTH_PAGE_CAPS["quick"] == 1
assert telegram.DEPTH_PAGE_CAPS["default"] == 3
assert telegram.DEPTH_PAGE_CAPS["deep"] == 6
def test_telegram_max_pages_override(monkeypatch):
pages_fetched = []
def mock_get(url, **kwargs):
params = kwargs.get("params", {})
pages_fetched.append(params.get("cursor"))
return {
"success": True,
"channel": MOCK_CHANNEL,
"posts": MOCK_POSTS_PAGE1,
"cursor": str(len(pages_fetched) * 10),
"has_more": True,
}
monkeypatch.setattr(telegram.http, "get", mock_get)
result = telegram.search_telegram(
"AI agents",
"2026-08-01",
"2026-08-21",
depth="quick",
token="test-key",
config={
"TELEGRAM_SOURCES": "testchannel",
"TELEGRAM_MAX_PAGES": "5",
},
)
assert len(pages_fetched) == 5
def test_http_error_gracefully_handled(monkeypatch):
def mock_get(url, **kwargs):
raise telegram.http.HTTPError("Connection failed", status_code=500)
monkeypatch.setattr(telegram.http, "get", mock_get)
result = telegram.search_telegram(
"AI agents",
"2026-08-01",
"2026-08-21",
token="test-key",
config={"TELEGRAM_SOURCES": "testchannel"},
)
assert result["items"] == []
def test_api_error_gracefully_handled(monkeypatch):
def mock_get(url, **kwargs):
return {"success": False, "error": "Channel not found"}
monkeypatch.setattr(telegram.http, "get", mock_get)
result = telegram.search_telegram(
"AI agents",
"2026-08-01",
"2026-08-21",
token="test-key",
config={"TELEGRAM_SOURCES": "testchannel"},
)
assert result["items"] == []
# ---- MAX_SOURCE_FETCHES cap ----
def test_telegram_capped_to_single_fetch():
assert pipeline.MAX_SOURCE_FETCHES.get("telegram") == 1
# ---- normalization ----
def test_normalize_telegram_posts():
from lib import normalize
items = [
{
"id": "123",
"handle": "testchannel",
"display_name": "Test Channel",
"text": "Test post about AI",
"url": "https://t.me/testchannel/123",
"date": "2026-08-20",
"engagement": {"views": 5000, "reactions": 100},
"relevance": 0.85,
"why_relevant": "Telegram @testchannel: Test post about AI",
},
]
normalized = normalize.normalize_source_items("telegram", items, "2026-08-01", "2026-08-21")
assert len(normalized) == 1
item = normalized[0]
assert item.source == "telegram"
assert item.item_id == "123"
assert "Test post about AI" in item.title