بناء بوت تليجرام ذكي بكلود AI

# إنشاء بيئة افتراضية
python -m venv telegram_bot_env
source telegram_bot_env/bin/activate   # Linux/Mac
# telegram_bot_env\Scripts\activate    # Windows

# تثبيت المكتبات
pip install python-telegram-bot==20.7
pip install anthropic
pip install python-dotenv
pip install aiohttp

TELEGRAM_BOT_TOKEN=7654321987:AAHxyz_your_actual_token_here
ANTHROPIC_API_KEY=sk-ant-api03-your_actual_key_here
BOT_NAME=SmartBot
MAX_HISTORY=20

import os
import logging
from dotenv import load_dotenv
from telegram import Update
from telegram.ext import (
    Application, CommandHandler, MessageHandler,
    filters, ContextTypes
)
import anthropic

load_dotenv()

# الإعدادات
TELEGRAM_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
ANTHROPIC_KEY  = os.getenv("ANTHROPIC_API_KEY")
MAX_HISTORY    = int(os.getenv("MAX_HISTORY", 20))

# تهيئة عميل كلود
claude = anthropic.Anthropic(api_key=ANTHROPIC_KEY)

# تخزين سجل المحادثات (في الذاكرة — استبدله بـ Redis للإنتاج)
conversation_history: dict[int, list] = {}

logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    level=logging.INFO
)

# ===== معالج /start =====
async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_name = update.effective_user.first_name
    welcome_text = (
        f"مرحباً {user_name}! أنا بوت ذكي مدعوم بـ Claude AI.\n\n"
        "يمكنني مساعدتك في:\n"
        "• الإجابة على أسئلتك بالعربية والإنجليزية\n"
        "• تحليل النصوص والمستندات\n"
        "• كتابة المحتوى والترجمة\n"
        "• البرمجة والأكواد\n\n"
        "أرسل لي أي رسالة للبدء، أو اكتب /help لمزيد من المعلومات."
    )
    await update.message.reply_text(welcome_text)

# ===== معالج /help =====
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    help_text = (
        "الأوامر المتاحة:\n\n"
        "/start — الصفحة الرئيسية\n"
        "/help — هذه الرسالة\n"
        "/clear — مسح سجل المحادثة\n"
        "/about — معلومات عن البوت\n\n"
        "يمكنك أيضاً إرسال أي رسالة نصية وسأرد عليك فوراً."
    )
    await update.message.reply_text(help_text)

# ===== معالج /clear =====
async def clear_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    conversation_history.pop(user_id, None)
    await update.message.reply_text("تم مسح سجل محادثتك. ابدأ من جديد!")

# ===== معالج الرسائل النصية العادية =====
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id  = update.effective_user.id
    user_msg = update.message.text

    # إضافة رسالة المستخدم للسجل
    if user_id not in conversation_history:
        conversation_history[user_id] = []
    conversation_history[user_id].append({"role": "user", "content": user_msg})

    # الاحتفاظ بآخر MAX_HISTORY رسالة فقط
    if len(conversation_history[user_id]) > MAX_HISTORY * 2:
        conversation_history[user_id] = conversation_history[user_id][-MAX_HISTORY * 2:]

    # إرسال "يكتب..." للمستخدم
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id,
        action="typing"
    )

    try:
        # استدعاء Claude API
        response = claude.messages.create(
            model="claude-opus-4-5",
            max_tokens=1024,
            system=(
                "أنت مساعد ذكي يتحدث العربية بطلاقة. "
                "كن مختصراً ومفيداً في ردودك. "
                "استخدم نقاطاً أو قوائم عند الحاجة لتنظيم المعلومات."
            ),
            messages=conversation_history[user_id]
        )

        bot_reply = response.content[0].text

        # إضافة رد البوت للسجل
        conversation_history[user_id].append({"role": "assistant", "content": bot_reply})

        await update.message.reply_text(bot_reply)

    except Exception as e:
        logging.error(f"خطأ في Claude API: {e}")
        await update.message.reply_text("حدث خطأ مؤقت. حاول مرة أخرى بعد لحظة.")

# ===== نقطة البداية =====
def main():
    app = Application.builder().token(TELEGRAM_TOKEN).build()

    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(CommandHandler("help",  help_command))
    app.add_handler(CommandHandler("clear", clear_command))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    print("البوت يعمل... اضغط Ctrl+C للإيقاف")
    app.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import CallbackQueryHandler

# ===== إرسال قائمة تفاعلية =====
async def menu_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    keyboard = [
        [
            InlineKeyboardButton("كتابة المحتوى", callback_data="mode_content"),
            InlineKeyboardButton("البرمجة",       callback_data="mode_code")
        ],
        [
            InlineKeyboardButton("الترجمة",       callback_data="mode_translate"),
            InlineKeyboardButton("التحليل",        callback_data="mode_analysis")
        ],
        [
            InlineKeyboardButton("محادثة عادية",  callback_data="mode_chat")
        ]
    ]
    reply_markup = InlineKeyboardMarkup(keyboard)

    await update.message.reply_text(
        "اختر وضع المساعدة المناسب:",
        reply_markup=reply_markup
    )

