app.py

import os
import sys
import gc
import json
import logging
import asyncio
import traceback
from datetime import datetime
from pathlib import Path
from typing import Dict, Optional, Set, List, Any
from collections import deque
from contextlib import asynccontextmanager
import tempfile
import telegram

# تنظیمات اولیه و مسیرها
BASE_DIR = Path(__file__).parent
temp_cache_dir = tempfile.gettempdir()

# تنظیم متغیرهای محیطی قبل از import کردن کتابخانه‌های هوش مصنوعی
os.environ["HF_HOME"] = str(temp_cache_dir)
if "TRANSFORMERS_CACHE" in os.environ:
    del os.environ["TRANSFORMERS_CACHE"]
os.environ["TRANSFORMERS_PARALLELISM"] = "false"
os.environ["TORCH_HOME"] = str(Path(temp_cache_dir) / "torch")

# کتابخانه‌های هوش مصنوعی
import torch
import transformers
from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM, AutoModelForMaskedLM

# کتابخانه‌های وب و تلگرام
import aiofiles
import aiodns
import httpx
import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, HTTPException, Response, Query
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from telegram import Update, ReplyKeyboardMarkup, InlineKeyboardMarkup, InlineKeyboardButton, BotCommand, Bot
from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, ContextTypes, filters, ApplicationBuilder
from telegram.error import NetworkError

# کتابخانه‌های دیگر
import hazm
from logging.handlers import RotatingFileHandler
from cachetools import TTLCache

# تنظیمات برنامه
class AppConfig:
    """تنظیمات اصلی برنامه"""

    # تنظیمات مسیرها
    BASE_DIR = Path(__file__).parent
    STATIC_DIR = BASE_DIR / "static"
    TEMPLATES_DIR = BASE_DIR / "templates"
    CACHE_DIR = Path(temp_cache_dir) / "cache"
    LOG_DIR = Path(temp_cache_dir) / "logs"

    # تنظیمات تلگرام
    TELEGRAM_TOKEN = "YOUR_TELEGRAM_BOT_TOKEN"  # توکن ربات تلگرام خود را اینجا وارد کنید
    CHAT_ID = "YOUR_CHAT_ID"  # شناسه چت خود را اینجا وارد کنید
    ADMIN_CHAT_ID = "YOUR_ADMIN_CHAT_ID"  # شناسه ادمین خود را اینجا وارد کنید
    ALLOWED_CHAT_IDS = {CHAT_ID, ADMIN_CHAT_ID}  # در صورت نیاز، شناسه‌های مجاز را اینجا وارد کنید

    # تنظیمات مدل
    MODEL_NAME = "HooshvareLab/bert-fa-base-uncased"
    MAX_LENGTH = 512
    MIN_LENGTH = 20
    TEMPERATURE = 0.7
    TOP_K = 50
    TOP_P = 0.92

    # تنظیمات سیستم
    MAX_RETRIES = 3
    CACHE_TTL = 86400  # 24 ساعت
    MAX_CONNECTIONS = 5
    MEMORY_THRESHOLD = 85
    CLEANUP_THRESHOLD = 80
    MAX_MESSAGE_LENGTH = 1024
    MAX_THREADS = os.cpu_count() or 4

    @classmethod
    def setup_directories(cls) -> None:
        """ایجاد دایرکتوری‌های مورد نیاز"""
        try:
            for path in [cls.CACHE_DIR, cls.LOG_DIR, cls.STATIC_DIR, cls.TEMPLATES_DIR]:
                path.mkdir(exist_ok=True, parents=True)
            logging.info("دایرکتوری‌ها با موفقیت ایجاد شدند")
        except Exception as e:
            logging.error(f"خطا در ایجاد دایرکتوری‌ها: {e}")
            raise

# تنظیمات لاگ
def setup_logging() -> None:
    """راه‌اندازی سیستم لاگ"""
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )

    # لاگ در فایل
    file_handler = RotatingFileHandler(
        AppConfig.LOG_DIR / 'app.log',
        maxBytes=10 ** 6,
        backupCount=5,
        encoding='utf-8'
    )
    file_handler.setFormatter(formatter)

    # لاگ در کنسول
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(formatter)

    logger.handlers = []
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)

