制作客户端前提条件：
干净的网络住宅IP代理，一般的梯子和自己搭的公网IP不行，可以使用cf的免费代理，然后在隐私模式下访问
https://my.telegram.org
写注册手机号(带+号和国际区号)telegram 的APP会出现一个验证码，将这个验证码输入登录，选→ API Development Tools，提交一个表单，通常只需要写这几项：
App title
Short name
平台写Desktop，然后提交获得一个API,这是操作个人账号的API
文档：https://docs.telethon.dev/en/stable/
uv pip install --upgrade telethon cryptg
安装cryptg会让加载加速(使用c解密)
客户端方法
@client.on(events.NewMessage(incoming=True))注册事件处理装饰器，只处理入站该函数中的一个参数，会被自动传入event对象，这个函数需要先被挂载(同步调用一次)
client.catch_up() 客户端离线期间“追赶”错过的更新，上次离线到现在中间
client.run_until_disconnected() 运行直到退出，异步调用，要先注册事件处理装饰器再执行此步
代码示范
简单的登录示范：
py
from telethon import TelegramClient
# 配置API ID和API Hash
API_ID = 应用的id
API_HASH = 应用hash
SESSION_NAME = 'leeken' # 会话名称,用来存储用户每个登录用户都会生成独立的名称
# 创建客户并登录
client = TelegramClient(SESSION_NAME, API_ID, API_HASH)
# 登录
client.start()
这会触发让你输入手机号(需带+号和国际区号)和验证码，然后输入正确后回生成你指定的名称'leeken'的session文件，生成的文件除非主动注销，否则身份信息一直生效
批量登录示范
py
from telethon import TelegramClient, errors
import env # 引入配置
import os
import asyncio
# 登录函数调用
async def login(phone: str):
# 补充+号
phone_norm = f'+{phone}'
# 拼接session文件路径
session_path = f'{env.SESSION_DIR}/{phone}'
# 创建客户端，如需代理，可在此处添加：proxy
client = TelegramClient(session_path, env.API_ID, env.API_HASH)
# 显式建立连接（不会触发 start 的交互式登录），这里不能使用async with client
await client.connect()
try:
# 未授权才进行验证码登录
if not await client.is_user_authorized():
try:
# 触发发送验证码（优先发到 Telegram 官方对话，必要时可尝试 force_sms）
await client.send_code_request(phone_norm)
# await client.send_code_request(phone_norm, force_sms=True)
except errors.FloodWaitError as e:
print(f'被限流，需要等待 {e.seconds} 秒后再试')
return
code = input('输入收到的验证码: ').strip()
try:
# 提交验证码完成登录
await client.sign_in(phone=phone_norm, code=code)
except errors.SessionPasswordNeededError:
# 如果账号开了二步验证
pwd = input('开启了两步验证，请输入密码: ').strip()
await client.sign_in(password=pwd)
print('已经成功登录，session已生成/更新')
finally:
# 显式断开连接，释放资源（即使发生异常也能执行）
await client.disconnect()
# 测试调用
if __name__ == '__main__':
# 如果没有 session 目录，则创建
os.makedirs(env.SESSION_DIR, exist_ok=True)
asyncio.run(login('8618681102500'))
复用身份文件
py
from telethon import TelegramClient #+导入客户端与事件
from env import API_ID, API_HASH
# 创建客户端时直接传入name就行，它会自动使用这个文件做登录
client = TelegramClient("leeken", API_ID, API_HASH)
# 获取自己的信息
async def get_me():
# 操作必须在客户端的上下文管理器中运行
async with client:
me = await client.get_me() # 打印自己的信息
print(me.stringify())
if __name__ == '__main__':
client.loop.run_until_complete(get_me())
实现监听信息，封装发送信息函数
py
# pip install telethon
import asyncio
from telethon import TelegramClient, events
from env import API_ID, API_HASH
# 假设已经登录了leeken
client = TelegramClient("leeken", API_ID, API_HASH)
# 归一化各种 peer 将字符串用户名、数字ID、实体对象转换为实体对象，
async def _resolve_peer(peer_like):
"""
归一化各种 peer：
- 字符串用户名（'flintpensky'、'me'）
- 数字ID（用户ID、群/频道ID，超级群/频道多为负数）
- 已是实体对象（InputPeer/Entity）
"""
# 如果已经是实体对象，直接返回
from telethon.tl.types import InputPeerUser, InputPeerChat, InputPeerChannel
if isinstance(peer_like, (InputPeerUser, InputPeerChat, InputPeerChannel)):
return peer_like
# 否则交给 Telethon 去解析（用户名/ID/链接都可以解析为实体）
return await client.get_entity(peer_like) # 可能走网络解析，拿到 access_hash 等
# get_entity 会把用户名/ID/链接解析成可用实体；纯数字ID需要Telethon有上下文，解析不到会抛错
# ========== 发送封装 ==========
async def send_text(peer, text: str):
"""
peer 可以是：
- 'username'（私聊/群的 @用户名）
- chat_id 整数（用户ID或群ID，超级群常见是负数如 -100xxxxxxxxxx）
- 从 client.get_entity(...) 得到的实体
"""
if not client.is_connected():
await client.connect()
# 这里的peer可以是用户名字符串、整数ID（用户ID/群ID）、或 实体对象（get_entity()/事件里取到的 entity，例如调用_resolve_peer得到的实体）
await client.send_message(peer, text)
# 发送文件
async def send_file(peer, file_path: str, caption: str | None = None):
if not client.is_connected():
await client.connect()
await client.send_file(peer, file=file_path, caption=caption)
# ========== 监听封装 ==========
async def attach_handlers():
@client.on(events.NewMessage(incoming=True))
async def on_new_message(event):
# 忽略自己发出的消息，避免回环
if event.out:
return
# 基本元信息
scope = "私聊" if event.is_private else ("群聊" if event.is_group else ("频道" if event.is_channel else "其它"))
sender = await event.get_sender()
text = event.raw_text or ""
print(
f"[{scope}] chat_id={event.chat_id} "
f"from={getattr(sender, 'username', None) or sender.id} "
f"text={text}"
)
# 私聊自动回复（按需保留/删除）
if scope == "私聊":
await send_text(event.chat_id, "已收到～")
# 如需保存对方发来的媒体，解注释：
# if event.photo or event.document:
# path = await event.download_media(file="downloads/")
# print(f"[保存媒体] {path}")
# ========== 主程序 ==========
async def main():
await client.connect() # 用已有 session 连接（不会触发登录交互）
if not await client.is_user_authorized():
print("会话无效或被吊销，请先重新登录生成 .session")
return
await attach_handlers() # 挂载事件处理
me = await client.get_me()
print(f"已上线：{me.id} {me.first_name} {me.last_name}")
# 示例：上线后给自己发一条消息（可删）
await send_text("me", "这是一个客户端测试发送")
print("开始监听，Ctrl+C 退出")
await client.run_until_disconnected() # 持续运行直到断开（会自动重连）
if __name__ == "__main__":
# 也可以使用client.loop.run_until_complete(main())
asyncio.run(main())
小demo
客户端线程池
py
from dataclasses import dataclass, field # 导入数据类装饰器和字段工厂函数
from telethon import TelegramClient, events,errors # 导入Telegram客户端和事件处理
from datetime import datetime, timezone, timedelta # 时间处理（UTC、时间窗） # 中文注释
import asyncio # 导入异步编程支持库
import contextlib # 导入上下文管理器工具
import env # 导入环境配置模块
from typing import Optional # 导入可选类型注解
from crowdpulse.rags.rag import chat_query # 导入rag查询
from log_ser.as_log import Logger
from fastapi_ser.fast_db import (
db_get_all_active_telegram_session,
db_list_user_telegram_session_ids,
db_update_telegram_session_is_active,
)
log = Logger(__name__)
@dataclass # 使用数据类装饰器，自动生成初始化方法等
class ClientEntry: # 定义客户端条目数据类
client: TelegramClient # Telegram客户端实例
user_id: int # 用户ID
telegram_id: int # Telegram账号唯一ID
telegram_username: str | None = None
task: Optional[asyncio.Task] = None # 可选的异步任务引用，默认为None
lock: asyncio.Lock = field(
default_factory=asyncio.Lock
) # 异步锁，用于保护共享资源，自动创建新锁实例
keywords: list[str] = field(
default_factory=list
) # 每个实例独立新列表，因为list是可变标量，因此需要独立新列表
monitor_chat_ids: list[int] = field(default_factory=list)
switch_all_chat: bool = False # 是否响应所有聊天
exclude_ids: list[int] = field(default_factory=list) # 排除的账号ID列表，主要用来防止自己的多个账号无限制互聊天
accept_time: datetime = field( # 新增：只接受该时间之后的消息 # 中文注释
default_factory=lambda: datetime.now(timezone.utc) - timedelta(hours=2)
)
# 定义客户端管理器类
class ClientManager:
def __init__(self): # 初始化方法
self._clients: dict[int, ClientEntry] = {} # 存储客户端条目的字典，键为telegram_id
self._global_lock = asyncio.Lock() # 全局异步锁，保护客户端字典的并发访问
self._shutting_down = False # 关机标志
def make_client(
self, telegram_id: int, proxy: dict | None = None, device_model: str | None = None
) -> TelegramClient: # 创建Telegram客户端的私有方法
"""
创建Telegram客户端（以 telegram_id 作为会话文件名）。
参数:
telegram_id (int): Telegram 唯一ID
proxy (dict | None): 代理配置
device_model (str | None): 设备型号
返回:
TelegramClient: Telethon 客户端实例
"""
# 先判断客户端是否已经在运行，如果已存在优先从线程池中返回实例
if self.is_running(telegram_id):
return self._clients[telegram_id].client
# # 构建会话文件路径
# session_path = Path(env.TG_SESSION_DIR, f"{telegram_id}.session").read_text()
return TelegramClient(
f"{env.TG_SESSION_DIR}/{telegram_id}.session",
env.TG_API_ID,
env.TG_API_HASH,
proxy=proxy,
device_model=device_model,
app_version="CrowdPulse 1.0.1",
) # 创建并返回Telegram客户端实例
def is_running(self, telegram_id: int) -> bool:
"""判断指定 telegram_id 的客户端是否在运行（任务存在且未结束）。"""
entry = self._clients.get(telegram_id)
return bool(entry and entry.task and not entry.task.done())
# 事件监听处理器
def _attach_handlers(
self, client: TelegramClient, telegram_id: int
): # 私有附加处理器方法，接收客户端和telegram_id
@client.on(
events.NewMessage(incoming=True)
) # 注册新消息事件处理器，只处理入站消息
async def on_msg(event): # 消息处理函数
if event.out: # 如果是自己发出的消息
return # 直接返回，不处理
# 取到当前entry - 通过telegram_id直接查找
entry = self._clients.get(telegram_id)
if not entry:
return # 未找到或已停止
# 安全获取发信人唯一ID（int 或 None）
sid = event.sender_id
# 如果发送人唯一ID在排除列表中则不处理
if sid is not None and sid in entry.exclude_ids:
log.info(f"{telegram_id} 检测到排除列表，不处理")
return
# 安全获取发送人信息
sender = await event.get_sender()
# 如果是bot则不处理
if getattr(sender, "bot", False):
return
text = event.raw_text or "" # 获取消息文本，处理空文本情况
# await log.info(f"[{telegram_id}] ({event.chat_id}) : {text}") # 打印消息信息
# 如果是私聊
if event.is_private:
result = await chat_query(entry.user_id,text)
await event.respond(result) # 发送自动回复
return
# 如果是群组，则检查是否在监控列表中
if event.is_group:
# 判断是否响应所有聊天
if entry.switch_all_chat:
# await log.info(f"{telegram_id}监听{event.chat_id}:{text}")
result = await chat_query(entry.user_id,text)
# 回复该消息
await event.respond(result)
return
# 再检查是否在监控列表中
elif event.chat_id in entry.monitor_chat_ids:
# await log.info(f"{telegram_id}监听{event.chat_id}:{text}")
# 如果消息是@我或回复我，或触发关键词则回复,这里的关键词从entry中获取
if event.message.mentioned or any(
keyword in text for keyword in entry.keywords
):
result = await chat_query(entry.user_id,text)
await event.respond(result)
# 启动一个账号
async def start(
self,
user_id: int,
telegram_id: int,
telegram_username: str | None = None,
keywords: list[str] | None = None,
monitor_chat_ids: list[int] | None = None,
switch_all_chat: bool = False,
proxy: dict | None = None,
device_model: str | None = None,
exclude_ids: list[int] | None = None,
) -> str: # 异步启动方法，仅以 telegram_id 标识客户端
"""
幂等启动：若已在运行直接返回提示；若失败，确保释放连接并回收entry。
"""
# 1) 先登记/获取 entry，但不要长时间持有全局锁
async with self._global_lock: # 获取全局锁，保护客户端字典操作
if self._shutting_down:
return "服务关机中" # 关机中，拒绝新启动
entry = self._clients.get(telegram_id) # 获取指定telegram_id的客户端条目
if entry and entry.task and not entry.task.done(): # 检查是否已在运行
return f"[{telegram_id}] 已在运行，忽略重复启动" # 返回已在运行的提示信息
if not entry: # 如果不存在该条目
# 创建新的客户端条目
entry = ClientEntry(
client=self.make_client(telegram_id, proxy, device_model),
user_id=user_id,
telegram_id=telegram_id,
telegram_username=telegram_username,
keywords=list(keywords or []),
monitor_chat_ids=list(monitor_chat_ids or []),
switch_all_chat=bool(switch_all_chat),
exclude_ids=list(exclude_ids or []),
)
self._clients[telegram_id] = (
entry # 暂时登记到管理字典，后续失败会在 finally 里清理
)
# 2) 进入专属锁，做真正的启动流程
runner_started = False # 标记后台任务是否成功启动
async with entry.lock: # 获取客户端条目的专属锁
# 2.1 清理残留任务
if entry.task and not entry.task.done(): # 如果存在未完成的任务
entry.task.cancel() # 取消该任务
with contextlib.suppress(asyncio.CancelledError): # 抑制取消异常
await entry.task # 等待任务结束
entry.task = None # 清空任务引用
# 2.2 显式连接
await entry.client.connect() # 建立Telegram连接
try:
# 授权校验
if not await entry.client.is_user_authorized(): # 检查是否已授权
await db_update_telegram_session_is_active(telegram_id,False,False)
await entry.client.disconnect() # 释放连接
return "身份信息失效，请重新登录" # 返回登录失效提示
# 只挂一次事件处理器（避免重复）
if not getattr(
entry.client, "_handlers_attached", False
): # 检查是否已附加处理器
self._attach_handlers(entry.client, telegram_id) # 附加事件处理器
entry.client._handlers_attached = True # 标记已附加处理器
# 2.5 后台常驻
entry.task = asyncio.create_task(
self._runner(entry), name=f"runner:{telegram_id}"
) # 创建后台运行任务
runner_started = True # 标记任务启动成功
return "success" # 返回成功状态
finally:
# 2.6 若 runner 未成功启动，断开连接并回收 entry，避免"半初始化"残留
if not runner_started: # 如果后台任务未能成功启动
with contextlib.suppress(Exception): # 抑制异常
await entry.client.disconnect() # 断开连接
# 从管理表回收未启动成功的 entry
async with self._global_lock: # 获取全局锁
cur = self._clients.get(telegram_id) # 获取当前条目
if cur is entry and (
entry.task is None or entry.task.done()
): # 确认是同一实例且任务未运行
self._clients.pop(telegram_id, None) # 从字典中移除该条目
async def stop(
self, telegram_id: int
) -> str: # 异步停止方法，接收telegram_id，返回状态字符串
"""
停止：先停任务/断开连接，再从表里删除，避免竞态。
"""
# 1) 抓到 entry 引用，不要急着从表里删
async with self._global_lock: # 获取全局锁
entry = self._clients.get(telegram_id) # 直接获取指定telegram_id的客户端条目
if not entry:
return f"账号{telegram_id}没有在运行" # 返回未运行提示
# 2) 专属锁内做真正停机
async with entry.lock: # 获取客户端条目的专属锁
if entry.task and not entry.task.done(): # 如果存在运行中的任务
entry.task.cancel() # 取消任务
with contextlib.suppress(asyncio.CancelledError): # 抑制取消异常
await entry.task # 等待任务结束
entry.task = None # 清空任务引用
with contextlib.suppress(Exception): # 抑制异常
await entry.client.disconnect() # 断开客户端连接
# 3) 现在再从表里删除，避免与同时 start 产生竞态
async with self._global_lock: # 获取全局锁
self._clients.pop(telegram_id, None) # 从字典中移除该条目
return "success" # 返回成功状态
async def send_text(
self, telegram_id: int, peer, text: str, reply_to: int | None = None
) -> str: # 异步发送文本方法，接收telegram_id、目标和文本内容，返回状态字符串
"""
发送：要求该账号已在运行；若断线自动重连。
参数：
telegram_id: Telegram账号ID
peer: 目标
text: 文本内容
reply_to: 回复消息的ID
返回：
- "success": 发送成功
- 其它字符串：错误提示（例如：主体不受支持）
"""
async with self._global_lock: # 获取全局锁
entry = self._clients.get(telegram_id) # 直接获取指定telegram_id的客户端条目
if not entry:
return f"账号{telegram_id}未运行，请先启动" # 返回未运行提示
# 确保连接
if not entry.client.is_connected(): # 检查是否已连接
await entry.client.connect() # 建立连接
# 可按需补：若未授权则提示
if not await entry.client.is_user_authorized(): # 检查是否已授权
with contextlib.suppress(Exception): # 抑制异常
await entry.client.disconnect() # 断开连接
return f"[{telegram_id}] 会话失效，请重新登录" # 返回会话失效提示
# 尝试解析 peer 为 Telegram 实体；若解析失败，给出自定义错误文本
try:
entity = await entry.client.get_input_entity(peer) # 获取目标实体信息
except Exception:
# 统一返回友好提示：仅支持 用户名/链接/群ID/群用户名；手机号必须在联系人中
return (
"您输入的主体不受支持，只能用户名或https://t.me/**，群ID，群用户名，如果是电话号码必须是联系人"
)
await entry.client.send_message(entity, text, reply_to=reply_to) # 发送文本消息
return "success" # 返回成功状态
def list_running(
self, user_id: int | None = None
) -> list[dict]: # 获取运行中的客户端列表方法，返回电话号码列表
"""
只返回"确实在跑"的账号（task 存在且未结束）
"""
result: list[dict] = []
for tid, entry in self._clients.items():
# 只取正在运行（task存在且未结束）
if not (entry.task and not entry.task.done()):
continue
# 如果指定了 user_id，则进行过滤
if user_id is not None and entry.user_id != user_id:
continue
# 组装返回信息
result.append(
{
"user_id": entry.user_id,
"telegram_id": entry.telegram_id,
"telegram_username": entry.telegram_username,
"keywords": list(entry.keywords) if entry.keywords else [],
"monitor_chat_ids": (
list(entry.monitor_chat_ids) if entry.monitor_chat_ids else []
),
"switch_all_chat": bool(entry.switch_all_chat),
"exclude_ids": list(entry.exclude_ids) if entry.exclude_ids else [],
}
)
return result # 中文注释：返回包含所需字段的字典列表
# 自动启动所有活跃的账号（从数据库中获取）
async def auto_start_telegram_clients(self) -> None:
"""
从数据库中获取所有is_active=True的账号并自动启动它们
"""
try:
# 从数据库获取所有活跃的Telegram账号
active_sessions = await db_get_all_active_telegram_session()
if not active_sessions:
await log.info("没有找到需要自动启动的活跃账号")
return
await log.info(f"发现 {len(active_sessions)} 个需要自动启动的账号")
# 逐个启动账号
for session in active_sessions:
try:
# 检查账号是否已经在运行
async with self._global_lock:
entry = self._clients.get(session.telegram_id)
if entry and entry.task and not entry.task.done():
await log.info(f"账号 {session.telegram_id} 已在运行中，跳过启动")
continue
# 获取用户自身的所有账号ID
exclude_ids = await db_list_user_telegram_session_ids(
session.user_id
)
# 启动账号（仅以 telegram_id 标识）
result = await self.start(
user_id=session.user_id,
telegram_id=session.telegram_id,
telegram_username=session.telegram_username,
keywords=session.keywords or [],
monitor_chat_ids=session.monitor_chat_ids or [],
switch_all_chat=session.switch_all_chat or False,
proxy=None,
device_model=session.device_model,
exclude_ids=exclude_ids,
)
if result == "success":
await log.info(f"账号 {session.telegram_id} 启动成功")
else:
await log.err(
f"账号 {session.telegram_id} 启动失败: {result}", kind="Telegram"
)
except Exception as e:
await log.err(f"启动账号 {session.telegram_id} 时发生异常: {e}", kind="Telegram")
except Exception as e:
await log.err(f"自动启动账号时发生异常: {e}", kind="Telegram")
# 修改某个telegram_id的keywords,switch_all_chat,monitor_chat_ids
async def update_config(
self,
telegram_id: int,
keywords: list[str] | None = None,
monitor_chat_ids: list[int] | None = None,
switch_all_chat: bool | None = None,
) -> str:
"""
修改指定telegram账号的配置信息
参数:
telegram_id (int): Telegram账号ID
keywords (list[str] | None): 要设置的关键字列表，为None则不修改
monitor_chat_ids (list[int] | None): 要监控的聊天ID列表，为None则不修改
switch_all_chat (bool | None): 是否响应所有聊天，为None则不修改
返回:
str: 操作结果，"success"表示成功
"""
# 先拿锁，避免并发修改
async with self._global_lock:
entry = self._clients.get(telegram_id) # 直接获取指定telegram_id的客户端条目
if not entry:
return "success"
# 再拿锁，避免并发修改
async with entry.lock:
if keywords is not None:
entry.keywords = (
keywords # 直接赋值，keywords已经是正确的类型 # 中文注释
)
if switch_all_chat is not None:
entry.switch_all_chat = bool(switch_all_chat)
if monitor_chat_ids is not None:
entry.monitor_chat_ids = monitor_chat_ids
return "success"
# 修改某个用户所有账号的exclude_ids
async def update_user_exclude_ids(
self, user_id: int, exclude_ids: list[int] | None = None
) -> str:
"""
修改某个用户所有账号的exclude_ids
当用户添加或删除了某个账号时，应调用此函数来更新该用户所有运行中账号的exclude_ids，
防止同一用户的不同账号互相聊天。
参数:
user_id (int): 用户ID
exclude_ids (list[int] | None): 要设置的排除账号ID列表。
如果为None，则从数据库获取该用户的所有账号ID作为排除列表。
返回:
str: 操作结果，"success"表示成功，其他字符串表示错误信息
"""
try:
# 获取该用户所有运行中的账号
running_accounts = []
async with self._global_lock:
for telegram_id, entry in self._clients.items():
# 只处理正在运行且属于指定用户的账号
if (
entry.task
and not entry.task.done()
and entry.user_id == user_id
):
running_accounts.append((telegram_id, entry))
if not running_accounts:
# 如果没有正在运行中的账号，直接返回成功
return "success"
# 如果exclude_ids为None，从数据库获取该用户的所有账号ID
if exclude_ids is None:
exclude_ids = await db_list_user_telegram_session_ids(user_id)
if not exclude_ids:
await log.info(f"用户 {user_id} 没有找到任何Telegram账号")
return "success" # 没有账号也算成功，不需要更新
# 更新每个运行中账号的exclude_ids
for telegram_id, entry in running_accounts:
async with entry.lock:
# 更新exclude_ids（排除自身账号ID）
old_exclude_ids = (
list(entry.exclude_ids) if entry.exclude_ids else []
)
entry.exclude_ids = list(exclude_ids)
# 记录变更（仅在有变化时记录）
if set(old_exclude_ids) != set(exclude_ids):
await log.info(
f"账号 {telegram_id} exclude_ids已更新: {old_exclude_ids} -> {exclude_ids}"
)
return "success"
except Exception as e:
await log.err(
f"更新用户 {user_id} 的exclude_ids时发生异常: {e}", kind="Telegram"
)
return f"更新失败: {str(e)}"
# 后台运行方法
async def _runner(self, entry: ClientEntry): # 私有后台运行方法，接收客户端条目
telegram_id = entry.telegram_id # 获取telegram_id
try:
await log.info(f"[{telegram_id}] 客户端启动") # 打印启动信息
await entry.client.run_until_disconnected() # 运行客户端直到断开连接
except asyncio.CancelledError: # 捕获取消异常
await log.info(f"[{telegram_id}] 客户端收到取消") # 打印取消信息
raise # 重新抛出异常
except Exception as ex: # 捕获其他异常
# 如果是身份和权限错误，则停止客户端
if isinstance(ex, (errors.UnauthorizedError, # 401 未授权 # 中文注释
errors.AuthKeyError, # 授权 key 类问题（基类） # 中文注释
errors.common.AuthKeyNotFound)):
await log.err(f"{telegram_id}客户端身份失效",kind="Telegram") # 打印异常信息
# 停止客户端
await self.stop(telegram_id)
# 更新数据库
await db_update_telegram_session_is_active(telegram_id,False,False)
return
await log.info(f"[{telegram_id}] runner 异常：{ex}") # 打印异常信息
finally:
with contextlib.suppress(Exception): # 抑制异常
await entry.client.disconnect() # 断开连接
await log.info(f"[{telegram_id}] 客户端结束") # 打印结束信息
# 停止所有客户端运行
async def stop_running(self):
self._shutting_down = True # ← 置位，阻止新的 start()
try:
# 先快照，避免长时间持有全局锁
async with self._global_lock:
items = list(self._clients.items())
for _, entry in items:
async with entry.lock:
if entry.task and not entry.task.done():
entry.task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await entry.task
entry.task = None
with contextlib.suppress(Exception):
await entry.client.disconnect()
# 清空管理表
async with self._global_lock:
self._clients.clear()
finally:
self._shutting_down = False # 关机状态复位
# @某个用户的“安全mention”工具（即使没有用户名也能@上）
def mention_user(self, user_id: int, name: str | None = None) -> str:
"""
生成可点击的 @ 提及（Markdown）
- 需要在 client 里 parse_mode='md' 或默认支持 Markdown
"""
# None/空串兜底
name = name or "user"
# 转义特殊字符
safe = (
name.replace("\\", "\\\\")
.replace("[", "\\[")
.replace("]", "\\]")
.replace("(", "\\(")
.replace(")", "\\)")
.replace("_", "\\_")
)
return f"[{safe}](tg://user?id={user_id})"
manager = ClientManager() # 实例化客户端管理器
fastapi实现调用和登录
py
from click.core import F
from pydantic import BaseModel, Field
from typing import Annotated,Optional
from fastapi import Response,APIRouter,Depends,UploadFile,Form
from db_ser.models import KolOrderModel
from fastapi_ser.routes.utils import Msg,random_device_model
from log_ser.as_log import Logger
import json
from fastapi_ser.jwtser import verify_auth
from telegram.client_manager import manager # 引入tg客户端管理器
from telethon import TelegramClient, errors
from uuid import uuid4
from fastapi_ser.routes.cache import set_tg_txid_cache,get_tg_txid_cache,delete_tg_txid_cache
import env
from fastapi_ser.fast_db import (
db_add_telegram_session,
db_check_telegram_session,
db_list_telegram_session,
db_transfer_telegram_session,
db_check_telegram_session_is_owner,
db_update_telegram_session_is_active,
db_get_user_telegram_session,
db_update_telegram_session_config
)
import os
log = Logger(__name__) # 日志对象
# 创建路由器
router = APIRouter(
prefix="/telegram", # 路由前缀
tags=["前台 / Telegram管理"] # API文档标签
)
# ====== 请求体模型 ======
# 发送验证码请求体
class SendCodeReq(BaseModel):
phone: Annotated[str,Field(..., description="国际区号+手机号码"),Msg("移动电话号码不能为空")]
force_sms: bool = Field(False, description="是否走短信通道")
# 校验验证码请求体
class VerifyCodeReq(BaseModel):
txid:Annotated[str,Field(..., description="第一次发码返回的事务ID"),Msg("事务ID不能为空")]
code:Annotated[str,Field(..., description="用户输入的验证码"),Msg("验证码不能为空")]
password: str | None = Field(None, description="若账号开启了2FA，这里带上密码（可选）")
# 登录第一步发送验证码
@router.post("/send_code",summary="登录第一步，发送验证码")
async def send_code(req: SendCodeReq,user_id:int=Depends(verify_auth(1))) -> str:
# 检查手机号格式前面是否有+号，如果存在，则去掉
if req.phone.startswith('+'):
phone = req.phone[1:]
else:
phone = req.phone
# 检查号码归属用户
check_user_id = await db_check_telegram_session(phone)
if check_user_id and check_user_id != user_id:
return Response(content="该Telegram号已绑定其他用户并登录，联系该用户用户转让给您", status_code=400)
client = TelegramClient(f'{env.TG_SESSION_DIR}/{phone}', env.TG_API_ID, env.TG_API_HASH) #构造客户端实例
# 显式建立连接
await client.connect()
try:
# 如果未登录才触发登录
if not await client.is_user_authorized():
# 补齐+号
plus_phone = f'+{phone}'
# 触发发码（通常发到应用内“Telegram 官方”对话，必要时尝试 force_sms）
try:
sent = await client.send_code_request(plus_phone, force_sms=req.force_sms) #发送验证码
txid = uuid4().hex # 生成事务ID
# 保存到redis
await set_tg_txid_cache(txid, sent.phone_code_hash, phone)
return txid # 返回事务ID，供后续提交验证码
except errors.FloodWaitError as e:
return Response(content=f"限流，请 {e.seconds}s 后重试", status_code=429) #被风控限流
except errors.PhoneNumberInvalidError:
return Response(content="手机号无效", status_code=400) #手机号格式或归属异常
except Exception as ex:
# 这里要同时写进err日志
log.err(f"tg发码失败：{ex}",kind="Telegram")
return Response(content=f"发码失败：{ex}", status_code=500) #其它异常
# 如果已登录，则返回已登录
return Response(content="该Telegram号已登录", status_code=400)
finally:
# 显式断开连接
await client.disconnect()
# 登录第二部，核验验证码+2FA密码
@router.post("/verify_code",summary="登录第二步，核验验证码+2FA密码")
async def verify_code(req: VerifyCodeReq,user_id:int=Depends(verify_auth(1))) -> Response:
# 根据txid读取缓存
cached_data = await get_tg_txid_cache(req.txid)
if not cached_data:
return Response(content="验证码已过期，重新发起登录", status_code=400)
phone_code_hash = cached_data["phone_code_hash"]
phone = cached_data["phone"]
# 随机生成一个设备型号
device_model = await random_device_model()
# 创建客户端实列
client = TelegramClient(f'{env.TG_SESSION_DIR}/{phone}', env.TG_API_ID, env.TG_API_HASH,device_model=device_model,app_version='CrowdPulse 1.0.1') #构造客户端实例
# 显式建立连接
await client.connect()
try:
# 如果未登录才触发登录
if not await client.is_user_authorized():
try:
# 继续用发码时保存的 phone_code_hash 完成登录
await client.sign_in( # 提交验证码完成登录
phone=phone,
code=req.code,
phone_code_hash=phone_code_hash
)
except errors.SessionPasswordNeededError:
# 需要二步验证密码
if not req.password:
return Response(content="该账号开启了二步验证，请提交password", status_code=401)
await client.sign_in(password=req.password) # 提交二步验证密码再次登录
except errors.PhoneCodeInvalidError:
return Response(content="验证码错误", status_code=400) #验证码不正确
except errors.PhoneCodeExpiredError:
return Response(content="验证码已过期", status_code=400) #验证码过期
except errors.FloodWaitError as e:
return Response(content=f"限流，请 {e.seconds}s 后重试", status_code=400) #限流重试
except Exception as ex:
# 这里要同时写进err日志
log.err(f"tg登录失败：{ex}",kind="Telegram")
return Response(content=f"登录失败：{ex}", status_code=500) #其它异常
# 走到这里表示登录成功,取登录用户信息
me = await client.get_me()
# 清理redis中的txid
await delete_tg_txid_cache(req.txid)
# 写入数据库,映射表
result = await db_add_telegram_session(dict(
user_id=user_id,
phone=phone,
telegram_id=me.id,
telegram_username=me.username or None,
device_model=device_model,
))
if result != "success":
return Response(content=result, status_code=500)
# 返回登录成功
return Response(content=json.dumps({
"id": me.id,
"username": me.username or None,
"name": f"{me.first_name or ''}{me.last_name or ''}".strip(),
}), status_code=200)
# 如果已登录，则返回已登录
return Response(content="该TG号已登录", status_code=400)
finally:
# 显式断开连接
await client.disconnect()
# 列出用户绑定的Telegram账号
@router.get("/list",summary="列出用户绑定的Telegram账号")
async def list_telegram_session(user_id:int=Depends(verify_auth(1))) -> list[dict]:
"""
列出用户绑定的Telegram账号
"""
return await db_list_telegram_session(user_id)
# 列出用户所有运行的账号
@router.get("/list_running",summary="列出所有运行的账号")
async def list_running(user_id:int=Depends(verify_auth(1))) -> list[dict]:
return manager.list_running(user_id)
# 转让Telegram账号
@router.post("/transfer",summary="转让Telegram账号")
async def transfer_telegram_session(phone:str,new_user:str,user_id:int=Depends(verify_auth(1))) -> Response:
"""
转让Telegram账号
"""
result = await db_transfer_telegram_session(phone,user_id,new_user)
if result != "success":
return Response(content=result, status_code=400)
return Response(content="转让成功", status_code=200)
# 检查登录状态
@router.get("/check_login_status",summary="检查登录状态")
async def auth_status(phone: str,response:Response,user_id:int=Depends(verify_auth(1))) -> bool|str:
"""
检查登录状态,该操作建议仅对未在运行的客户端去检查，已运行的客户端从list_running获取
"""
# 检查用户是否拥有该telegram号
is_owner = await db_check_telegram_session_is_owner(phone,user_id)
if not is_owner:
response.status_code = 400
return "您没有权限操作该Telegram账号"
client = TelegramClient(f'{env.TG_SESSION_DIR}/{phone}', env.TG_API_ID, env.TG_API_HASH)
# 显式建立连接
await client.connect()
try:
# 如果未登录才触发登录
if await client.is_user_authorized():
return True
return False
finally:
# 显式断开连接
await client.disconnect()
# 启动一个账号
@router.post("/client_start",summary="启动一个账号")
async def start_client(phone:str,user_id:int=Depends(verify_auth(1))) -> Response:
# 判断session文件是否存在
if not os.path.exists(f'{env.TG_SESSION_DIR}/{phone}.session'):
return Response(content="该Telegram账号不存在，请先登录", status_code=400)
# 取该用户该账号信息
session_info = await db_get_user_telegram_session(user_id,phone)
if not session_info:
return Response(content="您名下没有该Telegram账号", status_code=400)
# 启动账号,传递数据库中的信息
res = await manager.start(phone,user_id,session_info.telegram_id,session_info.telegram_username,session_info.keywords or None,session_info.monitor_chat_ids or None,session_info.switch_all_chat)
if res != "success":
return Response(content=res, status_code=400)
# 更新数据库中的激活状态
await db_update_telegram_session_is_active(phone,True)
return res
# 停止一个账号
@router.post("/client_stop",summary="停止一个账号")
async def stop_client(phone: str,user_id:int=Depends(verify_auth(1))) -> Response:
# 检查用户是否拥有该telegram号
is_owner = await db_check_telegram_session_is_owner(phone,user_id)
if not is_owner:
return Response(content="您没有权限操作该Telegram账号", status_code=400)
res = await manager.stop(phone)
if res != "success":
return Response(content=res, status_code=400)
# 更新数据库中的激活状态
await db_update_telegram_session_is_active(phone,False)
return Response(content="success", status_code=200)
# 取用户信息
@router.get("/client_get_user_info",summary="取用户信息")
async def get_user_info(phone: str,user_id:int=Depends(verify_auth(1))) -> Response:
"""
获取指定账号的用户信息
参数:
phone: 电话号码，作为账号标识符
返回:
Response: 包含用户信息的响应对象
"""
# 检查用户是否拥有该telegram号
is_owner = await db_check_telegram_session_is_owner(phone,user_id)
if not is_owner:
return Response(content="您没有权限操作该Telegram账号", status_code=400)
client = TelegramClient(f'{env.TG_SESSION_DIR}/{phone}', env.TG_API_ID, env.TG_API_HASH)
# 显式建立连接
await client.connect()
try:
# 如果未登录才触发登录
if await client.is_user_authorized():
me = await client.get_me()
return Response(content=json.dumps({
"id": me.id,
"username": me.username or None,
"first_name": (me.first_name or None),
"last_name": (me.last_name or None),
"phone": me.phone,
"premium": me.premium,
}, ensure_ascii=False, indent=2), status_code=200)
return Response(content="用户未登录", status_code=400)
except Exception as e:
return Response(content=str(e), status_code=400)
finally:
# 显式断开连接
await client.disconnect()
# 发送消息
@router.post("/client_send_message",summary="发送消息")
async def send_message(phone: str, message: str, peer: str="me",user_id:int=Depends(verify_auth(1))) -> Response:
"""
phone: 账号
message: 消息
peer: 接收方，默认me，即自己
"""
# 检查用户是否拥有该telegram号
is_owner = await db_check_telegram_session_is_owner(phone,user_id)
if not is_owner:
return Response(content="您没有权限操作该Telegram账号", status_code=400)
res = await manager.send_text(phone, peer, message)
if res != "success":
return Response(content=res, status_code=400)
return res
# 获取所有群组信息
@router.get("/client_get_all_groups",summary="指定账号，获取所有群组信息")
async def get_all_groups(phone: str,user_id:int=Depends(verify_auth(1))) -> Response:
# 检查用户是否拥有该telegram号
is_owner = await db_check_telegram_session_is_owner(phone,user_id)
if not is_owner:
return Response(content="您没有权限操作该Telegram账号", status_code=400)
try:
data = await manager.get_all_groups(phone)
return data
except Exception as e:
return Response(content=str(e), status_code=400)
# 修改配置模型
class UpdateConfigReq(BaseModel):
phone: Annotated[str,Field(..., description="账号"),Msg("账号不能为空")]
keywords: list[str] = Field([], description="触发关键词列表")
monitor_chat_ids: list[int] = Field([], description="监听群组列表")
switch_all_chat: bool = Field(False, description="是否响应所有聊天")
# 修改账号配置
@router.post("/client_update_config",summary="修改账号配置")
async def update_config(req: UpdateConfigReq,user_id:int=Depends(verify_auth(1))) -> Response:
"""
修改账号配置
"""
# 检查用户是否拥有该telegram号
is_owner = await db_check_telegram_session_is_owner(req.phone,user_id)
if not is_owner:
return Response(content="您没有权限操作该Telegram账号", status_code=400)
# 写入数据库
result = await db_update_telegram_session_config(req.phone, req.keywords, req.monitor_chat_ids, req.switch_all_chat)
if result != "success":
return Response(content=result, status_code=400)
# 更新内存中的配置
await manager.update_config(req.phone, req.keywords, req.monitor_chat_ids, req.switch_all_chat)
return Response(content="success", status_code=200)
StringSession 存储方式
telethon默认的存储session的格式是sqlite，除了身份验证还有一些缓存作用，以及拉取历史消息的游标等功能，使用StringSession存储则只存储一个身份验证密钥，极其轻量，但是需要自己维护游标等功能
ini
from telethon import TelegramClient
from telethon.sessions import StringSession,SQLiteSession
# 在内存中构建客户端实列
client = TelegramClient(StringSession(), env.TG_API_ID, env.TG_API_HASH)
# 假设这里执行了显式建立连接，并执行登录等操作
# 取出session字符串，这个字符串可以保存到数据库或文件
session_str = client.session.save()
# 第二次复用身份信息时，直接使用这个字符串
client = TelegramClient(StringSession(session_str), env.TG_API_ID, env.TG_API_HASH)
# 显式建立连接
await client.connect()
# 转存为SQLiteSession
# 创建文件
dst = SQLiteSession(f"{env.TG_SESSION_DIR}/{me.id}.session")
# 从上面的客户端中导入一些参数
dst.set_dc(client.session.dc_id, client.session.server_address, client.session.port)
dst.auth_key = client.session.auth_key
dst.takeout_id = getattr(client.session, "takeout_id", None) # 安全写入，因为这里通常为None
dst.save() # 保存
dst.close() # 关闭连接
# 显式断开连接
await client.disconnect()
# 后续使用sqlite方法再使用时：
TelegramClient(
f"{env.TG_SESSION_DIR}/{telegram_id}.session", # 这里直接填之之前保存的session文件路径即可
env.TG_API_ID,
env.TG_API_HASH,
proxy=proxy,
device_model=device_model,
app_version="CrowdPulse 1.0.1",
)