求助：为对接CustomGPT的API实现文件完整CRUD接口

我正在开发一个对接CustomGPT的API，部署在Google Cloud Run上，目标是让CustomGPT能访问我的文件库，实现文件搜索、读取、创建、修改、删除的完整CRUD能力，同时还支持文件缓存、用户偏好的存取。目前我已经搞定了/search_files和/get_file（读取）接口，但/create_file、/modify_file和/delete_file完全没思路，想请大家帮忙补全这些接口的实现！

先给大家看我现有的完整代码：

现有代码文件 #

main.py #

import os
import logging
from flask import Flask, request, jsonify
from firestore_utils import get_cached_file, cache_file, get_user_prefs, save_user_prefs
from google_api import get_file_content, search_files

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

API_KEY = os.environ.get("REDACTED")

@app.before_request
def require_api_key():
    if request.path == "/status":
        return  # allow health check
    key = request.headers.get("x-api-key")
    if not key or key != API_KEY:
        return jsonify({"error": "Unauthorized"}), 401

def log_request():
    logging.info(f"Incoming request: {request.method} {request.path}")

@app.route("/")
def root():
    return "Welcome to SudicGPT API. Visit /status to check status.", 200

@app.route("/status")
def health_check():
    return "SudicGPT API alive!", 200

@app.route("/get_file", methods=["POST"])
def get_file():
    data = request.json
    if not data or "user" not in data or "filename" not in data:
        return jsonify({"error": "Missing 'user' or 'filename' in request body"}), 400
    user = data["user"]
    filename = data["filename"]
    
    cached = get_cached_file(user, filename)
    if cached:
        return jsonify({"source": "cache", "content": cached})
    
    content = get_file_content(user, filename)
    if content:
        cache_file(user, filename, content)
        return jsonify({"source": "drive", "content": content})
    else:
        return jsonify({"error": "File not found"}), 404

@app.route("/search_files", methods=["POST"])
def search():
    data = request.json
    if not data or "user" not in data:
        return jsonify({"error": "Missing 'user' in request body"}), 400
    user = data["user"]
    query = data.get("query", "")
    results = search_files(user, query)
    return jsonify({"results": results})

@app.route("/get_user_prefs", methods=["POST"])
def get_prefs():
    data = request.json
    if not data or "user" not in data:
        return jsonify({"error": "Missing 'user' in request body"}), 400
    user = data["user"]
    prefs = get_user_prefs(user)
    return jsonify(prefs)

@app.route("/set_user_prefs", methods=["POST"])
def set_prefs():
    data = request.json
    if not data or "user" not in data or "prefs" not in data:
        return jsonify({"error": "Missing 'user' or 'prefs' in request body"}), 400
    user = data["user"]
    prefs = data["prefs"]
    if not isinstance(prefs, dict):
        return jsonify({"error": "'prefs' must be a JSON object/dictionary"}), 400
    save_user_prefs(user, prefs)
    return jsonify({"status": "ok"})

# if __name__ == "__main__":
#     app.run(host="0.0.0.0", port=8080, debug=True)

google_api.py #

from google.oauth2 import service_account
from googleapiclient.discovery import build
import os
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("google_api")

SCOPES = [
    "https://www.googleapis.com/auth/drive",
    "https://www.googleapis.com/auth/documents",
    "https://www.googleapis.com/auth/spreadsheets"
]
SERVICE_ACCOUNT_FILE = os.getenv("REDACTED", "service_account.json")
IMPERSONATE_USER = "REDACTED"

def get_credentials():
    return service_account.Credentials.from_service_account_file(
        SERVICE_ACCOUNT_FILE,
        scopes=SCOPES,
        subject=IMPERSONATE_USER
    )

def search_files(user, query):
    creds = get_credentials()
    drive_service = build("drive", "v3", credentials=creds)
    try:
        results = drive_service.files().list(
            q=f"name contains '{query}' and trashed = false",
            fields="files(id, name, mimeType)"
        ).execute()
        return results.get("files", [])
    except Exception as e:
        logger.exception("Error searching files")
        return []

def get_file_content(user, filename):
    # 假设这个方法已经实现了根据文件名读取Google Drive中文件的内容
    creds = get_credentials()
    # 这里省略具体的读取逻辑（比如区分文档、表格等不同类型文件的读取）
    # 实际要根据mimeType调用对应的API（Docs/Sheets Drive API）
    pass

