基于Python的Twikoo图床简单实现

利用服务器公网IP和域名，通过简单的Python代码实现Twikoo评论中的图片功能。

由于没有购买和注册图床服务，且没有绑定腾讯云服务，所以只能自行想办法实现Twikoo的评论带图片功能。

想起来之前工作中有利用FastAPI实现过在Streamlit大模型RAG检索对话中的图片显示功能，因此考虑在此复用。

查看Twikoo代码中的请求方式

Twikoo官方GitHub地址: twikoo

代码路径: src/server/function/twikoo/utils/image.js
代码地址: https://github.com/twikoojs/twikoo/blob/main/src/server/function/twikoo/utils/image.js
评论图床引用代码片段:

if (!config.IMAGE_CDN || !config.IMAGE_CDN_TOKEN) {
    throw new Error('未配置图片上传服务')
    }
    // tip: qcloud 图床走前端上传，其他图床走后端上传
    if (config.IMAGE_CDN === '7bu') {
    await fn.uploadImageToLskyPro({ photo, fileName, config, res, imageCdn: 'https://7bu.top' })
    } else if (config.IMAGE_CDN === 'smms') {
    await fn.uploadImageToSmms({ photo, fileName, config, res, imageCdn: 'https://smms.app/api/v2/upload' })
    } else if (isUrl(config.IMAGE_CDN)) {
    await fn.uploadImageToLskyPro({ photo, fileName, config, res, imageCdn: config.IMAGE_CDN })
    } else if (config.IMAGE_CDN === 'lskypro') {
    await fn.uploadImageToLskyPro({ photo, fileName, config, res, imageCdn: config.IMAGE_CDN_URL })
    } else if (config.IMAGE_CDN === 'piclist') {
    await fn.uploadImageToPicList({ photo, fileName, config, res, imageCdn: config.IMAGE_CDN_URL })
    } else {
    throw new Error('不支持的图片上传服务')
}

在源代码的第24行，有 if (isUrl(config.IMAGE_CDN))，即支持自定义图片功能的地址，因此只需要在方法中搞清楚请求头的相关参数即可。
请求代码如下：

async uploadImageToLskyPro({ photo, fileName, config, res, imageCdn }) {
    // 自定义兰空图床（v2）URL
    const formData = new FormData()
    formData.append('file', fn.base64UrlToReadStream(photo, fileName))
    if (process.env.TWIKOO_LSKY_STRATEGY_ID) {
        formData.append('strategy_id', parseInt(process.env.TWIKOO_LSKY_STRATEGY_ID))
    }
    const url = `${imageCdn}/api/v1/upload`
    let token = config.IMAGE_CDN_TOKEN
    if (!token.startsWith('Bearer')) {
        token = `Bearer ${token}`
    }
    const uploadResult = await axios.post(url, formData, {
        headers: {
            ...formData.getHeaders(),
            Authorization: token
        }
    })
    if (uploadResult.data.status) {
        res.data = uploadResult.data.data
        res.data.url = res.data.links.url
    } else {
        throw new Error(uploadResult.data.message)
    }
},

确定好请求头的相关参数，下面即可利用Python的FastAPI包实现图片服务，具体代码如下：

from fastapi import FastAPI, UploadFile, File, Form, HTTPException, Header, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from typing import Optional
import os

app = FastAPI()

UPLOAD_Folder = 'Saved_Images'

# 确保有一个目录来保存上传的图片
os.makedirs(UPLOAD_Folder, exist_ok=True)

# 确定图片在网页浏览的基础地址，这里默认/images，即`https://xxxx.com/images/file_name.jpg`
app.mount("/images", StaticFiles(directory=UPLOAD_Folder), name="static")


@app.post("/")
async def image_server():
    return {'message': 'Hello'}


@app.post("/api/v1/upload")
async def upload_image(
    file: UploadFile = File(...),  # 接收文件
    strategy_id: int = Form(None),  # 接收表单字段，设置默认值为None
    authorization: str = Header(...)  # 接收Authorization请求头
):
    if authorization != 'Bearer xxx':  # 这里可以设置相关鉴权的KEY等 --> 对应 `IMAGE_CDN_TOKEN`
        reponse = {
            'status': False,
            'data': {
                'links': {
                    'url': '--'
                }
            },
            'message': 'Toekn Invalid..'
        }
        return reponse
    # 读取上传的文件
    file_content = await file.read()

    # 保存图片到服务器
    cur_folder = authorization.replace('Bearer ', '')
    if cur_folder:
        os.makedirs(f'{UPLOAD_Folder}/{cur_folder}', exist_ok=True)
        file_path = f'{UPLOAD_Folder}/{cur_folder}/{file.filename}'
    else:
        os.makedirs(f'{UPLOAD_Folder}', exist_ok=True)
        file_path = f'{UPLOAD_Folder}/{file.filename}'
    with open(file_path, "wb") as f:
        f.write(file_content)
    
    upload_success = True  # 假设上传总是成功的，你可以根据实际情况修改这个值
    if not os.path.exists(file_path):
        upload_success = False
    message = "File and form data received successfully"
    data = {
        "links": {
            "url": f"https://{域名orIP地址}/images/{cur_folder}/{file.filename}"  # 假设的URL  --> 对应 `IMAGE_CDN`
        }
    }

    # 构造响应
    response = {
        "status": upload_success,
        "data": data if upload_success else None,
        "message": message if upload_success else "Upload failed"
    }

    # 如果上传失败，你可以抛出一个HTTPException
    if not upload_success:
        raise HTTPException(status_code=400, detail=response["message"])
    return response


# 在你的主程序中运行Uvicorn服务器
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8005)

至此，基于Pythond FastAPI的图片服务已经实现，附Python代码请求的脚本：

import requests
from fastapi import HTTPException
from typing import Optional

# 假设的服务器地址和端点
url = "http://127.0.0.1:8005/api/v1/upload"

# 要上传的文件路径
file_path = "star.png"

# 表单数据
form_data = {
    "strategy_id": 123  # 假设的表单字段值
}

# 请求头
headers = {
    "Authorization": "Bearer xxx"  # 替换为实际的鉴权令牌
}

# 打开文件
with open(file_path, "rb") as file:
    files = {"file": (file_path, file, "multipart/form-data")}

    # 发送请求
    try:
        response = requests.post(url, files=files, data=form_data, headers=headers)
        response.raise_for_status()  # 如果请求返回了失败的状态码，将抛出HTTPError异常
        print(response.json())  # 打印服务器响应
    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP error occurred: {http_err}")  # 打印HTTP错误
    except Exception as err:
        print(f"An error occurred: {err}")  # 打印其他错误