DouyinCommentBot：抖音直播评论自动化工具与 AI 内容助手整理

一、项目简介

DouyinCommentBot 是我整理过的一个基于 Python + ADB 的 Android 真机自动化工具。项目最早是为了练习 ADB 控制、SRT 时间轴解析、桌面 GUI 和多线程任务管理，后来逐步补了用途确认、发送上限、冷却时间、日志记录、内容清洗和异常处理。

这次重新整理时，我在原来的自动化工具基础上加了一层 AI 内容助手：不是让它做刷量或控评，而是用于自有直播间测试、评论素材草稿生成、SRT 台词改写、内容合规检查、运行日志总结和项目知识库问答。也就是说，ADB 负责“自动化流程验证”，AI 负责“内容准备和辅助分析”，两者之间加安全过滤和人工确认。

这个项目适合展示几个能力点：Python 脚本开发、ADB 真机自动化、Tkinter GUI、多线程、SRT 解析、配置化任务、日志队列、OpenAI / Claude API 调用、Prompt Engineering、RAG 检索和 Agent 工具调用设计。

二、使用边界

这个项目必须明确边界：它适合用于自有直播间测试、自动化流程验证、教学演示、ADB 操作练习、内容素材整理和多设备任务调度实验；不应用于刷量、控评、骚扰、批量营销、绕过平台规则或任何影响他人体验的用途。

所以我在设计里加了几个限制：启动前用途确认、单设备最大发送条数、冷却时间、内容长度限制、重复内容过滤、敏感词过滤、日志留存、AI 输出人工确认。AI 模块只生成候选内容，不直接跳过审核发送。

三、总体架构

项目分成六层：素材输入、AI 内容助手、安全过滤层、GUI 控制台、ADB 执行层和 Android 真机。素材可以来自 TXT、SRT 或人工输入的主题；AI 模块负责生成、改写和整理；过滤层负责频率、长度、重复度和敏感词检查；GUI 负责参数配置和运行控制；ADB 层负责设备操作；最后在真机里验证流程。

架构上我尽量让各层独立：AI 模块不直接操作手机，ADB 模块不关心内容怎么生成，GUI 只做状态展示和任务控制。这样后续即使不用 AI，也可以继续跑 TXT/SRT；即使用 AI，也必须经过安全过滤和人工确认。

四、核心功能

1. TXT 评论素材模式

TXT 模式读取 text 目录下的文本文件，每一行作为一条候选内容。程序会先清洗空行、过长内容和重复内容，然后根据配置随机选择。这个模式适合测试输入框、剪贴板、点击坐标和日志流程。

2. SRT 时间轴模式

SRT 模式会读取字幕文件，根据时间轴顺序发送内容。这个模式适合做直播脚本演示或弹幕时间轴测试。后面加入 AI 后，也可以让模型先把 SRT 台词改写成更口语化的评论草稿，再由人工审核后放入测试。

3. GUI 控制台

GUI 使用 Tkinter 实现，主要包含设备列表、文件选择、坐标配置、发送间隔、最大条数、开始/暂停/停止按钮和日志窗口。日志通过队列从工作线程传回主线程，避免界面卡死。

4. AI 内容助手

AI 内容助手主要做四件事：根据主题生成候选评论、把 SRT 台词改写成短句、检查内容是否重复或不合适、总结运行日志。它不会直接连接 ADB，所有输出都要进入待审核列表。

五、目录结构

DouyinCommentBot/
  app.py                    # GUI 启动入口
  adb_client.py             # ADB 设备控制封装
  task_runner.py            # 多线程任务调度
  srt_parser.py             # SRT 时间轴解析
  content_filter.py         # 长度、重复、敏感词过滤
  ai_assistant.py           # AI 内容生成和总结
  rag_store.py              # 项目文档检索
  prompts/
    comment_prompt.txt
    report_prompt.txt
  text/
    demo.txt
  logs/
  config.yaml

六、ADB 控制代码片段

