Gan, Jimmy
50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
import os
|
|
import pickle
|
|
import logging
|
|
from google.auth.transport.requests import Request
|
|
from google_auth_oauthlib.flow import InstalledAppFlow
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
SCOPES = [
|
|
"https://www.googleapis.com/auth/youtube.readonly",
|
|
"https://www.googleapis.com/auth/youtube.force-ssl"
|
|
]
|
|
CONFIG_DIR = os.environ.get("CONFIG_DIR", "/app/config")
|
|
_local_dir = os.path.dirname(os.path.abspath(__file__))
|
|
_local_config = os.path.join(os.path.dirname(_local_dir), "config")
|
|
if os.path.exists(os.path.join(_local_config, "client_secret.json")):
|
|
CONFIG_DIR = _local_config
|
|
|
|
CLIENT_SECRET = os.path.join(CONFIG_DIR, "client_secret.json")
|
|
TOKEN_FILE = os.path.join(CONFIG_DIR, "token.pickle")
|
|
|
|
|
|
def get_credentials():
|
|
"""Load or refresh OAuth credentials, triggering browser flow if needed."""
|
|
creds = None
|
|
if os.path.exists(TOKEN_FILE):
|
|
with open(TOKEN_FILE, "rb") as token:
|
|
creds = pickle.load(token)
|
|
|
|
if not creds or not creds.valid:
|
|
if creds and creds.expired and creds.refresh_token:
|
|
creds.refresh(Request())
|
|
else:
|
|
if not os.path.exists(CLIENT_SECRET):
|
|
raise FileNotFoundError(
|
|
f"client_secret.json not found at {CLIENT_SECRET}. "
|
|
"Download from Google Cloud Console."
|
|
)
|
|
flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRET, SCOPES)
|
|
# Print URL for manual opening
|
|
auth_url, _ = flow.authorization_url(prompt='consent', access_type='offline')
|
|
print(f"\nOpen this URL in your browser:\n{auth_url}\n")
|
|
creds = flow.run_local_server(port=0)
|
|
|
|
os.makedirs(os.path.dirname(TOKEN_FILE), exist_ok=True)
|
|
with open(TOKEN_FILE, "wb") as token:
|
|
pickle.dump(creds, token)
|
|
|
|
return creds
|