@clawhub-hnc87-7675bbc97d
自动化管理多平台电商店铺,实现选品扫描、商品上架、客服回复和订单处理全周期运营。
# solo-ecommerce-agent — 全平台全自动电商运营智能体
> 一人运营多平台店铺,无需人工干预。
> **关键词**:电商运营、自动上架、自动客服、自动发货、选品、订单处理
---
## 功能模块
| 模块 | 说明 | 自动化 |
|------|------|--------|
| 选品扫描 | 多平台热销榜单分析、机会品类推荐 | ✅ 每小时自动扫描 |
| 商品上架 | 商品信息生成、图文处理、一键发布 | ✅ 开启后自动上架 |
| 客服回复 | 买家咨询自动分类+回复,争议自动升级 | ✅ 每5分钟自动处理 |
| 订单处理 | 自动发货、物流录入、退款审核 | ✅ 每10分钟自动处理 |
| 日报汇总 | 每日23:00汇总运营数据发送通知 | ✅ 每日自动推送 |
---
## 快速开始
### 第一步:配置平台
编辑 `~/.qclaw/solo-ecommerce-data/config.json`,填入你的平台信息:
```json
{
"platform": "douyin", // douyin | taobao | pinduoduo | jingdong
"store_name": "你的店铺名",
"backend_url": "http://127.0.0.1:8080", // 平台API地址(如有)
"automation": {
"publish": { "enabled": true, "need_review": false },
"customer_service": { "enabled": true, "auto_reply": true },
"order": { "enabled": true, "auto_ship": true }
},
"enabled": true
}
```
### 第二步:开启定时任务
技能加载后,创建以下 cron 任务:
| 任务名 | 触发时间 | 作用 |
|--------|---------|------|
| 选品扫描 | 每小时整点 | 分析热销榜单,推荐机会品类 |
| 客服回复 | 每5分钟 | 自动处理买家消息 |
| 订单处理 | 每10分钟 | 自动发货+物流录入 |
| 日报汇总 | 每天23:00 | 推送当日运营数据 |
### 第三步:触发运营
**对话触发示例:**
- 「帮我扫描今天的选品机会」
- 「上架这款商品:[商品链接/信息]」
- 「检查今天的订单」
- 「生成今天运营日报」
---
## 脚本说明
| 脚本 | 功能 | 数据文件 |
|------|------|---------|
| `product_scanner.py` | 扫描热销榜单,生成推荐 | `recommendations.json` |
| `product_publisher.py` | 发布商品到店铺 | `products.json` |
| `customer_service.py` | 自动回复买家 | `customers.json` |
| `order_processor.py` | 处理订单+物流 | `orders.json` |
| `daily_report.py` | 汇总运营数据 | 日志文件 |
---
## 数据目录
`~/.qclaw/solo-ecommerce-data/`
```
solo-ecommerce-data/
├── config.json # 主配置文件(必填)
├── products.json # 商品列表
├── orders.json # 订单记录
├── customers.json # 客户对话记录
├── recommendations.json # 选品推荐
└── logs/
└── YYYY-MM-DD.log # 每日运行日志
```
---
## 平台接入说明
### 抖音小店
- 使用 Chrome CDP 浏览器自动化
- Chrome 需开启 `--remote-debugging-port=9222`
- 自动化流程:登录 → 商品管理 → 上架/客服/订单
### 其他平台
- 通过平台开放 API(需申请 AppKey/AppSecret)
- 或使用浏览器自动化模拟操作
- 具体接入方式根据平台文档配置
---
## 状态说明
- `enabled: false` → 技能休眠,所有定时任务跳过
- `enabled: true` → 全速运转
- 单个模块关闭 → 仅该模块跳过,其他模块继续
---
> 最后更新:2026-04-26
FILE:metadata.json
{
"name": "solo-ecommerce-agent",
"version": "1.0.0",
"description": "全平台全自动电商运营智能体,支持选品、上架、客服、订单全路线自动化",
"author": "HNC87",
"created": "2026-04-26",
"tags": ["ecommerce", "automation", "dropshipping", "customer-service", "order-management"],
"scripts": [
"product_scanner.py - 选品扫描",
"product_publisher.py - 商品上架",
"customer_service.py - 客服自动回复",
"order_processor.py - 订单处理",
"daily_report.py - 每日运营报告",
"init_agent.py - 初始化配置"
],
"data_dir": "solo-ecommerce-data",
"dependencies": {
"browser_automation": "xbrowser skill(浏览器自动化,必装)",
"python": ">=3.8"
}
}
FILE:README.md
# solo-ecommerce-agent
**全平台全自动电商运营智能体** — 一人运营多平台店铺,无需人工干预。
## 功能
| 模块 | 说明 | 自动化 |
|------|------|--------|
| 选品扫描 | 多平台热销榜单分析、机会品类推荐 | 每小时自动扫描 |
| 商品上架 | 商品信息生成、图文处理、一键发布 | 开启后自动上架 |
| 客服回复 | 买家咨询自动分类+回复,争议自动升级 | 每5分钟自动处理 |
| 订单处理 | 自动发货、物流录入、退款审核 | 每10分钟自动处理 |
| 日报汇总 | 每日23:00汇总运营数据推送通知 | 每日自动推送 |
## 支持平台
- 抖音小店(浏览器自动化)
- 淘宝/天猫(API 或浏览器自动化)
- 拼多多(API 或浏览器自动化)
- 京东(API 或浏览器自动化)
- 其他平台(可扩展)
## 安装
```bash
openclaw skill install solo-ecommerce-agent
```
## 配置
编辑 `~/.qclaw/solo-ecommerce-data/config.json`:
```json
{
"platform": "douyin",
"store_name": "你的店铺名",
"automation": {
"publish": { "enabled": true },
"customer_service": { "enabled": true },
"order": { "enabled": true }
},
"enabled": true
}
```
## 使用方式
### 对话触发(按需)
- 「帮我扫描今天的选品机会」
- 「上架这款商品:[商品信息]」
- 「检查今天的订单状态」
- 「生成今天运营日报」
### 全自动(定时)
配置 cron 任务后,智能体在后台自动运行:
- 每小时整点:选品扫描
- 每5分钟:客服回复
- 每10分钟:订单处理
- 每天23:00:运营日报
## 数据目录
`~/.qclaw/solo-ecommerce-data/`
```
solo-ecommerce-data/
├── config.json # 主配置文件
├── products.json # 商品列表
├── orders.json # 订单记录
├── customers.json # 客户对话
├── recommendations.json # 选品推荐
└── logs/ # 运行日志
```
## 系统要求
- Python 3.8+
- Chrome 浏览器(平台自动化用)
- OpenClaw 最新版
## License
MIT
FILE:references/platform_guides.md
# 平台操作指南
本文档描述如何通过浏览器自动化(xbrowser)操作各电商平台后台。
---
## 通用前提
1. **保持登录态**:提前在浏览器中登录店铺后台
2. **使用xbrowser**:通过OpenClaw的浏览器自动化功能操作
3. **稳定网络**:确保网络稳定,避免操作中断
---
## 淘宝/天猫(千牛后台)
### 后台地址
- 千牛工作台:https://myseller.taobao.com
### 商品发布流程
```
1. 打开:https://myseller.taobao.com
2. 导航:商品管理 → 发布商品
3. 填写:
- 选择类目(搜索框输入类目名)
- 商品标题(input[name="title"])
- 价格(input[type="price"])
- 库存(input[type="quantity"])
- 上传图片(input[type="file"])
- 详情页编辑器(iframe富文本)
4. 提交:点击"发布"按钮
```
### 客服回复流程
```
1. 打开:https://myseller.taobao.com
2. 导航:客服中心 → 待回复消息
3. 选择对话
4. 输入回复(textarea输入框)
5. 发送(点击"发送"按钮)
```
### 订单处理流程
```
1. 打开:https://myseller.taobao.com
2. 导航:交易管理 → 待发货
3. 点击"发货"
4. 选择物流公司
5. 输入快递单号
6. 确认发货
```
---
## 抖音小店(抖店后台)
### 后台地址
- 抖店工作台:https://fxg.jinritemai.com
### 商品发布流程
```
1. 打开:https://fxg.jinritemai.com
2. 导航:商品管理 → 商品创建
3. 填写:
- 商品标题
- 价格(售价)
- 库存
- 商品详情(详情页编辑器)
- 上传图片(主图/详情图)
4. 提交审核
```
### 客服回复流程
```
1. 打开:https://fxg.jinritemai.com
2. 导航:客服中心 → 飞鸽客服
3. 选择会话
4. 输入回复
5. 发送
```
### 订单处理流程
```
1. 打开:https://fxg.jinritemai.com
2. 导航:订单管理 → 待发货
3. 点击"发货"
4. 选择物流公司
5. 输入快递单号
6. 确认发货
```
---
## 拼多多(商家后台)
### 后台地址
- 拼多多商家后台:https://mms.pinduoduo.com
### 商品发布流程
```
1. 打开:https://mms.pinduoduo.com
2. 导航:商品管理 → 发布新商品
3. 填写:
- 商品标题
- 价格(拼团价)
- 库存
- 商品详情
- 上传图片
4. 提交审核
```
### 订单处理流程
```
1. 打开:https://mms.pinduoduo.com
2. 导航:订单查询 → 待发货
3. 点击"发货"
4. 录入物流信息
5. 确认发货
```
---
## 京东(京麦后台)
### 后台地址
- 京麦工作台:https://shop.jd.com
### 商品发布流程
```
1. 打开:https://shop.jd.com
2. 导航:商品管理 → 添加新商品
3. 填写商品信息
4. 提交审核
```
### 订单处理流程
```
1. 打开:https://shop.jd.com
2. 导航:订单管理 → 待发货
3. 处理发货
```
---
## 浏览器自动化注意事项
### 元素选择器
- 优先使用:`id`、`name`、`data-*` 属性
- 避免:动态生成的class名(可能变化)
- 兜底:XPath或CSS选择器
### 等待策略
- 使用 `snapshot` 确认页面加载完成
- 等待关键元素出现后再操作
- 处理弹窗/确认框
### 错误处理
- 网络超时:重试机制
- 登录失效:提醒用户重新登录
- 元素找不到:截图报错,转人工
### 反爬规避
- 模拟人工操作速度(不秒发)
- 随机间隔
- 避免高频操作同一页面
---
## API对接(如有密钥)
各平台开放API:
| 平台 | API地址 | 文档 |
|------|---------|------|
| 淘宝 | open.taobao.com | 淘宝开放平台 |
| 抖店 | developer.jinritemai.com | 抖店开放平台 |
| 拼多多 | open.pinduoduo.com | 多多进宝 |
| 京东 | open.jd.com | 京东宙斯 |
**API优势:**
- 稳定高效
- 无需浏览器
- 可批量操作
**API劣势:**
- 需申请权限
- 需配置密钥
- 有调用限制
---
## 快速验证
使用xbrowser测试后台操作:
```
用户:"打开淘宝店铺后台"
→ 启动浏览器
→ 导航到 myseller.taobao.com
→ 确认登录状态
→ 截图确认
```
测试完整流程后,再让智能体自动执行。
FILE:scripts/config_template.json
{
"version": "1.0.0",
"created": "2026-04-26",
"platforms": [],
"automation_scope": {
"auto_publish": false,
"auto_pricing": false,
"auto_reply": true,
"auto_ship": false,
"complaint_to_manual": true,
"refund_threshold": 50
},
"risk_controls": {
"publish_needs_review": true,
"ship_needs_confirm": true,
"complaint_auto_escalate": true,
"logistics_alert_days": 3
},
"schedules": {
"product_scan": "0 * * * *",
"customer_service": "*/5 * * * *",
"order_process": "*/10 * * * *",
"daily_report": "0 23 * * *"
}
}
FILE:scripts/customer_service.py
# -*- coding: utf-8 -*-
"""客服引擎 - 自动回复买家咨询"""
import json, os, sys, time
from pathlib import Path
DATA_DIR = Path.home() / ".qclaw" / "solo-ecommerce-data"
CONFIG_FILE = DATA_DIR / "config.json"
CUST_FILE = DATA_DIR / "customers.json"
LOG_FILE = DATA_DIR / "logs" / f"{time.strftime('%Y-%m-%d')}.log"
REPLIES = {
"咨询": "感谢您的咨询!小店商品均为正品,支持7天无理由退换,有任何问题随时联系客服~",
"尺码": "您好!尺码对照表已更新在商品详情页,建议您根据平时尺码结合详情页推荐选择~",
"库存": "您好,该商品有现货,付款后24小时内发货,请放心下单!",
"催发货": "您好,非常抱歉给您带来不便!您的订单已在打包中,预计今天发出,请耐心等待~",
"物流": "您好,您的包裹已发出,请点击订单详情查看实时物流,如有异常请联系我们~",
"售后": "您好,小店支持7天无理由退换货,请在订单中申请退款退货,我们会第一时间处理~",
}
def log(msg):
ts = time.strftime('%H:%M:%S')
line = f"[{ts}] {msg}"
print(line)
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\r\n")
def classify(text):
text = text or ""
if any(k in text for k in ["什么时候发","发货","发货时间"]): return "催发货"
if any(k in text for k in ["物流","到哪了","查物流"]): return "物流"
if any(k in text for k in ["尺码","大小","选哪个"]): return "尺码"
if any(k in text for k in ["有没有货","库存","还有吗"]): return "库存"
if any(k in text for k in ["退货","退款","售后","不满意"]): return "售后"
return "咨询"
def get_reply(msg_type):
return REPLIES.get(msg_type, REPLIES["咨询"])
def main():
cfg = load_config()
if not cfg or not cfg.get("enabled"):
log("[SKIP] 智能体未启用,跳过客服处理")
return
auto = cfg.get("automation", {}).get("customer_service", {})
if not auto.get("enabled"):
log("[SKIP] 客服功能未启用")
return
log(f"[START] 客服引擎启动")
# TODO: 接入平台消息API或浏览器自动化
log("[DONE] 客服功能待配置平台接入")
if __name__ == "__main__":
main()
FILE:scripts/daily_report.py
#!/usr/bin/env python3
"""
每日运营日报脚本
汇总选品、商品、客服、订单数据,生成日报
"""
import json
from datetime import datetime
from pathlib import Path
# 数据目录
DATA_DIR = Path.home() / "solo-ecommerce-data"
DATA_DIR.mkdir(exist_ok=True)
def generate_daily_report():
"""
生成每日运营日报
"""
today = datetime.now().strftime("%Y-%m-%d")
# 读取各模块数据
products = load_json(DATA_DIR / "products.json", [])
orders = load_json(DATA_DIR / "orders.json", [])
recommendations = load_json(DATA_DIR / "recommendations.json", [])
customers = load_json(DATA_DIR / "customers.json", {})
# 计算指标
total_products = len(products)
published_products = len([p for p in products if p.get("status") == "published"])
total_orders = len(orders)
total_amount = sum(o.get("amount", 0) for o in orders)
pending_shipment = len([o for o in orders if o.get("status") == "待发货"])
after_sale = len([o for o in orders if o.get("status") == "售后"])
total_customers = len(customers)
total_messages = sum(
len(c.get("messages", [])) for c in customers.values()
)
top_recommendations = sorted(
recommendations,
key=lambda x: x.get("opportunity_score", 0),
reverse=True
)[:5]
# 生成日报
report = f"""
# 电商运营日报 - {today}
## 📊 核心指标
| 指标 | 数值 |
|------|------|
| 在售商品 | {published_products}/{total_products} |
| 订单数 | {total_orders} |
| 销售额 | ¥{total_amount:.2f} |
| 客单价 | ¥{(total_amount/total_orders if total_orders else 0):.2f} |
| 待发货 | {pending_shipment} |
| 售后中 | {after_sale} |
| 客户数 | {total_customers} |
| 消息数 | {total_messages} |
## 🔍 选品推荐 TOP5
"""
for i, rec in enumerate(top_recommendations, 1):
score = rec.get("opportunity_score", 0)
keyword = rec.get("keyword", "未知")
price = rec.get("suggested_price", "未知")
report += f"{i}. **{keyword}** - 机会指数:{score},建议定价:{price}\n"
report += f"""
## 📦 待处理事项
- 待发货订单:{pending_shipment}笔
- 售后申请:{after_sale}笔
- 客服消息:{total_messages}条
## ⚠️ 异常提醒
"""
# 检查异常
if pending_shipment > 10:
report += f"- ⚠️ 待发货积压:{pending_shipment}笔,需及时处理\n"
if after_sale > 3:
report += f"- ⚠️ 售后积压:{after_sale}笔,需及时处理\n"
if pending_shipment <= 10 and after_sale <= 3:
report += "- ✅ 运营状态正常\n"
# 保存日报
report_file = DATA_DIR / "logs" / f"{today}.log"
report_file.parent.mkdir(exist_ok=True)
with open(report_file, "w", encoding="utf-8") as f:
f.write(report)
print(report)
print(f"\n日报已保存:{report_file}")
return report
def load_json(filepath, default=None):
"""加载JSON文件"""
if not filepath.exists():
return default if default is not None else []
try:
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, IOError):
return default if default is not None else []
def main():
"""主入口"""
print("=" * 50)
print("电商运营日报")
print("=" * 50)
generate_daily_report()
if __name__ == "__main__":
main()
FILE:scripts/init_agent.py
#!/usr/bin/env python3
"""
电商智能体初始化脚本
帮助用户快速配置智能体,创建必要的数据目录和配置文件
"""
import json
import os
from pathlib import Path
# 数据目录
DATA_DIR = Path.home() / "solo-ecommerce-data"
def init_data_directory():
"""初始化数据目录结构"""
DATA_DIR.mkdir(exist_ok=True)
(DATA_DIR / "logs").mkdir(exist_ok=True)
print(f"✅ 数据目录已创建:{DATA_DIR}")
# 创建初始数据文件
init_file(DATA_DIR / "products.json", [])
init_file(DATA_DIR / "orders.json", [])
init_file(DATA_DIR / "recommendations.json", [])
init_file(DATA_DIR / "customers.json", {})
print("✅ 数据文件已初始化")
def init_file(filepath, default_content):
"""初始化文件"""
if not filepath.exists():
with open(filepath, "w", encoding="utf-8") as f:
json.dump(default_content, f, ensure_ascii=False, indent=2)
print(f" - 创建:{filepath.name}")
def create_config(platforms, automation_config):
"""
创建配置文件
Args:
platforms: 平台列表 ["淘宝", "抖店"]
automation_config: 自动化配置
"""
config = {
"version": "1.0.0",
"created": get_timestamp(),
"platforms": platforms,
"automation_scope": {
"auto_publish": automation_config.get("auto_publish", False),
"auto_pricing": automation_config.get("auto_pricing", False),
"auto_reply": automation_config.get("auto_reply", True),
"auto_ship": automation_config.get("auto_ship", False),
"complaint_to_manual": True,
"refund_threshold": automation_config.get("refund_threshold", 50)
},
"risk_controls": {
"publish_needs_review": automation_config.get("publish_needs_review", True),
"ship_needs_confirm": automation_config.get("ship_needs_confirm", True),
"complaint_auto_escalate": True,
"logistics_alert_days": 3
},
"schedules": {
"product_scan": "0 * * * *",
"customer_service": "*/5 * * * *",
"order_process": "*/10 * * * *",
"daily_report": "0 23 * * *"
}
}
config_file = DATA_DIR / "config.json"
with open(config_file, "w", encoding="utf-8") as f:
json.dump(config, f, ensure_ascii=False, indent=2)
print(f"✅ 配置文件已创建:{config_file}")
return config_file
def get_timestamp():
"""获取当前时间戳"""
from datetime import datetime
return datetime.now().isoformat()
def setup_cron_jobs():
"""
设置定时任务
注意:需要通过OpenClaw的cron系统配置
"""
print("\n定时任务配置:")
print("需要在OpenClaw中配置以下cron任务:")
print("1. 选品扫描:每小时执行(cron: 0 * * * *)")
print("2. 客服回复:每5分钟执行(cron: */5 * * * *)")
print("3. 订单处理:每10分钟执行(cron: */10 * * * *)")
print("4. 数据复盘:每天23:00执行(cron: 0 23 * * *)")
def interactive_setup():
"""交互式配置"""
print("=" * 50)
print("电商智能体初始化向导")
print("=" * 50)
# 初始化数据目录
init_data_directory()
# 平台配置
print("\n请选择运营平台(多选用逗号分隔):")
print("1. 淘宝/天猫")
print("2. 抖音小店")
print("3. 拼多多")
print("4. 京东")
print("5. 其他")
platform_input = input("\n输入选项(如:1,2,3):").strip()
platform_map = {
"1": "淘宝",
"2": "抖店",
"3": "拼多多",
"4": "京东",
"5": "其他"
}
platforms = []
for p in platform_input.split(","):
p = p.strip()
if p in platform_map:
platforms.append(platform_map[p])
print(f"\n已选择平台:{', '.join(platforms)}")
# 自动化配置
print("\n请配置自动化范围:")
auto_reply = input("自动回复客服消息?(y/n,默认y):").strip().lower() != "n"
auto_publish = input("自动上架商品?(y/n,默认n):").strip().lower() == "y"
auto_ship = input("自动处理发货?(y/n,默认n):").strip().lower() == "y"
if auto_publish:
publish_review = input("上架前需要人工审核?(y/n,默认y):").strip().lower() != "n"
else:
publish_review = True
if auto_ship:
ship_confirm = input("发货前需要人工确认?(y/n,默认y):").strip().lower() != "n"
else:
ship_confirm = True
refund_threshold = input("自动处理退款上限(元,默认50):").strip()
try:
refund_threshold = int(refund_threshold) if refund_threshold else 50
except ValueError:
refund_threshold = 50
# 创建配置
config = {
"auto_reply": auto_reply,
"auto_publish": auto_publish,
"auto_ship": auto_ship,
"publish_needs_review": publish_review,
"ship_needs_confirm": ship_confirm,
"refund_threshold": refund_threshold
}
create_config(platforms, config)
# 设置定时任务提示
setup_cron_jobs()
print("\n" + "=" * 50)
print("✅ 初始化完成!")
print("=" * 50)
print("\n下一步:")
print("1. 保持浏览器已登录店铺后台")
print("2. 运行脚本开始自动运营")
print("3. 查看日报了解运营情况")
return DATA_DIR
def main():
"""主入口"""
import sys
if len(sys.argv) > 1 and sys.argv[1] == "--quick":
# 快速初始化(无交互)
init_data_directory()
create_default_config()
print("✅ 快速初始化完成")
else:
# 交互式初始化
interactive_setup()
def create_default_config():
"""创建默认配置"""
config = {
"version": "1.0.0",
"created": get_timestamp(),
"platforms": ["淘宝", "抖店"],
"automation_scope": {
"auto_publish": False,
"auto_pricing": False,
"auto_reply": True,
"auto_ship": False
},
"risk_controls": {
"publish_needs_review": True,
"ship_needs_confirm": True
}
}
config_file = DATA_DIR / "config.json"
with open(config_file, "w", encoding="utf-8") as f:
json.dump(config, f, ensure_ascii=False, indent=2)
if __name__ == "__main__":
main()
FILE:scripts/order_processor.py
# -*- coding: utf-8 -*-
"""订单引擎 - 自动处理发货、物流录入、售后申请"""
import json, os, sys, time
from pathlib import Path
DATA_DIR = Path.home() / ".qclaw" / "solo-ecommerce-data"
CONFIG_FILE = DATA_DIR / "config.json"
ORDER_FILE = DATA_DIR / "orders.json"
LOG_FILE = DATA_DIR / "logs" / f"{time.strftime('%Y-%m-%d')}.log"
def log(msg):
ts = time.strftime('%H:%M:%S')
line = f"[{ts}] {msg}"
print(line)
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\r\n")
def load_config():
if not CONFIG_FILE.exists(): return None
return json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
def load_orders():
if not ORDER_FILE.exists(): return []
return json.loads(ORDER_FILE.read_text(encoding="utf-8"))
def save_orders(orders):
ORDER_FILE.write_text(json.dumps(orders, ensure_ascii=False, indent=2), encoding="utf-8")
def main():
cfg = load_config()
if not cfg or not cfg.get("enabled"):
log("[SKIP] 智能体未启用,跳过订单处理")
return
auto = cfg.get("automation", {}).get("order", {})
if not auto.get("enabled"):
log("[SKIP] 订单功能未启用")
return
log(f"[START] 订单引擎启动,平台: {cfg.get('platform')}")
orders = load_orders()
# TODO: 接入平台订单API
log(f"[DONE] 订单功能待配置,共 {len(orders)} 条订单记录")
if __name__ == "__main__":
main()
FILE:scripts/product_publisher.py
# -*- coding: utf-8 -*-
"""上架引擎 - 自动生成商品信息并发布到店铺"""
import json, os, sys, time
from pathlib import Path
DATA_DIR = Path.home() / ".qclaw" / "solo-ecommerce-data"
CONFIG_FILE = DATA_DIR / "config.json"
PROD_FILE = DATA_DIR / "products.json"
LOG_FILE = DATA_DIR / "logs" / f"{time.strftime('%Y-%m-%d')}.log"
def log(msg):
ts = time.strftime('%H:%M:%S')
line = f"[{ts}] {msg}"
print(line)
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\r\n")
def load_config():
if not CONFIG_FILE.exists(): return None
return json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
def load_products():
if not PROD_FILE.exists(): return []
return json.loads(PROD_FILE.read_text(encoding="utf-8"))
def save_products(products):
PROD_FILE.write_text(json.dumps(products, ensure_ascii=False, indent=2), encoding="utf-8")
def generate_title(name, keywords, attrs, scene):
parts = [keywords[:8], attrs[:8], scene[:8]]
title = name + " " + " ".join(p for p in parts if p)
return title[:30]
def main():
cfg = load_config()
if not cfg:
log("[ERROR] 配置文件不存在")
return
if not cfg.get("enabled"):
log("[SKIP] 智能体未启用")
return
auto = cfg.get("automation", {}).get("publish", {})
if not auto.get("enabled"):
log("[SKIP] 上架功能未启用")
return
log(f"[START] 上架引擎启动,平台: {cfg.get('platform')}")
# TODO: 接入平台API或浏览器自动化
log("[DONE] 上架功能待配置具体商品信息")
if __name__ == "__main__":
main()
FILE:scripts/product_scanner.py
# -*- coding: utf-8 -*-
"""选品扫描引擎 - 扫描多平台热销榜单,推荐机会品类"""
import json, os, sys, time
from pathlib import Path
DATA_DIR = Path.home() / ".qclaw" / "solo-ecommerce-data"
DATA_DIR.mkdir(parents=True, exist_ok=True)
CONFIG_FILE = DATA_DIR / "config.json"
REC_FILE = DATA_DIR / "recommendations.json"
LOG_FILE = DATA_DIR / "logs" / f"{time.strftime('%Y-%m-%d')}.log"
def log(msg):
ts = time.strftime('%H:%M:%S')
line = f"[{ts}] {msg}"
print(line)
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\r\n")
def load_config():
if not CONFIG_FILE.exists():
return None
return json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
def save_recommendations(recs):
REC_FILE.write_text(json.dumps(recs, ensure_ascii=False, indent=2), encoding="utf-8")
def main():
cfg = load_config()
if not cfg:
log("[ERROR] 配置文件不存在,请先配置电商智能体")
return
if not cfg.get("enabled"):
log("[SKIP] 智能体未启用,跳过选品扫描")
return
platform = cfg.get("platform", "unknown")
log(f"[START] 选品扫描启动,平台: {platform}")
# TODO: 接入真实API或爬虫
# 1. 抓取平台热销榜单
# 2. 分析关键词趋势
# 3. 计算机会指数
# 4. 生成推荐清单
demo_recs = [
{"rank": 1, "category": "待配置", "score": 0.0,
"reason": "请配置平台后获取真实数据", "suggested_price": "待定"},
]
save_recommendations(demo_recs)
log(f"[DONE] 推荐清单已保存,共 {len(demo_recs)} 条")
if __name__ == "__main__":
main()抖音内容自动化运营技能。跨平台(Windows/macOS/Linux),一键安装,自动 clone 后端代码并配置,流水线执行:抓取AI量化视频→AI改写→发布长图文→自动回复评论。支持 Cron 定时任务。
---
name: douyin-automation
description: 抖音内容自动化运营技能。跨平台(Windows/macOS/Linux),一键安装,自动 clone 后端代码并配置,流水线执行:抓取AI量化视频→AI改写→发布长图文→自动回复评论。支持 Cron 定时任务。
---
# Douyin-Automation 抖音自动化运营
## 🚀 一键安装(2条命令)
```bash
# 1. 安装 skill(自动安装所有依赖)
clawhub install douyin-auto-hnc
# 2. 全自动引导(自动 clone 后端 + 配置路径 + 健康检查)
python ~/.qclaw/skills/douyin-automation/scripts/setup.py
```
> **macOS/Linux 用户**:路径为 `~/.qclaw/skills/douyin-automation/scripts/setup.py`
---
## 完整流程
```
安装 skill → setup.py (自动 clone 后端) → start-backend.py (启动服务)
↓
run-pipeline.py (执行发布)
```
### 第1步:安装
```bash
clawhub install douyin-auto-hnc
```
### 第2步:运行 setup.py(全自动)
```bash
python ~/.qclaw/skills/douyin-automation/scripts/setup.py
```
setup.py 自动完成:
- 从 GitHub clone `douyin-agent-master` 后端代码到 `~/douyin/`
- 复制 creator-tools 到 `~/.openclaw/douyin-creator-tools/`
- 安装 Python 依赖(requests, playwright 等)
- 交互式确认端口和路径配置
- 生成 `CONFIG.md`
- 运行健康检查
### 第3步:启动服务(一键)
```bash
python ~/.qclaw/skills/douyin-automation/scripts/start-backend.py
```
- 自动启动 Chrome(带 `--remote-debugging-port`)
- 自动启动 FastAPI 后端(端口 8080)
- 如果端口已被占用则跳过(已运行)
### 第4步:执行流水线
```bash
# 正式运行
python ~/.qclaw/skills/douyin-automation/scripts/run-pipeline.py
# 试运行(不实际发布)
python ~/.qclaw/skills/douyin-automation/scripts/run-pipeline.py --dry-run
# 禁用 AI 优化(直接发布原始内容)
python ~/.qclaw/skills/douyin-automation/scripts/run-pipeline.py --no-ai
```
### 第5步:配置定时任务(可选)
```bash
# 抖音运营流水线,每 6 小时执行
openclaw cron add "DOUYIN-PIPELINE-6H" \
--cron "0 */6 * * *" \
--message "执行: python ~/.qclaw/skills/douyin-automation/scripts/run-pipeline.py"
# 抖音评论回复,每 30 分钟执行
openclaw cron add "DOUYIN-COMMENTS-30M" \
--cron "*/30 * * * *" \
--message "执行: python ~/.qclaw/skills/douyin-automation/scripts/run-pipeline.py --comments-only"
```
---
## 系统架构
```
GitHub: HNC87/douyin-agent-master
↓ clone 到 ~/douyin/
↓
douyin-agent-master/backend/ (FastAPI :8080)
douyin-agent-master/orchestrator/douyin_full_orchestrator.py
↓
douyin-creator-tools/
publish-douyin-article.mjs → 发布到抖音
export-douyin-comments.mjs → 导出未回复评论
reply-douyin-comments.mjs → 自动回复
↓
OpenClaw Gateway (http://127.0.0.1:28789)
→ AI 改写内容(openclaw/default 模型)
```
---
## 手动前提条件
**必须提前准备(setup.py 无法自动化):**
1. **Chrome 浏览器**(已安装)
2. **抖音账号已登录 Chrome**(首次运行 setup.py 后,用 Chrome 手动扫码登录一次)
3. **OpenClaw Gateway 运行中**(AI 改写需要)
**可选(提高自动化程度):**
- Python 3.11+
- Node.js(用于 creator-tools 脚本)
---
## 配置说明
所有路径集中在 `~/.qclaw/skills/douyin-automation/CONFIG.md`:
| 键 | 默认值 | 说明 |
|---|---|---|
| `chrome_cdp_port` | `9222` | Chrome 调试端口 |
| `agent_port` | `8080` | FastAPI 后端端口 |
| `openclaw_gateway` | `http://127.0.0.1:28789` | AI 网关地址 |
| `douyin_home` | `~/douyin` | 项目根目录 |
重新配置:
```bash
python ~/.qclaw/skills/douyin-automation/scripts/setup.py
```
---
## 详细内容
- [发布规则与安全过滤](references/publishing.md)
- [评论导出与自动回复](references/comment-reply.md)
- [Qenda AI 封面生成](references/cover-ai.md)
- [编排器参数配置](references/config-reference.md)
---
## 常见问题
| 问题 | 解决 |
|------|------|
| "No items to publish" | 确认 douyin-agent 已抓取并改写视频内容到 DB |
| "CONFIG.md not found" | 运行 `python scripts/setup.py` |
| Chrome CDP 连接失败 | 确保 Chrome 已退出,重新运行 `start-backend.py` |
| AI 改写失败 | 检查 OpenClaw Gateway 是否运行 |
| 登录态失效 | 重新用 Chrome 扫码登录 creator.douyin.com |
---
## 更新 skill
```bash
clawhub update douyin-auto-hnc
```
FILE:CONFIG.md
# Douyin Automation Configuration
> Run `python scripts/setup.py` to auto-detect and generate paths, or edit the JSON block below manually.
```json
{
"douyin_home": "REQUIRED - run setup.py or edit manually",
"agent_backend": "",
"orchestrator": "",
"chatgroup_db": "",
"uploads_dir": "",
"cookies_file": "",
"creator_tools": "",
"comments_output": "",
"chrome_cdp_port": 9222,
"agent_port": 8080,
"openclaw_gateway": "http://127.0.0.1:28789",
"openclaw_model": "openclaw/default"
}
```
## Setup
```bash
# Auto-detect paths (interactive wizard)
python scripts/setup.py
# Or edit the JSON block above directly
```
FILE:references/comment-reply.md
# 评论导出与自动回复
## 完整流程
编排器 `douyin_full_orchestrator.py` 的 `run()` 在发布完成后自动执行:
```
export_comments() → build_reply_plan() → reply_comments()
```
## 路径
- **导出脚本:** `{creator_tools}/src/export-douyin-comments.mjs`
- **回复脚本:** `{creator_tools}/src/reply-douyin-comments.mjs`
- **输出目录:** `{comments_output}`
以上变量取自 CONFIG.md,由 `scripts/setup.py` 自动解析。
## 单独导出评论
```bash
node "{creator_tools}/src/export-douyin-comments.mjs" \
--out "{comments_output}/unreplied-comments.json" \
--limit 50
```
## 单独执行回复
```bash
node "{creator_tools}/src/reply-douyin-comments.mjs" \
--limit 20 \
--keep-open \
--out "{comments_output}/reply-result.json" \
--timeout 600000 \
-- "{comments_output}/auto-reply-plan.json"
```
> 注意:回复操作需要已登录 Chrome 浏览器,不要加 `--headless`。
## 回复分类规则
| 类型 | 关键词 | 回复风格 |
|------|--------|---------|
| q(问答) | how/what/why/怎么/如何/请问/教程 | 引导看视频/下次覆盖 |
| t(感谢) | thank/great/helpful/有用 | 感谢支持 |
| n(负面) | bad/wrong/fake/垃圾/骗人 | 中性感谢 |
| f(关注) | follow/粉丝/关注 | 欢迎关注 |
| d(默认) | 其他 | 简短肯定 |
## 数据库查询
```bash
sqlite3 "{chatgroup_db}" "SELECT * FROM comments LIMIT 10"
```
comments 表字段:`item_id`、`username`、`text`、`reply_status`(pending/sent/failed)
FILE:references/config-reference.md
# 编排器配置参数详解
## 配置位置
编排器常量定义在 `douyin_full_orchestrator.py` 顶部。所有路径通过 CONFIG.md 管理。
## 关键常量
```python
# 路径(从 CONFIG.md 读取,或硬编码覆盖)
AGENT_DB_PATH = "{chatgroup_db}"
CREATOR_TOOLS_DIR = "{creator_tools}"
CREATOR_OUTPUT = "{comments_output}"
UPLOADS_DIR = "{uploads_dir}"
# 发布频率(见 references/publishing.md)
MIN_SCORE = 0
MAX_ITEMS = 1
MIN_PUBLISH_INTERVAL_H = 1
MAX_DAILY_PUBLISH = 3
# AI 优化(从 CONFIG.md 读取)
OPENCLAW_GATEWAY = "{openclaw_gateway}"
OPENCLAW_TOKEN = "<从 openclaw.json 获取>"
OPENCLAW_MODEL = "{openclaw_model}"
AI_OPTIMIZE_TIMEOUT = 30 # 秒
# 小红书(已禁用)
XHS_ENABLED = False
```
## 路径解析
编排器启动时从 CONFIG.md 解析路径。如果没有 CONFIG.md,会回退到脚本中的硬编码默认值。
推荐的修改方式:**在编排器顶部添加覆盖变量**,不直接改硬编码常量。
```python
# === 用户配置覆盖区 ===
# MAX_DAILY_PUBLISH = 5
# XHS_ENABLED = True
# AI_OPTIMIZE_ENABLED = False
```
## 状态字段
monitor_items 表中与发布相关的字段:
| 字段 | 含义 |
|------|------|
| `article_published` | 是否已发文章(0/1) |
| `imagetext_published` | 是否已发长图文(0/1) |
| `publish_status` | published / failed:xxx / NULL |
| `publish_time` | 发布时间(ISO 格式) |
| `transcript_status` | pending / processing / full |
| `rank_score` | 内容质量评分 |
FILE:references/cover-ai.md
# Qenda AI 封面生成
## 封面优先级
1. item 专属封面:`{uploads_dir}/cover_{item_id}/cover.jpg`
2. Qenda AI 生成(基于标题,9:16 竖版,4K)
3. 通用封面:`{uploads_dir}/*.jpg`
以上路径取自 CONFIG.md。
## Qenda API
- 端点:`https://api.ai6700.com/api/v1/media/generate`
- 模型:`wan2.7-image`
- 同步等待,最长 120s(轮询每 5s 一次,共 24 次)
- 输出尺寸:9:16(竖版抖音封面)
## 生成提示词模板
```
抖音视频封面,标题文字「{clean_title}」,
深色科技感背景配渐变光效,左上角标注「AI量化」,
整体氛围专业权威,适合金融科技主题,9:16竖版,4K高清
```
clean_title = 标题中移除 emoji、截断到 40 字。
## 常见失败
| 状态 | 原因 | 解决 |
|------|------|------|
| submit failed | API Key 无效或余额不足 | 检查 Qenda 账户 |
| poll timeout | 生成耗时 >120s | 手动延长超时或使用已有封面 |
| 无生成权限 | 账户额度用完 | 联系 Qenda 续费 |
FILE:references/publishing.md
# 抖音发布配置与安全规则
## 发布参数
| 参数 | 默认值 | 说明 |
|------|--------|------|
| MIN_SCORE | 0 | 最低 rank_score 阈值 |
| MAX_ITEMS | 1 | 每次最多发布条数 |
| MIN_PUBLISH_INTERVAL_H | 1 | 两次发布最小间隔(小时) |
| MAX_DAILY_PUBLISH | 3 | 每日最大发布条数 |
## 选题关键词(标题必须包含其一)
AI, 量化, 交易, Python, 编程, 金融, 股票, 程序化,
回测, 策略, 算法, 机器人, 自动化, 大模型, LLM, GPT,
deepseek, DeepSeek, 机器学习, 期货, 外汇, MT5, MT4,
指标, 因子, 对冲, 高频, 模型, 开户, 软件, 工具,
理财, 投资, 炒股
## 排除关键词(标题包含其一则跳过)
搞笑, 段子, 整活, 沙雕, 奇葩, 恶搞, 离谱答案, 牛逼克拉斯,
笑话, 娱乐, 吃瓜, 八卦, 相亲, 情侣, 闺蜜, 婆媳,
柔顺剂, 广告, 金纺, 拉手姐, 超燃, 开学了, 作业
## 金融违规词(命中则跳过,不发布)
收割, 韭菜, 荐股, 牛股, 涨停, 抄底, 满仓, 加仓,
买入信号, 卖出信号, 稳赚, 暴赚, 翻倍, 内部消息,
庄家, 主力, 操纵, 散户, 黑马, 妖股
## 内容安全高危词(命中则跳过)
收割散户, 精准收割, 量化收割, 割韭菜, 血亏,
倾家荡产, 家破人亡, 跳楼, 爆仓
## AI 优化策略
调用 OpenClaw Gateway(地址见 CONFIG.md `openclaw_gateway`),模型见 `openclaw_model`。
**系统提示词核心约束:**
- 绝对不包含投资建议、荐股、买卖信号
- 用客观科普视角代替煽动性表述
- 强调风险意识,避免暗示"跟随操作能赚钱"
**失败回退:** AI 调用失败时,使用 `clean_rewrite_text()` 正则清洗脚本。
## 抖音长图文
- 标题最多 30 字
- 正文最多 8000 字
- 配乐关键词:星际穿越、电子科技、未来科技、AI、科技感、数码科技
- 话题标签(随机选 3 个):AI量化交易、量化交易、Python量化、AI交易、量化投资、程序化交易、金融科技
## 小红书发布(已暂停)
小红书发布功能已内置但账号封禁后禁用(`XHS_ENABLED = False`)。
标签:AI量化、量化交易、Python、金融科技、投资理财、程序员、搞钱
FILE:scripts/run-pipeline.py
#!/usr/bin/env python3
"""Douyin Automation Pipeline Runner - Cross-platform.
Reads CONFIG.md for paths, then runs the orchestrator.
Supports --dry-run and --no-ai flags.
"""
import argparse
import json
import re
import subprocess
import sys
import time
from pathlib import Path
SKILL_DIR = Path(__file__).resolve().parent.parent
CONFIG_FILE = SKILL_DIR / "CONFIG.md"
def load_config():
"""Load and parse CONFIG.md JSON block."""
if not CONFIG_FILE.exists():
print("ERROR: CONFIG.md not found. Run 'python scripts/setup.py' first.")
sys.exit(1)
text = CONFIG_FILE.read_text(encoding="utf-8")
m = re.search(r"```json\s*([\s\S]*?)\s*```", text)
if not m:
print("ERROR: No JSON block found in CONFIG.md.")
sys.exit(1)
config = json.loads(m.group(1))
if config.get("douyin_home") == "REQUIRED - run setup.py or edit manually":
print("ERROR: CONFIG.md not configured. Run 'python scripts/setup.py' first.")
sys.exit(1)
return config
def main():
parser = argparse.ArgumentParser(description="Douyin Automation Pipeline")
parser.add_argument("--dry-run", action="store_true", help="Dry run, no actual publishing")
parser.add_argument("--no-ai", action="store_true", help="Disable AI text optimization")
args = parser.parse_args()
config = load_config()
orchestrator = config["orchestrator"]
# Check orchestrator exists
if not Path(orchestrator).exists():
print(f"ERROR: Orchestrator not found: {orchestrator}")
print("Run 'python scripts/setup.py' to update paths.")
sys.exit(1)
# Build command
cmd = [sys.executable, orchestrator]
if args.dry_run:
cmd.append("--dry-run")
if args.no_ai:
cmd.append("--no-ai")
print(f"[{time.strftime('%H:%M:%S')}] Douyin Pipeline START")
print(f"[Config] Orchestrator: {orchestrator}")
if args.dry_run:
print("[Config] Mode: DRY RUN")
if args.no_ai:
print("[Config] AI: DISABLED")
print()
start = time.time()
result = subprocess.run(cmd, cwd=str(Path(orchestrator).parent))
duration = round(time.time() - start)
print()
status = "DONE" if result.returncode == 0 else "FAILED"
print(f"[{time.strftime('%H:%M:%S')}] {status} (exit={result.returncode}, {duration}s)")
sys.exit(result.returncode)
if __name__ == "__main__":
main()
FILE:scripts/setup.py
#!/usr/bin/env python3
"""
Douyin Automation Setup Wizard - Full bootstrap.
Handles: clone backend → install deps → configure paths → health check.
Cross-platform (Windows/macOS/Linux).
"""
import json
import os
import re
import subprocess
import sys
from pathlib import Path
SKILL_DIR = Path(__file__).resolve().parent.parent
CONFIG_FILE = SKILL_DIR / "CONFIG.md"
BACKEND_REPO = "https://github.com/HNC87/douyin-agent-master.git"
SKILL_REPO = "https://github.com/HNC87/douyin-automation-skill.git"
def run(cmd, cwd=None, timeout=60, check=False):
"""Run shell command, return (success, stdout+stderr)."""
try:
r = subprocess.run(
cmd, shell=True, cwd=cwd,
capture_output=True, text=True, timeout=timeout
)
out = (r.stdout + "\n" + r.stderr).strip()
if r.returncode == 0:
return True, out
return False, out
except subprocess.TimeoutExpired:
return False, "Command timed out"
except Exception as e:
return False, str(e)
def find_git_auth():
"""Get GitHub auth token if available."""
# Check gh CLI
ok, out = run("gh auth token 2>/dev/null")
if ok and out.strip():
return out.strip()
# Check environment
for env in ["GITHUB_TOKEN", "GH_TOKEN"]:
t = os.environ.get(env, "")
if t:
return t
# Check git credential store
cfg = os.path.expanduser("~/.git-credentials")
if Path(cfg).exists():
content = Path(cfg).read_text()
m = re.search(r"https://([^:]+):([^@]+)@", content)
if m:
return m.group(2)
return None
def git_clone_with_auth(url, dest, depth=1):
"""Clone using auth token if available."""
token = find_git_auth()
if token:
auth_url = url.replace("https://", f"https://{token}@")
else:
auth_url = url
# Try with auth first
ok, out = run(f'git clone {"--depth 1" if depth > 1 else ""} {auth_url} "{dest}"', timeout=120)
if ok:
return True, out
# Fallback: try without auth (public repo)
if token:
ok2, out2 = run(f'git clone {"--depth 1" if depth > 1 else ""} {url} "{dest}"', timeout=120)
return ok2, out2
return ok, out
def find_project():
"""Find existing douyin-agent-master."""
home = Path.home()
candidates = [
home / "douyin" / "douyin-agent-master",
home / "douyin-agent-master",
Path("D:/douyin/douyin-agent-master"),
home / "projects" / "douyin-agent-master",
home / "Documents" / "douyin-agent-master",
home / "dev" / "douyin-agent-master",
home / "code" / "douyin-agent-master",
]
for p in candidates:
if p.is_dir() and (p / "backend").exists():
return p.parent
return None
def find_creator_tools():
"""Find existing douyin-creator-tools."""
home = Path.home()
for base in [home / ".openclaw", home / ".qclaw", home]:
ct = base / "douyin-creator-tools"
if ct.is_dir():
return ct
ct2 = base / "workspace" / "douyin-creator-tools"
if ct2.is_dir():
return ct2
return None
def check_chrome_cdp(port=9222):
"""Check if Chrome with remote debugging is accessible."""
import urllib.request
try:
r = urllib.request.urlopen(f"http://localhost:{port}/json", timeout=3)
data = json.loads(r.read())
return True, f"Chrome CDP active (port {port}), {len(data)} tabs"
except Exception:
return False, f"Chrome CDP not accessible on port {port}"
def check_backend(port=8080):
"""Check if backend API is running."""
import urllib.request
try:
r = urllib.request.urlopen(f"http://localhost:{port}/health", timeout=3)
return True, f"Backend API active (port {port})"
except Exception:
return False, f"Backend API not running on port {port}"
def read_config():
if not CONFIG_FILE.exists():
return {}
text = CONFIG_FILE.read_text(encoding="utf-8")
m = re.search(r"```json\s*([\s\S]*?)\s*```", text)
if m:
try:
return json.loads(m.group(1))
except json.JSONDecodeError:
return {}
return {}
def write_config(config):
lines = [
"# Douyin Automation Configuration",
"",
f"> Auto-generated by setup.py on {__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M')}.",
"> Run `python scripts/setup.py` to reconfigure.",
"",
"```json",
json.dumps(config, indent=2, ensure_ascii=False),
"```",
]
CONFIG_FILE.write_text("\n".join(lines), encoding="utf-8")
return CONFIG_FILE
def generate_config(douyin_home, creator_tools=None, cdp_port=9222,
agent_port=8080, gateway="http://127.0.0.1:28789"):
dh = Path(douyin_home)
agent_backend = dh / "douyin-agent-master" / "backend"
ct = Path(creator_tools) if creator_tools else Path.home() / ".openclaw" / "douyin-creator-tools"
return {
"douyin_home": str(dh),
"agent_backend": str(agent_backend),
"orchestrator": str(dh / "orchestrator"),
"chatgroup_db": str(agent_backend / "app" / "chatgroup.db"),
"uploads_dir": str(agent_backend / "uploads"),
"cookies_file": str(agent_backend / "douyin_cookies.json"),
"creator_tools": str(ct),
"comments_output": str(ct / "comments-output"),
"chrome_cdp_port": cdp_port,
"agent_port": agent_port,
"openclaw_gateway": gateway,
"openclaw_model": "openclaw/default",
}
def validate(config):
"""Validate paths and report status."""
print("\n=== Validation ===")
checks = {
"Backend directory": Path(config["agent_backend"]),
"Orchestrator": Path(config["orchestrator"]),
"Chatgroup DB": Path(config["chatgroup_db"]),
"Creator tools": Path(config["creator_tools"]),
}
results = []
for name, path in checks.items():
ok = path.exists()
tag = "OK" if ok else "MISS"
print(f" [{tag:4}] {name}: {path}")
results.append(ok)
# Service checks
ok_cdp, msg_cdp = check_chrome_cdp(config["chrome_cdp_port"])
tag = "OK" if ok_cdp else "MISS"
print(f" [{tag:4}] Chrome CDP ({msg_cdp})")
ok_api, msg_api = check_backend(config["agent_port"])
tag = "OK" if ok_api else "MISS"
print(f" [{tag:4}] Backend API ({msg_api})")
return all(results)
def main():
print("=" * 60)
print(" Douyin Automation - Setup Wizard")
print("=" * 60)
print(f"OS: {sys.platform} | Python: {sys.version.split()[0]}")
print(f"Home: {Path.home()}")
print()
existing = read_config()
# Step 1: Find or clone backend
print("=== Step 1: Douyin Backend ===")
project = find_project()
if project:
print(f" [FOUND] {project}")
else:
print(f" [NEW] Cloning from GitHub...")
home = Path.home()
dest = home / "douyin" / "douyin-agent-master"
dest.parent.mkdir(parents=True, exist_ok=True)
ok, out = git_clone_with_auth(BACKEND_REPO, str(dest.parent))
if ok:
# Rename if cloned as douyin-agent-master directly
actual = home / "douyin-agent-master"
if actual.is_dir() and not dest.is_dir():
dest = actual
print(f" [OK] Cloned to: {dest}")
project = dest
else:
print(f" [WARN] Clone failed: {out[:200]}")
print(f" Manual: git clone {BACKEND_REPO} ~/douyin/")
print(f" Or download from: https://github.com/HNC87/douyin-agent-master")
# Still prompt for manual path
manual = input("\n Enter path to douyin-agent-master: ").strip()
if manual:
project = Path(manual)
else:
print(" Skipped. You can run setup.py again later.")
project = None
# Step 2: Find or clone creator-tools
print("\n=== Step 2: Creator Tools ===")
ct = find_creator_tools()
if ct:
print(f" [FOUND] {ct}")
else:
print(f" [NEW] Cloning creator-tools...")
ct_dest = Path.home() / ".openclaw" / "douyin-creator-tools"
ct_dest.parent.mkdir(parents=True, exist_ok=True)
ok, out = git_clone_with_auth(
"https://github.com/HNC87/douyin-automation-skill.git",
str(ct_dest.parent / "douyin-automation-skill")
)
if ok:
# Creator tools are in the skill repo
src = ct_dest.parent / "douyin-automation-skill" / "assets"
if src.exists():
import shutil
ct_dest.mkdir(parents=True, exist_ok=True)
for f in src.iterdir():
if f.suffix in [".mjs", ".js"]:
shutil.copy(f, ct_dest / f.name)
print(f" [OK] Copied to: {ct_dest}")
ct = ct_dest
else:
print(f" [WARN] Could not auto-clone: {out[:200]}")
print(f" Manual: git clone {SKILL_REPO} ~/.openclaw/douyin-creator-tools")
ct = Path.home() / ".openclaw" / "douyin-creator-tools"
# Step 3: Configure
if project:
print("\n=== Step 3: Configuration ===")
defaults = {
"cdp": existing.get("chrome_cdp_port", 9222),
"agent": existing.get("agent_port", 8080),
"gateway": existing.get("openclaw_gateway", "http://127.0.0.1:28789"),
}
cdp = input(f" Chrome CDP port [{defaults['cdp']}]: ").strip()
agent = input(f" Backend API port [{defaults['agent']}]: ").strip()
gw = input(f" OpenClaw Gateway [{defaults['gateway']}]: ").strip()
config = generate_config(
project,
ct,
int(cdp) if cdp else defaults["cdp"],
int(agent) if agent else defaults["agent"],
gw if gw else defaults["gateway"],
)
write_config(config)
print(f"\n [OK] Config saved: {CONFIG_FILE}")
# Step 4: Validate
ok = validate(config)
# Step 5: Install dependencies
print("\n=== Step 4: Dependencies ===")
req = project / "douyin-agent-master" / "backend" / "requirements.txt"
if req.exists():
print(f" Installing Python deps from {req}...")
ok_deps, out_deps = run(f'"{sys.executable}" -m pip install -r "{req}"', timeout=120)
if ok_deps:
print(f" [OK] Dependencies installed")
else:
print(f" [WARN] Some deps may have issues: {out_deps[:200]}")
else:
print(f" [SKIP] No requirements.txt found")
# Step 5: Quick start guide
print("\n=== Next Steps ===")
if ok:
print(" All checks passed!")
print(" Start backend: python scripts/start-backend.py")
print(" Run pipeline: python scripts/run-pipeline.py")
print(" Health check: python scripts/status-check.py")
else:
print(" Some paths missing. Check your setup.")
print(" Start Chrome with: chrome.exe --remote-debugging-port=9222")
print(" Then run: python scripts/start-backend.py")
else:
print("\n [DONE] Skipped backend setup.")
print(" Run `python scripts/setup.py` again once you have the backend.")
if __name__ == "__main__":
main()
FILE:scripts/start-backend.py
#!/usr/bin/env python3
"""
Start Douyin Automation Backend - Chrome CDP + FastAPI.
Cross-platform. Handles port conflicts gracefully.
"""
import json
import os
import re
import signal
import socket
import subprocess
import sys
import time
from pathlib import Path
SKILL_DIR = Path(__file__).resolve().parent.parent
# Load config
def load_config():
cfg = SKILL_DIR / "CONFIG.md"
if not cfg.exists():
return None
text = cfg.read_text(encoding="utf-8")
m = re.search(r"```json\s*([\s\S]*?)\s*```", text)
if m:
try:
return json.loads(m.group(1))
except json.JSONDecodeError:
return None
return None
def check_port(port):
"""Check if port is in use."""
s = socket.socket()
try:
s.bind(("127.0.0.1", port))
return False # free
except OSError:
return True # in use
finally:
s.close()
def wait_for_port(port, timeout=10):
"""Wait for port to become available (Chrome) or ready (API)."""
start = time.time()
while time.time() - start < timeout:
s = socket.socket()
try:
s.settimeout(1)
s.connect(("127.0.0.1", port))
return True
except (socket.timeout, ConnectionRefusedError, OSError):
time.sleep(0.5)
finally:
s.close()
return False
def get_chrome_path():
"""Find Chrome executable."""
home = Path.home()
if sys.platform == "win32":
candidates = [
Path(os.environ.get("ProgramFiles", "")) / "Google/Chrome/Application/chrome.exe",
Path(os.environ.get("ProgramFiles(x86)", "")) / "Google/Chrome/Application/chrome.exe",
Path(os.environ.get("LOCALAPPDATA", "")) / "Google/Chrome/Application/chrome.exe",
Path("C:/Program Files/Google/Chrome/Application/chrome.exe"),
]
elif sys.platform == "darwin":
candidates = [
Path("/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"),
Path(home / "Applications/Google Chrome.app/Contents/MacOS/Google Chrome"),
]
else:
candidates = [
Path("/usr/bin/google-chrome"),
Path("/usr/bin/chromium-browser"),
Path("/usr/bin/chromium"),
home / "snap/chromium/current/usr/lib/chromium-browser/chromium-browser",
]
for p in candidates:
if p.exists():
return str(p)
return None
def start_chrome(cdp_port, dry_run=False):
"""Launch Chrome with remote debugging."""
chrome = get_chrome_path()
if not chrome:
return False, "Chrome not found. Install Chrome first."
user_data_dir = Path.home() / ".openclaw" / "chrome-douyin-profile"
user_data_dir.mkdir(parents=True, exist_ok=True)
args = [
chrome,
f"--remote-debugging-port={cdp_port}",
f"--user-data-dir={user_data_dir}",
"--no-first-run",
"--no-default-browser-check",
"--disable-popup-blocking",
"https://creator.douyin.com",
]
if dry_run:
return True, " ".join(args)
try:
subprocess.Popen(
args,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
# Wait for Chrome to start
time.sleep(2)
if wait_for_port(cdp_port, timeout=10):
return True, f"Chrome started (CDP port {cdp_port})"
return False, "Chrome started but CDP not responding"
except Exception as e:
return False, f"Failed to start Chrome: {e}"
def start_backend(backend_dir, port):
"""Start FastAPI backend."""
python = sys.executable
main_py = Path(backend_dir) / "main.py"
if not main_py.exists():
main_py = Path(backend_dir) / "app" / "main.py"
if not main_py.exists():
return False, f"main.py not found in {backend_dir}"
cmd = f'"{python}" "{main_py}"'
try:
subprocess.Popen(
cmd,
cwd=str(Path(backend_dir).parent),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
if wait_for_port(port, timeout=15):
return True, f"Backend started (port {port})"
return False, "Backend started but port not responding"
except Exception as e:
return False, f"Failed to start backend: {e}"
def main():
print("=" * 50)
print(" Douyin Automation - Start Services")
print("=" * 50)
config = load_config()
if not config:
print("ERROR: CONFIG.md not found. Run `python scripts/setup.py` first.")
sys.exit(1)
cdp_port = config.get("chrome_cdp_port", 9222)
backend_port = config.get("agent_port", 8080)
backend_dir = config.get("agent_backend")
print(f" Chrome CDP port: {cdp_port}")
print(f" Backend port: {backend_port}")
print(f" Backend dir: {backend_dir}")
print()
# Chrome
print(f"[1/2] Chrome CDP (port {cdp_port})...")
if check_port(cdp_port):
print(f" Already running on port {cdp_port}")
else:
ok, msg = start_chrome(cdp_port)
print(f" {msg}")
if not ok:
print(f" Manual: Start Chrome with --remote-debugging-port={cdp_port}")
# Backend
print(f"\n[2/2] Backend API (port {backend_port})...")
if check_port(backend_port):
print(f" Already running on port {backend_port}")
else:
ok, msg = start_backend(backend_dir, backend_port)
print(f" {msg}")
if not ok:
print(f" Check: cd {backend_dir} && python main.py")
print()
print("=" * 50)
print(" Ready! Run `python scripts/run-pipeline.py` to execute.")
print(" Press Ctrl+C to stop this script (services keep running).")
print("=" * 50)
if __name__ == "__main__":
main()
FILE:scripts/status-check.py
#!/usr/bin/env python3
"""Douyin Automation Status Check - Cross-platform.
Reads CONFIG.md and checks: database, Chrome CDP, cover images, creator tools.
Works on Windows, macOS, and Linux.
"""
import json
import re
import sqlite3
import sys
import urllib.request
import urllib.error
from pathlib import Path
SKILL_DIR = Path(__file__).resolve().parent.parent
CONFIG_FILE = SKILL_DIR / "CONFIG.md"
def load_config():
"""Load and parse CONFIG.md JSON block."""
if not CONFIG_FILE.exists():
return None
text = CONFIG_FILE.read_text(encoding="utf-8")
m = re.search(r"```json\s*([\s\S]*?)\s*```", text)
if not m:
return None
config = json.loads(m.group(1))
if config.get("douyin_home") == "REQUIRED - run setup.py or edit manually":
return None
return config
def check_db(config):
"""Check database and return pending/published counts."""
db_path = config["chatgroup_db"]
if not Path(db_path).exists():
return None, f"NOT FOUND: {db_path}"
try:
conn = sqlite3.connect(db_path)
pending = conn.execute(
"SELECT COUNT(*) FROM monitor_items WHERE imagetext_published=0 "
"AND transcript_status='full' AND rank_score>=0"
).fetchone()[0]
today = conn.execute(
"SELECT COUNT(*) FROM monitor_items WHERE (imagetext_published=1 OR article_published=1) "
"AND publish_time >= date('now','localtime','+8 hours')"
).fetchone()[0]
conn.close()
return (pending, today), None
except Exception as e:
return None, f"DB Error: {e}"
def check_cdp(config):
"""Check Chrome CDP connection."""
port = config.get("chrome_cdp_port", 9222)
try:
url = f"http://localhost:{port}/json/version"
req = urllib.request.urlopen(url, timeout=3)
data = json.loads(req.read())
return True, data.get("Browser", "Unknown")
except Exception:
return False, f"CDP port {port} not responding"
def check_covers(config):
"""Count cover images in uploads directory."""
uploads = Path(config.get("uploads_dir", ""))
if not uploads.exists():
return 0
count = sum(1 for f in uploads.rglob("*") if f.suffix.lower() in (".jpg", ".jpeg", ".png", ".webp"))
return count
def check_tools(config):
"""Check creator tools scripts."""
ct_dir = Path(config.get("creator_tools", ""))
scripts = {
"publish-douyin-article.mjs": ct_dir / "src" / "publish-douyin-article.mjs",
"export-douyin-comments.mjs": ct_dir / "src" / "export-douyin-comments.mjs",
"reply-douyin-comments.mjs": ct_dir / "src" / "reply-douyin-comments.mjs",
}
return {name: path.exists() for name, path in scripts.items()}
def main():
print("=== Douyin Automation Status ===")
print(f"OS: {__import__('os').name} / {__import__('sys').platform}")
print()
config = load_config()
if not config:
print("[CONFIG] NOT CONFIGURED - run 'python scripts/setup.py'")
sys.exit(1)
print(f"[CONFIG] Project: {config['douyin_home']}")
# Database
print()
db_result, db_error = check_db(config)
if db_error:
print(f"[DB] {db_error}")
else:
pending, today = db_result
color_pending = "!" if pending > 0 else "ok"
print(f"[DB] Pending: {pending} | Today published: {today} {color_pending}")
# Chrome CDP
cdp_ok, cdp_info = check_cdp(config)
cdp_status = "OK" if cdp_ok else "DOWN"
if cdp_ok:
print(f"[CDP] {cdp_status} - {cdp_info}")
else:
print(f"[CDP] {cdp_status} - {cdp_info}")
# Covers
covers = check_covers(config)
print(f"[Cover] Image files: {covers}")
# Creator Tools
tools = check_tools(config)
for name, exists in tools.items():
status = "OK" if exists else "MISSING"
print(f"[Tools] {name}: {status}")
# Orchestrator
orch = Path(config.get("orchestrator", ""))
print(f"[Orchestrator] {'OK' if orch.exists() else 'MISSING'}: {config.get('orchestrator', '?')}")
# Agent Backend
backend = Path(config.get("agent_backend", ""))
print(f"[Agent Backend] {'OK' if backend.exists() else 'MISSING'}: {config.get('agent_backend', '?')}")
print()
all_ok = (
db_result is not None
and cdp_ok
and all(tools.values())
and orch.exists()
)
if all_ok:
print("All systems operational. Ready to publish.")
else:
print("Some issues detected. Review above.")
print("Run 'python scripts/setup.py' to reconfigure.")
if __name__ == "__main__":
main()