解决博客零评论困境：AI 评论机器人部署全记录

前言

写博客的人大多有过这种体验：精心打磨一篇文章发布后，评论区却长期空白。对读者而言，”零评论”往往意味着内容无人问津，反而降低了互动意愿；对博主来说，手动维护评论既耗时又难以持续，尤其是当站点积累了数十篇甚至上百篇文章后，逐篇补充评论几乎成了不可能完成的任务。

我目前维护的个人博客基于 Hexo 框架，使用 anzhiyu 主题，评论系统选用 Twikoo 并通过 Docker 独立部署。随着文章数量增加，我逐渐意识到需要一个自动化的方案来解决评论冷启动问题——不是批量注册马甲账号刷量，而是让 AI 以真实读者的视角参与讨论，生成有实质内容、角度各异的评论，从而营造自然的社区氛围。

这个方案的核心思路是：利用大模型能力生成拟人化评论内容，同时通过模拟真实用户的浏览器 UA、IP 地址、Gravatar 头像等细节，让 AI 评论在数据层面与真实用户无异。系统不修改 Hexo 主题或 Twikoo 源码，而是以独立 Python 脚本的形式运行，通过读取 Twikoo 本地数据库获取现有评论，再经 HTTPS 接口提交新评论或回复。

本文将完整记录该系统的部署过程，从环境准备、模块配置到定时任务运行，力求每一步都可复现。Twikoo 本身的部署已有大量文档，此处不再赘述；代码实现仅展示关键结构与思路，具体细节可根据实际需求调整补充

代码架构

1
2
3
4
5
6
7
8
ai_bot/                          # Python 包
├── __init__.py                  # 空文件，标记包
├── config.py                    # 配置中心（API密钥、URL、比例）
├── ai_service.py                # AI生成服务（昵称、评论、回复）
├── twikoo_client.py             # Twikoo交互层（读db、写接口）
├── auto_comment.py              # 新评论逻辑
├── reply_bot.py                 # 回复评论逻辑
└── main.py                      # 入口调度（CLI、批量、比例控制）

关键设计

分层隔离

  • ai_service 只负责内容生成，不感知 Twikoo
  • twikoo_client 只负责数据读写，不感知 AI
  • main 负责编排，业务逻辑下沉到 auto_comment / reply_bot

双通道读写

  • 读：直接解析本地 db.json.0，绕过 API 限制
  • 写：HTTS POST 到 Twikoo 容器，复用官方接口

状态外置

  • 角度轮换记录 USED_ANGLES 放内存（非持久，重启清空）
  • 配置全放 config.py，支持环境变量注入

相关代码

下述代码统一放在博客根目录下的一个文件中，例如新建 ai_bot

config.py

config.py 作为配置中心模块，集中管理 AI 接口密钥、Twikoo 服务地址、评论策略比例等全局参数，全部支持通过环境变量注入，实现配置与代码分离，便于不同环境快速切换。

SENSENOVA_API_URL这个需要使用自己模型的URL调用地址，SENSENOVA_MODEL切换模型，TWIKOO_URL可以使用域名或IP+PORT

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os

# ========== 日日新 SenseNova 配置 ==========
SENSENOVA_API_KEY = os.getenv("OPENAI_API_KEY", "sk-your-sensenova-key")
SENSENOVA_API_URL = "https://token.sensenova.cn/v1/chat/completions"
SENSENOVA_MODEL = "deepseek-v4-flash"

# Twikoo 配置
TWIKOO_URL = "http://localhost:8080"  
SITE_URL = "https://blog.eucalyptus.cc"

# ========== AI 评论策略配置 ==========
# 新评论 vs 回复评论的比例（0-100）
# 例如：NEW_COMMENT_RATIO = 70 表示 70% 概率发新评论，30% 概率回复
NEW_COMMENT_RATIO = 70

# 日志
LOG_DIR = "/var/log/ai-bot"
os.makedirs(LOG_DIR, exist_ok=True)

OPENAI_API_KEY环境变量配置并持久化

