"""Разовая авторизация: долгоживущий токен в базе. Автоматизировать целиком нельзя — Meta требует, чтобы человек подтвердил доступ в браузере. Есть два пути, оба заканчиваются 60-дневным токеном. Короткий (рекомендуется). В дашборде Meta сгенерировать маркер и вставить: python setup_auth.py instagram --paste python setup_auth.py threads --paste Instagram: «Настройка API для входа в Instagram» → шаг 2 → «Сгенерировать маркер». Threads: «Настройки» сценария Threads → «Генератор маркеров пользователя». Ни секрет приложения, ни redirect_uri при этом не нужны. Длинный (OAuth) — если маркера в дашборде нет. Скрипт печатает ссылку и принимает код из адресной строки: python setup_auth.py instagram python setup_auth.py threads """ from __future__ import annotations import sys from datetime import timedelta import requests from auth import _parse, TOKEN_TTL_DAYS import db CONFIG = { "instagram": { "app_id_env": "META_IG_APP_ID", "app_secret_env": "META_IG_APP_SECRET", "authorize": "https://www.instagram.com/oauth/authorize", "short_token": "https://api.instagram.com/oauth/access_token", "long_token": "https://graph.instagram.com/access_token", "exchange_grant": "ig_exchange_token", "scope": "instagram_business_basic,instagram_business_manage_insights", }, "threads": { "app_id_env": "META_THREADS_APP_ID", "app_secret_env": "META_THREADS_APP_SECRET", "authorize": "https://threads.net/oauth/authorize", "short_token": "https://graph.threads.net/oauth/access_token", "long_token": "https://graph.threads.net/access_token", "exchange_grant": "th_exchange_token", "scope": "threads_basic,threads_manage_insights", }, } class AuthSetupError(RuntimeError): """Первичная авторизация не удалась.""" def _config(platform: str) -> dict: try: return CONFIG[platform] except KeyError: raise AuthSetupError( "неизвестная платформа {!r}, ожидались instagram или threads".format( platform)) from None def authorize_url(platform: str, app_id: str, redirect_uri: str) -> str: cfg = _config(platform) return ("{}?client_id={}&redirect_uri={}&scope={}&response_type=code".format( cfg["authorize"], app_id, redirect_uri, cfg["scope"])) def exchange_code(db_path, platform: str, code: str, app_id: str, app_secret: str, redirect_uri: str, now: str, session=None) -> str: """Меняет код авторизации на долгоживущий токен и кладёт его в базу.""" cfg = _config(platform) session = session or requests # Meta приписывает к коду в адресной строке хвост #_ — с ним обмен падает code = code.strip().split("#")[0] def _hide(text: str) -> str: return str(text).replace(app_secret, "***") try: short = session.post(cfg["short_token"], data={ "client_id": app_id, "client_secret": app_secret, "grant_type": "authorization_code", "redirect_uri": redirect_uri, "code": code}, timeout=60).json() except (requests.RequestException, ValueError) as exc: raise AuthSetupError(_hide(exc)) from None if not short.get("access_token"): raise AuthSetupError("код не принят: {}. Получи свежий — он живёт " "считаные минуты и срабатывает один раз".format( _hide(short.get("error_message") or short.get("error", "без описания")))) try: long_lived = session.get(cfg["long_token"], params={ "grant_type": cfg["exchange_grant"], "client_secret": app_secret, "access_token": short["access_token"]}, timeout=60).json() except (requests.RequestException, ValueError) as exc: raise AuthSetupError(_hide(exc)) from None token = long_lived.get("access_token") if not token: raise AuthSetupError("часовой токен не удалось обменять на долгий: {}".format( _hide((long_lived.get("error") or {}).get("message", "без описания")))) expires_at = (_parse(now) + timedelta( seconds=long_lived.get("expires_in", TOKEN_TTL_DAYS * 86400)) ).strftime("%Y-%m-%dT%H:%M:%SZ") db.save_token(db_path, platform, token, issued_at=now, expires_at=expires_at) return token def app_credentials(platform: str, env) -> tuple: """ID, секрет и redirect_uri для платформы. Instagram и Threads — два разных приложения внутри одного приложения Meta, у каждого свой ID и свой секрет; общий у них только redirect_uri. """ cfg = _config(platform) names = (cfg["app_id_env"], cfg["app_secret_env"], "META_REDIRECT_URI") values = tuple(env.get(name) for name in names) missing = [name for name, value in zip(names, values) if not value] if missing: raise AuthSetupError("в .env не заполнены: {}".format(", ".join(missing))) return values def save_manual_token(db_path, platform: str, token: str, now: str) -> str: """Кладёт в базу маркер, выданный дашбордом Meta. Дашборд отдаёт сразу долгоживущий маркер, поэтому обмен кода не нужен — и, в отличие от OAuth, этот путь не требует ни секрета приложения, ни redirect_uri (у Threads он к тому же не сохраняется). """ _config(platform) token = token.strip() if not token: raise AuthSetupError("маркер пустой — скопируй его из дашборда целиком") expires_at = (_parse(now) + timedelta(days=TOKEN_TTL_DAYS)).strftime( "%Y-%m-%dT%H:%M:%SZ") db.save_token(db_path, platform, token, issued_at=now, expires_at=expires_at) return token def main(argv=None) -> int: from datetime import datetime, timezone import os from pathlib import Path from dotenv import load_dotenv argv = argv if argv is not None else sys.argv[1:] if not argv or argv[0] not in CONFIG: print("укажи платформу: python setup_auth.py instagram|threads [--paste]") return 2 platform = argv[0] load_dotenv(Path(__file__).resolve().parent / ".env") db_path = default_db_path() db.init_db(db_path) now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") # Дашборд Meta выдаёт готовый 60-дневный маркер — это короче, чем OAuth, # и не упирается в redirect_uri, который у Threads не сохраняется. if "--paste" in argv[1:]: from getpass import getpass print("\nМаркер из дашборда Meta ({}). Ввод скрыт — " "он не попадёт ни на экран, ни в историю команд.\n".format(platform)) try: save_manual_token(db_path, platform, getpass("Вставь маркер: "), now) except AuthSetupError as exc: print("\nНе вышло: {}".format(exc)) return 1 print("\nГотово: маркер {} сохранён, живёт 60 дней и дальше " "обновляется сам.".format(platform)) return 0 try: app_id, app_secret, redirect_uri = app_credentials(platform, os.environ) except AuthSetupError as exc: print(exc) return 2 print("\n1. Открой ссылку в браузере (нужен VPN — сайты Meta из России " "не открываются) и подтверди доступ:\n") print(" " + authorize_url(platform, app_id, redirect_uri)) print("\n2. После подтверждения браузер уйдёт на redirect_uri, и в адресной " "строке будет ?code=… — скопируй код целиком.\n") code = input("Вставь код: ").strip() try: exchange_code(db_path, platform, code, app_id, app_secret, redirect_uri, now) except AuthSetupError as exc: print("\nНе вышло: {}".format(exc)) return 1 print("\nГотово: токен {} сохранён, живёт 60 дней и дальше обновляется сам.".format( platform)) return 0 def default_db_path(): """Путь к базе: на Pi задаётся через env, на Маке — обычное место для данных.""" import os from pathlib import Path env_path = os.environ.get("META_DB_PATH") if env_path: return Path(env_path) return Path.home() / "Library" / "Application Support" / "meta-puller" / "meta.db" if __name__ == "__main__": sys.exit(main())