web_django_template/site/social/oauth_meta.py

"""Facebook Login for Business (Instagram API with Facebook Login).

Docs:
https://developers.facebook.com/documentation/instagram-platform/instagram-api-with-facebook-login/business-login-for-instagram

Uses response_type=token (tokens arrive in the URL fragment). A thin callback
page reads the fragment and POSTs tokens to the server.
"""

from __future__ import annotations

import json
import logging
from datetime import timedelta
from urllib.parse import urlencode

import requests
from django.conf import settings
from django.urls import reverse
from django.utils import timezone

from social.crypto import decrypt_tokens
from social.models import Platform, SocialAppCredentials

logger = logging.getLogger(__name__)

GRAPH_VERSION = "v21.0"
GRAPH = f"https://graph.facebook.com/{GRAPH_VERSION}"
AUTHORIZE_URL = f"https://www.facebook.com/{GRAPH_VERSION}/dialog/oauth"

# Permissions for Page posting + Instagram content publish via Facebook Login.
DEFAULT_SCOPES = (
    "instagram_basic",
    "instagram_content_publish",
    "pages_show_list",
    "pages_read_engagement",
    "pages_manage_posts",
)

# Triggers Instagram Professional onboarding inside Facebook Login for Business.
IG_ONBOARDING_EXTRAS = {"setup": {"channel": "IG_API_ONBOARDING"}}


class MetaOAuthError(RuntimeError):
    """Raised when Facebook Login for Business or Page/IG lookup fails."""


def get_app_credentials() -> SocialAppCredentials | None:
    """Meta App ID/Secret — stored on the facebook credentials row (shared)."""
    fb = SocialAppCredentials.objects.filter(platform=Platform.FACEBOOK).first()
    if fb and fb.is_configured:
        return fb
    # Back-compat if credentials were saved from the Instagram form earlier.
    return SocialAppCredentials.objects.filter(platform=Platform.INSTAGRAM).first()


def load_client_credentials() -> tuple[str, str]:
    app = get_app_credentials()
    if not app or not app.client_id or not app.encrypted_client_secret:
        raise MetaOAuthError(
            "Meta app credentials are not saved yet. "
            "Enter Meta App ID and App Secret on the Connect Facebook or Instagram form."
        )
    try:
        secret = decrypt_tokens(app.encrypted_client_secret)
    except ValueError as exc:
        raise MetaOAuthError(
            "Could not decrypt Meta app secret. Re-enter the App Secret."
        ) from exc
    if not secret:
        raise MetaOAuthError("Meta app secret is empty — save it again.")
    return app.client_id.strip(), secret.strip()


def credentials_configured() -> bool:
    app = get_app_credentials()
    return bool(app and app.is_configured)


def redirect_uri() -> str:
    """Must match Valid OAuth Redirect URIs (Facebook Login for Business settings)."""
    base = (settings.PUBLIC_SITE_URL or "").rstrip("/")
    if not base:
        raise MetaOAuthError(
            "PUBLIC_SITE_URL is not set; cannot build Meta OAuth redirect URI."
        )
    return f"{base}{reverse('social:meta_oauth_callback')}"


def authorization_url(*, state: str, scopes: tuple[str, ...] = DEFAULT_SCOPES) -> str:
    client_id, _secret = load_client_credentials()
    params = {
        "client_id": client_id,
        "display": "page",
        "extras": json.dumps(IG_ONBOARDING_EXTRAS, separators=(",", ":")),
        "redirect_uri": redirect_uri(),
        "response_type": "token",
        "scope": ",".join(scopes),
        "state": state,
    }
    return f"{AUTHORIZE_URL}?{urlencode(params)}"


def fetch_pages(user_access_token: str) -> list[dict]:
    """
    GET /me/accounts — Pages the user can manage, with linked IG business accounts.

    Each item: id, name, access_token (page), instagram_business_account (optional).
    """
    response = requests.get(
        f"{GRAPH}/me/accounts",
        params={
            "fields": "id,name,access_token,instagram_business_account{id,username,name}",
            "access_token": user_access_token,
        },
        timeout=30,
    )
    if response.status_code >= 400:
        logger.warning(
            "Meta /me/accounts failed: %s %s",
            response.status_code,
            response.text[:500],
        )
        raise MetaOAuthError(
            f"Could not list Facebook Pages ({response.status_code}). "
            "Confirm pages_show_list was granted and the user manages a Page."
        )
    data = response.json().get("data") or []
    if not data:
        raise MetaOAuthError(
            "No Facebook Pages found for this user. Create a Page and link a "
            "Professional Instagram account, then try again."
        )
    return data


def facebook_token_blob(page: dict, *, user_token: str = "") -> dict:
    page_token = page.get("access_token") or ""
    if not page_token:
        raise MetaOAuthError(f"Page {page.get('id')} missing access_token.")
    blob: dict = {
        "access_token": page_token,
        "page_id": str(page["id"]),
        "user_access_token": user_token,
        "auth_type": "facebook_login_for_business",
    }
    ig = page.get("instagram_business_account") or {}
    if ig.get("id"):
        blob["ig_user_id"] = str(ig["id"])
    return blob


def instagram_token_blob(page: dict, *, user_token: str = "") -> dict:
    ig = page.get("instagram_business_account") or {}
    ig_id = str(ig.get("id") or "").strip()
    if not ig_id:
        raise MetaOAuthError(
            f"Page “{page.get('name') or page.get('id')}” has no linked "
            "Instagram Professional account."
        )
    page_token = page.get("access_token") or ""
    if not page_token:
        raise MetaOAuthError("Page access token missing — required for Instagram Graph API.")
    username = (ig.get("username") or "").strip()
    name = (ig.get("name") or "").strip()
    label = f"@{username}" if username else (name or "Instagram account")
    return {
        "access_token": page_token,
        "page_id": str(page["id"]),
        "ig_user_id": ig_id,
        "username": username,
        "label": label,
        "user_access_token": user_token,
        "auth_type": "facebook_login_for_business",
    }


def expires_at_from_fragment(expires_in: str | int | None) -> str | None:
    try:
        seconds = int(expires_in or 0)
    except (TypeError, ValueError):
        return None
    if seconds <= 0:
        return None
    return (timezone.now() + timedelta(seconds=seconds)).isoformat()