From f95e085e7b3902aa7d932f9125db6c6b68de725b Mon Sep 17 00:00:00 2001 From: wsb1224 Date: Wed, 15 Oct 2025 16:46:31 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AC=AC=E4=B8=80=E6=AC=A1=E6=8F=90=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .idea/.gitignore | 5 + .idea/.name | 1 + .idea/inspectionProfiles/Project_Default.xml | 7 ++ .../inspectionProfiles/profiles_settings.xml | 6 + .idea/misc.xml | 7 ++ .idea/modules.xml | 8 ++ .idea/mp_chajian.iml | 8 ++ README.md | 72 +++++++++++ get_media_id.py | 57 +++++++++ get_permanent_media.py | 62 ++++++++++ get_token.py | 39 ++++++ main.py | 113 ++++++++++++++++++ requirements.txt | 4 + upload_draft.py | 75 ++++++++++++ 14 files changed, 464 insertions(+) create mode 100644 .idea/.gitignore create mode 100644 .idea/.name create mode 100644 .idea/inspectionProfiles/Project_Default.xml create mode 100644 .idea/inspectionProfiles/profiles_settings.xml create mode 100644 .idea/misc.xml create mode 100644 .idea/modules.xml create mode 100644 .idea/mp_chajian.iml create mode 100644 README.md create mode 100644 get_media_id.py create mode 100644 get_permanent_media.py create mode 100644 get_token.py create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 upload_draft.py diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..10b731c --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,5 @@ +# 默认忽略的文件 +/shelf/ +/workspace.xml +# 基于编辑器的 HTTP 客户端请求 +/httpRequests/ diff --git a/.idea/.name b/.idea/.name new file mode 100644 index 0000000..bb2d223 --- /dev/null +++ b/.idea/.name @@ -0,0 +1 @@ +mp_chajian \ No newline at end of file diff --git a/.idea/inspectionProfiles/Project_Default.xml b/.idea/inspectionProfiles/Project_Default.xml new file mode 100644 index 0000000..9c69411 --- /dev/null +++ b/.idea/inspectionProfiles/Project_Default.xml @@ -0,0 +1,7 @@ + + + + \ No newline at end of file diff --git a/.idea/inspectionProfiles/profiles_settings.xml b/.idea/inspectionProfiles/profiles_settings.xml new file mode 100644 index 0000000..105ce2d --- /dev/null +++ b/.idea/inspectionProfiles/profiles_settings.xml @@ -0,0 +1,6 @@ + + + + \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000..3b261f3 --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,7 @@ + + + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..51ab974 --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/mp_chajian.iml b/.idea/mp_chajian.iml new file mode 100644 index 0000000..960f332 --- /dev/null +++ b/.idea/mp_chajian.iml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..203e301 --- /dev/null +++ b/README.md @@ -0,0 +1,72 @@ +# 微信公众号插件API + +这是一个基于FastAPI构建的微信公众号插件API服务，提供以下功能： + +1. 获取微信公众号access_token +2. 上传临时图片到微信公众号素材库 +3. 上传永久图片到微信公众号素材库 +4. 上传文章草稿到微信公众号 + +## 安装依赖 + +```bash +pip install -r requirements.txt +``` + +## 启动服务 + +```bash +python main.py +``` + +服务将在 `http://localhost:8000` 上运行。 + +## API文档 + +启动服务后，可以通过以下地址访问API文档： + +- Swagger UI: http://localhost:8000/docs +- ReDoc: http://localhost:8000/redoc + +## API接口说明 + +### 1. 获取access_token + +- **URL**: `/token` +- **方法**: POST +- **参数**: + - `appid` (string): 公众号的APPID + - `appsecret` (string): 公众号的AppSecret +- **返回**: access_token + +### 2. 上传临时图片 + +- **URL**: `/upload_temp_image` +- **方法**: POST +- **参数**: + - `access_token` (string): 公众号的access_token + - `image_url` (string): 图片的URL地址 +- **返回**: media_id + +### 3. 上传永久图片 + +- **URL**: `/upload_permanent_image` +- **方法**: POST +- **参数**: + - `access_token` (string): 公众号的access_token + - `image_url` (string): 图片的URL地址 +- **返回**: 永久素材的media_id + +### 4. 上传文章草稿 + +- **URL**: `/upload_draft` +- **方法**: POST +- **参数**: + - `access_token` (string): 公众号的access_token + - `title` (string): 文章标题 + - `content` (string): 文章内容（HTML格式） + - `cover_media_id` (string): 封面图片的media_id + - `author` (string): 作者 + - `digest` (string, optional): 文章摘要 + - `show_cover_pic` (int, optional): 是否显示封面，0不显示，1显示，默认为1 +- **返回**: 草稿的media_id \ No newline at end of file diff --git a/get_media_id.py b/get_media_id.py new file mode 100644 index 0000000..951d5d5 --- /dev/null +++ b/get_media_id.py @@ -0,0 +1,57 @@ +import requests +import json +from io import BytesIO + + +def upload_image_to_wechat(access_token, image_url): + """ + 上传图片到微信公众号素材库 + + 参数: + access_token (str): 公众号的access_token + image_url (str): 图片的URL地址 + + 返回: + str: 成功时返回media_id + None: 失败时返回None + """ + if not access_token or not image_url: + return None + + # 微信上传临时素材接口 + upload_url = f"https://api.weixin.qq.com/cgi-bin/media/upload?access_token={access_token}&type=image" + + try: + # 先从URL下载图片 + image_response = requests.get(image_url, timeout=15) + image_response.raise_for_status() # 检查请求是否成功 + + # 将图片内容转换为文件对象 + image_file = BytesIO(image_response.content) + + # 构造表单数据 + files = { + 'media': ('image.jpg', image_file, 'image/jpeg') + } + + # 上传图片到微信服务器 + response = requests.post(upload_url, files=files, timeout=15) + result = json.loads(response.text) + + # 检查是否返回错误 + if "errcode" in result and result["errcode"] != 0: + return None + + # 返回media_id + media_id = result.get("media_id") + if media_id: + return media_id + else: + return None + + except requests.exceptions.RequestException: + return None + except json.JSONDecodeError: + return None + except Exception: + return None diff --git a/get_permanent_media.py b/get_permanent_media.py new file mode 100644 index 0000000..0fa1252 --- /dev/null +++ b/get_permanent_media.py @@ -0,0 +1,62 @@ +import requests +import json +from io import BytesIO + + +def upload_permanent_image_to_wechat(access_token, image_url): + """ + 上传永久图片到微信公众号素材库 + + 参数: + access_token (str): 公众号的access_token + image_url (str): 图片的URL地址 + + 返回: + str: 成功时返回永久素材的media_id + None: 失败时返回None + """ + if not access_token or not image_url: + return None + + # 微信上传永久素材接口（图片类型） + upload_url = f"https://api.weixin.qq.com/cgi-bin/material/add_material?access_token={access_token}&type=image" + + try: + # 先从URL下载图片 + image_response = requests.get(image_url, timeout=15) + image_response.raise_for_status() # 检查请求是否成功 + + # 检查图片大小（微信限制永久图片不超过2MB） + image_size = len(image_response.content) + if image_size > 2 * 1024 * 1024: # 2MB + return None + + # 将图片内容转换为文件对象 + image_file = BytesIO(image_response.content) + + # 构造表单数据 + files = { + 'media': ('image.jpg', image_file, 'image/jpeg') + } + + # 上传图片到微信服务器 + response = requests.post(upload_url, files=files, timeout=15) + result = json.loads(response.text) + + # 检查是否返回错误 + if "errcode" in result and result["errcode"] != 0: + return None + + # 返回media_id + media_id = result.get("media_id") + if media_id: + return media_id + else: + return None + + except requests.exceptions.RequestException: + return None + except json.JSONDecodeError: + return None + except Exception: + return None diff --git a/get_token.py b/get_token.py new file mode 100644 index 0000000..d388092 --- /dev/null +++ b/get_token.py @@ -0,0 +1,39 @@ +import requests +import json + + +def get_wechat_token(appid, appsecret): + """ + 通过微信公众号的APPID和AppSecret获取access_token + + 参数: + appid (str): 公众号的APPID + appsecret (str): 公众号的AppSecret + + 返回: + str: 成功时返回access_token + None: 失败时返回None + """ + if not appid or not appsecret: + return None + + # 微信获取token的接口 + url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={appid}&secret={appsecret}" + + try: + # 发送GET请求 + response = requests.get(url, timeout=10) + # 解析JSON响应 + result = json.loads(response.text) + + # 检查是否返回了错误信息 + if "errcode" in result and result["errcode"] != 0: + return None + + # 返回获取到的token + return result.get("access_token") + + except requests.exceptions.RequestException: + return None + except json.JSONDecodeError: + return None diff --git a/main.py b/main.py new file mode 100644 index 0000000..6a2b805 --- /dev/null +++ b/main.py @@ -0,0 +1,113 @@ +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=8000) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..4f8f147 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +fastapi>=0.68.0 +uvicorn>=0.15.0 +requests>=2.25.1 +pydantic>=1.8.0 \ No newline at end of file diff --git a/upload_draft.py b/upload_draft.py new file mode 100644 index 0000000..e95f27b --- /dev/null +++ b/upload_draft.py @@ -0,0 +1,75 @@ +import requests +import json + + +def upload_article_draft(access_token, title, content, cover_media_id, author, digest=None, show_cover_pic=1): + """ + 上传文章草稿到微信公众号草稿箱 + + 参数: + access_token (str): 公众号的access_token + title (str): 文章标题 + content (str): 文章内容（HTML格式） + cover_media_id (str): 封面图片的media_id + author (str): 作者 + digest (str, optional): 文章摘要，默认为None + show_cover_pic (int, optional): 是否显示封面，0不显示，1显示，默认为1 + + 返回: + str: 成功时返回草稿的media_id + None: 失败时返回None + """ + # 验证必填参数 + if not all([access_token, title, content, cover_media_id, author]): + return None + + # 微信上传草稿接口 + url = f"https://api.weixin.qq.com/cgi-bin/draft/add?access_token={access_token}" + + # 构造请求数据 + article_data = { + "articles": [ + { + "title": title, + "content": content, + "thumb_media_id": cover_media_id, + "author": author, + "show_cover_pic": show_cover_pic + } + ] + } + + # 添加可选参数 + if digest: + article_data["articles"][0]["digest"] = digest + + try: + # 发送POST请求 + response = requests.post( + url, + data=json.dumps(article_data, ensure_ascii=False).encode('utf-8'), + headers={'Content-Type': 'application/json'}, + timeout=15 + ) + + # 解析响应结果 + result = json.loads(response.text) + + # 检查是否有错误 + if "errcode" in result and result["errcode"] != 0: + return None + + # 返回草稿的media_id + media_id = result.get("media_id") + if media_id: + return media_id + else: + return None + + except requests.exceptions.RequestException: + return None + except json.JSONDecodeError: + return None + except Exception: + return None +