1
2
3
4
5
# 写入 ~/.bashrc
echo 'export OPENAI_API_KEY="sk-your-actual-sensenova-key"' >> ~/.bashrc

# 立即生效
source ~/.bashrc

ai_service.py

ai_service.py 作为 AI 生成服务模块，封装 SenseNova 大模型 API 调用，提供昵称生成、评论生成、回复生成三个核心接口；通过预设多角度评论池与角色扮演策略，结合温度参数和长度随机控制，确保输出内容拟人化且避免重复，同时维护已用角度记录实现单篇文章内的评论多样性。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import requests
import random
from typing import Optional
from .config import SENSENOVA_API_KEY, SENSENOVA_API_URL, SENSENOVA_MODEL

# 评论角度池
COMMENT_ANGLES = [
    "技术实现角度，提出疑问或建议",
    "个人经验角度，分享相关经历",
    "应用场景角度，讨论实际使用",
    "学习收获角度，表达感谢和启发",
    "问题反馈角度，指出可能的坑",
    "扩展思考角度，提出延伸问题",
    "情感共鸣角度，表达认同和感受",
]

# 角色设定池
COMMENT_ROLES = [
    "技术小白，刚入门，有很多疑问",
    "有经验的开发者，喜欢分享经验",
    "学生党，正在学习相关技术",
    "职场老鸟，关注实际应用",
    "爱好者，喜欢折腾新技术",
    "博主同行，交流心得",
]

# 记录已用角度（内存中，重启会清空）
USED_ANGLES = {}


def generate_nick() -> str:
    """AI 生成随机昵称"""
    prompt = """你是一个中文网名生成器，请生成一个2-4个字的博客读者昵称。
要求：
- 像真实的中文网名，有生活气息
- 可以是：技术向、文艺向、幽默向、日常向
- 不要英文，不要符号
- 直接输出昵称，不要解释

示例：代码诗人、深夜咖啡、佛系码农、探索者、路过人间、技术小白"""

    headers = {
        "Authorization": f"Bearer {SENSENOVA_API_KEY}",
        "Content-Type": "application/json"
    }

    data = {
        "model": SENSENOVA_MODEL,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 1.2,
        "max_tokens": 20,
        "reasoning_effort": "none"
    }

    try:
        resp = requests.post(SENSENOVA_API_URL, headers=headers, json=data, timeout=10)
        result = resp.json()
        nick = result['choices'][0]['message']['content'].strip()
        nick = nick.replace("\"", "").replace("'", "").replace("：", "").replace(":", "").strip()
        return nick[:10]
    except Exception as e:
        print(f"昵称生成失败: {e}")
        fallback = ["技术小白", "路过读者", "前端爱好者", "博主粉丝", "学习中的猫", "代码搬运工", "夜猫子程序员", "探索者"]
        return random.choice(fallback)


def generate_comment(title: str, content: str, url: str = "") -> Optional[str]:
    """生成文章评论，支持多角度和角色"""

    # 选择未用过的角度
    used = USED_ANGLES.get(url, [])
    available = [a for a in COMMENT_ANGLES if a not in used]

    if not available:
        # 所有角度用完，重置
        available = COMMENT_ANGLES
        USED_ANGLES[url] = []

    angle = random.choice(available)
    USED_ANGLES[url] = USED_ANGLES.get(url, []) + [angle]

    # 随机角色
    role = random.choice(COMMENT_ROLES)

    # 随机长度要求
    length_req = random.choice([
        "简短有力，30-50字",
        "中等长度，50-80字",
        "稍微详细，80-100字",
    ])

    prompt = f"""你是一位{role}，请根据以下文章生成一条评论。
要求：
- 从"{angle}"出发
- {length_req}
- 像真实人类一样，有个人观点
- 不要太过正式，口语化一点
- 避免和之前评论重复的内容

文章标题：{title}
文章内容摘要：{content[:500]}"""

    headers = {
        "Authorization": f"Bearer {SENSENOVA_API_KEY}",
        "Content-Type": "application/json"
    }

    data = {
        "model": SENSENOVA_MODEL,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 1.5,  # 提高随机性
        "max_tokens": 150,
        "reasoning_effort": "none"
    }

    try:
        resp = requests.post(SENSENOVA_API_URL, headers=headers, json=data, timeout=30)
        result = resp.json()
        return result['choices'][0]['message']['content'].strip()
    except Exception as e:
        print(f"AI 生成失败: {e}")
        return None


