基于Python3.x+Django2.x实现Telegram用户授权及消息发送问题
解决Telethon中"phone_code_hash"缺失问题及持久化授权方案
刚好之前做过类似的Telegram授权功能,来帮你理清这个问题~
什么是phone_code_hash?
这是Telegram服务器在你请求发送验证码时返回的唯一验证标识,用来关联你的"验证码请求"和后续的"登录验证"动作,防止恶意重复请求或者验证码滥用——简单说就是服务器用来确认"你现在输入的验证码,确实是刚才给你那个手机号发的"。
如何获取phone_code_hash?
你不能直接调用sign_in,必须先调用send_code_request方法触发验证码发送,这个方法的返回值里就包含phone_code_hash。结合Django视图的示例如下:
from telethon import TelegramClient from django.http import JsonResponse from django.conf import settings import asyncio # 初始化Telethon客户端(建议把API信息存在Django settings里) client = TelegramClient('temp_session', settings.TELEGRAM_API_ID, settings.TELEGRAM_API_HASH) async def request_verify_code(request): phone_number = request.POST.get('phone') if not phone_number: return JsonResponse({'error': '手机号不能为空'}, status=400) # 确保客户端已连接 if not client.is_connected(): await client.connect() # 发送验证码请求,获取包含phone_code_hash的响应 code_response = await client.send_code_request(phone_number) # 把关键信息存在Django Session里,供后续验证使用 request.session['tg_phone'] = phone_number request.session['tg_code_hash'] = code_response.phone_code_hash return JsonResponse({'message': '验证码已发送,请查收'}) # 如果你用的是同步视图,需要用sync_to_async包装 from asgiref.sync import sync_to_async request_verify_code_sync = sync_to_async(request_verify_code)
接下来是验证登录的视图,必须传入之前保存的phone_code_hash:
async def verify_and_sign_in(request): phone = request.session.get('tg_phone') code_hash = request.session.get('tg_code_hash') input_code = request.POST.get('code') if not all([phone, code_hash, input_code]): return JsonResponse({'error': '缺少必要参数'}, status=400) try: # 这里必须传入phone_code_hash! await client.sign_in(phone=phone, code=input_code, phone_code_hash=code_hash) # 登录成功后,会话会自动保存到本地session文件(或你自定义的存储) return JsonResponse({'success': True, 'message': '登录成功'}) except Exception as e: return JsonResponse({'error': str(e)}, status=400) verify_and_sign_in_sync = sync_to_async(verify_and_sign_in)
实现多日多次发送消息的持久化授权
要支持跨多天发送消息,核心是持久化Telethon的会话数据,而不是每次请求都重新登录。这里给你两种适配Django场景的方案:
方案1:数据库存储会话(推荐多服务器/分布式场景)
Telethon支持自定义会话存储,你可以把会话数据存在Django数据库中,方便多进程/多服务器共享:
- 先定义Django模型存储会话:
from django.db import models class TelegramUserSession(models.Model): user = models.OneToOneField('auth.User', on_delete=models.CASCADE) session_data = models.TextField() # 序列化后的会话数据 updated_at = models.DateTimeField(auto_now=True)
- 自定义Telethon会话类:
from telethon.sessions import Session import pickle class DjangoDbSession(Session): def __init__(self, user): self.user = user super().__init__() self._load_session() def _load_session(self): try: session_obj = TelegramUserSession.objects.get(user=self.user) self._dc_id, self._server_address, self._port, self._auth_key = pickle.loads(session_obj.session_data.encode('latin1')) except TelegramUserSession.DoesNotExist: pass def save(self): # 序列化会话数据并存入数据库 session_data = pickle.dumps((self._dc_id, self._server_address, self._port, self._auth_key)) TelegramUserSession.objects.update_or_create( user=self.user, defaults={'session_data': session_data.decode('latin1')} ) def delete(self): TelegramUserSession.objects.filter(user=self.user).delete()
- 初始化客户端时使用自定义会话:
# 针对当前登录的Django用户初始化客户端 user = request.user client = TelegramClient(DjangoDbSession(user), settings.TELEGRAM_API_ID, settings.TELEGRAM_API_HASH)
这样用户登录后的会话会永久存在数据库,下次请求直接加载即可,无需重复验证。
方案2:本地会话文件(适合单服务器部署)
如果你的应用是单服务器运行,可以把会话文件按用户ID命名存在固定目录(比如telegram_sessions/user_123.session),每次请求时加载对应文件即可。但这种方式在多服务器环境下不适用,因为会话文件存在本地,其他节点无法访问。
额外注意事项
- 确保你的Telegram API ID和API Hash是在官方开发者平台注册的有效信息;
- Django中使用Telethon异步方法时,建议配置ASGI服务器(比如Daphne),或者用
sync_to_async包装异步函数适配同步视图; - 处理登录异常时,要区分"验证码错误""账号被限制""哈希过期"等不同情况,给用户更精准的提示。
内容的提问来源于stack exchange,提问作者Vladyslav
相关产品推荐
相关产品推荐

