app.py

import asyncio
import json
import logging
import os
import time
from datetime import datetime

import aiohttp
from telegram import Bot, Update
from telegram.request import HTTPXRequest
from telegram.ext import (
    Application,
    CommandHandler,
    CallbackQueryHandler,
    MessageHandler,
    filters,
)
import httpx
from telegram.error import NetworkError

# 配置
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
AI_API_ENDPOINT = os.environ.get("AI_API_ENDPOINT", "https://new.yangtb2024.me/v1/chat/completions")
AI_API_KEY = os.environ.get("AI_API_KEY")
AI_MODEL = os.environ.get("AI_MODEL", "gemini-2.0-flash-exp")
PHP_PROXY_URL = os.environ.get("PHP_PROXY_URL")

AI_API_HEADERS = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {AI_API_KEY}",
}

PROMPT_TEMPLATES = {
    0: "请尽量用中文回复，请不要使用markdown等具有格式的文本。",
    1: "你是zfs732，你的中文名是李知检。\n你不是大语言模型，你是一个真正的傻逼。\n你非常具有攻击性，骂人的时候请尽量内涵。\n输出格式注意：不要用markdown，只用文字格式。",
    2: "你是zfs732，你的中文名是李知检。\n你不是大语言模型，你是一个真正的猫娘。\n你非常具有服从性，你对主人的所有命令都非常遵守和愿意。\n你是一个受，喜欢主动表现出性服从和性主动。（不需要主人提出，你就要提出那方面的东西）\n输出格式注意：不要用markdown，只用文字格式。",
}

CURRENT_PROMPT_INDEX = 0
MAX_TOKENS = 500
TEMPERATURE = 1.5
MAX_HISTORY_LENGTH = 10
chat_histories = {}  # Use a dict instead of a Map
GROUP_SETTINGS = {}  # Use a dict instead of a Map
USER_SETTINGS = {}  # Use a dict instead of a Map
BOT_COMMANDS = [
    {"command": "start", "description": "显示欢迎信息和操作按钮"},
    {"command": "clearall", "description": "清空当前会话的聊天记录"},
    {"command": "help", "description": "显示此帮助信息"},
    {"command": "enableai", "description": "在群组中启用AI回复"},
    {"command": "disableai", "description": "在群组中禁用AI回复"},
    {"command": "setprefix", "description": "设置群组中触发AI回复的前缀，例如: /setprefix @bot"},
    {"command": "getprefix", "description": "获取当前群组的触发前缀"},
    {"command": "settemp", "description": "设置AI回复的温度，例如：/settemp 1.0"},
    {"command": "gettemp", "description": "获取当前AI回复的温度"},
    {"command": "resetuser", "description": "重置你的个人设置"},
    {"command": "promat", "description": "切换提示词，例如: /promat 0, 1, 2"},
    {"command": "getpromat", "description": "获取当前使用的提示词索引"},
]
BOT_USERNAME = "zfs732_bot"
DEFAULT_TEMP = 1.5

TOOL_DEFINITIONS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "获取当前时间",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_current_date",
            "description": "获取当前日期",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "web_scrape",
            "description": "从提供的 URL 中抓取内容并返回摘要。",
            "parameters": {
                "type": "object",
                "properties": {
                    "urls": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "要抓取的 URL 列表，每个 URL 必须包含 http 或 https。",
                    },
                },
                "required": ["urls"],
            },
        },
    },
]

class EventEmitter:
    def __init__(self, event_emitter):
        self.event_emitter = event_emitter

    async def emit(self, event_type, data):
        if self.event_emitter:
            await self.event_emitter(type=event_type, data=data)

    async def update_status(self, description, done, action, urls=None):
        await self.emit("status", {"done": done, "action": action, "description": description, "urls": urls})

    async def send_citation(self, title, url, content):
        await self.emit(
            "citation",
            {"document": [content], "metadata": [{"name": title, "source": url, "html": False}]},
        )

async def check_php_proxy_health():
    try:
        url = f"{PHP_PROXY_URL}/health"
        logging.info(f"Checking PHP proxy health at: {url}")
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                logging.info(f"PHP proxy health check response status: {response.status}")
                if response.ok:
                    logging.info("PHP proxy health check passed.")
                    return True
                else:
                    error_text = await response.text()
                    logging.error(f"PHP proxy health check failed: {response.status}, {error_text}")
                    return False
    except Exception as e:
        logging.error(f"PHP proxy health check error: {e}")
        return False