ADB 操作部分尽量封装成简单函数：检测设备、点击坐标、设置剪贴板、执行粘贴、发送按键。坐标使用比例配置，而不是写死像素，方便适配不同分辨率。

import subprocess
from dataclasses import dataclass

@dataclass
class DeviceConfig:
    serial: str
    input_x_ratio: float = 0.5
    input_y_ratio: float = 0.9
    send_x_ratio: float = 0.92
    send_y_ratio: float = 0.9

def run_adb(serial: str, *args: str) -> str:
    cmd = ["adb", "-s", serial, *args]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
    if result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or "adb command failed")
    return result.stdout.strip()

def tap(serial: str, x: int, y: int):
    run_adb(serial, "shell", "input", "tap", str(x), str(y))

def paste_text(serial: str, text: str):
    safe_text = text.replace("%", "%25").replace(" ", "%s")
    run_adb(serial, "shell", "input", "text", safe_text)

七、SRT 解析代码片段

SRT 文件格式看起来简单，但真实文件里会遇到空行、编号缺失、时间格式不统一、多行字幕等情况。所以解析时不只按固定行号读取，而是用正则提取时间段，再合并文本。

import re
from dataclasses import dataclass

@dataclass
class SubtitleLine:
    start_ms: int
    end_ms: int
    text: str

def to_ms(value: str) -> int:
    h, m, rest = value.split(":")
    s, ms = rest.split(",")
    return (int(h) * 3600 + int(m) * 60 + int(s)) * 1000 + int(ms)

def parse_srt(raw: str) -> list[SubtitleLine]:
    blocks = re.split(r"\n\s*\n", raw.strip())
    lines = []
    for block in blocks:
        match = re.search(r"(\d{2}:\d{2}:\d{2},\d{3})\s+-->\s+(\d{2}:\d{2}:\d{2},\d{3})", block)
        if not match:
            continue
        text = re.sub(r"^\d+\s*", "", block).split(match.group(0), 1)[-1]
        text = " ".join(x.strip() for x in text.splitlines() if x.strip())
        lines.append(SubtitleLine(to_ms(match.group(1)), to_ms(match.group(2)), text))
    return lines

八、AI 评论素材生成

AI 生成评论素材时，Prompt 不能写成“生成很多重复弹幕”。我的处理方式是让模型生成候选草稿，并要求内容自然、短、无攻击性、无诱导、无敏感信息。生成后还要经过本地过滤，再人工确认。

你是一个直播内容助理，只为自有直播间测试生成候选评论草稿。

要求：
1. 每条 8 到 20 个中文字符。
2. 语气自然，边界清晰，不诱导互动。
3. 不包含攻击、骚扰、营销、刷量、违规词。
4. 只输出 JSON 数组。

直播主题：{topic}
商品/内容要点：{points}
需要生成数量：{count}

