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

.gitignore

@@ -0,0 +1,3 @@
+.DS_Store
+__pycache__/
+*.pyc

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'])
+```

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

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);

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

install/menu_append.json

@@ -0,0 +1,5 @@
+{
+    "name": "publisher",
+    "label": "内容发布",
+    "url": "{{entire_url('publisher')}}"
+}

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
+    }
+}

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
+    }
+}

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
+    }
+}

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
+    }
+}

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']

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

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)

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

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'])

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()

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;"
+                }
+            }
+        ]
+    }
+}

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)

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)