From 622ff789f5da30606c06dc2c48bb4ac64fcfd92e Mon Sep 17 00:00:00 2001 From: yumoqing Date: Tue, 4 Aug 2026 08:36:19 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20honest=20platform=20classification=20?= =?UTF-8?q?=E2=80=94=20real=20API=20/=20restricted=20/=20browser-only?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - youtube, wechat: ✅ real API, ready to use - weibo, twitter: ⚠️ restricted (needs app review / paid tier) - douyin, bilibili, kuaishou, channels, xiaohongshu, toutiao: ❌ no API → raise NotImplementedError with clear browser-automation guidance - engine.py: skip browser-only platforms with descriptive error in log --- publisher/engine.py | 5 +- publisher/platforms.py | 216 ++++++++++++++++------------------------- 2 files changed, 86 insertions(+), 135 deletions(-) diff --git a/publisher/engine.py b/publisher/engine.py index 4738494..71705c2 100644 --- a/publisher/engine.py +++ b/publisher/engine.py @@ -5,7 +5,7 @@ import json, time, traceback import aiohttp from datetime import datetime -from .platforms import PUBLISHERS +from .platforms import PUBLISHERS, BROWSER_ONLY from .db import (get_pending_tasks, get_content, get_platform, update_task_status, update_content_status, write_publish_log) @@ -20,6 +20,9 @@ async def publish_one(session, task, content, platform, sor): if not publisher: return False, 'unsupported platform: ' + platform_code, '' + if platform_code in BROWSER_ONLY: + return False, 'browser-only: ' + platform_code + ' has no public API, needs Playwright/Selenium', '' + if content['content_type'] not in publisher['content_types']: return False, 'content type ' + content['content_type'] + ' not supported by ' + platform_code, '' diff --git a/publisher/platforms.py b/publisher/platforms.py index 5c4c974..e144d46 100644 --- a/publisher/platforms.py +++ b/publisher/platforms.py @@ -1,134 +1,41 @@ """ -多平台发布适配器 — 将统一内容格式转换为各平台 API 请求 +多平台发布适配器 — 按真实 API 可用性分三类: -支持的平台: douyin/kuaishou/bilibili/weibo/wechat/xiaohongshu/toutiao/youtube/twitter -新增平台: 添加 format_xxx() 函数并注册到 PUBLISHERS 字典即可 + ✅ 真实 API — 可立即使用: youtube, wechat + ⚠️ 受限 API — 需审核/付费: weibo, twitter + ❌ 无 API — 只能浏览器自动化: douyin, bilibili, kuaishou, xiaohongshu, toutiao, channels + +新增平台: 添加对应函数并 register() 即可。 """ -import json, time, hashlib, hmac, base64, re +import json, time from datetime import datetime -from urllib.parse import urlencode PUBLISHERS = {} +BROWSER_ONLY = set() # 无 API，只能浏览器自动化 -def register(code, name, fn, content_types): - PUBLISHERS[code] = {'name': name, 'publish': fn, 'content_types': content_types} - -# ---- 平台适配函数: 返回 (method, url, headers, body, content_type_hint) ---- +def register(code, name, fn, content_types, api_status='real'): + """ + api_status: 'real' (真实API) | 'restricted' (需审核/付费) | 'none' (浏览器自动化) + """ + PUBLISHERS[code] = { + 'name': name, 'publish': fn, + 'content_types': content_types, 'api_status': api_status + } + if api_status == 'none': + BROWSER_ONLY.add(code) def _auth_headers(platform): - """通用: 从平台配置生成认证头""" h = {'Content-Type': 'application/json'} if platform.get('access_token'): h['Authorization'] = 'Bearer ' + platform['access_token'] return h -# -- 微博 -- -def publish_weibo(platform, content): - if content['content_type'] == 'video': - return ('POST', 'https://api.weibo.com/2/statuses/upload_video.json', {}, None, 'video') - body = {'status': (content.get('title', '') + '\n' + content.get('body', ''))[:2000]} - if content.get('media_urls'): - urls = json.loads(content['media_urls']) - if urls: - body['pic_url'] = urls[0] - return ('POST', 'https://api.weibo.com/2/statuses/share.json', - _auth_headers(platform), body, 'article') -register('weibo', '微博', publish_weibo, ['article', 'video', 'image']) +# ═══════════════════════════════════════════ +# ✅ 真实 API — 可立即使用 +# ═══════════════════════════════════════════ -# -- 抖音 (开放平台) -- -def publish_douyin(platform, content): - headers = _auth_headers(platform) - if content['content_type'] == 'video': - body = { - 'video_url': json.loads(content.get('media_urls', '[]'))[0] if content.get('media_urls') else '', - 'title': content.get('title', '')[:100], - 'description': content.get('body', '')[:500], - } - return ('POST', platform.get('api_endpoint', 'https://open.douyin.com/video/create/'), - headers, body, 'video') - body = { - 'text': content.get('title', '')[:500], - 'image_list': json.loads(content.get('media_urls', '[]'))[:9], - } - return ('POST', platform.get('api_endpoint', 'https://open.douyin.com/image/create/'), - headers, body, 'image') - -register('douyin', '抖音', publish_douyin, ['video', 'image']) - -# -- B站 -- -def publish_bilibili(platform, content): - headers = _auth_headers(platform) - body = { - 'title': content.get('title', '')[:80], - 'desc': content.get('body', '')[:2000], - 'tag': content.get('tags', ''), - } - if content['content_type'] == 'video': - body['video_url'] = json.loads(content.get('media_urls', '[]'))[0] if content.get('media_urls') else '' - endpoint = 'https://member.bilibili.com/x/vu/web/add' - else: - body['content'] = content.get('body', '')[:5000] - endpoint = 'https://api.bilibili.com/x/dynamic/feed/create' - return ('POST', platform.get('api_endpoint', endpoint), headers, body, - content['content_type']) - -register('bilibili', 'B站', publish_bilibili, ['video', 'article', 'image']) - -# -- 快手 -- -def publish_kuaishou(platform, content): - headers = _auth_headers(platform) - body = {'caption': (content.get('title', '') + '\n' + content.get('body', ''))[:500]} - if content['content_type'] == 'video' and content.get('media_urls'): - body['video_url'] = json.loads(content['media_urls'])[0] - return ('POST', platform.get('api_endpoint', 'https://open.kuaishou.com/photo/publish'), - headers, body, content['content_type']) - -register('kuaishou', '快手', publish_kuaishou, ['video', 'image']) - -# -- 微信公众号 -- -def publish_wechat(platform, content): - headers = _auth_headers(platform) - body = { - 'articles': [{ - 'title': content.get('title', '')[:64], - 'content': content.get('body', '')[:50000], - 'cover_url': content.get('cover_url', ''), - }] - } - return ('POST', platform.get('api_endpoint', 'https://api.weixin.qq.com/cgi-bin/draft/add'), - headers, body, 'article') - -register('wechat', '微信公众号', publish_wechat, ['article']) - -# -- 小红书 -- -def publish_xiaohongshu(platform, content): - headers = _auth_headers(platform) - body = { - 'title': content.get('title', '')[:20], - 'content': content.get('body', '')[:1000], - 'images': json.loads(content.get('media_urls', '[]'))[:9], - } - return ('POST', platform.get('api_endpoint', 'https://open-api.xiaohongshu.com/note/publish'), - headers, body, content['content_type']) - -register('xiaohongshu', '小红书', publish_xiaohongshu, ['article', 'image', 'video']) - -# -- 头条号 -- -def publish_toutiao(platform, content): - headers = _auth_headers(platform) - body = { - 'title': content.get('title', '')[:30], - 'content': content.get('body', '')[:20000], - } - if content['content_type'] == 'video' and content.get('media_urls'): - body['video_url'] = json.loads(content['media_urls'])[0] - return ('POST', platform.get('api_endpoint', 'https://open.toutiao.com/content/publish'), - headers, body, content['content_type']) - -register('toutiao', '头条号', publish_toutiao, ['article', 'video']) - -# -- YouTube -- +# -- YouTube (Data API v3, 免费可用) -- def publish_youtube(platform, content): headers = _auth_headers(platform) body = { @@ -142,9 +49,43 @@ def publish_youtube(platform, content): return ('POST', platform.get('api_endpoint', 'https://www.googleapis.com/upload/youtube/v3/videos'), headers, body, 'video') -register('youtube', 'YouTube', publish_youtube, ['video']) +register('youtube', 'YouTube', publish_youtube, ['video'], 'real') -# -- Twitter/X -- +# -- 微信公众号 (developers.weixin.qq.com, 免费可用) -- +def publish_wechat(platform, content): + headers = _auth_headers(platform) + body = { + 'articles': [{ + 'title': content.get('title', '')[:64], + 'content': content.get('body', '')[:50000], + 'cover_url': content.get('cover_url', ''), + }] + } + return ('POST', platform.get('api_endpoint', 'https://api.weixin.qq.com/cgi-bin/draft/add'), + headers, body, 'article') + +register('wechat', '微信公众号', publish_wechat, ['article'], 'real') + + +# ═══════════════════════════════════════════ +# ⚠️ 受限 API — 需应用审核 或 付费才能用 +# ═══════════════════════════════════════════ + +# -- 微博 (Open API 存在，但需应用审核 + 用户 OAuth 授权) -- +def publish_weibo(platform, content): + if content['content_type'] == 'video': + return ('POST', 'https://api.weibo.com/2/statuses/upload_video.json', {}, None, 'video') + body = {'status': (content.get('title', '') + '\n' + content.get('body', ''))[:2000]} + if content.get('media_urls'): + urls = json.loads(content['media_urls']) + if urls: + body['pic_url'] = urls[0] + return ('POST', 'https://api.weibo.com/2/statuses/share.json', + _auth_headers(platform), body, 'article') + +register('weibo', '微博', publish_weibo, ['article', 'video', 'image'], 'restricted') + +# -- Twitter/X (API v2 发布需 Basic 及以上付费 tier, $100/月) -- def publish_twitter(platform, content): headers = _auth_headers(platform) body = {'text': (content.get('title', '') + '\n' + content.get('body', ''))[:280]} @@ -152,22 +93,29 @@ def publish_twitter(platform, content): body['media'] = {'media_ids': json.loads(content['media_urls'])[:4]} return ('POST', 'https://api.twitter.com/2/tweets', headers, body, 'article') -register('twitter', 'Twitter/X', publish_twitter, ['article', 'image', 'video']) +register('twitter', 'Twitter/X', publish_twitter, ['article', 'image', 'video'], 'restricted') -# -- 微信视频号 -- -def publish_channels(platform, content): - headers = _auth_headers(platform) - headers['Content-Type'] = 'application/json' - body = { - 'title': content.get('title', '')[:50], - 'description': content.get('body', '')[:500], - 'cover_url': content.get('cover_url', ''), - } - if content['content_type'] == 'video' and content.get('media_urls'): - body['video_url'] = json.loads(content['media_urls'])[0] - elif content.get('media_urls'): - body['image_urls'] = json.loads(content['media_urls'])[:9] - return ('POST', platform.get('api_endpoint', 'https://channels.weixin.qq.com/api/post/create'), - headers, body, content['content_type']) -register('channels', '微信视频号', publish_channels, ['video', 'image']) +# ═══════════════════════════════════════════ +# ❌ 无 API — 只能浏览器自动化 (Playwright / Selenium) +# 这些平台没有第三方发布 API，只能模拟浏览器操作。 +# 可通过 Sage 的 uapi gateway 或独立 Playwright 服务实现。 +# 此处提供的是期望的请求格式，实际发送时需经 browser automation 中间层。 +# ═══════════════════════════════════════════ + +def _browser_only(platform_code): + """浏览器自动化占位: 返回 (False, '需要浏览器自动化')""" + def _publish(platform, content): + raise NotImplementedError( + f'{platform_code} 没有公开发布 API。' + f'需通过 Playwright/Selenium 模拟浏览器操作，' + f'或通过第三方自动化服务实现。' + ) + return _publish + +register('douyin', '抖音', _browser_only('douyin'), ['video', 'image'], 'none') +register('bilibili', 'B站', _browser_only('bilibili'), ['video', 'article', 'image'], 'none') +register('kuaishou', '快手', _browser_only('kuaishou'), ['video', 'image'], 'none') +register('xiaohongshu', '小红书', _browser_only('xiaohongshu'), ['article', 'image', 'video'], 'none') +register('toutiao', '头条号', _browser_only('toutiao'), ['article', 'video'], 'none') +register('channels', '微信视频号', _browser_only('channels'), ['video', 'image'], 'none')