Python代码实现Twikoo品评中的图片显示 - IT评测·应用市场-qidao123.com

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('不支持的图片上传服务')
}

复制代码

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)
}
},

复制代码

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)

复制代码

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}") # 打印其他错误

复制代码