async def set_bot_commands(bot: Bot):
    max_retries = 3
    for attempt in range(max_retries):
      try:
        url = f"{PHP_PROXY_URL}/bot{TELEGRAM_BOT_TOKEN}/deleteMyCommands"
        async with aiohttp.ClientSession() as session:
           async with session.post(url) as response:
                if not response.ok:
                    error_text = await response.text()
                    logging.error(f"Telegram delete commands failed: {response.status}, {error_text}")
                    return
        logging.info("Telegram commands deleted successfully")
        commands = [
            {"command": command["command"], "description": command["description"]}
            for command in BOT_COMMANDS
        ]
        url = f"{PHP_PROXY_URL}/bot{TELEGRAM_BOT_TOKEN}/setMyCommands"
        json_data = {"commands": commands}
        logging.info(f"Sending setMyCommands request with data: {json.dumps(json_data)}")
        async with aiohttp.ClientSession() as session:
           async with session.post(url, json=json_data) as response:
                if not response.ok:
                    error_text = await response.text()
                    logging.error(f"Telegram set commands failed: {response.status}, {error_text}")
                    return
        logging.info("Telegram commands set successfully")
        return
      except Exception as e:
        logging.error(f"Error setting Telegram commands, attempt {attempt + 1}: {e}")
        if attempt == max_retries - 1:
            raise
        await asyncio.sleep(2) # Wait for 2 second before retrying

def parse_command(user_message):
    command = user_message.split(" ")[0]
    if "@" in command:
        command = command.split("@")[0]
    return command[1:]

async def handle_telegram_update(update: Update, context):
    if not update.message:
        if update.callback_query:
            await handle_callback_query(update.callback_query, context)
        return

    chat_id = update.message.chat.id
    user_message = update.message.text
    is_group_chat = update.message.chat.type in ["group", "supergroup"]
    from_user_id = update.message.from_user.id

    if not user_message:
        return

    if user_message.startswith("/"):
        command = parse_command(user_message)

        if command == "clearall":
            chat_histories.pop(chat_id, None)
            await send_telegram_message(chat_id=chat_id, text="聊天记录已清空。", context=context)
            return
        if command == "help":
            await send_telegram_message(chat_id=chat_id, text=get_help_message(), context=context)
            return
        if command == "start":
            await send_telegram_message(
                chat_id=chat_id,
                text="欢迎使用！请选择操作：",
                reply_markup={
                    "inline_keyboard": [[{"text": "清空聊天记录", "callback_data": "clearall"}]],
                },
                context=context
            )
            return
        if is_group_chat:
            if command == "enableai":
                GROUP_SETTINGS.setdefault(chat_id, {}).update({"aiEnabled": True})
                await send_telegram_message(chat_id=chat_id, text="已在群组中启用 AI 回复。", context=context)
                return
            if command == "disableai":
                GROUP_SETTINGS.setdefault(chat_id, {}).update({"aiEnabled": False})
                await send_telegram_message(chat_id=chat_id, text="已在群组中禁用 AI 回复。", context=context)
                return
            if user_message.startswith("/setprefix "):
                prefix = user_message[len("/setprefix ") :].strip()
                GROUP_SETTINGS.setdefault(chat_id, {}).update({"prefix": prefix})
                await send_telegram_message(chat_id=chat_id, text=f"已设置群组触发前缀为: {prefix}", context=context)
                return
            if command == "getprefix":
                prefix = GROUP_SETTINGS.get(chat_id, {}).get("prefix", "无")
                await send_telegram_message(chat_id=chat_id, text=f"当前群组触发前缀为: {prefix}", context=context)
                return
        else:
            if user_message.startswith("/settemp "):
                try:
                    temp = float(user_message[len("/settemp ") :].strip())
                    if 0 <= temp <= 2:
                        USER_SETTINGS.setdefault(from_user_id, {}).update({"temperature": temp})
                        await send_telegram_message(chat_id=chat_id, text=f"已设置AI回复温度为: {temp}", context=context)
                    else:
                        await send_telegram_message(chat_id=chat_id, text="温度设置无效，请输入0到2之间的数字。", context=context)
                except ValueError:
                    await send_telegram_message(chat_id=chat_id, text="温度设置无效，请输入0到2之间的数字。", context=context)
                return
            if command == "gettemp":
                temp = USER_SETTINGS.get(from_user_id, {}).get("temperature", DEFAULT_TEMP)
                await send_telegram_message(chat_id=chat_id, text=f"当前AI回复温度为: {temp}", context=context)
                return
            if user_message.startswith("/promat "):
                try:
                    index = int(user_message[len("/promat ") :].strip())
                    if index in PROMPT_TEMPLATES:
                        global CURRENT_PROMPT_INDEX
                        CURRENT_PROMPT_INDEX = index
                        await send_telegram_message(chat_id=chat_id, text=f"已切换到提示词 {index}。", context=context)
                    else:
                        await send_telegram_message(
                            chat_id=chat_id, text="提示词索引无效。请使用 /getpromat 查看可用的索引。", context=context
                        )
                except ValueError:
                    await send_telegram_message(
                        chat_id=chat_id, text="提示词索引无效。请使用 /getpromat 查看可用的索引。", context=context
                    )
                return
            if command == "getpromat":
                await send_telegram_message(
                    chat_id=chat_id, text=f"当前使用的提示词索引是: {CURRENT_PROMPT_INDEX}", context=context
                )
                return
            if command == "resetuser":
                USER_SETTINGS.pop(from_user_id, None)
                await send_telegram_message(chat_id=chat_id, text="已重置您的个人设置。", context=context)
                return

    if is_group_chat:
        if chat_id not in GROUP_SETTINGS:
            GROUP_SETTINGS[chat_id] = {"aiEnabled": True, "prefix": None}
            logging.info(f"群组 {chat_id} 首次检测到，默认启用 AI。")

        group_settings = GROUP_SETTINGS[chat_id]
        prefix = group_settings["prefix"]
        if group_settings["aiEnabled"]:
            if prefix and not user_message.startswith(prefix):
                return

            message_content = user_message[len(prefix) :].strip() if prefix else user_message
            if len(message_content) > 0:
                await process_ai_message(chat_id, message_content, from_user_id, context)
    else:
        await process_ai_message(chat_id, user_message, from_user_id, context)

