ReplyHey 给自己的定位是 Autopilot让用户通过自动回复从 Reddit 寻找潜在客户。真正动手做这个主题之前要先把它从营销叙事翻译成工程问题监控哪些 subreddit、按什么标准筛选帖子、怎样生成不惹人厌的回复、如何控制频率以及怎样在 Reddit 的 API 限制和社区规则下安全运行。这些工作可以拆成一个最小可复现系统一个进程负责扫描一个队列负责存储一个审核步骤负责把关一个 API 调用负责真正提交回复。本文会按这条主线搭建一个 Python 示例从 Reddit 应用注册开始到扫描、过滤、生稿、审核、提交和排查结束。文章假定你已经掌握 Python 基础了解 HTTP 和 OAuth 的基本概念。先从整体设计开始。1. 先理解 ReplyHey 这类工具的技术本质1.1 自动回复不是“自动群发”而是“自动筛选 人工确认”“Autopilot”这个词容易让人产生误解是不是写一个脚本看到有人问“有没有推荐工具”就立刻冲上去回复自己的产品如果这样做技术难度确实很低但它不是可长期运行的系统因为 Reddit 用户对广告非常敏感subreddit 版务也会封禁频繁推广的账号。真正值得实现的是“自动筛选 草稿生成 人工确认”这条流水线。整个系统的本质可以概括为读取一个或多个 subreddit 的新帖和评论流。用规则判断该讨论是否存在真实需求。如果存在生成一段回复草稿。草稿进入审核队列由运营人员决定是否回复、修改后回复或放弃。只有审核通过的草稿才会通过 Reddit API 真正提交。所有扫描、审核、提交动作都写入日志和数据库。这样设计的好处是自动化处理了重复劳动但最终决定权仍然留在人手里出问题时可以回溯是哪一份草稿、哪次提交造成的。1.2 系统边界范围、频率、状态和审计技术实现前要先定义边界否则写着写着就会失控。范围不能扫描整个 Reddit要限定在跟产品场景相关的 subreddit。产品是发票工具就只看 r/freelance、r/smallbusiness、r/invoicing 等产品是开发者工具就只看 r/webdev、r/selfhosted 等。频率Reddit API 有速率限制程序内部也要设置自己的调度间隔。代码里宁可保守也不要把请求打满。状态每条帖子在数据库里必须有一条完整状态记录常见状态包括 pending、approved、rejected、replied、skipped、failed。有了状态才能做幂等处理避免重复回复。审计每条回复都要保留原始帖子标题、链接、生成时间、回复内容、操作人。这不仅是工程要求也是平台规则和口碑管理的需要。这三层约束会直接影响后面的数据库设计和主循环写法。1.3 平台规则对架构设计的影响Reddit 对自动化请求有明确要求虽然是社区平台API 规则更像一家严肃的 SaaS 服务。基础要求包括使用 OAuth2 认证不能再用匿名接口做业务操作。User-Agent 必须能识别调用方身份。对单个端点的请求需要控制频率遇到 429 必须退避。自动回复如果触发反垃圾机制账号会被限流或封禁。不同 subreddit 有不同的推广规则代码需要支持按 subreddit 配置黑名单。这些要求不是安全文档里的空话它们会直接影响架构。比如“不同 subreddit 有不同规则”意味着过滤层不能只做全局关键词匹配还要允许运营为每个 subreddit 单独配置允许/禁止名单“遇到 429 必须退避”意味着提交回复的任务不能像普通循环一样一梭子全部打出去而要有队列和重试间隔。这也是本文示例中把“扫描”和“提交”分成两个模块的原因。扫描模块需要高频读取提交模块需要低频写入二者混在一起会让限流处理非常难做。2. 环境准备与 Reddit API 接入2.1 注册应用与获取凭据要让脚本使用 Reddit 账号执行操作需要创建一组 API 凭据。访问 Reddit 的偏好页面下的 apps 页面创建一个 script 类型的应用提交后会得到两段关键内容client_id应用标识界面里通常显示为一串短 ID。client_secret应用密钥只有创建者能看到需要保存好。同时脚本还需要使用一个真实 Reddit 账号的用户名和密码。由于脚本通过 OAuth2 的 password grant 方式换取 token因此账号密码要作为环境变量传入不要写进代码。配置项说明示例REDDIT_CLIENT_ID应用 ID1a2b3c4d5eREDDIT_CLIENT_SECRET应用密钥一段长随机字符串REDDIT_USERNAME执行自动回复的账号名my_bot_accountREDDIT_PASSWORD对应账号密码不要写入代码或仓库REDDIT_USER_AGENT标识调用方的请求头invoice-tool/0.1 by my_bot_account需要注意一个坑不要用日常交流的主账号做自动化测试。自动回复即使合规也可能因为 subreddit 规则、图片处理、临时误报被版务封禁。推荐的做法是创建专用账号并设置一个能表明身份的用户名。2.2 安装依赖并用最小脚本验证连通性Python 生态里最常用的 Reddit API 客户端是 PRAW。下面的依赖清单适用于多数本地开发环境实际安装时以官方最新版本为准praw7.7.1 python-dotenv1.0.0配置信息放到.env文件读取时使用 python-dotenvREDDIT_CLIENT_ID1a2b3c4d5e REDDIT_CLIENT_SECRETyour_client_secret REDDIT_USERNAMEmy_bot_account REDDIT_PASSWORDyour_account_password REDDIT_USER_AGENTinvoice-tool/0.1 by my_bot_account初始化客户端的代码# reddit_client.py import os import praw from dotenv import load_dotenv load_dotenv() def build_reddit_client() - praw.Reddit: return praw.Reddit( client_idos.environ[REDDIT_CLIENT_ID], client_secretos.environ[REDDIT_CLIENT_SECRET], usernameos.environ[REDDIT_USERNAME], passwordos.environ[REDDIT_PASSWORD], user_agentos.environ[REDDIT_USER_AGENT], )验证连通性的最小脚本# check_connection.py from reddit_client import build_reddit_client reddit build_reddit_client() print(reddit.user.me())运行后如果能打印出账号名说明认证成功。这里最常见的坑是 User-Agent 设置过于简单例如直接写成python:mybot version 1.0。Reddit 官方要求 User-Agent 能够唯一识别调用方。推荐格式包含应用名、版本号和执行账号名。2.3 配置项目结构和环境变量为了让示例可维护建议先定义项目结构replyhey/ ├── .env ├── requirements.txt ├── reddit_client.py ├── schema.sql ├── scanner.py ├── draft.py ├── review.py ├── reply.py ├── database.py └── logs/database.py负责 SQLite 连接生产环境可以替换成 PostgreSQL 或 MySQL。.env必须加入.gitignore避免密钥提交到仓库。这里要说清环境变量和配置文件的选择环境变量适合放密钥配置文件适合放可变更的业务规则比如关键词、subreddit 名单、回复模板。二者不要混用。3. 核心实现扫描讨论、过滤相关帖子、生成草稿3.1 定义帖子模型与 SQLite 表先建一张表记录扫描到的候选帖子和后续状态。SQLite 在本地够用生产环境建议迁移到 PostgreSQL-- schema.sql CREATE TABLE IF NOT EXISTS submissions ( id TEXT PRIMARY KEY, subreddit TEXT NOT NULL, title TEXT NOT NULL, body TEXT NOT NULL DEFAULT , url TEXT, score INTEGER DEFAULT 0, relevance_score REAL DEFAULT 0, state TEXT DEFAULT pending, candidate_keywords TEXT DEFAULT , created_utc INTEGER, scanned_utc INTEGER, reviewed_by TEXT, reviewed_utc INTEGER, reply_text TEXT, reply_id TEXT, error_message TEXT ); CREATE INDEX IF NOT EXISTS idx_submissions_state ON submissions(state); CREATE INDEX IF NOT EXISTS idx_submissions_subreddit ON submissions(subreddit);id 就是 Reddit 帖子的 base36 id用它做主键可以天然去重。relevance_score 用来记录相关度排序依据candidate_keywords 记录命中了哪些关键词方便审核人员快速判断。这里的核心设计是“状态机”pending 表示已入队但未审核approved 表示人工确认可以回复rejected 表示放弃replied 表示成功提交skipped 表示因规则跳过failed 表示提交失败。所有后续逻辑都围绕这个状态流转。3.2 扫描指定 subreddit 并做关键词过滤扫描逻辑按 subreddit 分组对每个 subreddit 读取新帖不需要把一个 subreddit 的几千篇历史帖全部拉下来。示例中先读取最近 50 篇新帖再结合关键词判断# scanner.py import logging import sqlite3 import time from reddit_client import build_reddit_client logger logging.getLogger(__name__) SUBREDDITS [freelance, smallbusiness, invoicing] POSITIVE_KEYWORDS [invoice, client, payment, get paid, late payment] NEGATIVE_KEYWORDS [I sell invoice paper, invoice management software comparison] MIN_SCORE 2 SCAN_LIMIT 50 def relevance(text: str) - float: score 0 matched [] for kw in POSITIVE_KEYWORDS: if kw in text: score 2 matched.append(kw) for kw in NEGATIVE_KEYWORDS: if kw in text: score - 4 return score, matched def scan_once(conn: sqlite3.Connection) - int: reddit build_reddit_client() inserted 0 for sub_name in SUBREDDITS: sub reddit.subreddit(sub_name) for submission in sub.new(limitSCAN_LIMIT): text f{submission.title}\n{submission.selftext}.lower() score, matched relevance(text) if score MIN_SCORE: continue cur conn.execute( INSERT OR IGNORE INTO submissions (id, subreddit, title, body, url, score, relevance_score, candidate_keywords, created_utc, scanned_utc) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( submission.id, sub_name, submission.title, submission.selftext[:2000], submission.url, submission.score, score, ,.join(matched), int(submission.created_utc), int(time.time()), ), ) inserted cur.rowcount conn.commit() return inserted这里有几个关键判断。第一为什么不直接给所有帖子回复因为关键词匹配只是弱信号它只能表示帖子在讨论相关主题不能表示用户一定想要你的产品。第二为什么要有 NEGATIVE_KEYWORDS因为“发票管理软件对比”这类帖子里虽然出现关键词大概率是竞品调研不是商机。第三为什么只扫描新帖而不是全站搜索因为新帖时效性好也更容易控制 API 频率。最常踩的坑是直接对所有命中关键词的帖子生成回复完全忽略 subreddit 自身的推广规则。比如某些 subreddit 明令禁止自我推广你的关键词再准也不能回复。后续实现中要加入“subreddit 黑名单”和“每 subreddit 只允许回复 N 次/天”的策略。3.3 生成回复草稿并落库过滤后进入草稿生成阶段。为了说明完整流程先用模板生成草稿后续可以替换成大模型生成# draft.py import sqlite3 TEMPLATE ( 我们是做 {product} 的开发者正好看到你在讨论 {topic}。 先说明一下身份这个回复和我们的产品直接相关。 如果你的场景和下面的描述一致可以看一下官方文档{doc_url}。 如果没有帮助直接忽略即可。 ) def generate_draft(submission_row, product发票管理工具, doc_urlhttps://example.com/docs) - str: topic (submission_row[title] or ).strip() return TEMPLATE.format(productproduct, topictopic, doc_urldoc_url)草稿为什么要明文保存而不是直接提交因为人工审核需要看到完整内容而且如果生成模型出问题至少知道发给用户的是什么。不要把草稿只放在内存里否则程序重启后待审核队列就丢失了。生成草稿的函数可以单独写也可以合并在扫描流程中。推荐在扫描后生成草稿再进入人工审核接口。3.4 状态与幂等设计幂等是这个系统最容易忽略的地方。Reddit 帖子不会因为程序重启而消失同一篇帖子如果被扫描到两次不能插入两条记录更不能回复两次。使用INSERT OR IGNORE可以避免重复入库。提交回复前必须再检查一次数据库状态def can_reply(conn: sqlite3.Connection, submission_id: str) - bool: row conn.execute( SELECT state FROM submissions WHERE id ?, (submission_id,) ).fetchone() return row is not None and row[0] approved同时建议在事务里更新状态conn.execute( UPDATE submissions SET state ?, reply_id ?, error_message ? WHERE id ? AND state ?, (replied, reply_id, None, submission_id, approved), )这样即使提交过程中抛异常也不会把状态改乱。另一个坑是提交后没有保存 reply_id导致排查时不知道哪条评论是自己的。4. 人工审核环节不要绕过也不要侥幸4.1 命令行审核队列审核无需一开始就上 Web 界面命令行工具已经能支撑早期使用。示例代码# review.py import sqlite3 import sys def list_pending(conn: sqlite3.Connection): rows conn.execute( SELECT id, subreddit, title, relevance_score, reply_text FROM submissions WHERE state pending ORDER BY relevance_score DESC LIMIT 20 ).fetchall() for row in rows: print(f{row[id]} | {row[subreddit]} | score{row[relevance_score]}) print(f title: {row[title]}) print(f draft: {row[reply_text]}) def approve(conn: sqlite3.Connection, submission_id: str): conn.execute( UPDATE submissions SET state approved WHERE id ? AND state pending, (submission_id,), ) conn.commit() def reject(conn: sqlite3.Connection, submission_id: str): conn.execute( UPDATE submissions SET state rejected WHERE id ? AND state pending, (submission_id,), ) conn.commit()使用方法python review.py list python review.py approve 1abcde python review.py reject 1abcde为什么要这一步因为程序判断的是“话题相关”人判断的是“这个讨论是否适合以我们身份出现、语气是否合适、是否违反 subreddit 规则”。这一步是过滤器也是防线。4.2 审核通过后自动提交回复提交逻辑是独立脚本只处理 approved 状态的记录并且要限制每次运行最多提交几条# reply.py import sqlite3 import time from reddit_client import build_reddit_client MAX_REPLIES_PER_RUN 5 def reply_approved(conn: sqlite3.Connection) - int: reddit build_reddit_client() rows conn.execute( SELECT id, reply_text FROM submissions WHERE state approved LIMIT ?, (MAX_REPLIES_PER_RUN,), ).fetchall() replied 0 for row in rows: submission_id row[id] try: submission reddit.submission(idsubmission_id) comment submission.reply(row[reply_text]) conn.execute( UPDATE submissions SET state replied, reply_id ? WHERE id ?, (comment.id, submission_id), ) conn.commit() replied 1 except Exception as exc: conn.execute( UPDATE submissions SET state failed, error_message ? WHERE id ?, (str(exc), submission_id), ) conn.commit() time.sleep(2) return replied这里要关注几个现实限制。Reddit 会对新账号或低 Karma 账号的发布频率做限制即使 API 返回成功也未必能立刻看到效果。提交时如果帖子被锁定、删除或账号被封会抛异常要把异常信息记录到 error_message而不是静默失败。另一个高频坑是提交前没有检查该帖子是否已经收到过自己的回复。两种实现思路一是依赖数据库状态二是提交前读取 submission 的评论列表做匹配。推荐两种都做数据库状态作为主判断评论列表作为兜底。4.3 主循环与调度方式本地开发时可以用一个循环同时调用扫描和提交但生产环境建议拆开。扫描任务可以每 10 分钟跑一次提交任务可以每 1 到 2 分钟跑一次且提交任务必须比扫描任务更保守。# run_local.py import logging import sqlite3 import time from scanner import scan_once from reply import reply_approved logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, ) SCAN_INTERVAL 600 REPLY_INTERVAL 60 def main(): conn sqlite3.connect(replyhey.db) while True: try: inserted scan_once(conn) replied reply_approved(conn) if inserted or replied: logging.info(scan_inserted%s replied%s, inserted, replied) except Exception: logging.exception(unexpected error in main loop) time.sleep(min(SCAN_INTERVAL, REPLY_INTERVAL))这个循环只适合学习环境因为它把两类任务写在同一个进程里一旦扫描请求卡住提交任务也会被拖慢。生产环境应该使用任务队列例如 Redis 定时任务或者直接使用分布式调度平台。5. 运行验证、日志和常见问题排查5.1 本地运行和验证清单第一步先验证认证第二步验证扫描第三步验证审核第四步验证回复。不要第一次就把扫描和提交一起跑。完成本地验证后可以按以下清单确认是否具备真实性环境.env中客户端信息是否完整。认证执行python check_connection.py能打印账号名。扫描执行scan_once后看数据库是否新增记录。审核执行python review.py list能看到待审核草稿。提交手动指定一条已 approved 的帖子看 Reddit 网页端是否出现评论。幂等再次运行扫描不应插入重复记录再次运行提交不应重复回复。运行输出大致如下INFO root:scan_inserted12 replied0 INFO root:scan_inserted0 replied1如果出现 replied0 且没有异常通常是没有 approved 状态的记录需要先在审核队列里处理。5.2 日志字段与监控日志不要只记录“运行成功”至少记录字段含义示例subreddit来源板块freelancesubmission_id帖子 ID1abcdestate当前状态approvedaction操作reply_approvederror_message异常信息403 Forbiddenelapsed_ms耗时1203user操作账号my_bot_account日志格式可以选择 JSON 行便于接入 ELK 或云日志服务。生产环境建议增加告警连续提交失败超过 3 次、每个 subreddit 当日提交次数超过阈值、扫描插入数为 0 且持续超过 24 小时。5.3 常见 API 错误对照表现象错误特征常见原因处理建议认证失败401 Unauthorizedclient_id、secret、用户名或密码错误检查环境变量确认 script app 是否创建成功禁止访问403 Forbidden账号被封、subreddit 私有、无权限 用Python构建Reddit自动回复系统：从API接入到人工审核
ReplyHey 给自己的定位是 Autopilot让用户通过自动回复从 Reddit 寻找潜在客户。真正动手做这个主题之前要先把它从营销叙事翻译成工程问题监控哪些 subreddit、按什么标准筛选帖子、怎样生成不惹人厌的回复、如何控制频率以及怎样在 Reddit 的 API 限制和社区规则下安全运行。这些工作可以拆成一个最小可复现系统一个进程负责扫描一个队列负责存储一个审核步骤负责把关一个 API 调用负责真正提交回复。本文会按这条主线搭建一个 Python 示例从 Reddit 应用注册开始到扫描、过滤、生稿、审核、提交和排查结束。文章假定你已经掌握 Python 基础了解 HTTP 和 OAuth 的基本概念。先从整体设计开始。1. 先理解 ReplyHey 这类工具的技术本质1.1 自动回复不是“自动群发”而是“自动筛选 人工确认”“Autopilot”这个词容易让人产生误解是不是写一个脚本看到有人问“有没有推荐工具”就立刻冲上去回复自己的产品如果这样做技术难度确实很低但它不是可长期运行的系统因为 Reddit 用户对广告非常敏感subreddit 版务也会封禁频繁推广的账号。真正值得实现的是“自动筛选 草稿生成 人工确认”这条流水线。整个系统的本质可以概括为读取一个或多个 subreddit 的新帖和评论流。用规则判断该讨论是否存在真实需求。如果存在生成一段回复草稿。草稿进入审核队列由运营人员决定是否回复、修改后回复或放弃。只有审核通过的草稿才会通过 Reddit API 真正提交。所有扫描、审核、提交动作都写入日志和数据库。这样设计的好处是自动化处理了重复劳动但最终决定权仍然留在人手里出问题时可以回溯是哪一份草稿、哪次提交造成的。1.2 系统边界范围、频率、状态和审计技术实现前要先定义边界否则写着写着就会失控。范围不能扫描整个 Reddit要限定在跟产品场景相关的 subreddit。产品是发票工具就只看 r/freelance、r/smallbusiness、r/invoicing 等产品是开发者工具就只看 r/webdev、r/selfhosted 等。频率Reddit API 有速率限制程序内部也要设置自己的调度间隔。代码里宁可保守也不要把请求打满。状态每条帖子在数据库里必须有一条完整状态记录常见状态包括 pending、approved、rejected、replied、skipped、failed。有了状态才能做幂等处理避免重复回复。审计每条回复都要保留原始帖子标题、链接、生成时间、回复内容、操作人。这不仅是工程要求也是平台规则和口碑管理的需要。这三层约束会直接影响后面的数据库设计和主循环写法。1.3 平台规则对架构设计的影响Reddit 对自动化请求有明确要求虽然是社区平台API 规则更像一家严肃的 SaaS 服务。基础要求包括使用 OAuth2 认证不能再用匿名接口做业务操作。User-Agent 必须能识别调用方身份。对单个端点的请求需要控制频率遇到 429 必须退避。自动回复如果触发反垃圾机制账号会被限流或封禁。不同 subreddit 有不同的推广规则代码需要支持按 subreddit 配置黑名单。这些要求不是安全文档里的空话它们会直接影响架构。比如“不同 subreddit 有不同规则”意味着过滤层不能只做全局关键词匹配还要允许运营为每个 subreddit 单独配置允许/禁止名单“遇到 429 必须退避”意味着提交回复的任务不能像普通循环一样一梭子全部打出去而要有队列和重试间隔。这也是本文示例中把“扫描”和“提交”分成两个模块的原因。扫描模块需要高频读取提交模块需要低频写入二者混在一起会让限流处理非常难做。2. 环境准备与 Reddit API 接入2.1 注册应用与获取凭据要让脚本使用 Reddit 账号执行操作需要创建一组 API 凭据。访问 Reddit 的偏好页面下的 apps 页面创建一个 script 类型的应用提交后会得到两段关键内容client_id应用标识界面里通常显示为一串短 ID。client_secret应用密钥只有创建者能看到需要保存好。同时脚本还需要使用一个真实 Reddit 账号的用户名和密码。由于脚本通过 OAuth2 的 password grant 方式换取 token因此账号密码要作为环境变量传入不要写进代码。配置项说明示例REDDIT_CLIENT_ID应用 ID1a2b3c4d5eREDDIT_CLIENT_SECRET应用密钥一段长随机字符串REDDIT_USERNAME执行自动回复的账号名my_bot_accountREDDIT_PASSWORD对应账号密码不要写入代码或仓库REDDIT_USER_AGENT标识调用方的请求头invoice-tool/0.1 by my_bot_account需要注意一个坑不要用日常交流的主账号做自动化测试。自动回复即使合规也可能因为 subreddit 规则、图片处理、临时误报被版务封禁。推荐的做法是创建专用账号并设置一个能表明身份的用户名。2.2 安装依赖并用最小脚本验证连通性Python 生态里最常用的 Reddit API 客户端是 PRAW。下面的依赖清单适用于多数本地开发环境实际安装时以官方最新版本为准praw7.7.1 python-dotenv1.0.0配置信息放到.env文件读取时使用 python-dotenvREDDIT_CLIENT_ID1a2b3c4d5e REDDIT_CLIENT_SECRETyour_client_secret REDDIT_USERNAMEmy_bot_account REDDIT_PASSWORDyour_account_password REDDIT_USER_AGENTinvoice-tool/0.1 by my_bot_account初始化客户端的代码# reddit_client.py import os import praw from dotenv import load_dotenv load_dotenv() def build_reddit_client() - praw.Reddit: return praw.Reddit( client_idos.environ[REDDIT_CLIENT_ID], client_secretos.environ[REDDIT_CLIENT_SECRET], usernameos.environ[REDDIT_USERNAME], passwordos.environ[REDDIT_PASSWORD], user_agentos.environ[REDDIT_USER_AGENT], )验证连通性的最小脚本# check_connection.py from reddit_client import build_reddit_client reddit build_reddit_client() print(reddit.user.me())运行后如果能打印出账号名说明认证成功。这里最常见的坑是 User-Agent 设置过于简单例如直接写成python:mybot version 1.0。Reddit 官方要求 User-Agent 能够唯一识别调用方。推荐格式包含应用名、版本号和执行账号名。2.3 配置项目结构和环境变量为了让示例可维护建议先定义项目结构replyhey/ ├── .env ├── requirements.txt ├── reddit_client.py ├── schema.sql ├── scanner.py ├── draft.py ├── review.py ├── reply.py ├── database.py └── logs/database.py负责 SQLite 连接生产环境可以替换成 PostgreSQL 或 MySQL。.env必须加入.gitignore避免密钥提交到仓库。这里要说清环境变量和配置文件的选择环境变量适合放密钥配置文件适合放可变更的业务规则比如关键词、subreddit 名单、回复模板。二者不要混用。3. 核心实现扫描讨论、过滤相关帖子、生成草稿3.1 定义帖子模型与 SQLite 表先建一张表记录扫描到的候选帖子和后续状态。SQLite 在本地够用生产环境建议迁移到 PostgreSQL-- schema.sql CREATE TABLE IF NOT EXISTS submissions ( id TEXT PRIMARY KEY, subreddit TEXT NOT NULL, title TEXT NOT NULL, body TEXT NOT NULL DEFAULT , url TEXT, score INTEGER DEFAULT 0, relevance_score REAL DEFAULT 0, state TEXT DEFAULT pending, candidate_keywords TEXT DEFAULT , created_utc INTEGER, scanned_utc INTEGER, reviewed_by TEXT, reviewed_utc INTEGER, reply_text TEXT, reply_id TEXT, error_message TEXT ); CREATE INDEX IF NOT EXISTS idx_submissions_state ON submissions(state); CREATE INDEX IF NOT EXISTS idx_submissions_subreddit ON submissions(subreddit);id 就是 Reddit 帖子的 base36 id用它做主键可以天然去重。relevance_score 用来记录相关度排序依据candidate_keywords 记录命中了哪些关键词方便审核人员快速判断。这里的核心设计是“状态机”pending 表示已入队但未审核approved 表示人工确认可以回复rejected 表示放弃replied 表示成功提交skipped 表示因规则跳过failed 表示提交失败。所有后续逻辑都围绕这个状态流转。3.2 扫描指定 subreddit 并做关键词过滤扫描逻辑按 subreddit 分组对每个 subreddit 读取新帖不需要把一个 subreddit 的几千篇历史帖全部拉下来。示例中先读取最近 50 篇新帖再结合关键词判断# scanner.py import logging import sqlite3 import time from reddit_client import build_reddit_client logger logging.getLogger(__name__) SUBREDDITS [freelance, smallbusiness, invoicing] POSITIVE_KEYWORDS [invoice, client, payment, get paid, late payment] NEGATIVE_KEYWORDS [I sell invoice paper, invoice management software comparison] MIN_SCORE 2 SCAN_LIMIT 50 def relevance(text: str) - float: score 0 matched [] for kw in POSITIVE_KEYWORDS: if kw in text: score 2 matched.append(kw) for kw in NEGATIVE_KEYWORDS: if kw in text: score - 4 return score, matched def scan_once(conn: sqlite3.Connection) - int: reddit build_reddit_client() inserted 0 for sub_name in SUBREDDITS: sub reddit.subreddit(sub_name) for submission in sub.new(limitSCAN_LIMIT): text f{submission.title}\n{submission.selftext}.lower() score, matched relevance(text) if score MIN_SCORE: continue cur conn.execute( INSERT OR IGNORE INTO submissions (id, subreddit, title, body, url, score, relevance_score, candidate_keywords, created_utc, scanned_utc) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( submission.id, sub_name, submission.title, submission.selftext[:2000], submission.url, submission.score, score, ,.join(matched), int(submission.created_utc), int(time.time()), ), ) inserted cur.rowcount conn.commit() return inserted这里有几个关键判断。第一为什么不直接给所有帖子回复因为关键词匹配只是弱信号它只能表示帖子在讨论相关主题不能表示用户一定想要你的产品。第二为什么要有 NEGATIVE_KEYWORDS因为“发票管理软件对比”这类帖子里虽然出现关键词大概率是竞品调研不是商机。第三为什么只扫描新帖而不是全站搜索因为新帖时效性好也更容易控制 API 频率。最常踩的坑是直接对所有命中关键词的帖子生成回复完全忽略 subreddit 自身的推广规则。比如某些 subreddit 明令禁止自我推广你的关键词再准也不能回复。后续实现中要加入“subreddit 黑名单”和“每 subreddit 只允许回复 N 次/天”的策略。3.3 生成回复草稿并落库过滤后进入草稿生成阶段。为了说明完整流程先用模板生成草稿后续可以替换成大模型生成# draft.py import sqlite3 TEMPLATE ( 我们是做 {product} 的开发者正好看到你在讨论 {topic}。 先说明一下身份这个回复和我们的产品直接相关。 如果你的场景和下面的描述一致可以看一下官方文档{doc_url}。 如果没有帮助直接忽略即可。 ) def generate_draft(submission_row, product发票管理工具, doc_urlhttps://example.com/docs) - str: topic (submission_row[title] or ).strip() return TEMPLATE.format(productproduct, topictopic, doc_urldoc_url)草稿为什么要明文保存而不是直接提交因为人工审核需要看到完整内容而且如果生成模型出问题至少知道发给用户的是什么。不要把草稿只放在内存里否则程序重启后待审核队列就丢失了。生成草稿的函数可以单独写也可以合并在扫描流程中。推荐在扫描后生成草稿再进入人工审核接口。3.4 状态与幂等设计幂等是这个系统最容易忽略的地方。Reddit 帖子不会因为程序重启而消失同一篇帖子如果被扫描到两次不能插入两条记录更不能回复两次。使用INSERT OR IGNORE可以避免重复入库。提交回复前必须再检查一次数据库状态def can_reply(conn: sqlite3.Connection, submission_id: str) - bool: row conn.execute( SELECT state FROM submissions WHERE id ?, (submission_id,) ).fetchone() return row is not None and row[0] approved同时建议在事务里更新状态conn.execute( UPDATE submissions SET state ?, reply_id ?, error_message ? WHERE id ? AND state ?, (replied, reply_id, None, submission_id, approved), )这样即使提交过程中抛异常也不会把状态改乱。另一个坑是提交后没有保存 reply_id导致排查时不知道哪条评论是自己的。4. 人工审核环节不要绕过也不要侥幸4.1 命令行审核队列审核无需一开始就上 Web 界面命令行工具已经能支撑早期使用。示例代码# review.py import sqlite3 import sys def list_pending(conn: sqlite3.Connection): rows conn.execute( SELECT id, subreddit, title, relevance_score, reply_text FROM submissions WHERE state pending ORDER BY relevance_score DESC LIMIT 20 ).fetchall() for row in rows: print(f{row[id]} | {row[subreddit]} | score{row[relevance_score]}) print(f title: {row[title]}) print(f draft: {row[reply_text]}) def approve(conn: sqlite3.Connection, submission_id: str): conn.execute( UPDATE submissions SET state approved WHERE id ? AND state pending, (submission_id,), ) conn.commit() def reject(conn: sqlite3.Connection, submission_id: str): conn.execute( UPDATE submissions SET state rejected WHERE id ? AND state pending, (submission_id,), ) conn.commit()使用方法python review.py list python review.py approve 1abcde python review.py reject 1abcde为什么要这一步因为程序判断的是“话题相关”人判断的是“这个讨论是否适合以我们身份出现、语气是否合适、是否违反 subreddit 规则”。这一步是过滤器也是防线。4.2 审核通过后自动提交回复提交逻辑是独立脚本只处理 approved 状态的记录并且要限制每次运行最多提交几条# reply.py import sqlite3 import time from reddit_client import build_reddit_client MAX_REPLIES_PER_RUN 5 def reply_approved(conn: sqlite3.Connection) - int: reddit build_reddit_client() rows conn.execute( SELECT id, reply_text FROM submissions WHERE state approved LIMIT ?, (MAX_REPLIES_PER_RUN,), ).fetchall() replied 0 for row in rows: submission_id row[id] try: submission reddit.submission(idsubmission_id) comment submission.reply(row[reply_text]) conn.execute( UPDATE submissions SET state replied, reply_id ? WHERE id ?, (comment.id, submission_id), ) conn.commit() replied 1 except Exception as exc: conn.execute( UPDATE submissions SET state failed, error_message ? WHERE id ?, (str(exc), submission_id), ) conn.commit() time.sleep(2) return replied这里要关注几个现实限制。Reddit 会对新账号或低 Karma 账号的发布频率做限制即使 API 返回成功也未必能立刻看到效果。提交时如果帖子被锁定、删除或账号被封会抛异常要把异常信息记录到 error_message而不是静默失败。另一个高频坑是提交前没有检查该帖子是否已经收到过自己的回复。两种实现思路一是依赖数据库状态二是提交前读取 submission 的评论列表做匹配。推荐两种都做数据库状态作为主判断评论列表作为兜底。4.3 主循环与调度方式本地开发时可以用一个循环同时调用扫描和提交但生产环境建议拆开。扫描任务可以每 10 分钟跑一次提交任务可以每 1 到 2 分钟跑一次且提交任务必须比扫描任务更保守。# run_local.py import logging import sqlite3 import time from scanner import scan_once from reply import reply_approved logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, ) SCAN_INTERVAL 600 REPLY_INTERVAL 60 def main(): conn sqlite3.connect(replyhey.db) while True: try: inserted scan_once(conn) replied reply_approved(conn) if inserted or replied: logging.info(scan_inserted%s replied%s, inserted, replied) except Exception: logging.exception(unexpected error in main loop) time.sleep(min(SCAN_INTERVAL, REPLY_INTERVAL))这个循环只适合学习环境因为它把两类任务写在同一个进程里一旦扫描请求卡住提交任务也会被拖慢。生产环境应该使用任务队列例如 Redis 定时任务或者直接使用分布式调度平台。5. 运行验证、日志和常见问题排查5.1 本地运行和验证清单第一步先验证认证第二步验证扫描第三步验证审核第四步验证回复。不要第一次就把扫描和提交一起跑。完成本地验证后可以按以下清单确认是否具备真实性环境.env中客户端信息是否完整。认证执行python check_connection.py能打印账号名。扫描执行scan_once后看数据库是否新增记录。审核执行python review.py list能看到待审核草稿。提交手动指定一条已 approved 的帖子看 Reddit 网页端是否出现评论。幂等再次运行扫描不应插入重复记录再次运行提交不应重复回复。运行输出大致如下INFO root:scan_inserted12 replied0 INFO root:scan_inserted0 replied1如果出现 replied0 且没有异常通常是没有 approved 状态的记录需要先在审核队列里处理。5.2 日志字段与监控日志不要只记录“运行成功”至少记录字段含义示例subreddit来源板块freelancesubmission_id帖子 ID1abcdestate当前状态approvedaction操作reply_approvederror_message异常信息403 Forbiddenelapsed_ms耗时1203user操作账号my_bot_account日志格式可以选择 JSON 行便于接入 ELK 或云日志服务。生产环境建议增加告警连续提交失败超过 3 次、每个 subreddit 当日提交次数超过阈值、扫描插入数为 0 且持续超过 24 小时。5.3 常见 API 错误对照表现象错误特征常见原因处理建议认证失败401 Unauthorizedclient_id、secret、用户名或密码错误检查环境变量确认 script app 是否创建成功禁止访问403 Forbidden账号被封、subreddit 私有、无权限