Initial commit

site/social/oauth_meta.py

@@ -0,0 +1,185 @@
+"""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()