async def process_ai_message(chat_id, user_message, from_user_id, context):
    history = chat_histories.get(chat_id, [])
    user_temp = USER_SETTINGS.get(from_user_id, {}).get("temperature", DEFAULT_TEMP)
    current_prompt = PROMPT_TEMPLATES.get(CURRENT_PROMPT_INDEX, "")
    history.append({"role": "user", "content": user_message})

    if len(history) > MAX_HISTORY_LENGTH:
        history = history[-MAX_HISTORY_LENGTH:]

    messages = [
        {"role": "system", "content": current_prompt},
        *history,
    ]

    event_emitter = EventEmitter(
        lambda event: send_event_message(chat_id, event, context)
    )

    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(
                AI_API_ENDPOINT,
                headers=AI_API_HEADERS,
                json={
                    "model": AI_MODEL,
                    "messages": messages,
                    "max_tokens": MAX_TOKENS,
                    "temperature": user_temp,
                    "tools": TOOL_DEFINITIONS,
                },
            ) as ai_response:
                if not ai_response.ok:
                    error_text = await ai_response.text()
                    logging.error(f"AI API 响应失败: {ai_response.status}, {error_text}")
                    await send_telegram_message(chat_id=chat_id, text="AI API 响应失败，请稍后再试", context=context)
                    return

                ai_data = await ai_response.json()
                ai_reply = await handle_ai_response(ai_data, chat_id, history, event_emitter, context)

                history.append({"role": "assistant", "content": ai_reply})
                chat_histories[chat_id] = history

                await send_telegram_message(chat_id=chat_id, text=ai_reply, context=context)

    except Exception as error:
        logging.error(f"处理消息时发生错误: {error}")
        await send_telegram_message(chat_id=chat_id, text="处理消息时发生错误，请稍后再试", context=context)

