from fastapi import FastAPI, HTTPException import httpx import os from dotenv import load_dotenv from pydantic import BaseModel from telethon import TelegramClient, types from telethon.tl.functions.contacts import ImportContactsRequest import asyncio import hashlib from email_service import send_email from models import ApiResult, BotMessageRequest, ContactRequest, EmailRequest, LoginRequest, MessageRequest # Load biến môi trường từ file .env load_dotenv() # Thay thế bằng API ID và API Hash của bạn API_ID = int(os.getenv("API_ID")) API_HASH = os.getenv("API_HASH") # Token bot Telegram (Thay bằng token thực tế) BOT_TOKEN = os.getenv("BOT_TOKEN") TELEGRAM_API_URL = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage" clients = {} # Lưu trữ các client đang kết nối clients_lock = asyncio.Lock() # Đảm bảo thread-safe khi dùng nhiều request cùng lúc # Tên session file #SESSION_NAME = "telegram_session_user" # Khởi tạo FastAPI app = FastAPI() # Thư mục chứa session files SESSION_DIR = "sessions_tele" # Tự động tạo folder nếu chưa có if not os.path.exists(SESSION_DIR): os.makedirs(SESSION_DIR) # Khởi tạo Telegram client (sử dụng session đã lưu) #client = TelegramClient(SESSION_NAME, API_ID, API_HASH) # Kết nối Telegram khi API khởi động # @app.on_event("startup") # async def startup(): # """ Kết nối Telegram khi API khởi động """ # await client.connect() # Ngắt kết nối Telegram khi API tắt @app.on_event("shutdown") async def shutdown(): """ Ngắt kết nối Telegram khi API tắt """ await disconnect_all_clients() # add cliend async def get_client(phone: str) -> TelegramClient: """ Lấy hoặc tạo TelegramClient cho mỗi user dựa vào số điện thoại """ async with clients_lock: # Nếu client đã tồn tại và còn kết nối → trả về luôn if phone in clients: client = clients[phone] if client.is_connected(): return client # Nếu chưa có client hoặc đã disconnect → tạo mới session_name = generate_session_name(phone) client = TelegramClient(session_name, API_ID, API_HASH) # Kết nối client await client.connect() # Lưu vào pool để quản lý sau này clients[phone] = client return client async def disconnect_all_clients(): """ Ngắt kết nối tất cả client khi shutdown server hoặc cần reset """ async with clients_lock: for phone, client in clients.items(): if client.is_connected(): await client.disconnect() clients.clear() # Xoá toàn bộ sau khi disconnect # tạo Session_name def generate_session_name(phone: str) -> str: """ Tạo session_name duy nhất từ phone (MD5 hash) """ if not phone: raise ValueError("❌ Phone number is required!") hashed_phone = hashlib.md5(phone.encode()).hexdigest() session_path = os.path.join(SESSION_DIR, f"telegram_session_{hashed_phone}") return session_path # API đăng nhập @app.post("/api/integration/telegram/login/", response_model=ApiResult) async def login(request: LoginRequest): """ API đăng nhập vào Telegram """ try: client = await get_client(request.phone) # Kiểm tra kết nối Telegram if not client.is_connected(): await client.connect() # Kiểm tra đăng nhập if await client.is_user_authorized(): return ApiResult.success(data={ "session_name": client.session.filename },message="✅ Bạn đã đăng nhập rồi, không cần nhập mã OTP!") # Gửi mã OTP nếu chưa có if not request.otp: await client.send_code_request(request.phone) return ApiResult.success(data={ "session_name": client.session.filename },message="📩 Mã OTP đã được gửi, vui lòng nhập mã OTP để hoàn tất đăng nhập!") # Đăng nhập bằng mã OTP await client.sign_in(request.phone, request.otp) # Kiểm tra đăng nhập if await client.is_user_authorized(): return ApiResult.success(data={ "session_name": client.session.filename },message="🎉 Đăng nhập thành công! Session đã được lưu lại.") else: return ApiResult.error(message="❌ Đăng nhập thất bại! Hãy kiểm tra lại thông tin.", code=401) except Exception as e: return ApiResult.error(message=str(e), code=500) # API thêm danh bạ @app.post("/api/integration/telegram/add_contact/", response_model=ApiResult) async def add_contact(request: ContactRequest): """ API thêm danh bạ bằng username hoặc số điện thoại """ try: client = await get_client(request.phoneGui) # 🔹 Kiểm tra kết nối Telegram if not client.is_connected(): await client.connect() # 🔹 Kiểm tra đăng nhập if not await client.is_user_authorized(): return ApiResult.error(message="❌ Bạn chưa đăng nhập Telegram! Hãy gọi API /login/ trước.", code=401) if request.username: # 🔹 Tìm người dùng theo username entity = await client.get_entity(request.username) return ApiResult.success( data={ "user_id": entity.id, "username": entity.username }, message="✅ Đã tìm thấy người dùng!" ) elif request.phone: # 🔹 Thêm vào danh bạ nếu có số điện thoại contact = types.InputPhoneContact( client_id=0, phone=request.phone, first_name=request.first_name, last_name=request.last_name ) # 🔹 Gửi yêu cầu thêm vào danh bạ result = await client(ImportContactsRequest([contact])) # 🔹 Trả về thông tin người dùng nếu tìm thấy if result.users: user = result.users[0] return ApiResult.success( data={ "user_id": user.id, "username": user.username }, message="✅ Đã tìm thấy người dùng!" ) else: return ApiResult.error(message="⚠ Không tìm thấy người dùng với số điện thoại này.", code=404) else: return ApiResult.error(message="❌ Bạn phải nhập ít nhất username hoặc số điện thoại!", code=400) except Exception as e: return ApiResult.error(message=str(e), code=500) # API gửi tin nhắn @app.post("/api/integration/telegram/send_message/", response_model=ApiResult) async def send_message(request: MessageRequest): """ API gửi tin nhắn đến một người dùng hoặc nhóm """ try: client = await get_client(request.phoneGui) if not await client.is_user_authorized(): return ApiResult.error(message="❌ Bạn chưa đăng nhập Telegram! Hãy gọi API /login/ trước.", code=401) # Xác định người nhận tin nhắn if request.username: entity = await client.get_entity(request.username) elif request.user_id: entity = await client.get_entity(request.user_id) elif request.phone: entity = await client.get_entity(request.phone) else: return ApiResult.error(message="❌ Bạn phải nhập ít nhất một trong các trường: username, user_id hoặc phone!", code=400) # Gửi tin nhắn await client.send_message(entity, request.message) return ApiResult.success( message="📩 Tin nhắn đã được gửi thành công!", data={"recipient_id": entity.id, "recipient_name": entity.username or entity.first_name} ) except Exception as e: return ApiResult.error(message=str(e), code=500) # API gửi tin nhắn qua bot @app.post("/api/integration/telegram/bot/send_message/", response_model=ApiResult) async def send_message_via_bot(request: BotMessageRequest): """ Gửi tin nhắn qua bot Telegram """ try: client = await get_client(request.phone) # Gửi tin nhắn qua API của bot async with httpx.AsyncClient() as client: response = await client.post( TELEGRAM_API_URL, json={"chat_id": request.chat_id, "text": request.message} ) # Trả về kết quả if response.status_code == 200: return ApiResult.success( data=response.json(), message="📩 Tin nhắn đã được gửi thành công qua bot!" ) else: return ApiResult.error("❌ Gửi tin nhắn thất bại!", code=response.status_code) except Exception as e: return ApiResult.error(str(e), code=500) # API gửi email @app.post("/api/integration/email/send-email/", response_model=ApiResult) async def send_email_api(request: EmailRequest): return send_email(request.to_email, request.subject, request.body)