需求说明 #

我需要在现有代码基础上，补全以下3个接口：

1. /create_file：根据用户传入的文件名、内容创建新文件
2. /modify_file：修改指定文件的内容
3. /delete_file：删除指定文件

同时要和现有代码风格保持一致：包含参数校验、错误返回、日志记录，还要在合适的时机更新文件缓存。

完整实现方案 #

第一步：补全google_api.py的核心方法 #

首先要在google_api.py中补充创建、更新、删除文件的核心逻辑，因为现有代码只用到了搜索和读取：

# 继续在google_api.py中添加以下方法
def create_file(user, filename, content, mimeType="text/plain"):
    """创建新文件，默认创建普通文本文件，支持指定mimeType（比如application/vnd.google-apps.document对应Google文档）"""
    creds = get_credentials()
    drive_service = build("drive", "v3", credentials=creds)
    
    # 先检查是否存在同名文件（可选，根据需求决定是否覆盖）
    existing_files = drive_service.files().list(
        q=f"name='{filename}' and trashed = false",
        fields="files(id)"
    ).execute().get("files", [])
    if existing_files:
        logger.warning(f"File {filename} already exists for user {user}")
        return None  # 或者返回已存在的文件ID，根据需求调整
    
    # 创建文件元数据
    file_metadata = {
        "name": filename,
        "mimeType": mimeType
    }
    
    # 根据文件类型处理内容上传
    try:
        if mimeType == "text/plain":
            # 普通文本文件直接上传
            from googleapiclient.http import MediaIoBaseUpload
            import io
            media = MediaIoBaseUpload(io.BytesIO(content.encode()), mimetype=mimeType, resumable=True)
            file = drive_service.files().create(body=file_metadata, media_body=media, fields="id").execute()
            return file.get("id")
        elif mimeType == "application/vnd.google-apps.document":
            # 创建Google文档并写入内容
            docs_service = build("docs", "v1", credentials=creds)
            doc = docs_service.documents().create(body={"title": filename}).execute()
            doc_id = doc.get("documentId")
            # 插入内容到Google文档
            requests = [
                {
                    "insertText": {
                        "location": {"index": 1},
                        "text": content
                    }
                }
            ]
            docs_service.documents().batchUpdate(documentId=doc_id, body={"requests": requests}).execute()
            return doc_id
        # 可以继续扩展支持表格等其他类型
        return None
    except Exception as e:
        logger.exception(f"Error creating file {filename} for user {user}")
        return None

def update_file_content(user, filename, new_content):
    """更新指定文件的内容，需要先找到文件ID，再根据类型更新"""
    creds = get_credentials()
    drive_service = build("drive", "v3", credentials=creds)
    
    # 先根据文件名获取文件ID和类型
    files = drive_service.files().list(
        q=f"name='{filename}' and trashed = false",
        fields="files(id, mimeType)"
    ).execute().get("files", [])
    if not files:
        logger.warning(f"File {filename} not found for user {user}")
        return False
    
    file_id = files[0]["id"]
    mimeType = files[0]["mimeType"]
    
    try:
        if mimeType == "text/plain":
            # 更新普通文本文件内容
            from googleapiclient.http import MediaIoBaseUpload
            import io
            media = MediaIoBaseUpload(io.BytesIO(new_content.encode()), mimetype=mimeType, resumable=True)
            drive_service.files().update(fileId=file_id, media_body=media).execute()
        elif mimeType == "application/vnd.google-apps.document":
            # 更新Google文档内容：先清空原有内容，再插入新内容
            docs_service = build("docs", "v1", credentials=creds)
            # 先获取文档长度，清空内容
            doc = docs_service.documents().get(documentId=file_id).execute()
            content_length = doc.get("body", {}).get("content", [])[-1].get("endIndex", 1)
            requests = [
                {"deleteContentRange": {"range": {"startIndex": 1, "endIndex": content_length}}},
                {"insertText": {"location": {"index": 1}, "text": new_content}}
            ]
            docs_service.documents().batchUpdate(documentId=file_id, body={"requests": requests}).execute()
        # 其他类型文件同理扩展
        return True
    except Exception as e:
        logger.exception(f"Error updating file {filename} for user {user}")
        return False

