在Flask中实现YouTube Data API浏览器端授权以标记垃圾评论

问题背景 #

我已提取YouTube评论ID，希望通过markAsSpam接口将这些评论标记为垃圾内容。终端环境下以下代码可正常运行：

from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build

flow = InstalledAppFlow.from_client_secrets_file('client_secret.json', ["https://www.googleapis.com/auth/youtube.force-ssl"])
credentials = flow.run_console()
youtube = build('youtube', 'v3', credentials=credentials)
request = youtube.comments().markAsSpam(id='some_id')
request.execute()

该代码通过flow.run_console()在终端生成授权链接，用户访问链接复制授权码后输入终端完成操作。现需适配Flask框架，在浏览器中完成授权及评论标记流程。

解决方案 #

1. 配置Google OAuth客户端 #

在Google Cloud控制台的OAuth 2.0客户端ID设置中，添加Flask服务的回调地址，本地测试可设为http://localhost:5000/callback，确保该地址与代码中的REDIRECT_URI一致。

2. Flask项目结构 #

your_flask_app/
├── app.py
└── client_secret.json

3. 核心代码实现 #

from flask import Flask, redirect, request, session
from google_auth_oauthlib.flow import Flow
from googleapiclient.discovery import build
import os

app = Flask(__name__)
app.secret_key = 'your_secure_random_secret_key'  # 生产环境建议用环境变量存储

# 本地测试允许HTTP传输，生产环境需删除此配置并改用HTTPS
os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'

CLIENT_SECRETS_FILE = 'client_secret.json'
SCOPES = ["https://www.googleapis.com/auth/youtube.force-ssl"]
REDIRECT_URI = 'http://localhost:5000/callback'

# 替换为你提取的评论ID列表
TARGET_COMMENT_IDS = ['comment_id_1', 'comment_id_2']

def batch_mark_spam(credentials):
    """批量处理评论标记为垃圾"""
    youtube = build('youtube', 'v3', credentials=credentials)
    for comment_id in TARGET_COMMENT_IDS:
        try:
            youtube.comments().markAsSpam(id=comment_id).execute()
            print(f"评论 {comment_id} 已标记为垃圾")
        except Exception as e:
            print(f"标记评论 {comment_id} 失败: {str(e)}")

@app.route('/')
def init_auth():
    # 生成授权跳转链接
    flow = Flow.from_client_secrets_file(
        CLIENT_SECRETS_FILE,
        scopes=SCOPES,
        redirect_uri=REDIRECT_URI
    )
    auth_url, state = flow.authorization_url(
        access_type='offline',  # 获取离线凭证，支持后续无需用户重新授权的刷新操作
        include_granted_scopes='true'
    )
    session['state'] = state
    return redirect(auth_url)

@app.route('/callback')
def auth_callback():
    # 验证请求合法性并获取授权凭证
    state = session.get('state')
    flow = Flow.from_client_secrets_file(
        CLIENT_SECRETS_FILE,
        scopes=SCOPES,
        redirect_uri=REDIRECT_URI,
        state=state
    )
    flow.fetch_token(authorization_response=request.url)
    credentials = flow.credentials

    # 执行批量标记垃圾操作
    batch_mark_spam(credentials)

    return "评论标记垃圾操作完成，可查看终端输出详情。"

if __name__ == '__main__':
    app.run(debug=True)

关键注意事项 #

• 生产环境必须部署HTTPS，同时删除OAUTHLIB_INSECURE_TRANSPORT配置。
• app.secret_key需使用随机安全字符串，建议通过环境变量读取，避免硬编码泄露。
• TARGET_COMMENT_IDS可改为动态获取方式，比如从数据库读取、文件导入或前端表单提交。
• access_type='offline'获取的离线凭证支持刷新，适合批量处理大量评论的场景，无需用户重复授权。

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