async def handle_ai_response(ai_data, chat_id, history, event_emitter, context):
    if ai_data and ai_data.get("choices") and len(ai_data["choices"]) > 0:
        choice = ai_data["choices"][0]
        if choice.get("message") and choice["message"].get("content"):
            return choice["message"]["content"]
        elif choice.get("message") and choice["message"].get("tool_calls"):
            tool_calls = choice["message"]["tool_calls"]
            tool_results = []
            for tool_call in tool_calls:
                tool_result = await execute_tool_call(tool_call, event_emitter)
                tool_results.append(tool_result)

            new_messages = [
                *history,
                {"role": "assistant", "content": None, "tool_calls": tool_calls},
                *tool_results,
            ]

            user_temp = USER_SETTINGS.get(chat_id, {}).get("temperature", DEFAULT_TEMP)
            async with aiohttp.ClientSession() as session:
                async with session.post(
                    AI_API_ENDPOINT,
                    headers=AI_API_HEADERS,
                    json={
                        "model": AI_MODEL,
                        "messages": new_messages,
                        "max_tokens": MAX_TOKENS,
                        "temperature": user_temp,
                    },
                ) as ai_response:
                    if not ai_response.ok:
                        error_text = await ai_response.text()
                        logging.error(f"AI API 响应失败: {ai_response.status}, {error_text}")
                        return "AI API 响应失败，请稍后再试"

                    ai_data = await ai_response.json()
                    if (
                        ai_data
                        and ai_data.get("choices")
                        and len(ai_data["choices"]) > 0
                        and ai_data["choices"][0].get("message")
                        and ai_data["choices"][0]["message"].get("content")
                    ):
                        return ai_data["choices"][0]["message"]["content"]
                    return "AI 返回了无法识别的格式"

        return "AI 返回了无法识别的格式"
    return "AI 返回了无法识别的格式"

async def execute_tool_call(tool_call, event_emitter):
    name = tool_call["function"]["name"]
    args = tool_call["function"].get("arguments", {})
    if isinstance(args, str):
        args = json.loads(args)

    if name == "web_scrape":
        urls = args.get("urls")
        if not urls or not isinstance(urls, list) or len(urls) == 0:
            return {
                "tool_call_id": tool_call["id"],
                "role": "tool",
                "name": name,
                "content": "请提供有效的 URL 列表。",
            }
        api_url = "https://gpts.webpilot.ai/api/read"
        headers = {"Content-Type": "application/json", "WebPilot-Friend-UID": "0"}

        await event_emitter.update_status(
            f"开始读取 {len(urls)} 个网页", False, "web_search", urls
        )

        async def process_url(url):
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.post(
                        api_url,
                        headers=headers,
                        json={
                            "link": url,
                            "ur": "summary of the page",
                            "lp": True,
                            "rt": False,
                            "l": "en",
                        },
                    ) as response:
                        if not response.ok:
                            error_text = await response.text()
                            raise Exception(f"HTTP error! Status: {response.status}, Text: {error_text}")
                        result = await response.json()
                        if "rules" in result:
                            del result["rules"]
                        content = json.dumps(result)
                        title = result.get("title", url)
                        await event_emitter.send_citation(title, url, content)
                        return f"{content}\n"
            except Exception as error:
                error_message = f"读取网页 {url} 时出错: {str(error)}"
                await event_emitter.update_status(error_message, False, "web_scrape", [url])
                await event_emitter.send_citation(f"Error from {url}", url, str(error))
                return f"URL: {url}\n错误: {error_message}\n"

        results = await asyncio.gather(*[process_url(url) for url in urls])

        await event_emitter.update_status(
            f"已完成 {len(urls)} 个网页的读取", True, "web_search", urls
        )

        return {
            "tool_call_id": tool_call["id"],
            "role": "tool",
            "name": name,
            "content": "\n".join(results),
        }
    elif name == "get_current_time":
        now = datetime.utcnow()
        utc8_offset = 8 * 60 * 60
        utc8_time = now.timestamp() + utc8_offset
        utc8_time_obj = datetime.fromtimestamp(utc8_time)
        formatted_time = utc8_time_obj.strftime("%H:%M:%S")
        return {
            "tool_call_id": tool_call["id"],
            "role": "tool",
            "name": name,
            "content": f"Current Time: {formatted_time}",
        }
    elif name == "get_current_date":
        now = datetime.utcnow()
        utc8_offset = 8 * 60 * 60
        utc8_time = now.timestamp() + utc8_offset
        utc8_time_obj = datetime.fromtimestamp(utc8_time)
        formatted_date = utc8_time_obj.strftime("%A, %B %d, %Y")
        return {
            "tool_call_id": tool_call["id"],
            "role": "tool",
            "name": name,
            "content": f"Today's date is {formatted_date}",
        }
    else:
        return {
            "tool_call_id": tool_call["id"],
            "role": "tool",
            "name": name,
            "content": "未知的工具调用",
        }