# مدل هوش مصنوعی
class AIModel:
    """کلاس مدیریت مدل هوش مصنوعی"""

    def __init__(self):
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model = None
        self.tokenizer = None
        self.pipeline = None
        self._initialize_model()

    def _initialize_model(self):
        try:
            logging.info(f"در حال بارگذاری مدل {AppConfig.MODEL_NAME}...")

            # بارگذاری توکنایزر
            self.tokenizer = AutoTokenizer.from_pretrained(
                AppConfig.MODEL_NAME,
                cache_dir=AppConfig.CACHE_DIR,
                use_fast=True
            )

            # بارگذاری مدل
            if torch.cuda.is_available():
                self.model = AutoModelForCausalLM.from_pretrained(
                    AppConfig.MODEL_NAME,
                    cache_dir=AppConfig.CACHE_DIR,
                    torch_dtype=torch.float16,
                    low_cpu_mem_usage=True,
                )
            else:
                self.model = AutoModelForCausalLM.from_pretrained(
                    AppConfig.MODEL_NAME,
                    cache_dir=AppConfig.CACHE_DIR,
                    low_cpu_mem_usage=True,
                )
                logging.warning("CUDA در دسترس نیست. استفاده از CPU ممکن است کند باشد.")

            # ایجاد پایپلاین
            self.pipeline = pipeline(
                "text-generation",
                model=self.model,
                tokenizer=self.tokenizer,
                device=0 if torch.cuda.is_available() else -1,
            )

            logging.info("مدل با موفقیت بارگذاری شد")
            return True
        except Exception as e:
            logging.error(f"خطا در بارگذاری مدل: {e}")
            return False

    async def generate_response(self, text: str) -> str:
        try:
            # پیش پردازش متن
            # ... (می‌توانید توابع پیش پردازش مانند حذف ایموجی، اعداد و غیره را اضافه کنید)

            # تولید پاسخ
            result = self.pipeline(
                text,
                max_length=AppConfig.MAX_LENGTH,
                min_length=AppConfig.MIN_LENGTH,
                temperature=AppConfig.TEMPERATURE,
                top_k=AppConfig.TOP_K,
                top_p=AppConfig.TOP_P,
                do_sample=True,
                no_repeat_ngram_size=3,
                pad_token_id=self.tokenizer.eos_token_id,
                num_return_sequences=1,
            )

            response = result[0]["generated_text"]

            # پس پردازش پاسخ
            response = response.replace(text, "").strip()

            return response

        except Exception as e:
            logging.error(f"خطا در تولید پاسخ: {e}")
            return "متأسفم، در تولید پاسخ مشکلی پیش آمد."

# رابط کاربری تلگرام
class TelegramUI:
    """کلاس مدیریت رابط کاربری تلگرام"""

    @staticmethod
    def get_main_menu() -> InlineKeyboardMarkup:
        """ایجاد منوی اصلی"""
        keyboard = [
            [
                InlineKeyboardButton("💬 چت جدید", callback_data="new_chat"),
                InlineKeyboardButton("🔍 جستجو", callback_data="search")
            ],
            [
                InlineKeyboardButton("📚 آموزش", callback_data="learn"),
                InlineKeyboardButton("🎯 تمرین", callback_data="practice")
            ],
            [
                InlineKeyboardButton("⚙️ تنظیمات", callback_data="settings"),
                InlineKeyboardButton("❓ راهنما", callback_data="help")
            ]
        ]
        return InlineKeyboardMarkup(keyboard)

    @staticmethod
    def get_main_menu_with_icons() -> InlineKeyboardMarkup:
        """ایجاد منوی اصلی با آیکون"""
        keyboard = [
            [
                InlineKeyboardButton("💬", callback_data="new_chat"),
                InlineKeyboardButton("🔍", callback_data="search")
            ],
            [
                InlineKeyboardButton("📚", callback_data="learn"),
                InlineKeyboardButton("🎯", callback_data="practice")
            ],
            [
                InlineKeyboardButton("⚙️", callback_data="settings"),
                InlineKeyboardButton("❓", callback_data="help")
            ]
        ]
        return InlineKeyboardMarkup(keyboard)

    @staticmethod
    def get_chat_keyboard() -> ReplyKeyboardMarkup:
        """ایجاد صفحه کلید چت"""
        keyboard = [
            ["💭 چت عمومی", "❓ سوال و جواب"],
            ["📚 آموزش", "🎯 تمرین"],
            ["🏠 منوی اصلی"]
        ]
        return ReplyKeyboardMarkup(keyboard, resize_keyboard=True, is_persistent=True)

