"""Сбор прироста подписчиков и упоминаний канала в волт (для контент-планёрки). Зачем отдельный сборщик. TGStat на тарифе S отдаёт **только последние ~10 дней** истории подписчиков и ~30 дней упоминаний — всё, что старше, недоступно безвозвратно. Поэтому данные надо не «спрашивать по требованию», а копить: каждый прогон дописывает свежий срез в накопительный файл волта. Отсюда же выбор частоты: планёрка бегает дважды в неделю (вс и ср), окно API — 10 дней, значит пропусков не будет даже если один прогон упадёт. Отдельный ежедневный демон не нужен. Почему в волт, а не в базу: планёрка — headless Claude, которому запрещён Bash (он работает только Read/Glob/Grep/Edit/Write). Достучаться до SQLite он не может, а markdown-файл в волте прочитает и он, и Наташа с телефона. Только stdlib: на Pi у content-planner нет venv, скрипт зовётся системным python3. Запуск: python3 growth.py <корень волта> [ГГГГ-ММ-ДД] Печатает в stdout короткую сводку для лога, в stderr — проблемы. Код возврата ненулевой только если не удалось вообще ничего забрать. """ from __future__ import annotations import json import os import sys import time import urllib.parse import urllib.request from datetime import date, datetime, timedelta, timezone from pathlib import Path API_BASE = "https://api.tgstat.ru" CHANNEL = "natashhhh" TIMEOUT = 30 REL_DIR = Path("projects") / "channel" JSON_NAME = "{channel} {source} прирост и упоминания канала.json" MD_NAME = "{channel} {source} прирост и упоминания канала.md" # Сколько дней прироста показывать в markdown. Накопительный JSON хранит всё. MD_WINDOW_DAYS = 90 class TGStatError(Exception): pass def _api(method: str, params: dict, token: str) -> object: """GET к TGStat. Токен в лог не утекает: в тексте ошибки его нет.""" query = urllib.parse.urlencode({**params, "token": token}) url = f"{API_BASE}/{method}?{query}" try: with urllib.request.urlopen(url, timeout=TIMEOUT) as resp: data = json.loads(resp.read().decode("utf-8")) except Exception as exc: # сеть, таймаут, битый JSON — наверх одним типом raise TGStatError(f"{method}: {type(exc).__name__}: {exc}") from exc if data.get("status") != "ok": raise TGStatError(f"{method}: {data.get('error') or data.get('message') or data}") return data.get("response") def fetch_subscribers(token: str, channel: str = CHANNEL, days: int = 30) -> list: """История подписчиков по дням. Фактически TGStat отдаёт ~10 последних дней.""" now = int(time.time()) resp = _api( "channels/subscribers", { "channelId": f"@{channel}", "startDate": now - days * 86400, "endDate": now, "group": "day", }, token, ) items = resp if isinstance(resp, list) else (resp or {}).get("subscribers") or [] out = [] for it in items: period = it.get("period") count = it.get("participants_count") if period is None or count is None: continue out.append((str(period)[:10], int(count))) return out def fetch_mentions(token: str, channel: str = CHANNEL, days: int = 30) -> tuple: """Упоминания канала за окно. Грабля TGStat: у этого эндпоинта окно задаётся ``startDate``/``endDate``. Параметр ``startTime`` (который работает в channels/posts) здесь **молча игнорируется** — запрос за любой период вернёт один и тот же свежий кусок, и это легко принять за «упоминаний нет». Лимит — не больше 50 за запрос. """ now = int(time.time()) items: list = [] channels: dict = {} for offset in range(0, 200, 50): resp = _api( "channels/mentions", { "channelId": f"@{channel}", "startDate": now - days * 86400, "endDate": now, "limit": 50, "offset": offset, "extended": 1, }, token, ) if not isinstance(resp, dict): # пустой ответ приходит списком break batch = resp.get("items") or [] for ch in resp.get("channels") or []: channels[ch.get("id")] = ch items.extend(batch) if len(batch) < 50: break return items, channels def merge_subscribers(stored: dict, fresh: list) -> dict: """Слить свежий срез в накопленный. Свежее значение важнее: TGStat уточняет вчерашние цифры задним числом.""" out = dict(stored) for day, count in fresh: out[day] = int(count) return out def merge_mentions(stored: dict, items: list, channels: dict) -> dict: """Слить упоминания. Ключ — mentionId, повтор перезаписывает (у канала могло измениться число подписчиков).""" out = dict(stored) for it in items: mid = it.get("mentionId") or it.get("postId") if mid is None: continue ch = channels.get(it.get("channelId")) or {} posted = it.get("postDate") out[str(mid)] = { "date": date.fromtimestamp(posted).isoformat() if posted else None, "username": (ch.get("username") or "").lstrip("@") or None, "title": ch.get("title"), "subscribers": ch.get("participants_count"), "link": it.get("postLink"), } return out def _deltas(subs: dict) -> list: """[(дата, число, дельта к предыдущему дню)] по возрастанию даты.""" rows = [] prev = None for day in sorted(subs): count = subs[day] rows.append((day, count, None if prev is None else count - prev)) prev = count return rows def week_summary(subs: dict, mentions: dict, today: str) -> str: """Короткая сводка за 7 дней — идёт в лог прогона.""" rows = [r for r in _deltas(subs) if r[2] is not None] start = (date.fromisoformat(today) - timedelta(days=7)).isoformat() week = [r for r in rows if r[0] > start] if not week: return "прироста за неделю нет в данных" total = sum(r[2] for r in week) by_day = {} for m in mentions.values(): if m.get("date") and m["date"] > start: by_day.setdefault(m["date"], []).append(m) with_mentions = sum(r[2] for r in week if r[0] in by_day) return ( f"за 7 дней {total:+d} подписчиков, из них {with_mentions:+d} в дни с " f"упоминаниями ({len(by_day)} дн.)" ) def render_md(subs: dict, mentions: dict, today: str) -> str: """Накопленные данные → markdown для волта и для планёрки.""" rows = _deltas(subs) cutoff = (date.fromisoformat(today) - timedelta(days=MD_WINDOW_DAYS)).isoformat() rows = [r for r in rows if r[0] >= cutoff] by_day: dict = {} for m in mentions.values(): if m.get("date"): by_day.setdefault(m["date"], []).append(m) L = [ "---", "tags:", " - type/source", " - project/channel", f"date: {today}", "---", "", "# Прирост канала и упоминания — накопительный срез", "", f"Обновлено: {today}. Собирает `content-planner/growth.py` на каждом прогоне " "планёрки (вс и ср).", "", "> TGStat на тарифе S отдаёт только ~10 дней истории подписчиков и ~30 дней " "упоминаний. Этот файл — накопленная история: всё, что старше окна API, " "существует только здесь.", "", f"**За 7 дней:** {week_summary(subs, mentions, today)}.", "", "## Подписчики по дням", "", "| Дата | Подписчиков | Δ | Упоминания в этот день |", "|---|---|---|---|", ] for day, count, delta in rows: d = "—" if delta is None else f"{delta:+d}" tag = ", ".join( f"@{m['username']} ({m['subscribers'] or '?'})" for m in by_day.get(day, []) if m.get("username") ) L.append(f"| {day} | {count} | {d} | {tag} |") L += ["", "## Упоминания (новые сверху)", "", "| Дата | Канал | Их подписчиков | Ссылка |", "|---|---|---|---|"] for m in sorted(mentions.values(), key=lambda x: x.get("date") or "", reverse=True): if not m.get("date"): continue L.append( f"| {m['date']} | @{m.get('username') or '—'} | " f"{m.get('subscribers') or '—'} | {m.get('link') or ''} |" ) L.append("") return "\n".join(L) def load(path: Path) -> dict: try: with open(path, encoding="utf-8") as f: data = json.load(f) except (FileNotFoundError, ValueError): return {"subscribers": {}, "mentions": {}} data.setdefault("subscribers", {}) data.setdefault("mentions", {}) return data def main(argv: list) -> int: if len(argv) < 2: print("usage: growth.py <корень волта> [ГГГГ-ММ-ДД]", file=sys.stderr) return 2 vault = Path(argv[1]) today = argv[2] if len(argv) > 2 else datetime.now(timezone.utc).strftime("%Y-%m-%d") token = os.environ.get("TGSTAT_TOKEN") if not token: print("growth: нет TGSTAT_TOKEN — пропускаю сбор", file=sys.stderr) return 1 out_dir = vault / REL_DIR if not out_dir.is_dir(): print(f"growth: нет папки {out_dir}", file=sys.stderr) return 1 json_path, md_path = out_dir / JSON_NAME, out_dir / MD_NAME data = load(json_path) failed = [] try: data["subscribers"] = merge_subscribers(data["subscribers"], fetch_subscribers(token)) except TGStatError as exc: failed.append(f"подписчики: {exc}") try: items, channels = fetch_mentions(token) data["mentions"] = merge_mentions(data["mentions"], items, channels) except TGStatError as exc: failed.append(f"упоминания: {exc}") for line in failed: print(f"growth: {line}", file=sys.stderr) # Полный провал — не трогаем файлы: старый срез полезнее затёртого пустым. if len(failed) == 2: return 1 with open(json_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=1, sort_keys=True) with open(md_path, "w", encoding="utf-8") as f: f.write(render_md(data["subscribers"], data["mentions"], today)) print(week_summary(data["subscribers"], data["mentions"], today)) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv))