Django项目中Google Drive公开图片频繁触发403错误求助

我正在开发一个Django项目，实现用户上传图片到Google Drive并设置公开访问，后续按用户维度展示所有上传图片的功能。目前已完成文件夹创建、文件上传及file-id获取，但查看图片时遇到严重问题：尝试多种URL格式后，图片仅能查看几次就弹出403 Forbidden错误，多次尝试无法解决。

尝试过的URL格式：

• 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}

项目代码：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)
        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/"
        )
        creds = flow.fetch_token(code=authorization_code)
        print("creds in fetch_credentials",creds)
        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)
            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:
            self.creds = self.load_credentials()
            if self.creds is not None:
                self.service = build('drive', 'v3', credentials=self.creds)
        except Exception as e:
            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:
            self.creds = self.load_credentials()
            if self.creds is not None:
                self.service = build('drive', 'v3', credentials=self.creds)
        except Exception as e:
            print(f"An error occurred: {str(e)}")

    def logout(self):
        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)

问题根源与修复方案 #

1. 核心问题：权限设置错误

当前代码仅给文件夹设置了公开权限，但Google Drive的权限默认不会自动继承到子文件。即使文件夹是公开的，里面的图片文件依然是私有状态，外部访问时就会触发403。

2. 次要问题：代码冗余与命名错误

• 上传时用临时文件路径作为文件名，导致Drive中文件名异常
• 重复创建Drive实例调用权限设置方法，完全没必要

具体修复步骤

(1) 修改shareWithEveryone方法，简化参数

def shareWithEveryone(self, file_id):
    payload = {
        "role": "reader",
        "type": "anyone"
    }
    return self.service.permissions().create(fileId=file_id, body=payload).execute()

(2) 修复upload_file方法，给图片文件本身设置权限

def upload_file(self, folder_id, file_name):
    try:
        # 用文件原名作为Drive中的文件名，而非临时路径
        file_metadata = {
            'name': file_name.name,
            '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()
        file_id = _file.get('id')
        
        # 给上传的图片文件设置公开权限，而非仅设置文件夹
        self.shareWithEveryone(file_id)
        
        return file_id
    except RefreshError as e:
        self.creds = self.load_credentials()
        if self.creds is not None:
            self.service = build('drive', 'v3', credentials=self.creds)
    except Exception as e:
        print(f"An error occurred: {str(e)}")

(3) 统一Scopes（可选）

load_credentials中写死的scopes是https://www.googleapis.com/auth/drive，和类定义的SCOPES不一致，建议统一为drive.file（足够操作自身创建的文件），避免权限范围混乱。

验证修复后的访问URL

推荐使用稳定的嵌入格式：
https://drive.google.com/uc?export=view&id={file-id}
修复权限后，该URL可稳定提供图片访问，不会再出现403错误。

内容的提问来源于stack exchange，提问作者sodmzs2