# ===== معالج ضغط الأزرار =====
async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()  # إيقاف رمز التحميل على الزر

    mode_labels = {
        "mode_content":   "كتابة المحتوى",
        "mode_code":      "مساعد برمجة",
        "mode_translate": "مترجم ذكي",
        "mode_analysis":  "محلل نصوص",
        "mode_chat":      "محادثة عامة"
    }

    selected = mode_labels.get(query.data, "محادثة عامة")
    user_id  = query.from_user.id

    # تحديث System Prompt بناءً على الاختيار
    context.user_data["mode"] = query.data

    await query.edit_message_text(
        f"تم اختيار: {selected}\n\nأرسل رسالتك الآن وسأساعدك في هذا الوضع."
    )

# أضف في main():
# app.add_handler(CommandHandler("menu", menu_command))
# app.add_handler(CallbackQueryHandler(button_callback))

async def moderate_group_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # تجاهل الرسائل من المشرفين
    chat_member = await context.bot.get_chat_member(
        update.effective_chat.id,
        update.effective_user.id
    )
    if chat_member.status in ["administrator", "creator"]:
        return

    message_text = update.message.text or ""
    if not message_text:
        return

    # تحليل الرسالة بكلود
    analysis = claude.messages.create(
        model="claude-haiku-4-5",
        max_tokens=100,
        system=(
            "أنت مشرف ذكي لمجموعة تليجرام. "
            "صنّف الرسالة التالية بكلمة واحدة فقط:\n"
            "SPAM: إعلان أو ترويج غير مرخص\n"
            "OFFENSIVE: محتوى مسيء أو مخالف\n"
            "FLOOD: رسائل متكررة بلا معنى\n"
            "OK: رسالة عادية مسموح بها"
        ),
        messages=[{"role": "user", "content": message_text}]
    )

    verdict = analysis.content[0].text.strip().upper()

    if verdict in ["SPAM", "OFFENSIVE", "FLOOD"]:
        # حذف الرسالة
        await update.message.delete()

        # تحذير المستخدم
        user_mention = update.effective_user.mention_html()
        warning_msg = await context.bot.send_message(
            chat_id=update.effective_chat.id,
            text=f"⚠️ {user_mention} تم حذف رسالتك لمخالفة قواعد المجموعة.",
            parse_mode="HTML"
        )

        # حذف رسالة التحذير بعد 10 ثوانٍ
        import asyncio
        await asyncio.sleep(10)
        await warning_msg.delete()

import asyncio
from datetime import datetime
from telegram.ext import Application

CHANNEL_ID = "@your_channel_username"  # أو رقم Channel ID

async def generate_and_post_daily_brief(bot):
    """تولّد وتنشر ملخصاً يومياً في القناة"""

    # توليد المحتوى بكلود
    today = datetime.now().strftime("%A %d %B %Y")
    response = claude.messages.create(
        model="claude-opus-4-5",
        max_tokens=600,
        system=(
            "أنت محرر نشرة إخبارية احترافية متخصص في التقنية والذكاء الاصطناعي. "
            "اكتب منشوراً موجزاً وجذاباً لقناة تليجرام. "
            "استخدم Emojis باعتدال وقسّم المحتوى بوضوح."
        ),
        messages=[{
            "role": "user",
            "content": f"اكتب ملخص يوم {today} في مجال الذكاء الاصطناعي والتقنية."
        }]
    )

    post_text = response.content[0].text

    # نشر في القناة
    await bot.send_message(
        chat_id=CHANNEL_ID,
        text=post_text,
        parse_mode="Markdown"
    )
    print(f"تم النشر بنجاح في {datetime.now()}")

async def scheduled_poster():
    """حلقة الجدولة — تنشر كل يوم في الساعة 9 صباحاً"""
    app = Application.builder().token(TELEGRAM_TOKEN).build()
    async with app:
        while True:
            now = datetime.now()
            # حساب الوقت حتى الساعة 9 صباحاً
            if now.hour == 9 and now.minute == 0:
                await generate_and_post_daily_brief(app.bot)
            await asyncio.sleep(60)  # فحص كل دقيقة

if __name__ == "__main__":
    asyncio.run(scheduled_poster())

import sqlite3
import json
from datetime import datetime

def init_db():
    conn = sqlite3.connect("bot_conversations.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS conversations (
            user_id   INTEGER PRIMARY KEY,
            messages  TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
    """)
    conn.commit()
    conn.close()

def get_history(user_id: int) -> list:
    conn = sqlite3.connect("bot_conversations.db")
    row = conn.execute(
        "SELECT messages FROM conversations WHERE user_id = ?", (user_id,)
    ).fetchone()
    conn.close()
    return json.loads(row[0]) if row else []

def save_history(user_id: int, messages: list):
    conn = sqlite3.connect("bot_conversations.db")
    conn.execute("""
        INSERT INTO conversations (user_id, messages, updated_at)
        VALUES (?, ?, ?)
        ON CONFLICT(user_id) DO UPDATE SET
            messages = excluded.messages,
            updated_at = excluded.updated_at
    """, (user_id, json.dumps(messages, ensure_ascii=False), datetime.now().isoformat()))
    conn.commit()
    conn.close()

def main_webhook():
    app = Application.builder().token(TELEGRAM_TOKEN).build()

    # أضف الـ Handlers هنا...

    WEBHOOK_URL = "https://yourdomain.com/webhook"
    PORT = int(os.getenv("PORT", 8443))

    app.run_webhook(
        listen="0.0.0.0",
        port=PORT,
        url_path="/webhook",
        webhook_url=WEBHOOK_URL,
        # للـ SSL الذاتي:
        # cert="./cert.pem",
        # key="./private.key"
    )