# ربات تلگرام
class TelegramBot:
    """کلاس اصلی ربات تلگرام"""

    def __init__(self, ai_model: AIModel):
        self.model = ai_model
        self.ui = TelegramUI()
        self.application = (
            ApplicationBuilder()
            .token(AppConfig.TELEGRAM_TOKEN)
            .build()
        )
        self._setup_handlers()
        self.normalizer = hazm.Normalizer()
        self.message_queue = asyncio.Queue()
        asyncio.create_task(self.process_message_queue())
        self._set_my_commands()

    def _set_my_commands(self):
        """تنظیم دستورات ربات"""
        bot = Bot(AppConfig.TELEGRAM_TOKEN)

        async def set_commands():
            await bot.set_my_commands([
                BotCommand("help", "نمایش راهنما"),
                BotCommand("model", "نمایش اطلاعات مدل"),
                BotCommand("status", "نمایش وضعیت سرور"),
            ])

        asyncio.create_task(set_commands())

    def _setup_handlers(self):
        """تنظیم هندلرهای ربات"""
        self.application.add_handler(CallbackQueryHandler(self.button_callback))
        self.application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_message))
        self.application.add_handler(CommandHandler("help", self.help_command))
        self.application.add_handler(CommandHandler("model", self.model_command))
        self.application.add_handler(CommandHandler("status", self.status_command))

    async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """پردازش دستور راهنما"""
        help_text = """
راهنمای استفاده از ربات:

/help - نمایش راهنما
/model - نمایش اطلاعات مدل
/status - نمایش وضعیت سرور

برای ارسال پیام، کافیست متن خود را تایپ کرده و ارسال کنید.
"""
        await update.message.reply_text(help_text)

    async def model_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """پردازش دستور نمایش اطلاعات مدل"""
        model_info = f"""
مدل: {AppConfig.MODEL_NAME}
حداکثر طول ورودی: {AppConfig.MAX_LENGTH}
دمای تولید: {AppConfig.TEMPERATURE}
"""
        await update.message.reply_text(model_info)

    async def status_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """پردازش دستور نمایش وضعیت سرور"""
        try:
            # بررسی وضعیت سرور
            status_text = "وضعیت سرور: نرمال"
        except Exception as e:
            logging.error(f"خطا در بررسی وضعیت سرور: {e}")
            status_text = "خطا در بررسی وضعیت سرور"
        await update.message.reply_text(status_text)

    async def button_callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """پردازش دکمه‌های منو"""
        query = update.callback_query
        await query.answer()

        if query.data == "new_chat":
            await query.message.reply_text(
                "💭 نوع چت را انتخاب کنید:",
                reply_markup=self.ui.get_chat_keyboard()
            )
        elif query.data == "help":
            help_text = """
📚 راهنمای استفاده:

1️⃣ برای شروع گفتگو، متن خود را تایپ و ارسال کنید.
2️⃣ نوع گفتگو را انتخاب کنید
3️⃣ سؤال یا پیام خود را ارسال کنید
4️⃣ منتظر پاسخ بمانید

❗️ نکات مهم:
• سؤالات خود را واضح و دقیق مطرح کنید
• از منوی تنظیمات برای شخصی‌سازی استفاده کنید
• برای بازگشت به منوی اصلی از دکمه "🏠 منوی اصلی" استفاده کنید
"""
            await query.message.edit_text(help_text, reply_markup=self.ui.get_main_menu_with_icons())
        elif query.data == "settings":
            await query.message.reply_text("⚙️ در حال حاضر تنظیمات خاصی در دسترس نیست.")
        elif query.data == "learn":
            await query.message.reply_text("📚 در حال حاضر محتوای آموزشی‌ای در دسترس نیست.")
        elif query.data == "practice":
            await query.message.reply_text("🎯 در حال حاضر تمرینی در دسترس نیست.")
        elif query.data == "search":
            await query.message.reply_text("🔍 در حال حاضر قابلیتی برای جستجو در دسترس نیست.")
        else:
            await query.message.edit_text("منوی اصلی:", reply_markup=self.ui.get_main_menu_with_icons())

    async def handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """پردازش پیام‌های دریافتی"""
        if update.message.text == "🏠 منوی اصلی":
            await update.message.reply_text("منوی اصلی:", reply_markup=self.ui.get_main_menu_with_icons())
            return

        await self.message_queue.put((update, context))

    async def process_message_queue(self):
        """پردازش صف پیام‌ها"""
        while True:
            update, context = await self.message_queue.get()
            try:
                # نرمال‌سازی متن
                normalized_text = self.normalizer.normalize(update.message.text)

                # بررسی طول پیام
                if len(normalized_text) > AppConfig.MAX_MESSAGE_LENGTH:
                    await update.message.reply_text(
                        "پیام شما بیش از حد طولانی است. لطفاً آن را کوتاه‌تر کنید."
                    )
                    continue

                # تولید پاسخ
                response = await self.model.generate_response(normalized_text)

                # ارسال پاسخ
                if len(response) > AppConfig.MAX_MESSAGE_LENGTH:
                    for i in range(0, len(response), AppConfig.MAX_MESSAGE_LENGTH):
                        await update.message.reply_text(f"🤖 {response[i:i+AppConfig.MAX_MESSAGE_LENGTH]}")
                else:
                    await update.message.reply_text(f"🤖 {response}")

            except Exception as e:
                logging.error(f"خطا در پردازش پیام: {e}")
                await update.message.reply_text("متأسفانه خطایی رخ داد. لطفاً دوباره تلاش کنید.")
            finally:
                self.message_queue.task_done()

    async def run(self):
        """اجرای ربات"""
        try:
            self.application.run_polling(
                allowed_updates=Update.ALL_TYPES,
                drop_pending_updates=True
            )

        except NetworkError as e:
            logging.error(f"خطای شبکه: {e}")
            await asyncio.sleep(5)  # صبر برای 5 ثانیه
            await self.run()  # تلاش مجدد برای اتصال
        except Exception as e:
            logging.error(f"خطا در اجرای ربات: {e}")
            raise

