194 lines
5.9 KiB
Python
Executable File
194 lines
5.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Telethon Session 创建脚本（安全版本）
|
||
|
||
这个脚本会在独立的安全目录中创建 Telegram session 文件
|
||
防止与其他项目冲突或被意外删除
|
||
"""
|
||
|
||
import asyncio
|
||
import os
|
||
from pathlib import Path
|
||
|
||
from telethon import TelegramClient
|
||
from telethon.errors import SessionPasswordNeededError
|
||
|
||
from env_loader import load_env
|
||
|
||
load_env()
|
||
|
||
API_ID = int(os.getenv("TELEGRAM_API_ID", "0") or 0)
|
||
API_HASH = os.getenv("TELEGRAM_API_HASH", "")
|
||
SESSION_BASE = os.path.expanduser(os.getenv("TELEGRAM_SESSION_PATH", str(Path.home() / "telegram_sessions" / "funstat_bot")))
|
||
SESSION_PATH = Path(SESSION_BASE)
|
||
SESSION_DIR = SESSION_PATH.parent
|
||
|
||
if not API_ID or not API_HASH:
|
||
raise RuntimeError("请在 .env 中设置 TELEGRAM_API_ID 和 TELEGRAM_API_HASH")
|
||
|
||
async def create_session():
|
||
"""创建 Telegram session 文件"""
|
||
|
||
print("=" * 60)
|
||
print("🚀 Telegram Session 创建工具（安全版本）")
|
||
print("=" * 60)
|
||
print()
|
||
|
||
# 创建 session 目录
|
||
SESSION_DIR.mkdir(parents=True, exist_ok=True)
|
||
print(f"📁 Session 目录: {SESSION_DIR}")
|
||
print()
|
||
|
||
# 创建客户端
|
||
client = TelegramClient(str(SESSION_PATH), API_ID, API_HASH)
|
||
|
||
print("📱 正在连接到 Telegram...")
|
||
await client.connect()
|
||
|
||
if not await client.is_user_authorized():
|
||
print()
|
||
print("=" * 60)
|
||
print("需要登录到你的 Telegram 账号")
|
||
print("=" * 60)
|
||
print()
|
||
|
||
# 请求手机号
|
||
phone = input("请输入你的手机号 (格式: +86xxxxxxxxxx): ")
|
||
|
||
try:
|
||
await client.send_code_request(phone)
|
||
print()
|
||
print("✅ 验证码已发送到你的 Telegram 客户端")
|
||
print(" (请检查你的 Telegram 应用)")
|
||
print()
|
||
|
||
# 请求验证码
|
||
code = input("请输入收到的验证码: ")
|
||
|
||
try:
|
||
await client.sign_in(phone, code)
|
||
except SessionPasswordNeededError:
|
||
# 如果账号设置了两步验证
|
||
print()
|
||
print("⚠️ 你的账号启用了两步验证")
|
||
password = input("请输入你的两步验证密码: ")
|
||
await client.sign_in(password=password)
|
||
|
||
except Exception as e:
|
||
print(f"❌ 登录失败: {e}")
|
||
await client.disconnect()
|
||
return False
|
||
|
||
# 验证登录成功
|
||
me = await client.get_me()
|
||
print()
|
||
print("=" * 60)
|
||
print("✅ 登录成功!")
|
||
print("=" * 60)
|
||
print(f"账号信息:")
|
||
print(f" - 用户名: @{me.username if me.username else '未设置'}")
|
||
print(f" - 姓名: {me.first_name} {me.last_name if me.last_name else ''}")
|
||
print(f" - 手机号: {me.phone}")
|
||
print(f" - ID: {me.id}")
|
||
print()
|
||
|
||
# 设置文件权限（仅当前用户可读写）
|
||
session_file = Path(f"{SESSION_PATH}.session")
|
||
if session_file.exists():
|
||
os.chmod(session_file, 0o600)
|
||
print(f"✅ Session 文件已创建: {session_file}")
|
||
print(f"✅ 文件权限已设置: 600 (仅你可读写)")
|
||
print()
|
||
|
||
# 测试与 @openaiw_bot 的连接
|
||
print("=" * 60)
|
||
print("🔍 正在测试与 @openaiw_bot 的连接...")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
# 查找 bot
|
||
bot_entity = await client.get_entity("@openaiw_bot")
|
||
print(f"✅ 找到 BOT: {bot_entity.first_name}")
|
||
print(f" BOT ID: {bot_entity.id}")
|
||
print()
|
||
|
||
# 发送测试消息
|
||
print("📤 发送测试消息: /start")
|
||
await client.send_message(bot_entity, "/start")
|
||
|
||
print("⏳ 等待 BOT 响应 (最多 10 秒)...")
|
||
|
||
# 等待响应
|
||
async def wait_for_response():
|
||
async for message in client.iter_messages(bot_entity, limit=1):
|
||
return message
|
||
|
||
try:
|
||
response = await asyncio.wait_for(wait_for_response(), timeout=10.0)
|
||
if response:
|
||
print()
|
||
print("=" * 60)
|
||
print("✅ 收到 BOT 响应:")
|
||
print("=" * 60)
|
||
print(response.text[:500]) # 显示前500个字符
|
||
print()
|
||
if len(response.text) > 500:
|
||
print(f"... (还有 {len(response.text) - 500} 个字符)")
|
||
print()
|
||
except asyncio.TimeoutError:
|
||
print("⚠️ 10秒内未收到响应，但连接正常")
|
||
print()
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {e}")
|
||
print()
|
||
|
||
await client.disconnect()
|
||
|
||
# 创建 .gitignore
|
||
gitignore_path = SESSION_DIR / ".gitignore"
|
||
if not gitignore_path.exists():
|
||
with open(gitignore_path, 'w') as f:
|
||
f.write("*.session\n")
|
||
f.write("*.session-journal\n")
|
||
print(f"✅ 创建了 .gitignore 防止意外提交")
|
||
print()
|
||
|
||
# 显示总结
|
||
print("=" * 60)
|
||
print("🎉 完成!")
|
||
print("=" * 60)
|
||
print()
|
||
print(f"Session 文件位置:")
|
||
print(f" {session_file}")
|
||
print()
|
||
print(f"安全检查:")
|
||
import stat
|
||
file_stat = session_file.stat()
|
||
perms = stat.filemode(file_stat.st_mode)
|
||
print(f" - 文件权限: {perms}")
|
||
print(f" - 文件大小: {file_stat.st_size} 字节")
|
||
print()
|
||
print(f"备份建议:")
|
||
print(f" cp {session_file} {session_file}.backup.$(date +%Y%m%d)")
|
||
print()
|
||
print("现在你可以在 MCP 服务器中使用这个 session 文件了！")
|
||
print("MCP 服务器会自动从这个安全位置读取 session。")
|
||
print()
|
||
|
||
return True
|
||
|
||
if __name__ == "__main__":
|
||
# 检查是否已安装 telethon
|
||
try:
|
||
import telethon
|
||
print(f"✅ Telethon 版本: {telethon.__version__}")
|
||
print()
|
||
except ImportError:
|
||
print("❌ 未安装 Telethon")
|
||
print("请先运行: pip install telethon")
|
||
exit(1)
|
||
|
||
# 运行创建流程
|
||
asyncio.run(create_session())
|