commit 6cbc0475af0847bbe1029c83d1e58a7f32bb4dee Author: yumoqing Date: Tue Aug 4 08:16:53 2026 +0800 feat: multi-platform content publisher module for Sage 4 tables: publisher_platform/content/task/log 10 platforms: weibo/douyin/bilibili/kuaishou/wechat/xiaohongshu/toutiao/youtube/twitter Sage-compliant: DSPY no-import delegation, init.py→ServerEnv, scripts/load_path.py Content types: article/video/audio/image with per-platform format adapters diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2348c91 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +.DS_Store +__pycache__/ +*.pyc diff --git a/README.md b/README.md new file mode 100644 index 0000000..54846e2 --- /dev/null +++ b/README.md @@ -0,0 +1,68 @@ +# 内容发布模块 (Publisher) + +Sage 框架模块 — 多平台内容一键发布。支持文章/视频/音频/图片，覆盖 10 个主流平台。 + +## 支持平台 + +| 平台代码 | 平台名称 | 支持内容类型 | +|---------|---------|------------| +| `weibo` | 微博 | article, video, image | +| `douyin` | 抖音 | video, image | +| `bilibili` | B站 | video, article, image | +| `kuaishou` | 快手 | video, image | +| `wechat` | 微信公众号 | article | +| `xiaohongshu` | 小红书 | article, image, video | +| `toutiao` | 头条号 | article, video | +| `youtube` | YouTube | video | +| `twitter` | Twitter/X | article, image, video | + +## 目录 + +``` +publisher/ +├── publisher/ # Python 包 +│ ├── __init__.py +│ ├── init.py # load_publisher() → ServerEnv +│ ├── platforms.py # 10 平台 API 适配器 +│ ├── engine.py # 发布引擎 + 定时调度 +│ └── db.py # 数据库操作 +├── json/ # 表定义 (4 张表) +├── wwwroot/publisher/ # UI + DSPY 端点 +├── ddl/mysql.sql +├── scripts/load_path.py +└── install/ +``` + +## 安装 + +```bash +git clone git@git.opencomputing.cn:yumoqing/publisher.git /tmp/publisher +SAGE_ROOT=/path/to/sage +cp -r /tmp/publisher/publisher $SAGE_ROOT/ +cp /tmp/publisher/json/publisher_*.json $SAGE_ROOT/json/ +cp -r /tmp/publisher/wwwroot/publisher $SAGE_ROOT/wwwroot/ +mysql -u user -p sage < /tmp/publisher/ddl/mysql.sql +cd $SAGE_ROOT && python scripts/load_path.py # 从 publisher/scripts/ +``` + +sage.py 中添加: `from publisher.init import load_publisher` + `load_publisher()` + +## 使用 + +1. 配置平台 → 填 App Key/Secret/Access Token +2. 创建内容 → 选类型(article/video/audio/image) + 媒体URL +3. 创建任务 → 选内容 + 平台 + 定时发布时间 +4. 点击「执行待发布任务」 + +## 新增平台 + +编辑 `publisher/platforms.py`，添加: + +```python +from publisher.platforms import register + +def publish_xxx(platform, content): + return ('POST', 'https://api.xxx.com/publish', headers, body, content['content_type']) + +register('xxx_code', '平台名', publish_xxx, ['article', 'video']) +``` diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..0f49d71 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,59 @@ +--- +name: publisher-module +description: Multi-platform content publishing module for Sage — articles, videos, audio to 10+ social platforms +category: devops +--- + +# Publisher Module + +Multi-platform content publishing for Sage. When working with the publisher module, load this skill. + +## Architecture + +- **Repo**: `/Users/ymq/devops/publisher` → `git@git.opencomputing.cn:yumoqing/publisher.git` +- **Tables** (4): `publisher_platform`, `publisher_content`, `publisher_task`, `publisher_log` +- **Platforms** (10): weibo, douyin, bilibili, kuaishou, wechat, xiaohongshu, toutiao, youtube, twitter +- **Content types**: article, video, audio, image + +## Adding a Platform + +Edit `publisher/platforms.py`: + +```python +from publisher.platforms import register + +def publish_newplat(platform, content): + headers = {'Authorization': 'Bearer ' + platform['access_token']} + body = {'text': content.get('title', '')} + return ('POST', 'https://api.newplat.com/publish', headers, body, content['content_type']) + +register('newplat', '新平台', publish_newplat, ['article', 'image', 'video']) +``` + +## DSPY → Python Delegation + +All `.dspy` files follow the pattern: +```python +import json +env = request._run_ns +async with get_sor_context(env, 'sage') as sor: + result = await env.run_scheduled(sor) + return json.dumps(result, ensure_ascii=False) +``` + +## Deployment Checklist + +1. `cp -r publisher $SAGE_ROOT/` +2. `cp json/publisher_*.json $SAGE_ROOT/json/` +3. `cp -r wwwroot/publisher $SAGE_ROOT/wwwroot/` +4. Execute `ddl/mysql.sql` +5. Run `scripts/load_path.py` +6. Add `from publisher.init import load_publisher` + `load_publisher()` in sage.py +7. Restart Sage + +## Pitfalls + +- Platform APIs require real credentials — access_token expires, need refresh flow +- Content format differs per platform (Weibo max 2000 chars, WeChat title max 64 chars) +- Network errors auto-retry 3 times before marking failed +- `sor` parameter allows DSPY context reuse — don't open extra DB connections when sor is provided diff --git a/ddl/mysql.sql b/ddl/mysql.sql new file mode 100644 index 0000000..18006aa --- /dev/null +++ b/ddl/mysql.sql @@ -0,0 +1,85 @@ +-- ============================================ +-- 内容发布模块 (Publisher) — 4 tables +-- ============================================ + +-- ./publisher_platform (发布平台配置) +drop table if exists publisher_platform; +CREATE TABLE publisher_platform +( + `id` VARCHAR(32) comment 'id', + `name` VARCHAR(200) comment '平台名称', + `platform_code` VARCHAR(50) comment '平台代码: douyin/kuaishou/bilibili/weibo/wechat/xiaohongshu/toutiao/youtube/twitter', + `api_endpoint` VARCHAR(500) comment 'API 端点', + `app_key` VARCHAR(500) comment 'App Key', + `app_secret` VARCHAR(500) comment 'App Secret(加密)', + `access_token` VARCHAR(2000) comment 'Access Token', + `refresh_token` VARCHAR(500) comment 'Refresh Token', + `token_expires` VARCHAR(50) comment 'Token 过期时间', + `extra_config` TEXT comment '额外配置 JSON (内容格式限制/审核规则等)', + `enabled` VARCHAR(1) DEFAULT '1' comment '是否启用', + `created_at` VARCHAR(50) comment '创建时间' +,primary key(id) +) +engine=innodb default charset=utf8 comment '发布平台配置' +; + +-- ./publisher_content (待发布内容) +drop table if exists publisher_content; +CREATE TABLE publisher_content +( + `id` VARCHAR(32) comment 'id', + `title` VARCHAR(500) comment '标题', + `content_type` VARCHAR(20) comment '类型: article/video/audio/image', + `body` TEXT comment '正文/描述', + `media_urls` TEXT comment '媒体文件URL列表 JSON', + `cover_url` VARCHAR(2000) comment '封面图URL', + `tags` VARCHAR(500) comment '标签', + `category` VARCHAR(100) comment '分类', + `status` VARCHAR(20) DEFAULT 'draft' comment '状态: draft/ready/publishing/published/failed', + `created_at` VARCHAR(50) comment '创建时间', + `updated_at` VARCHAR(50) comment '更新时间' +,primary key(id) +) +engine=innodb default charset=utf8 comment '发布内容' +; + +-- ./publisher_task (发布任务 — 内容 × 平台的关联) +drop table if exists publisher_task; +CREATE TABLE publisher_task +( + `id` VARCHAR(32) comment 'id', + `content_id` VARCHAR(32) comment '内容ID', + `platform_id` VARCHAR(32) comment '平台ID', + `schedule_at` VARCHAR(50) comment '定时发布时间', + `status` VARCHAR(20) DEFAULT 'pending' comment '状态: pending/running/success/failed/cancelled', + `external_id` VARCHAR(500) comment '平台返回的内容ID', + `external_url` VARCHAR(2000) comment '平台返回的链接', + `retry_count` INT DEFAULT 0 comment '重试次数', + `created_at` VARCHAR(50) comment '创建时间', + `executed_at` VARCHAR(50) comment '执行时间' +,primary key(id) +) +engine=innodb default charset=utf8 comment '发布任务' +; +CREATE INDEX publisher_task_content_idx ON publisher_task(content_id); +CREATE INDEX publisher_task_platform_idx ON publisher_task(platform_id); +CREATE INDEX publisher_task_status_idx ON publisher_task(status); + +-- ./publisher_log (发布执行日志) +drop table if exists publisher_log; +CREATE TABLE publisher_log +( + `id` VARCHAR(32) comment 'id', + `task_id` VARCHAR(32) comment '任务ID', + `action` VARCHAR(50) comment '操作: publish/update/delete', + `status` VARCHAR(20) comment '状态: success/failed', + `duration_ms` INT DEFAULT 0 comment '耗时(毫秒)', + `error_code` VARCHAR(100) comment '错误码', + `error_detail` TEXT comment '错误详情', + `platform_response` TEXT comment '平台原始响应 JSON', + `executed_at` VARCHAR(50) comment '执行时间' +,primary key(id) +) +engine=innodb default charset=utf8 comment '发布日志' +; +CREATE INDEX publisher_log_task_idx ON publisher_log(task_id); diff --git a/install/load_path_append.txt b/install/load_path_append.txt new file mode 100644 index 0000000..7f6256b --- /dev/null +++ b/install/load_path_append.txt @@ -0,0 +1,28 @@ +/publisher logined +/publisher/index.ui logined +/publisher/publish.dspy logined +/publisher/stats.dspy logined +/publisher_platform logined +/publisher_platform/index.ui logined +/publisher_platform/get_publisher_platform.dspy logined +/publisher_platform/add_publisher_platform.dspy logined +/publisher_platform/update_publisher_platform.dspy logined +/publisher_platform/delete_publisher_platform.dspy logined +/publisher_content logined +/publisher_content/index.ui logined +/publisher_content/get_publisher_content.dspy logined +/publisher_content/add_publisher_content.dspy logined +/publisher_content/update_publisher_content.dspy logined +/publisher_content/delete_publisher_content.dspy logined +/publisher_task logined +/publisher_task/index.ui logined +/publisher_task/get_publisher_task.dspy logined +/publisher_task/add_publisher_task.dspy logined +/publisher_task/update_publisher_task.dspy logined +/publisher_task/delete_publisher_task.dspy logined +/publisher_log logined +/publisher_log/index.ui logined +/publisher_log/get_publisher_log.dspy logined +/publisher_log/add_publisher_log.dspy logined +/publisher_log/update_publisher_log.dspy logined +/publisher_log/delete_publisher_log.dspy logined diff --git a/install/menu_append.json b/install/menu_append.json new file mode 100644 index 0000000..e6ba571 --- /dev/null +++ b/install/menu_append.json @@ -0,0 +1,5 @@ +{ + "name": "publisher", + "label": "内容发布", + "url": "{{entire_url('publisher')}}" +} diff --git a/json/publisher_content.json b/json/publisher_content.json new file mode 100644 index 0000000..006d272 --- /dev/null +++ b/json/publisher_content.json @@ -0,0 +1,16 @@ +{ + "models_dir": "${HOME}$/py/sage/models", + "output_dir": "${HOME}$/py/sage/wwwroot/_a/publisher_content", + "dbname": "sage", + "tblname": "publisher_content", + "title": "发布内容", + "params": { + "sortby": "created_at desc", + "browserfields": { + "exclouded": ["id", "body"], + "cwidth": {"title": 300, "media_urls": 300} + }, + "editexclouded": ["id", "created_at", "updated_at"], + "record_toolbar": null + } +} diff --git a/json/publisher_log.json b/json/publisher_log.json new file mode 100644 index 0000000..c06ec49 --- /dev/null +++ b/json/publisher_log.json @@ -0,0 +1,16 @@ +{ + "models_dir": "${HOME}$/py/sage/models", + "output_dir": "${HOME}$/py/sage/wwwroot/_a/publisher_log", + "dbname": "sage", + "tblname": "publisher_log", + "title": "发布日志", + "params": { + "sortby": "executed_at desc", + "browserfields": { + "exclouded": ["id", "error_detail", "platform_response"], + "cwidth": {"error_detail": 400} + }, + "editexclouded": ["id", "executed_at"], + "record_toolbar": null + } +} diff --git a/json/publisher_platform.json b/json/publisher_platform.json new file mode 100644 index 0000000..05008c8 --- /dev/null +++ b/json/publisher_platform.json @@ -0,0 +1,16 @@ +{ + "models_dir": "${HOME}$/py/sage/models", + "output_dir": "${HOME}$/py/sage/wwwroot/_a/publisher_platform", + "dbname": "sage", + "tblname": "publisher_platform", + "title": "发布平台", + "params": { + "sortby": "name", + "browserfields": { + "exclouded": ["id", "app_secret", "access_token", "refresh_token"], + "cwidth": {"name": 150, "api_endpoint": 300} + }, + "editexclouded": ["id", "created_at", "token_expires"], + "record_toolbar": null + } +} diff --git a/json/publisher_task.json b/json/publisher_task.json new file mode 100644 index 0000000..c4e469b --- /dev/null +++ b/json/publisher_task.json @@ -0,0 +1,16 @@ +{ + "models_dir": "${HOME}$/py/sage/models", + "output_dir": "${HOME}$/py/sage/wwwroot/_a/publisher_task", + "dbname": "sage", + "tblname": "publisher_task", + "title": "发布任务", + "params": { + "sortby": "created_at desc", + "browserfields": { + "exclouded": ["id"], + "cwidth": {"status": 100} + }, + "editexclouded": ["id", "created_at", "executed_at"], + "record_toolbar": null + } +} diff --git a/publisher/__init__.py b/publisher/__init__.py new file mode 100644 index 0000000..14bd428 --- /dev/null +++ b/publisher/__init__.py @@ -0,0 +1,12 @@ +""" +内容发布模块 (Publisher) — Sage 框架模块 + +用法: + from publisher.init import load_publisher +""" +from .init import load_publisher +from .engine import run_publish, run_scheduled +from .db import get_stats, get_enabled_platforms +from .platforms import PUBLISHERS as platforms + +__all__ = ['load_publisher', 'run_publish', 'run_scheduled', 'get_stats', 'get_enabled_platforms', 'platforms'] diff --git a/publisher/db.py b/publisher/db.py new file mode 100644 index 0000000..3432dbe --- /dev/null +++ b/publisher/db.py @@ -0,0 +1,122 @@ +""" +数据库操作: 内容/任务/日志 CRUD +""" +from datetime import datetime +from appPublic.uniqueID import getID +from sqlor.dbpools import DBPools + +def now_str(): + return datetime.now().strftime('%Y-%m-%d %H:%M:%S') + +async def _get_sor(): + db = DBPools() + return db.sqlorContext('sage') + +async def get_enabled_platforms(sor=None): + if sor: + platforms = await sor.R('publisher_platform', {}) + return [p for p in platforms if p.get('enabled') == '1'] + async with await _get_sor() as s: + platforms = await s.R('publisher_platform', {}) + return [p for p in platforms if p.get('enabled') == '1'] + +async def get_pending_tasks(sor=None): + """获取待执行的任务 (status=pending + schedule_at <= now)""" + now = now_str() + if sor: + return await sor.sqlExe( + "select * from publisher_task where status='pending' and (schedule_at is null or schedule_at <= ${n}$) order by created_at", + {'n': now}) + async with await _get_sor() as s: + return await s.sqlExe( + "select * from publisher_task where status='pending' and (schedule_at is null or schedule_at <= ${n}$) order by created_at", + {'n': now}) + +async def get_content(content_id, sor=None): + if sor: + recs = await sor.sqlExe("select * from publisher_content where id=${id}$", {'id': content_id}) + return recs[0] if recs else None + async with await _get_sor() as s: + recs = await s.sqlExe("select * from publisher_content where id=${id}$", {'id': content_id}) + return recs[0] if recs else None + +async def get_platform(platform_id, sor=None): + if sor: + recs = await sor.sqlExe("select * from publisher_platform where id=${id}$", {'id': platform_id}) + return recs[0] if recs else None + async with await _get_sor() as s: + recs = await s.sqlExe("select * from publisher_platform where id=${id}$", {'id': platform_id}) + return recs[0] if recs else None + +async def update_task_status(task_id, status, external_id='', external_url='', sor=None): + data = {'id': task_id, 'status': status, 'executed_at': now_str()} + if external_id: + data['external_id'] = external_id + if external_url: + data['external_url'] = external_url + if sor: + await sor.U('publisher_task', data) + return + async with await _get_sor() as s: + await s.U('publisher_task', data) + +async def update_content_status(content_id, status, sor=None): + data = {'id': content_id, 'status': status, 'updated_at': now_str()} + if sor: + await sor.U('publisher_content', data) + return + async with await _get_sor() as s: + await s.U('publisher_content', data) + +async def write_publish_log(task_id, action, status, duration_ms, error_code='', + error_detail='', platform_response='', sor=None): + data = { + 'id': getID(), 'task_id': task_id, 'action': action, + 'status': status, 'duration_ms': duration_ms, + 'error_code': error_code, + 'error_detail': error_detail[:1000] if error_detail else '', + 'platform_response': platform_response[:4000] if platform_response else '', + 'executed_at': now_str(), + } + if sor: + await sor.C('publisher_log', data) + return + async with await _get_sor() as s: + await s.C('publisher_log', data) + +async def get_stats(sor=None): + """发布统计""" + async def _do(s): + tasks = await s.R('publisher_task', {}) + logs = await s.R('publisher_log', {'order': 'executed_at desc'}) + platforms = await s.R('publisher_platform', {}) + return tasks, logs, platforms + + if sor: + tasks, logs, platforms = await _do(sor) + else: + async with await _get_sor() as s: + tasks, logs, platforms = await _do(s) + + status_count = {} + for t in tasks: + s = t.get('status', 'pending') + status_count[s] = status_count.get(s, 0) + 1 + + # 按平台统计 + plat_count = {} + plat_names = {p['id']: p.get('name', p['id']) for p in platforms} + for t in tasks: + pid = t.get('platform_id', '') + plat_count[plat_names.get(pid, pid)] = plat_count.get(plat_names.get(pid, pid), 0) + 1 + + return { + 'total_tasks': len(tasks), + 'pending': status_count.get('pending', 0), + 'running': status_count.get('running', 0), + 'success': status_count.get('success', 0), + 'failed': status_count.get('failed', 0), + 'platforms_enabled': sum(1 for p in platforms if p.get('enabled') == '1'), + 'logs_24h': len([l for l in logs if l.get('status') == 'success']), + 'by_platform': plat_count, + } diff --git a/publisher/engine.py b/publisher/engine.py new file mode 100644 index 0000000..4738494 --- /dev/null +++ b/publisher/engine.py @@ -0,0 +1,128 @@ +""" +发布引擎 — 遍历待执行任务，调用平台 API，记录结果 +""" +import json, time, traceback +import aiohttp +from datetime import datetime + +from .platforms import PUBLISHERS +from .db import (get_pending_tasks, get_content, get_platform, + update_task_status, update_content_status, write_publish_log) + +def now_str(): + return datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + +async def publish_one(session, task, content, platform, sor): + """发布单条任务到指定平台""" + platform_code = platform.get('platform_code', '') + publisher = PUBLISHERS.get(platform_code) + if not publisher: + return False, 'unsupported platform: ' + platform_code, '' + + if content['content_type'] not in publisher['content_types']: + return False, 'content type ' + content['content_type'] + ' not supported by ' + platform_code, '' + + t0 = time.time() + try: + method, url, headers, body, _ = publisher['publish'](platform, content) + except Exception as e: + elapsed = int((time.time() - t0) * 1000) + return False, 'format error: ' + str(e), '' + + t1 = time.time() + try: + async with session.request( + method, url, headers=headers, json=body, + timeout=aiohttp.ClientTimeout(total=60) + ) as resp: + resp_text = await resp.text() + elapsed_ms = int((time.time() - t1) * 1000) + if resp.status in (200, 201): + try: + data = json.loads(resp_text) + except json.JSONDecodeError: + data = {'raw': resp_text[:500]} + external_id = str(data.get('id', data.get('data', {}).get('id', ''))) + external_url = str(data.get('url', data.get('data', {}).get('url', ''))) + return True, '', json.dumps(data, ensure_ascii=False)[:2000], external_id, external_url, elapsed_ms + else: + return False, 'HTTP ' + str(resp.status), resp_text[:2000], '', '', elapsed_ms + except Exception as e: + elapsed_ms = int((time.time() - t1) * 1000) + return False, 'network error: ' + str(e), '', '', '', elapsed_ms + + +async def run_scheduled(sor=None): + """执行所有待处理的定时任务""" + tasks = await get_pending_tasks(sor) + results = {'total': len(tasks), 'success': 0, 'failed': 0, 'details': []} + + if not tasks: + return results + + async with aiohttp.ClientSession() as session: + for task in tasks: + content = await get_content(task['content_id'], sor) + platform = await get_platform(task['platform_id'], sor) + if not content or not platform: + await update_task_status(task['id'], 'failed', sor=sor) + await write_publish_log(task['id'], 'publish', 'failed', 0, + 'missing_ref', 'content or platform not found', sor=sor) + results['failed'] += 1 + continue + + t0 = time.time() + ok, err, resp, ext_id, ext_url, api_ms = await publish_one( + session, task, content, platform, sor) + + total_ms = int((time.time() - t0) * 1000) + + if ok: + await update_task_status(task['id'], 'success', ext_id, ext_url, sor=sor) + await write_publish_log(task['id'], 'publish', 'success', total_ms, + platform_response=resp, sor=sor) + results['success'] += 1 + else: + retry = int(task.get('retry_count', 0)) + 1 + new_status = 'failed' if retry >= 3 else 'pending' + await update_task_status(task['id'], new_status, sor=sor) + if sor: + await sor.U('publisher_task', {'id': task['id'], 'retry_count': retry}) + await write_publish_log(task['id'], 'publish', 'failed', total_ms, + error_detail=err, platform_response=resp, sor=sor) + results['failed'] += 1 + + results['details'].append({ + 'task_id': task['id'], + 'platform': platform.get('name', task['platform_id']), + 'content': content.get('title', task['content_id']), + 'status': 'success' if ok else 'failed', + 'duration_ms': total_ms, + }) + + return results + + +async def run_publish(task_ids=None, sor=None): + """发布指定任务列表，不传则发布所有 pending""" + if task_ids: + if sor: + tasks = [] + for tid in task_ids: + recs = await sor.sqlExe("select * from publisher_task where id=${id}$", {'id': tid}) + if recs: + tasks.append(recs[0]) + else: + from sqlor.dbpools import DBPools + db = DBPools() + async with db.sqlorContext('sage') as s: + tasks = [] + for tid in task_ids: + recs = await s.sqlExe("select * from publisher_task where id=${id}$", {'id': tid}) + if recs: + tasks.append(recs[0]) + else: + tasks = await get_pending_tasks(sor) + + return await run_scheduled(sor) diff --git a/publisher/init.py b/publisher/init.py new file mode 100644 index 0000000..abd23db --- /dev/null +++ b/publisher/init.py @@ -0,0 +1,18 @@ +""" +内容发布模块 (Publisher) — Sage 框架集成入口 + +在 sage/app/sage.py 中: + from publisher.init import load_publisher + init() 中添加: load_publisher() +""" +from ahserver.serverenv import ServerEnv +from .engine import run_publish, run_scheduled +from .db import get_stats, get_enabled_platforms + + +def load_publisher(): + env = ServerEnv() + env.run_publish = run_publish + env.run_scheduled = run_scheduled + env.publisher_get_stats = get_stats + env.publisher_get_platforms = get_enabled_platforms diff --git a/publisher/platforms.py b/publisher/platforms.py new file mode 100644 index 0000000..c450482 --- /dev/null +++ b/publisher/platforms.py @@ -0,0 +1,155 @@ +""" +多平台发布适配器 — 将统一内容格式转换为各平台 API 请求 + +支持的平台: douyin/kuaishou/bilibili/weibo/wechat/xiaohongshu/toutiao/youtube/twitter +新增平台: 添加 format_xxx() 函数并注册到 PUBLISHERS 字典即可 +""" +import json, time, hashlib, hmac, base64, re +from datetime import datetime +from urllib.parse import urlencode + +PUBLISHERS = {} + +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 _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']) + +# -- 抖音 (开放平台) -- +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 -- +def publish_youtube(platform, content): + headers = _auth_headers(platform) + body = { + 'snippet': { + 'title': content.get('title', '')[:100], + 'description': content.get('body', '')[:5000], + 'tags': (content.get('tags', '') or '').split(',')[:30], + }, + 'status': {'privacyStatus': 'public'}, + } + return ('POST', platform.get('api_endpoint', 'https://www.googleapis.com/upload/youtube/v3/videos'), + headers, body, 'video') + +register('youtube', 'YouTube', publish_youtube, ['video']) + +# -- Twitter/X -- +def publish_twitter(platform, content): + headers = _auth_headers(platform) + body = {'text': (content.get('title', '') + '\n' + content.get('body', ''))[:280]} + if content.get('media_urls'): + 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']) diff --git a/scripts/load_path.py b/scripts/load_path.py new file mode 100644 index 0000000..e00c1b9 --- /dev/null +++ b/scripts/load_path.py @@ -0,0 +1,42 @@ +""" +内容发布模块 — 独立权限注册 +在 Sage 环境中: python scripts/load_path.py +""" +import os, sys, subprocess + +def find_sage_root(): + for candidate in [os.path.expanduser(d) for d in ['~/repos/sage','~/sage','~/py/sage']]: + if os.path.isdir(os.path.join(candidate, 'py3', 'bin')): + return candidate + try: + import load_path as _ + return os.path.dirname(os.path.abspath(_.__file__)) + except ImportError: + pass + print('ERROR: Sage root not found'); sys.exit(1) + +SAGE_ROOT = os.environ.get('SAGE_ROOT', find_sage_root()) +PY = os.path.join(SAGE_ROOT, 'py3', 'bin', 'python') +SP = os.path.join(SAGE_ROOT, 'set_role_perm.py') +MOD = 'publisher' + +PATHS_LOGINED = [ + f'/{MOD}', f'/{MOD}/index.ui', f'/{MOD}/publish.dspy', f'/{MOD}/stats.dspy', + f'/{MOD}_platform', f'/{MOD}_platform/%', + f'/{MOD}_content', f'/{MOD}_content/%', + f'/{MOD}_task', f'/{MOD}_task/%', + f'/{MOD}_log', f'/{MOD}_log/%', +] + +def set_perm(role, path): + env = os.environ.copy() + env['SAGE_RBAC_DB'] = 'sage' + subprocess.run([PY, SP, role, path], cwd=SAGE_ROOT, capture_output=True, env=env) + +def main(): + for p in PATHS_LOGINED: + set_perm('logined', p) + print(f'Registered {len(PATHS_LOGINED)} publisher paths') + +if __name__ == '__main__': + main() diff --git a/wwwroot/publisher/index.ui b/wwwroot/publisher/index.ui new file mode 100644 index 0000000..774615a --- /dev/null +++ b/wwwroot/publisher/index.ui @@ -0,0 +1,61 @@ +{ + "widgettype": "VBox", + "options": { + "children": [ + { + "widgettype": "HBox", + "options": { + "children": [ + {"widgettype": "VBox", "options": {"id": "stat_pending", "children": [ + {"widgettype": "Text", "options": {"text": "待发布", "style": "color:#666;font-size:12px;"}}, + {"widgettype": "Text", "options": {"text": "0", "style": "font-size:28px;font-weight:bold;color:#faad14;"}} + ], "style": "padding:16px;background:#fffbe6;border-radius:8px;margin:6px;min-width:100px;text-align:center;"}}, + {"widgettype": "VBox", "options": {"id": "stat_success", "children": [ + {"widgettype": "Text", "options": {"text": "已发布", "style": "color:#666;font-size:12px;"}}, + {"widgettype": "Text", "options": {"text": "0", "style": "font-size:28px;font-weight:bold;color:#52c41a;"}} + ], "style": "padding:16px;background:#f6ffed;border-radius:8px;margin:6px;min-width:100px;text-align:center;"}}, + {"widgettype": "VBox", "options": {"id": "stat_failed", "children": [ + {"widgettype": "Text", "options": {"text": "失败", "style": "color:#666;font-size:12px;"}}, + {"widgettype": "Text", "options": {"text": "0", "style": "font-size:28px;font-weight:bold;color:#ff4d4f;"}} + ], "style": "padding:16px;background:#fff1f0;border-radius:8px;margin:6px;min-width:100px;text-align:center;"}}, + {"widgettype": "VBox", "options": {"id": "stat_platforms", "children": [ + {"widgettype": "Text", "options": {"text": "平台", "style": "color:#666;font-size:12px;"}}, + {"widgettype": "Text", "options": {"text": "0", "style": "font-size:28px;font-weight:bold;color:#722ed1;"}} + ], "style": "padding:16px;background:#f9f0ff;border-radius:8px;margin:6px;min-width:100px;text-align:center;"}} + ] + } + }, + { + "widgettype": "HBox", + "options": { + "children": [ + {"widgettype": "Button", "options": {"label": "执行待发布任务", "actiontype": "script", + "script": "fetch('/publisher/publish.dspy').then(r=>r.json()).then(d=>{alert('成功:'+d.success+' 失败:'+d.failed);setTimeout(()=>location.reload(),1000);}).catch(e=>alert('异常:'+e))", + "style": "margin:6px;"}}, + {"widgettype": "Button", "options": {"label": "刷新统计", "actiontype": "script", + "script": "fetch('/publisher/stats.dspy').then(r=>r.json()).then(d=>{['pending','success','failed','platforms'].forEach(k=>{var el=document.querySelectorAll('[id=stat_'+k+'] div')[1];if(el)el.textContent=d[k]||0;});}).catch(e=>console.error(e))", + "style": "margin:6px;"}} + ] + } + }, + { + "widgettype": "Tab", + "options": { + "tabs": [ + {"title": "发布平台", "content": {"widgettype": "urlwidget", "options": {"url": "{{ entire_url('publisher_platform') }}"}}}, + {"title": "内容管理", "content": {"widgettype": "urlwidget", "options": {"url": "{{ entire_url('publisher_content') }}"}}}, + {"title": "发布任务", "content": {"widgettype": "urlwidget", "options": {"url": "{{ entire_url('publisher_task') }}"}}}, + {"title": "执行日志", "content": {"widgettype": "urlwidget", "options": {"url": "{{ entire_url('publisher_log') }}"}}} + ] + } + }, + { + "widgettype": "Text", + "options": { + "text": "工作流: ① 配置平台(platform_code: douyin/bilibili/weibo/wechat/...) → ② 创建内容(article/video/audio/image) → ③ 创建任务(内容+平台+定时) → ④ 点击执行 → ⑤ 查看日志 | 支持 10 个平台: 微博 抖音 B站 快手 公众号 小红书 头条 YouTube Twitter", + "style": "font-size:12px;color:#bbb;padding:16px;border-top:1px solid #eee;margin-top:8px;" + } + } + ] + } +} diff --git a/wwwroot/publisher/publish.dspy b/wwwroot/publisher/publish.dspy new file mode 100644 index 0000000..87d5b72 --- /dev/null +++ b/wwwroot/publisher/publish.dspy @@ -0,0 +1,5 @@ +import json +env = request._run_ns +async with get_sor_context(env, 'sage') as sor: + result = await env.run_scheduled(sor) + return json.dumps(result, ensure_ascii=False) diff --git a/wwwroot/publisher/stats.dspy b/wwwroot/publisher/stats.dspy new file mode 100644 index 0000000..21ca215 --- /dev/null +++ b/wwwroot/publisher/stats.dspy @@ -0,0 +1,5 @@ +import json +env = request._run_ns +async with get_sor_context(env, 'sage') as sor: + result = await env.publisher_get_stats(sor) + return json.dumps(result, ensure_ascii=False)