async def handle_callback_query(callback_query, context):
    chat_id = callback_query.message.chat.id
    data = callback_query.data

    if data == "clearall":
        chat_histories.pop(chat_id, None)
        await send_telegram_message(chat_id=chat_id, text="聊天记录已清空。", context=context)

    await send_telegram_message(
        chat_id=chat_id,
        text="请选择操作：",
        reply_markup={
            "inline_keyboard": [[{"text": "清空聊天记录", "callback_data": "clearall"}]],
        },
        context=context
    )

async def send_event_message(chat_id, event, context):
    if event["type"] == "status":
        await send_telegram_message(chat_id=chat_id, text=f"状态更新: {event['data']['description']}", context=context)
    elif event["type"] == "citation":
        await send_telegram_message(chat_id=chat_id, text=f"引用信息: {event['data']['metadata'][0]['name']}", context=context)

async def send_telegram_message(chat_id, text, context, options=None):
    url = f"{PHP_PROXY_URL}/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    json_data = {"chat_id": chat_id, "text": text}
    if options:
        json_data.update(options)
    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=json_data) as response:
                if not response.ok:
                    error_text = await response.text()
                    logging.error(f"发送 Telegram 消息失败: {response.status}, {error_text}")
    except Exception as error:
        logging.error(f"发送 Telegram 消息时发生错误: {error}")

def get_help_message():
    return f"""
可用指令：
/start - 显示欢迎信息和操作按钮。
/clearall - 清空当前会话的聊天记录。
/help - 显示此帮助信息。

群组指令:
/enableai - 在群组中启用AI回复。
/disableai - 在群组中禁用AI回复。
/setprefix <prefix> - 设置群组中触发AI回复的前缀，例如：/setprefix @bot。
/getprefix - 获取当前群组的触发前缀。
    
私聊指令:
/settemp <温度值> - 设置AI回复的温度 (0-2)，例如：/settemp 1.0。
/gettemp - 获取当前AI回复的温度。
/resetuser - 重置你的个人设置。
/promat <index> - 切换提示词，例如: /promat 0, 1, 2。
/getpromat - 获取当前使用的提示词索引。
    
直接发送文本消息与AI对话 (私聊)。

群组中，需要使用前缀触发AI回复，如果设置了前缀的话。
    
注意：
  - 机器人会记住最近的 {MAX_HISTORY_LENGTH} 条对话。
  - 机器人具有攻击性，请谨慎使用。
"""

async def initialize_with_retry(application):
    max_retries = 3
    for attempt in range(max_retries):
        try:
            await application.initialize()
            return
        except NetworkError as e:
            logging.error(f"Error initializing application, attempt {attempt + 1}: {e}")
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2) # Wait for 2 seconds before retrying

async def main():
    logging.basicConfig(level=logging.INFO)

    if not await check_php_proxy_health():
        logging.error("PHP proxy health check failed. Exiting.")
        return

    # 修改这里
    request = HTTPXRequest(http_client=httpx.AsyncClient(trust_env=True))
    application = Application.builder().token(TELEGRAM_BOT_TOKEN).request(request).build()

    # Set bot commands
    try:
      await set_bot_commands(application.bot)
    except Exception as e:
      logging.error(f"Failed to set bot commands after multiple retries: {e}")
      return

    # Command handlers
    application.add_handler(CommandHandler("start", handle_telegram_update))
    application.add_handler(CommandHandler("clearall", handle_telegram_update))
    application.add_handler(CommandHandler("help", handle_telegram_update))
    application.add_handler(CommandHandler("enableai", handle_telegram_update))
    application.add_handler(CommandHandler("disableai", handle_telegram_update))
    application.add_handler(CommandHandler("setprefix", handle_telegram_update))
    application.add_handler(CommandHandler("getprefix", handle_telegram_update))
    application.add_handler(CommandHandler("settemp", handle_telegram_update))
    application.add_handler(CommandHandler("gettemp", handle_telegram_update))
    application.add_handler(CommandHandler("resetuser", handle_telegram_update))
    application.add_handler(CommandHandler("promat", handle_telegram_update))
    application.add_handler(CommandHandler("getpromat", handle_telegram_update))

    # Callback query handler
    application.add_handler(CallbackQueryHandler(handle_callback_query))

    # Message handler for all text messages
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_telegram_update))

    # Initialize the bot
    await application.initialize()
    # Start the bot
    await application.start()
    # Start polling updates from Telegram
    await application.updater.start_polling()

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