"""Клиенты Instagram и Threads: свои посты и их метрики. Оба API — из семейства Meta Graph, но живут на разных хостах, под разными токенами и отдают insights в разной форме. Общее — пагинация курсорами, формат ошибок и политика повторов — вынесено в MetaClient. """ from __future__ import annotations from typing import Callable, Iterator import requests IG_BASE = "https://graph.instagram.com/v23.0" TH_BASE = "https://graph.threads.net/v1.0" # Метрики, которые Meta отдаёт по своему посту. impressions не запрашиваем: # для медиа, созданных после 2 июля 2024, метрика выведена из строя, её # заменил views. IG_METRICS_COMMON = ("views", "reach", "likes", "comments", "saved", "shares", "total_interactions") IG_METRICS_FEED_ONLY = ("follows", "profile_visits") TH_METRICS = ("views", "likes", "replies", "reposts", "quotes", "shares") IG_MEDIA_FIELDS = ("id,caption,media_type,media_product_type,permalink," "timestamp,like_count,comments_count") TH_MEDIA_FIELDS = "id,text,media_type,permalink,timestamp,is_quote_post" PAGE_SIZE = 25 MAX_RETRIES = 4 BACKOFF_BASE_SECONDS = 2 class MetaAPIError(RuntimeError): """Запрос к Meta не удался и повторы не помогли.""" class TokenExpiredError(MetaAPIError): """Код 190: токен протух. Лечится переавторизацией, а не повтором.""" class MetaClient: """Общая механика: запрос с повторами, пагинация, чистка токена из ошибок.""" base_url = "" def __init__(self, token: str, sleep: Callable[[float], None] = None, max_retries: int = MAX_RETRIES, session=None): self._token = token self._sleep = sleep if sleep is not None else __import__("time").sleep self._max_retries = max_retries self._session = session or requests.Session() def _mask(self, text: str) -> str: """Токен не должен попасть ни в лог, ни в алерт, ни в текст ошибки.""" return text.replace(self._token, "***") if self._token else text def get(self, path: str, params: dict) -> dict: url = "{}/{}".format(self.base_url, path.lstrip("/")) query = dict(params, access_token=self._token) last_error = "" for attempt in range(self._max_retries): try: resp = self._session.get(url, params=query, timeout=60) except requests.RequestException as exc: last_error = self._mask(str(exc)) else: payload = self._payload(resp) error = payload.get("error") or {} if error.get("code") == 190: raise TokenExpiredError(self._mask( error.get("message", "токен протух"))) if resp.status_code < 400 and "error" not in payload: return payload last_error = "HTTP {}: {}".format( resp.status_code, self._mask(error.get("message", "без описания"))) if resp.status_code < 500 and resp.status_code != 429: break # запрос неверен сам по себе — повтор не поможет if attempt < self._max_retries - 1: self._sleep(BACKOFF_BASE_SECONDS * (2 ** attempt)) raise MetaAPIError("{} — {}".format(self._mask(url), last_error)) @staticmethod def _payload(resp) -> dict: try: payload = resp.json() except ValueError: return {"error": {"message": "не-JSON ответ"}} return payload if isinstance(payload, dict) else {"data": payload} def _paginate(self, path: str, fields: str, since: str) -> Iterator[dict]: """Листает страницы, пока посты новее `since`. Останавливаться на первой странице нельзя: одна тред-серия способна вытолкнуть 30-дневное окно за пределы 25 постов. """ params = {"fields": fields, "limit": PAGE_SIZE} while True: payload = self.get(path, params) for item in payload.get("data", []): if item.get("timestamp", "") < since: return yield item after = (payload.get("paging") or {}).get("cursors", {}).get("after") if not after or not payload.get("data"): return params = dict(params, after=after) def _insights(self, post_id: str, metrics) -> dict: payload = self.get("{}/insights".format(post_id), {"metric": ",".join(metrics)}) return {m["name"]: self._metric_value(m) for m in payload.get("data", [])} @staticmethod def _metric_value(metric: dict): """Instagram кладёт число в values[0].value, Threads — в total_value.value.""" if "total_value" in metric: return (metric["total_value"] or {}).get("value") values = metric.get("values") or [{}] return values[0].get("value") class InstagramClient(MetaClient): base_url = IG_BASE def account(self) -> dict: data = self.get("me", {"fields": "id,username,followers_count,media_count"}) return {"username": data.get("username"), "followers": data.get("followers_count"), "media_count": data.get("media_count")} def posts(self, since: str) -> Iterator[dict]: for item in self._paginate("me/media", IG_MEDIA_FIELDS, since): yield { "post_id": item["id"], "caption": item.get("caption") or "", "media_type": item.get("media_type"), "media_product_type": item.get("media_product_type"), "permalink": item.get("permalink"), "timestamp": item.get("timestamp"), "likes": item.get("like_count"), "comments": item.get("comments_count"), } def insights(self, post_id: str, product_type: str = None) -> dict: metrics = list(IG_METRICS_COMMON) if product_type == "FEED": metrics += list(IG_METRICS_FEED_ONLY) return self._insights(post_id, metrics) class ThreadsClient(MetaClient): base_url = TH_BASE def account(self) -> dict: data = self.get("me", {"fields": "id,username"}) insights = self.get("me/threads_insights", {"metric": "followers_count"}) followers = next( (self._metric_value(m) for m in insights.get("data", []) if m.get("name") == "followers_count"), None) # media_count у Threads API нет — не выдумываем, считаем по своей базе return {"username": data.get("username"), "followers": followers, "media_count": None} def posts(self, since: str) -> Iterator[dict]: for item in self._paginate("me/threads", TH_MEDIA_FIELDS, since): yield { "post_id": item["id"], "text": item.get("text") or "", "media_type": item.get("media_type"), "is_quote_post": int(bool(item.get("is_quote_post"))), "permalink": item.get("permalink"), "timestamp": item.get("timestamp"), } def insights(self, post_id: str, product_type: str = None) -> dict: # у тредов нет деления на ленту и рилсы — product_type принимаем ради # единого интерфейса и игнорируем return self._insights(post_id, TH_METRICS)