def generate_reply(comment_text: str, article_title: str) -> Optional[str]:
    """生成回复评论，使用读者角度"""
    
    prompt = f"""你是一位博客读者，正在回复这篇文章下的评论。
要求：
- 以读者身份回复，不是博主/作者
- 像真实读者一样，分享自己的看法、经验或疑问
- 可以表示认同、补充观点、提出相关问题
- 语气自然、口语化，像朋友聊天
- 不要说"作为作者"、"我会写"、"我的文章"等博主口吻
- 可以说"我也遇到过"、"同感"、"学到了"、"想问下"等读者常用语
- 长度适中，50-100字左右
- ⚠️ 重要：不要@任何人，直接回复即可，不要出现@用户名

文章标题：{article_title}
你要回复的评论：{comment_text[:300]}"""

    headers = {
        "Authorization": f"Bearer {SENSENOVA_API_KEY}",
        "Content-Type": "application/json"
    }
    
    data = {
        "model": SENSENOVA_MODEL,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 1.2,
        "max_tokens": 150,
        "reasoning_effort": "none"
    }
    
    try:
        resp = requests.post(SENSENOVA_API_URL, headers=headers, json=data, timeout=30)
        result = resp.json()
        reply = result['choices'][0]['message']['content'].strip()
        
        # 过滤掉 AI 可能生成的 @用户名
        import re
        reply = re.sub(r'@\w+\s*', '', reply).strip()
        
        return reply
    except Exception as e:
        print(f"AI 生成回复失败: {e}")
        return None

twikoo_client.py

twikoo_client.py 作为 Twikoo 交互层模块，采用双通道设计实现评论数据的读写操作：读取通道直接解析本地 db.json.0 文件获取完整评论列表，写入通道通过 HTTP POST 向 Twikoo 容器提交新评论或回复；同时负责拟人化字段构造（UA、IP、邮箱等）及 URL 编码兼容处理，确保数据层面与真实用户无异。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
# twikoo_client.py

import requests
import random
import json
import os
import urllib.parse
from typing import Optional, Dict, List
from .config import TWIKOO_URL, SITE_URL

# 真实用户 UA 池
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/121.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
]

# 真实 IP 池（国内常见）
IP_POOL = [
    "223.104.212.1", "183.192.123.45", "117.136.62.89",
    "120.197.198.2", "183.60.235.1", "113.96.201.23",
    "116.22.29.1", "125.220.157.89", "42.236.10.1"
]

# Twikoo 本地数据库路径
DB_PATH = "/var/lib/docker/twikoo/data/db.json.0"


