104 lines
4.3 KiB
Python
104 lines
4.3 KiB
Python
|
|
import os
|
|
from django.conf import settings
|
|
import requests
|
|
from dotenv import load_dotenv
|
|
from django.core.management.base import BaseCommand
|
|
|
|
# Load .env variables
|
|
load_dotenv()
|
|
|
|
class Command(BaseCommand):
|
|
help = 'Update Facebook long-lived access tokens and page access token'
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.app_id = settings.FACEBOOK_APP_ID
|
|
self.app_secret = settings.FACEBOOK_APP_SECRET
|
|
self.page_id = settings.FACEBOOK_PAGE_ID
|
|
self.graph_api_version = settings.FACEBOOK_GRAPH_VERSION_API
|
|
self.page_access_token = settings.FACEBOOK_ACCESS_TOKEN
|
|
self.long_lived_token = settings.FACEBOOK_ACCESS_TOKEN
|
|
|
|
def handle(self, *args, **kwargs):
|
|
"""Handle the token refresh and update .env file."""
|
|
if self.refresh_access_tokens():
|
|
self.stdout.write(self.style.SUCCESS("Successfully refreshed Facebook tokens."))
|
|
else:
|
|
self.stdout.write(self.style.ERROR("Failed to refresh Facebook tokens."))
|
|
|
|
def _exchange_short_to_long_lived_token(self, short_lived_token):
|
|
"""Exchange short-lived token for long-lived token."""
|
|
try:
|
|
url = f"https://graph.facebook.com/{self.graph_api_version}/oauth/access_token"
|
|
params = {
|
|
"grant_type": "fb_exchange_token",
|
|
"client_id": self.app_id,
|
|
"client_secret": self.app_secret,
|
|
"fb_exchange_token": short_lived_token,
|
|
}
|
|
response = requests.get(url, params=params)
|
|
response.raise_for_status()
|
|
long_lived_token = response.json().get("access_token")
|
|
self.stdout.write(self.style.SUCCESS("Successfully exchanged for long-lived user access token."))
|
|
return long_lived_token
|
|
except requests.exceptions.RequestException as e:
|
|
self.stdout.write(self.style.ERROR(f"Error exchanging short-lived token: {e}"))
|
|
return None
|
|
|
|
def _get_page_access_token(self, user_token):
|
|
"""Retrieve Page Access Token."""
|
|
try:
|
|
url = f"https://graph.facebook.com/{self.graph_api_version}/{self.page_id}"
|
|
params = {
|
|
"fields": "access_token",
|
|
"access_token": user_token,
|
|
}
|
|
response = requests.get(url, params=params)
|
|
response.raise_for_status()
|
|
page_access_token = response.json().get("access_token")
|
|
self.stdout.write(self.style.SUCCESS("Successfully obtained page access token."))
|
|
return page_access_token
|
|
except requests.exceptions.RequestException as e:
|
|
self.stdout.write(self.style.ERROR(f"Error retrieving page access token: {e}"))
|
|
return None
|
|
|
|
def _update_env_variable(self, key, value):
|
|
"""Update a variable in the .env file."""
|
|
with open('.env', 'r') as file:
|
|
lines = file.readlines()
|
|
|
|
with open('.env', 'w') as file:
|
|
updated = False
|
|
for line in lines:
|
|
if line.startswith(key):
|
|
file.write(f"{key}={value}\n")
|
|
updated = True
|
|
else:
|
|
file.write(line)
|
|
if not updated:
|
|
file.write(f"{key}={value}\n")
|
|
|
|
def refresh_access_tokens(self):
|
|
"""Refresh long-lived user access token and page access token."""
|
|
if not self.long_lived_token:
|
|
self.stdout.write(self.style.ERROR("No valid long-lived user access token found."))
|
|
return False
|
|
|
|
# Refresh long-lived user token (optional, based on expiry)
|
|
refreshed_user_token = self._exchange_short_to_long_lived_token(self.long_lived_token)
|
|
if refreshed_user_token:
|
|
self.long_lived_token = refreshed_user_token
|
|
|
|
# Refresh page access token
|
|
# page_access_token = self._get_page_access_token(self.long_lived_token)
|
|
# if page_access_token:
|
|
# self.page_access_token = page_access_token
|
|
# # Update tokens in .env file
|
|
self._update_env_variable("FACEBOOK_ACCESS_TOKEN", self.long_lived_token)
|
|
# self._update_env_variable("FACEBOOK_PAGE_ACCESS_TOKEN", self.page_access_token)
|
|
return True
|
|
|
|
self.stdout.write(self.style.ERROR("Failed to refresh page access token."))
|
|
return False
|