# برنامه FastAPI
app = FastAPI(title="AI Assistant", lifespan=lifespan)

# پیوست کردن فایل‌های استاتیک
app.mount("/static", StaticFiles(directory=AppConfig.STATIC_DIR), name="static")
templates = Jinja2Templates(directory=AppConfig.TEMPLATES_DIR)

# HTML پیش‌فرض
DEFAULT_HTML = """
<!DOCTYPE html>
<html dir="rtl" lang="fa">
<head>
    <meta charset="UTF-8">
    <title>دستیار هوشمند</title>
    <style>
        body {
            font-family: Tahoma, Arial;
            background-color: #f5f5f5;
            margin: 0;
            padding: 20px;
        }
        .container {
            max-width: 800px;
            margin: 0 auto;
            background: white;
            padding: 20px;
            border-radius: 10px;
            box-shadow: 0 2px 4px rgba(0,0,0,0.1);
        }
        h1 { color: #1a73e8; }
        .status { margin: 20px 0; }
        .chat-container {
            margin-top: 20px;
            border: 1px solid #e0e0e0;
            border-radius: 8px;
            padding: 10px;
            min-height: 300px;
        }
    </style>
</head>
<body>
    <div class="container">
        <h1>دستیار هوشمند</h1>
        <div class="status">
            <p>برای استفاده از دستیار هوشمند، به ربات تلگرام مراجعه کنید یا از رابط وب استفاده نمایید.</p>
        </div>
        <div class="chat-container" id="chat-container">
        </div>
        <input type="text" id="message-input" placeholder="پیام خود را بنویسید..." style="width: 100%; padding: 10px; margin-top: 10px;">
        <button onclick="sendMessage()" style="margin-top: 10px; padding: 10px 20px;">ارسال</button>
    </div>
    <script>
        const ws = new WebSocket(`ws://${window.location.host}/ws`);
        const chatContainer = document.getElementById('chat-container');
        const messageInput = document.getElementById('message-input');

        ws.onmessage = function(event) {
            const response = JSON.parse(event.data);
            appendMessage('bot', response.response);
        };

        function appendMessage(type, text) {
            const messageDiv = document.createElement('div');
            messageDiv.style.padding = '10px';
            messageDiv.style.margin = '5px';
            messageDiv.style.borderRadius = '5px';

            if (type === 'user') {
                messageDiv.style.backgroundColor = '#e3f2fd';
                messageDiv.style.marginLeft = '20%';
            } else {
                messageDiv.style.backgroundColor = '#f5f5f5';
                messageDiv.style.marginRight = '20%';
            }

            messageDiv.textContent = text;
            chatContainer.appendChild(messageDiv);
            chatContainer.scrollTop = chatContainer.scrollHeight;
        }

        function sendMessage() {
            const message = messageInput.value.trim();
            if (message) {
                appendMessage('user', message);
                ws.send(message);
                messageInput.value = '';
            }
        }

        messageInput.addEventListener('keypress', function(e) {
            if (e.key === 'Enter') {
                sendMessage();
            }
        });
    </script>
</body>
</html>
"""

