我正在开发一个 Django 项目，我允许用户在具有公共访问权限的谷歌驱动器上上传他们的图像，然后我以用户方式显示所有上传的图像。
一切完成后，我可以创建文件夹，然后在其中上传文件并获取文件ID。但在查看该图像时，我陷入了困境。
我用了很多方法都没有收到 403 Forbidden Error 但还是没有成功！
https://drive.google.com/file/d/{file-id}/preview
https://drive.google.com/uc?id={file-id}&export=download
https://drive.google.com/uc?export=view&id={file-id}
文件（只是一个图像）我只能查看几次，但随后我开始收到这个烦人的 403 禁止错误。
我对这个错误感到非常沮丧，请帮助。
这是我来自 utils.py 文件的完整代码
from __future__ import print_function
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload, MediaIoBaseDownload
from google_auth_oauthlib.flow import InstalledAppFlow
from google.oauth2.credentials import Credentials
from google.auth.exceptions import RefreshError
from google.auth.transport.requests import Request
from django.conf import settings
from django.shortcuts import redirect
import argparse
import base64
import json
import io, os
import shutil
import requests
from pebble import concurrent
TOKEN_DIR = "oauth_tokens/drive"
class Drive:
SCOPES = ['https://www.googleapis.com/auth/drive.file']
extensions = {"csv": "text/csv", "mpeg": "video/mpeg", "mpg": "video/mpeg", "tiff": "image/tiff",
"wmv": "video/x-ms-wmv", "xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
"zip": "application/x-zip-compressed"}
def __init__(self,email):
self.email = email
self.creds = self.load_credentials()
if self.creds is not None:
self.service = build('drive', 'v3', credentials=self.creds)
else:
self.service = None
def generate_authorization_url(self):
print("inside generate_authorization_url")
flow = InstalledAppFlow.from_client_secrets_file(
settings.GOOGLE_OAUTH2_CLIENT_SECRETS_JSON, self.SCOPES, redirect_uri="http://localhost:8000/google_drive_callback/"
)
print("flow in generate_authorization_url", flow)
# Generate the authorization URL
authorization_url, _ = flow.authorization_url(
access_type='offline',
include_granted_scopes='false'
)
print("authorization_url in generate_authorization_url ",authorization_url)
return authorization_url
def fetch_credentials(self, authorization_code):
print("inside fetch_credentials")
flow = InstalledAppFlow.from_client_secrets_file(
settings.GOOGLE_OAUTH2_CLIENT_SECRETS_JSON, self.SCOPES, redirect_uri="http://localhost:8000/google_drive_callback/"
)
# Exchange the authorization code for credentials
creds = flow.fetch_token(code=authorization_code)
print("creds in fetch_credentials",creds)
# Serialize the credentials to JSON and save them to a file
creds_json = {
"token": creds['access_token'],
"refresh_token": creds.get('refresh_token',"1//0gTWnxxxxRAAGBASNwF-xxxxxxx-JFK_Yg"),
"scopes": creds['scope'],
"expires_at":creds['expires_at'],
"token_type":creds['token_type'],
"token_uri": "https://oauth2.googleapis.com/token",
"client_id": "175xxxxn.apps.googleusercontent.com",
"client_secret": "GOCSPX-XXXXXX"
}
print("creds_json in fetch_credentials",creds_json)
token_path = os.path.join(TOKEN_DIR, f"{self.email}_token.json")
try:
os.makedirs(os.path.dirname(token_path), exist_ok=True)
with open(token_path, 'w') as token_file:
json.dump(creds_json, token_file)
print("token.json updated.")
except Exception as e:
print("Error writing credentials to token.json:", str(e))
return creds
def load_credentials(self):
token_path = os.path.join(TOKEN_DIR, f"{self.email}_token.json")
if not os.path.exists(token_path):
print("file not found in load_credentials")
creds_json = {"token": "ya29.a0AfB_byD7t9wWRGsuERfEt3xxxxxS92ZW_MDFcPcuuKrT8IoRfcFWinRZ-xSYxxYXwjtnCJlvxxxxnSOS2xBtXm7F05eagPHyg6F9xUD7rmz_uZzAqv5EugxQKaCgYKAYgSARISFQGOcNnCv4v3mTcbXRtdT0T4ZPtQpQ0171", "refresh_token": "1//0gOgL_UZ_xxxxx-L9Ir0W_-0kLqhgtnosiiWBESLDUQwP21VIl1j6Q_sPrO34tl-wZzJfXK4RJl0-PAmxCQ9ws", "scopes": ["https://www.googleapis.com/auth/drive"], "expires_at": 1696676452.3799534, "token_type": "Bearer", "token_uri": "https://oauth2.googleapis.com/token", "client_id": "175XXXXX.apps.googleusercontent.com", "client_secret": "GOCSPX-XXXXXX"}
try:
with open(token_path, 'w') as token_file:
json.dump(creds_json, token_file)
print("token.json created in load_credentials.")
except Exception as e:
print("Error writing credentials to token.json:", str(e))
if os.path.exists(token_path):
print("token_path found in load_credentials")
creds = Credentials.from_authorized_user_file(token_path, scopes=self.SCOPES)
# Check if the credentials need refreshing
if creds and creds.expired and creds.refresh_token:
try:
creds.refresh(Request())
print("creds.refresh(Request()) completed")
except Exception as e:
print(f"Error refreshing token in load_credentials: {str(e)}")
return None
return creds
return None
def shareWithEveryone(self, folderId, payload):
payload = {
"role": "reader",
"type": "anyone"
}
return self.service.permissions().create(fileId=folderId, body=payload).execute()
def create_folder(self, folder_name):
try:
folder_id = None
query = "mimeType='application/vnd.google-apps.folder' and trashed=false and name='" + folder_name + "'"
results = self.service.files().list(
pageSize=1, q=query, fields="files(id, name)").execute()
folders = results.get('files', [])
if folders:
folder_id = folders[0]['id']
else:
file_metadata = {
'name': folder_name,
'mimeType': 'application/vnd.google-apps.folder'
}
folder_file = self.service.files().create(body=file_metadata,
fields='id').execute()
folder_id = folder_file.get('id')
return folder_id
except RefreshError as e:
# Handle RefreshError by reauthorizing
self.creds = self.load_credentials()
if self.creds is not None:
self.service = build('drive', 'v3', credentials=self.creds)
# You can log the error, retry the operation, or take other actions as needed
except Exception as e:
# Handle other exceptions as appropriate
print(f"An error occurred: {str(e)}")
def upload_file(self, folder_id, file_name):
try:
file_metadata = {
'name': file_name.temporary_file_path(),
'parents': [folder_id]
}
media = MediaFileUpload(file_name.temporary_file_path(),
mimetype=self.extensions[file_name.name.split(".")[1]], resumable=True)
_file = self.service.files().create(body=file_metadata,
media_body=media, fields='id').execute()
aas = Drive(email=self.email).shareWithEveryone(folder_id, file_metadata)
file_id = _file.get('id')
return file_id
except RefreshError as e:
# Handle RefreshError by reauthorizing
self.creds = self.load_credentials()
if self.creds is not None:
self.service = build('drive', 'v3', credentials=self.creds)
# You can log the error, retry the operation, or take other actions as needed
except Exception as e:
# Handle other exceptions as appropriate
print(f"An error occurred: {str(e)}")
def logout(self):
# Revoke Google token (use the user's refresh token)
if self.creds and self.creds.refresh_token:
try:
token_info_url = 'https://oauth2.googleapis.com/revoke'
payload = {
'token': self.creds.refresh_token
}
response = requests.post(token_info_url, data=payload)
if response.status_code == 200:
print("Token Drive revoked successfully")
else:
print("Failed to revoke token:", response.status_code)
except Exception as e:
print("Error while revoking token:", str(e))
我称这个函数为
drive = Drive(email=request.user)
file-id= drive.upload_file(folder_id,filename)
这就是我获取文件 ID 的方法，我将其与上述 URL 一起使用以在网页中查看。
我也有同样的问题。两天前显示谷歌驱动器图像工作正常，但昨天开始收到 403 错误。没有对代码进行任何更新。从谷歌驱动器获取文字文件仍然有效，但图像不行。