import os
import json
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def generate_comment_candidates(topic: str, points: list[str], count: int = 20) -> list[str]:
    prompt = COMMENT_PROMPT.format(topic=topic, points="、".join(points), count=count)
    resp = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[
            {"role": "system", "content": "你是谨慎的内容草稿助手，不能生成骚扰、刷量或违规内容。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.5,
    )
    data = json.loads(resp.choices[0].message.content)
    return [x.strip() for x in data if isinstance(x, str) and x.strip()]

九、内容过滤和去重

AI 生成的内容不能直接使用，所以本地做一层过滤。过滤包含长度、重复、敏感词、最近窗口去重。项目里我把过滤函数写成纯函数，方便用 pytest 测。

class ContentFilter:
    def __init__(self, banned_words: set[str], max_len: int = 30):
        self.banned_words = banned_words
        self.max_len = max_len
        self.recent = []

    def clean(self, text: str) -> str:
        return " ".join(text.replace("\n", " ").split())

    def allow(self, text: str) -> bool:
        text = self.clean(text)
        if not text or len(text) > self.max_len:
            return False
        if text in self.recent:
            return False
        if any(word in text for word in self.banned_words):
            return False
        return True

    def push_recent(self, text: str, window: int = 20):
        self.recent.append(text)
        self.recent = self.recent[-window:]

十、RAG 知识库问答

这个项目也适合加一个小型 RAG：把 README、使用说明、常见报错、ADB 命令、设备连接问题放进知识库。遇到“设备 offline 怎么办”“为什么输入框点不到”“SRT 解析失败怎么办”这类问题时，先检索本地文档，再让 AI 根据资料回答。

def build_answer_prompt(question: str, chunks: list[str]) -> str:
    context = "\n---\n".join(chunks[:5])
    return f"""
你是项目文档助手，只能根据资料回答。资料不足时请说明需要人工检查。

资料：
{context}

问题：{question}
"""

def answer_question(question: str):
    chunks = rag_store.search(question, top_k=5)
    prompt = build_answer_prompt(question, [c.text for c in chunks])
    return llm.complete(prompt)

十一、Agent 调度思路

Agent 不负责发送内容，只负责辅助排查和生成任务建议。比如它可以调用读取日志、检查设备、解析 SRT、检查配置这些只读工具，然后给出建议。这样可以展示 Agent 思路，又不会让 AI 直接操作手机。

TOOLS = {
    "list_devices": list_devices,
    "read_recent_logs": read_recent_logs,
    "validate_srt": validate_srt,
    "check_config": check_config,
}

def run_diagnosis_agent(question: str):
    plan = planner.plan(question, available_tools=list(TOOLS))
    observations = []
    for step in plan.steps:
        if step.tool not in TOOLS:
            continue
        result = TOOLS[step.tool](**step.arguments)
        observations.append({"tool": step.tool, "result": result})
    return summarizer.summarize(question, observations)

十二、GUI 任务线程

GUI 程序不能直接在主线程里跑循环，否则界面会卡死。所以任务执行放在线程里，日志通过 queue 传回主界面。停止按钮通过 Event 控制。

import queue
import threading
import time

class TaskRunner:
    def __init__(self):
        self.log_queue = queue.Queue()
        self.stop_event = threading.Event()

    def start(self, device, messages, interval):
        self.stop_event.clear()
        thread = threading.Thread(
            target=self._run,
            args=(device, messages, interval),
            daemon=True,
        )
        thread.start()

    def _run(self, device, messages, interval):
        for message in messages:
            if self.stop_event.is_set():
                self.log_queue.put("任务已停止")
                break
            self.log_queue.put(f"准备发送：{message}")
            # send_message_by_adb(device, message)
            time.sleep(interval)

十三、测试用例

测试重点不放在真实发送，而是放在解析、过滤、配置和任务状态。这样不连接手机也能测试核心逻辑。

def test_content_filter_blocks_duplicate():
    f = ContentFilter(banned_words=set(), max_len=20)
    assert f.allow("今天状态不错")
    f.push_recent("今天状态不错")
    assert not f.allow("今天状态不错")

def test_srt_parser_extracts_text():
    raw = """1
00:00:01,000 --> 00:00:03,000
欢迎来到直播间
"""
    lines = parse_srt(raw)
    assert len(lines) == 1
    assert lines[0].text == "欢迎来到直播间"

十四、项目复盘

这个项目从一个 ADB 小脚本整理成工具包，最大的收获是把“能跑”和“可控”分开看。自动化脚本很容易写到能执行，但要变成可维护工具，就要补配置、日志、暂停、停止、异常处理、测试和使用边界。

加入 AI 后，项目的展示面更完整：AI 可以生成素材草稿、改写 SRT、总结日志、回答使用问题，但核心自动化仍然由确定性代码控制。这样既能体现大模型 API、Prompt、RAG、Agent 的理解，也不会把项目写成不合规的批量操作工具。

如果后续继续完善，我会把 AI 输出做成“候选池”，每条内容都显示来源、生成时间、风险提示和人工审核状态；把日志总结做成日报；把设备诊断 Agent 做成只读工具；再用 Docker 打包一个本地服务版本，方便在不同电脑上复现。