def delete_file(user, filename):
    """删除指定文件"""
    creds = get_credentials()
    drive_service = build("drive", "v3", credentials=creds)
    
    # 先获取文件ID
    files = drive_service.files().list(
        q=f"name='{filename}' and trashed = false",
        fields="files(id)"
    ).execute().get("files", [])
    if not files:
        logger.warning(f"File {filename} not found for user {user}")
        return False
    
    file_id = files[0]["id"]
    try:
        drive_service.files().delete(fileId=file_id).execute()
        return True
    except Exception as e:
        logger.exception(f"Error deleting file {filename} for user {user}")
        return False

第二步：在main.py中添加CRUD接口 #

接下来在main.py中添加对应的Flask路由，和现有接口风格保持一致：

# 先在main.py顶部导入新增的方法
from google_api import get_file_content, search_files, create_file, update_file_content, delete_file

# 然后在main.py中添加以下路由

@app.route("/create_file", methods=["POST"])
def create_file_route():
    data = request.json
    # 参数校验
    if not data or "user" not in data or "filename" not in data or "content" not in data:
        return jsonify({"error": "Missing 'user', 'filename' or 'content' in request body"}), 400
    
    user = data["user"]
    filename = data["filename"]
    content = data["content"]
    mimeType = data.get("mimeType", "text/plain")
    
    # 调用Google API创建文件
    file_id = create_file(user, filename, content, mimeType)
    if not file_id:
        return jsonify({"error": "Failed to create file (may already exist)"}), 500
    
    # 创建成功后更新缓存
    cache_file(user, filename, content)
    return jsonify({"status": "ok", "file_id": file_id, "message": f"File {filename} created successfully"})

@app.route("/modify_file", methods=["POST"])
def modify_file_route():
    data = request.json
    # 参数校验
    if not data or "user" not in data or "filename" not in data or "new_content" not in data:
        return jsonify({"error": "Missing 'user', 'filename' or 'new_content' in request body"}), 400
    
    user = data["user"]
    filename = data["filename"]
    new_content = data["new_content"]
    
    # 调用Google API更新文件内容
    success = update_file_content(user, filename, new_content)
    if not success:
        return jsonify({"error": "Failed to modify file or file not found"}), 404
    
    # 更新成功后更新缓存
    cache_file(user, filename, new_content)
    return jsonify({"status": "ok", "message": f"File {filename} modified successfully"})

@app.route("/delete_file", methods=["POST"])
def delete_file_route():
    data = request.json
    # 参数校验
    if not data or "user" not in data or "filename" not in data:
        return jsonify({"error": "Missing 'user' or 'filename' in request body"}), 400
    
    user = data["user"]
    filename = data["filename"]
    
    # 调用Google API删除文件
    success = delete_file(user, filename)
    if not success:
        return jsonify({"error": "Failed to delete file or file not found"}), 404
    
    # 删除成功后清除缓存（需要在firestore_utils.py中补充clear_cached_file方法）
    # clear_cached_file(user, filename)
    return jsonify({"status": "ok", "message": f"File {filename} deleted successfully"})

补充说明 #

• 上面的delete_file_route中，需要你在firestore_utils.py中补充一个clear_cached_file方法，用来删除已删除文件的缓存，示例实现如下：

  # firestore_utils.py中新增
  def clear_cached_file(user, filename):
      # 假设缓存存在Firestore的某个集合中，这里根据你的缓存逻辑实现删除
      db = firestore.Client()
      doc_ref = db.collection("file_cache").document(f"{user}_{filename}")
      doc_ref.delete()
  

关键注意事项 #

• 权限检查：确保服务账号（SERVICE_ACCOUNT_FILE）拥有Google Drive的创建、修改、删除权限，并且IMPERSONATE_USER对应的账号也有对应的文件操作权限
• 文件类型兼容：如果需要支持更多文件类型（比如Google表格），可以继续扩展create_file和update_file_content方法，调用对应的Sheets API
• 错误处理：每个接口都做了参数校验和异常捕获，返回明确的错误信息，同时用logger.exception记录详细日志，方便排查问题
• 缓存一致性：文件创建、修改后及时更新缓存，删除文件后清除缓存，避免CustomGPT读取到旧数据

内容来源于stack exchange