113 lines
3.6 KiB
Python
113 lines
3.6 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
import uvicorn
|
|
|
|
from get_token import get_wechat_token
|
|
from get_media_id import upload_image_to_wechat
|
|
from get_permanent_media import upload_permanent_image_to_wechat
|
|
from upload_draft import upload_article_draft
|
|
|
|
app = FastAPI(
|
|
title="微信公众号插件API",
|
|
description="提供微信公众号相关功能的API接口",
|
|
version="1.0.0"
|
|
)
|
|
|
|
# 定义请求模型
|
|
class TokenRequest(BaseModel):
|
|
appid: str
|
|
appsecret: str
|
|
|
|
class ImageUploadRequest(BaseModel):
|
|
access_token: str
|
|
image_url: str
|
|
|
|
class DraftUploadRequest(BaseModel):
|
|
access_token: str
|
|
title: str
|
|
content: str
|
|
cover_media_id: str
|
|
author: str
|
|
digest: Optional[str] = None
|
|
show_cover_pic: Optional[int] = 1
|
|
|
|
# 定义响应模型
|
|
class TokenResponse(BaseModel):
|
|
success: bool
|
|
token: Optional[str] = None
|
|
error: Optional[str] = None
|
|
|
|
class MediaResponse(BaseModel):
|
|
success: bool
|
|
media_id: Optional[str] = None
|
|
url: Optional[str] = None
|
|
error: Optional[str] = None
|
|
|
|
class DraftResponse(BaseModel):
|
|
success: bool
|
|
media_id: Optional[str] = None
|
|
error: Optional[str] = None
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "微信公众号插件API服务已启动"}
|
|
|
|
@app.post("/token", response_model=TokenResponse)
|
|
async def get_token(request: TokenRequest):
|
|
"""获取微信公众号access_token"""
|
|
try:
|
|
token = get_wechat_token(request.appid, request.appsecret)
|
|
if token:
|
|
return TokenResponse(success=True, token=token)
|
|
else:
|
|
return TokenResponse(success=False, error="获取token失败")
|
|
except Exception as e:
|
|
return TokenResponse(success=False, error=str(e))
|
|
|
|
@app.post("/upload_temp_image", response_model=MediaResponse)
|
|
async def upload_temp_image(request: ImageUploadRequest):
|
|
"""上传临时图片到微信公众号素材库"""
|
|
try:
|
|
media_id = upload_image_to_wechat(request.access_token, request.image_url)
|
|
if media_id:
|
|
return MediaResponse(success=True, media_id=media_id)
|
|
else:
|
|
return MediaResponse(success=False, error="上传临时图片失败")
|
|
except Exception as e:
|
|
return MediaResponse(success=False, error=str(e))
|
|
|
|
@app.post("/upload_permanent_image", response_model=MediaResponse)
|
|
async def upload_permanent_image(request: ImageUploadRequest):
|
|
"""上传永久图片到微信公众号素材库"""
|
|
try:
|
|
media_id = upload_permanent_image_to_wechat(request.access_token, request.image_url)
|
|
if media_id:
|
|
return MediaResponse(success=True, media_id=media_id)
|
|
else:
|
|
return MediaResponse(success=False, error="上传永久图片失败")
|
|
except Exception as e:
|
|
return MediaResponse(success=False, error=str(e))
|
|
|
|
@app.post("/upload_draft", response_model=DraftResponse)
|
|
async def upload_draft(request: DraftUploadRequest):
|
|
"""上传文章草稿到微信公众号"""
|
|
try:
|
|
media_id = upload_article_draft(
|
|
request.access_token,
|
|
request.title,
|
|
request.content,
|
|
request.cover_media_id,
|
|
request.author,
|
|
request.digest,
|
|
request.show_cover_pic
|
|
)
|
|
if media_id:
|
|
return DraftResponse(success=True, media_id=media_id)
|
|
else:
|
|
return DraftResponse(success=False, error="上传文章草稿失败")
|
|
except Exception as e:
|
|
return DraftResponse(success=False, error=str(e))
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=5050) |