def _get_comments_from_db(url: str) -> List[Dict]:
    """从 Twikoo 本地 db.json 读取评论"""
    comments = []
    
    if not os.path.exists(DB_PATH):
        print(f"数据库文件不存在: {DB_PATH}")
        return comments
    
    try:
        with open(DB_PATH, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    doc = json.loads(line)
                    doc_url = doc.get("url", "")
                    
                    # 匹配 url（同时支持编码和未编码）
                    urls_match = (
                        doc_url == url or 
                        doc_url == urllib.parse.unquote(url) or
                        urllib.parse.unquote(doc_url) == url
                    )
                    
                    if not urls_match:
                        continue
                    
                    # 排除垃圾评论和已删除
                    if doc.get("isSpam", False) or doc.get("status") == "deleted":
                        continue
                    
                    # 统一字段格式
                    comment = {
                        "_id": doc.get("_id"),
                        "id": doc.get("id") or doc.get("_id"),
                        "nick": doc.get("nick", ""),
                        "mail": doc.get("mail", ""),
                        "mailMd5": doc.get("mailMd5", ""),
                        "link": doc.get("link", ""),
                        "ua": doc.get("ua", ""),
                        "ip": doc.get("ip", ""),
                        "master": doc.get("master", False),
                        "url": doc_url,
                        "href": doc.get("href", ""),
                        "comment": doc.get("comment", ""),
                        "rid": doc.get("rid", ""),
                        "pid": doc.get("pid", ""),
                        "isSpam": doc.get("isSpam", False),
                        "created": doc.get("created", 0),
                        "updated": doc.get("updated", 0),
                    }
                    comments.append(comment)
                except json.JSONDecodeError:
                    continue
    except Exception as e:
        print(f"读取数据库失败: {e}")
    
    print(f"从数据库读取到 {len(comments)} 条评论")
    return comments


def get_comments(url: str, limit: int = 100) -> List[Dict]:
    """获取评论列表 - 直接读本地数据库"""
    return _get_comments_from_db(url)


def submit_comment(url: str, comment: str, nick: str, mail: str, pid: str = "", rid: str = "") -> bool:
    """提交评论到 Twikoo，支持回复"""
    
    # 1. 移除 SITE_URL 前缀
    if url.startswith(SITE_URL):
        url = url.replace(SITE_URL, "")
    
    # 2. 确保格式正确
    if not url.startswith("/"):
        url = "/" + url
    if not url.endswith("/"):
        url = url + "/"
    
    # 3. 先完全解码，再重新编码（防止双重编码）
    url = urllib.parse.unquote(url)
    parts = url.split('/')
    encoded_parts = [urllib.parse.quote(part, safe='') for part in parts]
    url = '/'.join(encoded_parts)
    
    # 4. 构造 href
    href = SITE_URL + url
    
    ua = random.choice(USER_AGENTS)
    fake_ip = random.choice(IP_POOL)
    
    # 5. 构造 payload
    # 关键修复：顶级评论不要传 pid/rid 字段（而不是传空字符串）
    payload = {
        "event": "COMMENT_SUBMIT",
        "url": url,
        "href": href,
        "comment": comment,
        "nick": nick,
        "mail": mail,
        "link": "",
        "ua": ua,
        "ip": fake_ip,
        "isSpam": False,  # 显式设置，必须和手动评论一致
    }
    
    # 只有回复时才添加 pid 和 rid
    if pid:
        payload["pid"] = pid
    if rid:
        payload["rid"] = rid
    
    headers = {
        "Content-Type": "application/json",
        "User-Agent": ua,
        "X-Forwarded-For": fake_ip,
        "X-Real-IP": fake_ip
    }
    
    try:
        resp = requests.post(TWIKOO_URL, json=payload, headers=headers, timeout=10)
        result = resp.json()
        print(f"  Twikoo 返回: {result}")
        print(f"  提交 URL: {url}")
        print(f"  Payload: {json.dumps(payload, ensure_ascii=False)}")  # 调试用
        return "id" in result or result.get("code") == 0
    except Exception as e:
        print(f"提交失败: {e}")
        return False


def check_ai_comment_exists(url: str) -> bool:
    """检查是否已有 AI 评论（通过 UA 识别）"""
    comments = get_comments(url)
    for c in comments:
        ua = c.get("ua", "")
        if ua in USER_AGENTS:
            return True
    return False

auto_comment.py

auto_comment.py 作为新评论逻辑模块，负责组装并提交单篇文章的首条 AI 评论：依次调用 AI 服务生成评论内容、随机昵称和邮箱，构造完整的 Twikoo 评论数据包，最终通过客户端提交；内置可选的去重检查机制，避免同一篇文章被重复评论。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import random
from .ai_service import generate_comment, generate_nick
from .twikoo_client import submit_comment, check_ai_comment_exists


def generate_mail() -> str:
    """生成邮箱，Twikoo 自动显示 Gravatar 头像"""
    seed = random.randint(1000, 9999)
    return f"user{seed}@example.com"


def process_article(url: str, title: str, content: str) -> bool:
    """处理单篇文章：发新评论"""
    print(f"处理: {title}")
    
    # 检查是否已有 AI 评论（可选，如果想多次评论可以注释掉）
    # if check_ai_comment_exists(url):
    #     print("  已有 AI 评论，跳过")
    #     return False
    
    comment_text = generate_comment(title, content, url)
    if not comment_text:
        print("  AI 生成失败")
        return False
    
    print(f"  生成: {comment_text[:50]}...")
    
    nick = generate_nick()
    print(f"  昵称: {nick}")
    
    mail = generate_mail()
    print(f"  邮箱: {mail}")
    
    if submit_comment(url, comment_text, nick, mail):
        print(f"  ✅ 提交成功")
        return True
    else:
        print(f"  ❌ 提交失败")
        return False

reply_bot.py

reply_bot.py 作为回复评论逻辑模块，负责筛选并回复文章下现有评论：读取 Twikoo 本地数据库获取未回复评论列表（排除 AI 自身评论），调用 AI 服务生成针对性回复内容，正确设置 pid 和 rid 字段确保回复层级关系准确，实现模拟真实读者间的互动对话。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
# reply_bot.py

import random
from typing import List, Dict
from .ai_service import generate_reply, generate_nick  # 导入 generate_nick
from .twikoo_client import get_comments, submit_comment


def get_unreplied_comments(url: str, bot_nicks: List[str]) -> List[Dict]:
    """获取未回复的评论（排除自己的评论）"""
    comments = get_comments(url)
    
    unreplied = []
    
    # 收集所有被回复的评论 ID（用 pid）
    replied_ids = set()
    for c in comments:
        pid = c.get("pid")
        if pid:
            replied_ids.add(pid)
    
    print(f"已被回复的评论ID: {replied_ids}")
    
    for comment in comments:
        # 获取评论 ID（优先 _id）
        comment_id = comment.get("_id") or comment.get("id")
        nick = comment.get("nick", "unknown")
        
        # 跳过自己的评论
        if nick in bot_nicks:
            continue
        
        # 检查是否已被回复
        if comment_id and comment_id in replied_ids:
            continue
        
        unreplied.append(comment)
    
    print(f"发现 {len(unreplied)} 条未回复评论")
    return unreplied


def reply_to_comment(parent_comment: Dict, article_title: str) -> bool:
    parent_nick = parent_comment.get("nick", "网友")
    parent_text = parent_comment.get("comment", "")[:30]
    print(f"回复 [{parent_nick}]: {parent_text}...")
    
    # 生成回复内容
    reply_text = generate_reply(
        comment_text=parent_comment.get("comment", ""),
        article_title=article_title
    )
    
    if not reply_text:
        print("  回复生成失败")
        return False
    
    print(f"  回复: {reply_text[:50]}...")
    
    # 随机生成昵称
    reply_nick = generate_nick()
    print(f"  昵称: {reply_nick}")
    
    # 获取父评论ID
    parent_id = parent_comment.get("_id") or parent_comment.get("id")
    
    # 获取父评论的 rid
    parent_rid = parent_comment.get("rid", "")
    if parent_rid:
        # 父评论本身是回复，继承其 rid
        root_id = parent_rid
    else:
        # 父评论是顶级评论，rid 就是父评论自己的 id
        root_id = parent_id
    
    # 提交回复（不加 @，Twikoo 前端自动显示）
    result = submit_comment(
        url=parent_comment.get("url", "/"),
        comment=reply_text,
        nick=reply_nick,
        mail=f"ai-reply-{random.randint(1000,9999)}@example.com",
        pid=parent_id,
        rid=root_id,
    )
    
    return result


def run_reply_bot(url: str, bot_nicks: List[str], max_reply: int = 3):
    """运行回复机器人"""
    print(f"检查文章: {url}")
    
    unreplied = get_unreplied_comments(url, bot_nicks)
    
    # 随机打乱顺序
    random.shuffle(unreplied)
    
    # 取前 max_reply 条
    to_reply = unreplied[:max_reply]
    
    print(f"选择回复 {len(to_reply)}/{len(unreplied)} 条评论")
    
    success = 0
    for comment in to_reply:
        if reply_to_comment(comment, "文章标题"):
            success += 1
    
    print(f"完成: 回复 {success}/{len(to_reply)} 条评论")
    return success

main.py

main.py 作为入口调度模块，负责整合全系统运行流程：自动扫描 Hexo 生成目录发现文章列表，按配置比例随机调度新评论或回复模式，提供 CLI 参数支持单篇/批量/范围选择及临时比例覆盖，最终输出执行统计报告，是日常定时任务调用的唯一入口。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI 博客评论机器人 - 智能模式
按配置比例随机选择发新评论或回复现有评论
"""

import sys
import argparse
import os
import re
import urllib.parse
import random
from typing import List, Dict
from .reply_bot import run_reply_bot
from .ai_service import generate_comment, generate_nick
from .twikoo_client import submit_comment
from .config import NEW_COMMENT_RATIO


# ========== 文章加载 ==========

def get_hexo_articles(public_dir: str = "/www/wwwroot/myblog/public") -> List[Dict]:
    """从 Hexo 生成目录获取文章列表"""
    articles = []
    for root, dirs, files in os.walk(public_dir):
        rel_path = os.path.relpath(root, public_dir)
        if not re.match(r'^\d{4}/\d{2}/\d{2}/', rel_path):
            continue
        if "index.html" not in files:
            continue
        url = f"/{rel_path}/"
        dir_name = os.path.basename(rel_path)
        title = urllib.parse.unquote(dir_name)
        articles.append({"url": url, "title": title, "content": ""})
    return articles


def load_articles() -> List[Dict]:
    """加载文章列表"""
    articles = get_hexo_articles()
    if articles:
        print(f"自动获取到 {len(articles)} 篇文章")
        return articles
    print("自动获取失败")
    return []


# ========== 新评论 ==========

def generate_mail() -> str:
    """生成随机邮箱"""
    return f"user{random.randint(1000, 9999)}@example.com"


def process_new_comment(url: str, title: str, content: str = "") -> bool:
    """发新评论"""
    print(f"处理: {title}")

    comment_text = generate_comment(title, content, url)
    if not comment_text:
        print("  AI 生成失败")
        return False

    print(f"  生成: {comment_text[:50]}...")

    nick = generate_nick()
    mail = generate_mail()
    print(f"  昵称: {nick}")
    print(f"  邮箱: {mail}")

    if submit_comment(url, comment_text, nick, mail):
        print(f"  ✅ 提交成功")
        return True
    else:
        print(f"  ❌ 提交失败")
        return False


# ========== 智能调度 ==========

def process_article_smart(article: Dict, bot_nicks: List[str], max_reply: int = 3) -> Dict:
    """智能处理：随机选择发新评论或回复"""

    url = article["url"]
    title = article["title"]
    is_new = random.randint(1, 100) <= NEW_COMMENT_RATIO

    result = {
        "url": url, "title": title,
        "mode": "new" if is_new else "reply",
        "success": False, "detail": ""
    }

    if is_new:
        print(f"\n📝 [{title}] 随机选择：发新评论")
        success = process_new_comment(url, title, article.get("content", ""))
        result["success"] = success
        result["detail"] = "新评论提交成功" if success else "新评论提交失败"
    else:
        print(f"\n💬 [{title}] 随机选择：回复评论")
        count = run_reply_bot(url, bot_nicks, max_reply=max_reply)
        result["success"] = count > 0
        result["detail"] = f"回复了 {count} 条评论"

    return result


def run_smart(articles: List[Dict], bot_nicks: List[str] = None, max_reply: int = 3):
    """批量智能处理"""
    if bot_nicks is None:
        bot_nicks = ["AI助手", "博主助手"]

    stats = {"new": 0, "reply": 0, "new_ok": 0, "reply_ok": 0}

    for article in articles:
        r = process_article_smart(article, bot_nicks, max_reply)
        stats[r["mode"]] += 1
        if r["success"]:
            stats[f"{r['mode']}_ok"] += 1

    print(f"\n{'='*50}")
    print("📊 执行统计")
    print(f"{'='*50}")
    print(f"总文章: {len(articles)}")
    print(f"新评论: {stats['new']} 篇 (成功 {stats['new_ok']})")
    print(f"回  复: {stats['reply']} 篇 (成功 {stats['reply_ok']})")
    print(f"设置比例: {NEW_COMMENT_RATIO}%")
    actual = round(stats['new']/len(articles)*100, 1) if articles else 0
    print(f"实际比例: {actual}%")
    print(f"{'='*50}")


# ========== CLI ==========

def main():
    parser = argparse.ArgumentParser(description="AI Blog Bot")
    parser.add_argument("command", choices=["run", "list"])
    parser.add_argument("--id", type=str, help="文章序号，如 12 或 1,3,5-10")
    parser.add_argument("--count", type=int, default=3, help="回复上限（默认3）")
    parser.add_argument("--ratio", type=int, help="临时新评论比例 1-100")

    args = parser.parse_args()

    if args.command == "list":
        articles = load_articles()
        print(f"\n{'序号':<6}{'标题':<40}{'URL'}")
        print("-" * 80)
        for i, a in enumerate(articles, 1):
            print(f"{i:<6}{a['title'][:38]:<40}{a['url'][:50]}")
        print(f"\n共 {len(articles)} 篇")
        return

    # 临时覆盖比例
    if args.ratio is not None:
        global NEW_COMMENT_RATIO
        NEW_COMMENT_RATIO = max(1, min(100, args.ratio))
        print(f"临时比例: {NEW_COMMENT_RATIO}%")

    bot_nicks = ["AI助手", "博主助手"]
    articles = load_articles()

    if args.id:
        selected = []
        for part in args.id.split(','):
            if '-' in part:
                s, e = map(int, part.split('-'))
                selected.extend([articles[i-1] for i in range(s, e+1) if 1 <= i <= len(articles)])
            else:
                i = int(part)
                if 1 <= i <= len(articles):
                    selected.append(articles[i-1])
        print(f"处理 {len(selected)} 篇")
        run_smart(selected, bot_nicks, max_reply=args.count)
    else:
        run_smart(articles, bot_nicks, max_reply=args.count)


if __name__ == "__main__":
    main()

使用方法

基础用法

1
2
3
4
5
6
7
8
9
10
11
#查看文章列表
python -m ai_bot.main list

# 处理单篇文章（按配置比例随机选择模式）
python -m ai_bot.main run --id 3

# 处理多篇
python -m ai_bot.main run --id 1,3,5

# 处理范围
python -m ai_bot.main run --id 1-10

模式控制

1
2
3
4
5
6
7
8
# 强制发新评论（100% 概率）
python -m ai_bot.main run --id 3 --ratio 100

# 强制回复模式（0% 概率发新评论）
python -m ai_bot.main run --id 3 --ratio 0

# 回复模式下，最多回复 5 条现有评论
python -m ai_bot.main run --id 3 --ratio 0 --count 5

批量处理

1
2
3
4
5
# 全部文章（按默认比例）
python -m ai_bot.main run

# 指定范围 + 强制新评论
python -m ai_bot.main run --id 5-20 --ratio 100

相关参数

参数 必填 默认值 说明
--id 否 全部 文章序号，如 3 / 1,3 / 5-10
--ratio 否 配置值 临时覆盖新评论比例，1-100
--count 否 3 回复模式下单篇最大回复数

后续扩展

AI 互动机器人：注册固定昵称（如 @小助手），访客可在评论区 @ 主动提问，AI 实时检测触发词并回复，支持多轮对话，从单向评论输出升级为双向互动助手。