@asynccontextmanager
async def lifespan(app: FastAPI):
    """مدیریت چرخه حیات برنامه"""
    # راه‌اندازی
    AppConfig.setup_directories()
    setup_logging()

    # ایجاد نمونه از مدل هوش مصنوعی
    app.state.model = AIModel()
    app.state.is_ready = asyncio.Event()

    # راه‌اندازی ربات تلگرام
    app.state.bot = TelegramBot(app.state.model)
    asyncio.create_task(app.state.bot.run())

    yield

    # خاتمه
    tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    for task in tasks:
        task.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    await app.state.bot.application.shutdown()
    logging.info("برنامه با موفقیت خاتمه یافت")

@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
    """صفحه اصلی"""
    try:
        # می‌توانید اطلاعات بیشتری را به تمپلیت ارسال کنید
        return templates.TemplateResponse(
            "index.html",
            {
                "request": request,
                "bot_status": "فعال" if request.app.state.is_ready.is_set() else "در حال آماده‌سازی"
            }
        )
    except Exception as e:
        logging.error(f"خطا در نمایش صفحه اصلی: {e}")
        return HTMLResponse(content=DEFAULT_HTML)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """اتصال WebSocket"""
    await websocket.accept()

    try:
        while True:
            # دریافت پیام
            text = await websocket.receive_text()

            # تولید پاسخ
            response = await app.state.model.generate_response(text)

            # ارسال پاسخ
            await websocket.send_json({
                "response": response
            })
    except WebSocketDisconnect:
        logging.info("اتصال WebSocket قطع شد")
    except Exception as e:
        logging.error(f"خطا در WebSocket: {e}")
        await websocket.close()

@app.exception_handler(404)
async def not_found_handler(request: Request, exc: Exception):
    """پردازش خطای 404"""
    return HTMLResponse(content=DEFAULT_HTML, status_code=404)

@app.exception_handler(500)
async def server_error_handler(request: Request, exc: Exception):
    """پردازش خطای 500"""
    return HTMLResponse(content=DEFAULT_HTML, status_code=500)

if __name__ == "__main__":
    # اجرای برنامه
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=7860,
        reload=False,
        workers=1,
        log_level="info"
    )