起因
我的 Telegram 账号前阵子被双向限制了：陌生人搜不到我、给我发消息也发不过来，可我时不时需要在外面留一个能联系上我的口子。另外还有个小痛点：平时在论坛发帖要贴图，得开电脑上的图床客户端上传再复制链接，手机上截了图想发帖就更折腾。
两个需求凑一块，我决定自己写一个 Telegram 机器人：
传图：我把图片发给机器人，它传到我自建的图床上，把直链发回来。这个功能只给我自己用。
联系中继：别人给机器人留言，机器人转给我，我回复之后再由机器人送回去。对方不用加好友，我也不用暴露账号。
取名风波
机器人要先在 BotFather 那里注册一个用户名，必须以 bot 结尾、全网唯一。我想用的名字基本全军覆没：Fokienia、Fraxinus 的各种组合都被人注册了，短一点的也一个不剩。带数字凑长度的我又不想用。
最后翻到梧桐的拉丁学名 Firmiana，试了一下，FirmianaBot 一次通过。梧桐这个意象我本来就喜欢，算是因祸得福。
方案：为什么用「话题制」
动手前我去 GitHub 上看了一圈同类的联系机器人项目，重点确认了一件事：怎么把「访客是谁」认准。
有一种偷懒做法是靠转发消息时携带的来源信息读对方 ID。这条路是死的：Telegram 早就允许用户在隐私设置里隐藏转发来源，人家一开这个设置，机器人就读不到来源，整个对应关系全乱。所以映射关系必须自己存：访客第一次留言时，把他的数字 ID 记进数据库——数字 ID 不随改名、换头像变化，认这个最稳。
呈现方式上我选了话题制：建一个开了话题功能的超级群当管理群，每个访客第一次留言，机器人就在群里给他开一个专属话题，话题名就是他的名字和 ID，里面再钉一张访客名片。我在话题里打字回复，机器人就把话回送给他。访客多了也不会串线，比所有消息挤在一个聊天里清爽得多。
实现要点
语言和框架：Python + python-telegram-bot，几百行一个文件。
通讯方式用长轮询：机器人主动去 Telegram 拉消息，不用开公网端口、不用配域名和证书，扔在自己的小服务器上用 Docker 常驻就行。
两个功能严格隔离：传图只认我自己的数字 ID（白名单），联系中继对所有人开放；访客发的斜杠命令不中继，防捣乱。
防刷：每个访客每分钟最多中继 8 条，超了静默丢弃；话题里有 /ban、/unban、/info 三个管理命令，拉黑的人发什么都石沉大海。
该存的东西存 SQLite：访客 ID、话题 ID、拉黑状态，就这么点。
两把钥匙（机器人 token、图床令牌）只写在服务器的配置文件里，不进代码，这样代码可以随便公开——文末就附了全文。
图床那边我给机器人单独生成了一个新令牌，没跟电脑上的客户端共用。这样哪天想收回机器人的图床权限，吊销它一个，不影响别处。
踩的三个坑
坑一：配置留空，程序直接崩。 管理群的 ID 是要等机器人跑起来之后才能拿到的，所以配置里先留空。结果程序读到「空字符串」去转数字，当场报错，容器反复崩溃重启。默认值只能防「没有这一项」，防不住「有这一项但空着」。改法很小：取值时空串当 0 处理。教训是读配置一律防空。
坑二：日志把 token 打出来了。 框架默认的日志级别会把完整的请求网址打出来，而网址里就带着机器人 token。我排查问题时盯着日志看了半天才反应过来：这串东西等于钥匙，不能让它躺在日志里。处理分两步：先去 BotFather 把旧 token 作废、生成新的（旧的从此无效）；再在代码里把那一类日志的级别关掉。以后贴任何日志出来之前，我都会先想一遍里面有没有钥匙。
坑三：图床令牌没有网页入口。 我用的开源图床后台里只有接口说明页，翻遍了设置也没找到令牌管理——它根本没有网页版入口，令牌只能调用接口创建。好在创建的方法跟我之前给电脑客户端弄令牌时一模一样：用账号密码调一次创建接口，拿回令牌填进配置。找入口找了两圈，算是给后来人提个醒。
验收
我私聊机器人发一张截图，秒回直链加一条现成的 Markdown 代码，点开图片正常。
拿小号演访客：点开始，机器人回联系信箱的欢迎语；发一句话，管理群里立刻多出一个以小号名字命名的话题，名片和消息都在；我在话题里回复，小号那边原样收到。
两条线全通，收工。
完整代码
配置全走环境变量（.env），代码里没有任何真实信息，可以放心贴出来。部署就是把这几个文件放一个目录，填好 .env，docker compose up -d --build 一条命令的事。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
FirmianaBot (@FirmianaBot)
功能一（仅主人）：私聊机器人发图片/文件 -> 上传兰空图床 -> 回直链 + Markdown。
功能二（对外）：双向联系·话题制中继。访客私聊机器人发消息，消息进入管理群里
该访客的专属话题；主人在话题内回复，机器人回送给访客。
配置全部来自环境变量，见 .env.example。token 不进代码。
"""
import asyncio
import html
import logging
import os
import sqlite3
import time
from collections import defaultdict, deque
import httpx
from telegram import Update
from telegram.constants import ParseMode
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
# ---------------- 配置 ----------------
BOT_TOKEN = os.environ["BOT_TOKEN"]
OWNER_ID = int(os.environ.get("OWNER_ID", "0") or "0")
ADMIN_GROUP_ID = int(os.environ.get("ADMIN_GROUP_ID", "0") or "0")
LSKY_BASE = os.environ.get("LSKY_BASE", "https://img.fokienia.com").rstrip("/")
LSKY_TOKEN = os.environ.get("LSKY_TOKEN", "")
DB_PATH = os.environ.get("DB_PATH", "/data/bot.db")
RATE_LIMIT = 8 # 每位访客在 RATE_WINDOW 秒内最多中继多少条
RATE_WINDOW = 60
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logging.getLogger("httpx").setLevel(logging.WARNING)
log = logging.getLogger("firmiana")
# ---------------- 数据库 ----------------
def _conn():
conn = sqlite3.connect(DB_PATH)
conn.execute(
"""CREATE TABLE IF NOT EXISTS visitors(
user_id INTEGER PRIMARY KEY,
topic_id INTEGER,
username TEXT,
first_name TEXT,
banned INTEGER DEFAULT 0,
created_at INTEGER
)"""
)
return conn
def get_visitor(user_id):
conn = _conn()
row = conn.execute(
"SELECT user_id, topic_id, username, first_name, banned FROM visitors WHERE user_id=?",
(user_id,),
).fetchone()
conn.close()
return row
def upsert_visitor(user):
conn = _conn()
conn.execute(
"""INSERT INTO visitors(user_id, username, first_name, created_at)
VALUES(?,?,?,?)
ON CONFLICT(user_id) DO UPDATE SET
username=excluded.username, first_name=excluded.first_name""",
(user.id, user.username or "", user.first_name or "", int(time.time())),
)
conn.commit()
conn.close()
def set_topic(user_id, topic_id):
conn = _conn()
conn.execute("UPDATE visitors SET topic_id=? WHERE user_id=?", (topic_id, user_id))
conn.commit()
conn.close()
def visitor_by_topic(topic_id):
conn = _conn()
row = conn.execute(
"SELECT user_id, banned FROM visitors WHERE topic_id=?", (topic_id,)
).fetchone()
conn.close()
return row
def set_banned(user_id, banned):
conn = _conn()
conn.execute(
"UPDATE visitors SET banned=? WHERE user_id=?", (1 if banned else 0, user_id)
)
conn.commit()
conn.close()
def get_stats():
conn = _conn()
total = conn.execute("SELECT COUNT(*) FROM visitors").fetchone()[0]
banned = conn.execute("SELECT COUNT(*) FROM visitors WHERE banned=1").fetchone()[0]
conn.close()
return total, banned
# ---------------- 防刷 ----------------
_hits = defaultdict(deque)
def rate_ok(user_id):
now = time.time()
q = _hits[user_id]
while q and now - q[0] > RATE_WINDOW:
q.popleft()
if len(q) >= RATE_LIMIT:
return False
q.append(now)
return True
# ---------------- 话题管理 ----------------
_topic_lock = asyncio.Lock()
async def ensure_topic(bot, user):
"""返回该访客在管理群里的话题 id，没有就新建。"""
row = get_visitor(user.id)
if row and row[1]:
return row[1]
async with _topic_lock:
row = get_visitor(user.id)
if row and row[1]:
return row[1]
name = (user.first_name or "访客").strip()
if user.username:
name += f" (@{user.username})"
name = (name + f" · {user.id}")[:128]
topic = await bot.create_forum_topic(chat_id=ADMIN_GROUP_ID, name=name)
set_topic(user.id, topic.message_thread_id)
card = (
f"访客名片\n"
f"名字：{user.first_name or '-'}\n"
f"用户名：@{user.username or '无'}\n"
f"数字 ID：{user.id}\n"
f"在本话题内直接回复即可回送给对方。\n"
f"管理命令：/ban 拉黑 · /unban 解封 · /info 查看"
)
await bot.send_message(
chat_id=ADMIN_GROUP_ID,
message_thread_id=topic.message_thread_id,
text=card,
)
return topic.message_thread_id
# ---------------- 图床 ----------------
async def upload_lsky(filename, data):
async with httpx.AsyncClient(timeout=60) as client:
r = await client.post(
f"{LSKY_BASE}/api/v1/upload",
headers={"Authorization": f"Bearer {LSKY_TOKEN}"},
files={"file": (filename, data)},
)
r.raise_for_status()
j = r.json()
return j["data"]["links"]["url"]
async def owner_media(update: Update, context: ContextTypes.DEFAULT_TYPE):
msg = update.effective_message
try:
if msg.photo:
f = await msg.photo[-1].get_file()
filename = f"photo_{f.file_unique_id}.jpg"
else:
doc = msg.document
f = await doc.get_file()
filename = doc.file_name or f"file_{doc.file_unique_id}"
data = bytes(await f.download_as_bytearray())
url = await upload_lsky(filename, data)
except Exception as e: # noqa: BLE001
log.exception("upload failed")
await msg.reply_text(f"上传失败：{e}")
return
await msg.reply_text(url)
await msg.reply_text(
f"<code>})</code>", parse_mode=ParseMode.HTML
)
# ---------------- 命令 ----------------
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
user = update.effective_user
chat = update.effective_chat
if chat.type != "private":
return
if user.id == OWNER_ID:
await update.effective_message.reply_text(
"主人控制台\n"
"· 私聊发图片/文件 = 上传图床回链接\n"
"· 访客消息在管理群的话题里回\n"
"· /stats 看统计 · /id 看自己的数字 ID · 在群里发 /chatid 可看群 ID"
)
return
upsert_visitor(user)
await update.effective_message.reply_text(
"你好，这里是 Fraxinus 的联系信箱。\n"
"直接把要说的话发在这里就行，对方会通过这个机器人回复你，"
"不用加好友也能联系上。\n"
"（请勿发送广告或骚扰信息，频繁刷屏会被拉黑。）"
)
async def cmd_id(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.effective_message.reply_text(f"你的数字 ID：{update.effective_user.id}")
async def cmd_chatid(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.effective_user.id != OWNER_ID:
return
msg = update.effective_message
text = f"本会话 ID：{update.effective_chat.id}"
if getattr(msg, "message_thread_id", None):
text += f"\n话题 ID：{msg.message_thread_id}"
await msg.reply_text(text)
async def cmd_stats(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.effective_user.id != OWNER_ID:
return
total, banned = get_stats()
await update.effective_message.reply_text(
f"累计访客 {total} 人，其中拉黑 {banned} 人。"
)
def _topic_visitor(update):
"""在管理群话题里，根据话题找到访客。找不到返回 None。"""
msg = update.effective_message
if not getattr(msg, "message_thread_id", None):
return None
return visitor_by_topic(msg.message_thread_id)
async def cmd_ban(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.effective_user.id != OWNER_ID:
return
row = _topic_visitor(update)
if not row:
await update.effective_message.reply_text("这个话题还没绑定访客。")
return
set_banned(row[0], True)
await update.effective_message.reply_text(f"已拉黑访客 {row[0]}，其消息将不再中继。")
async def cmd_unban(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.effective_user.id != OWNER_ID:
return
row = _topic_visitor(update)
if not row:
await update.effective_message.reply_text("这个话题还没绑定访客。")
return
set_banned(row[0], False)
await update.effective_message.reply_text(f"已解封访客 {row[0]}。")
async def cmd_info(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.effective_user.id != OWNER_ID:
return
row = _topic_visitor(update)
if not row:
await update.effective_message.reply_text("这个话题还没绑定访客。")
return
v = get_visitor(row[0])
await update.effective_message.reply_text(
f"访客 {v[0]}\n用户名：@{v[2] or '无'}\n名字：{v[3] or '-'}\n状态：{'已拉黑' if v[4] else '正常'}"
)
# ---------------- 消息分发 ----------------
async def private_dispatch(update: Update, context: ContextTypes.DEFAULT_TYPE):
msg = update.effective_message
user = update.effective_user
if user.id == OWNER_ID:
if msg.photo or msg.document:
await owner_media(update, context)
else:
await msg.reply_text("图片/文件发我这里就传图床；访客对话请去管理群的话题里回。")
return
# 访客：登记、查拉黑、防刷，然后中继进话题
upsert_visitor(user)
row = get_visitor(user.id)
if row and row[4]:
return # 已拉黑：静默丢弃
if not rate_ok(user.id):
return # 超速：静默丢弃
try:
topic_id = await ensure_topic(context.bot, user)
await context.bot.copy_message(
chat_id=ADMIN_GROUP_ID,
from_chat_id=msg.chat_id,
message_id=msg.message_id,
message_thread_id=topic_id,
)
except Exception: # noqa: BLE001
log.exception("relay visitor->group failed")
async def group_dispatch(update: Update, context: ContextTypes.DEFAULT_TYPE):
msg = update.effective_message
chat = update.effective_chat
if ADMIN_GROUP_ID and chat.id != ADMIN_GROUP_ID:
return
if update.effective_user and update.effective_user.id == context.bot.id:
return # 机器人自己发进话题的消息（名片、转发件）不要回送
if not getattr(msg, "is_topic_message", False):
return
if not getattr(msg, "message_thread_id", None):
return
row = visitor_by_topic(msg.message_thread_id)
if not row:
return
try:
await context.bot.copy_message(
chat_id=row[0],
from_chat_id=msg.chat_id,
message_id=msg.message_id,
)
except Exception: # noqa: BLE001
log.exception("relay group->visitor failed")
try:
await msg.reply_text("（这条没能送达对方：对方可能已停用机器人或账号异常。）")
except Exception: # noqa: BLE001
pass
def main():
if not OWNER_ID:
log.warning("OWNER_ID 未设置：主人功能（传图/管理命令）不会对任何人开放")
if not ADMIN_GROUP_ID:
log.warning("ADMIN_GROUP_ID 未设置：先建管理群，把机器人拉进去后用 /chatid 拿到群 ID 再填回 .env")
app = Application.builder().token(BOT_TOKEN).build()
app.add_handler(CommandHandler("start", cmd_start))
app.add_handler(CommandHandler("id", cmd_id))
app.add_handler(CommandHandler("chatid", cmd_chatid))
app.add_handler(CommandHandler("stats", cmd_stats))
app.add_handler(CommandHandler("ban", cmd_ban))
app.add_handler(CommandHandler("unban", cmd_unban))
app.add_handler(CommandHandler("info", cmd_info))
app.add_handler(
MessageHandler(filters.ChatType.PRIVATE & ~filters.COMMAND, private_dispatch)
)
app.add_handler(
MessageHandler(filters.ChatType.GROUPS & ~filters.COMMAND, group_dispatch)
)
log.info("FirmianaBot 启动，长轮询中…")
app.run_polling(allowed_updates=Update.ALL_TYPES, drop_pending_updates=True)
if __name__ == "__main__":
main()
小结
跑通之后回头看，这个机器人本身没什么黑科技，难的全是边角：名字被人占了、配置空值会崩、日志会漏钥匙、图床令牌藏在接口里。折腾的乐趣也在这儿。
使用说明就一句：想联系我，在 Telegram 里搜 @FirmianaBot，点开始，直接把话留那儿就行，我看到会回。发广告和刷屏的会被拉黑，机器人自己也带限速，别试。
评论