qqbot/main.py

from flask import Flask, request, abort, send_file
import pandas as pd
from openai import OpenAI
import requests
import time
import re
from wakeonlan import send_magic_packet
from ai_pic import story_start, story_start_p2p, story_start_p2p_sese, story_start2
from msg_send_save import (send_private_message_word,save_group_message,send_group_message_pic,send_group_message_word,
                           send_group_poke,del_group_message,send_group_vioce)
from chat_model import AI_chat,AI_get_picprompt,AI_lora_getpic_prompt,AI_sendphoto_ornot,get_ja_trans
from special_date import get_week,get_v50,get_holidays
from tts import get_free_tts,get_gpu_tts
from config import config_get
import random
import os

app = Flask(__name__)

# 读取王石列表，存为stones数据df文件
stones = pd.read_csv('other/stone.csv')
if 'Unnamed: 0' in stones.columns:
    stones = stones.drop('Unnamed: 0', axis=1)
for i in range(len(stones)):
    final_stat = "\n"
    abilities = stones.loc[i, 'stats'].split("&")
    for j in range(len(abilities)):
        ability = abilities[j].split("#")
        if ability[2] != "Nan":
            final_stat += ability[2]+":"
        if ability[0].endswith("%"):
            final_stat += ability[0].split("%")[0] + ability[1] +"% "
        else:
            final_stat += ability[0] + ability[1] + " "
    stones.loc[i,"stat"] = final_stat

# QQ机器人基础板块，包括人格，绘图风格和名称自我意识
my_name = config_get("my_name")
prompt_max = config_get("prompt_max")
prompt_less = config_get("prompt_less")

# 一个比较没用的变量，用来限制绘画cd的，现在默认是0就是无cd
last_request_time = 0
# 最近一次的心跳上报时间，用来记录服务状态
last_heartbeat_time = time.time()

def on_my_message(group_id, user_id, user_name, raw_message, message):
    """
    每次触发猫猫关键词，收到@使用这个方法来进行回应
    """
    global last_request_time
    # 先对解码后消息进行判断，如果第一个解码消息为@，则触发命令模式，遍历完成若没有命令触发则再触发人格对话
    if message[0]['type'] == "at" and message[1]['type'] == "text":
        command = message[1]['data']['text'].strip(" ")
        if str(command).startswith("/"):
            on_command(group_id, user_id, command)
        elif command=="v你50":
            print(send_group_message_pic(group_id, user_id, 'v50.png'))
        else:
            AI_ans = AI_chat(group_id, user_name+"："+raw_message, prompt_less)
            send_group_message_word(group_id, AI_ans)
            save_group_message(group_id, my_name, AI_ans)
            # 根据条件触发AI绘画
            if AI_sendphoto_ornot("猫猫:"+AI_ans) == "是":
                current_time = time.time()
                # 判断距离上次请求图片的时间间隔是否小于60秒（1分钟）
                if current_time-last_request_time >= 0:
                    last_request_time = current_time
                    send_group_message_pic(group_id, user_id, story_start(AI_get_picprompt(group_id)))
    else:
        AI_ans = AI_chat(group_id, user_name + "：" + raw_message, prompt_less)
        send_group_message_word(group_id, AI_ans)
        save_group_message(group_id, my_name, AI_ans)
        # 根据条件触发AI绘画
        if AI_sendphoto_ornot(AI_ans) == "是":
            current_time = time.time()
            # 判断距离上次请求图片的时间间隔是否小于60秒（1分钟）
            if current_time - last_request_time >= 0:
                last_request_time = current_time
                send_group_message_pic(group_id, user_id, story_start(AI_get_picprompt(group_id)))

def on_command(group_id, user_id,command):
    """
    每当接到命令符号时，开启这个功能进行回应
    """
    global last_request_time
    if command == "/菜单":
        home = """欢迎使用命令功能:
/菜单：显示功能介绍
/王石：查询王石效果
/猫猫画画：调用自定义AI绘画功能
/整点瑟瑟：不用我多说了吧
/语音合成+角色名称：合成AI语音（测试用，目前支持诺艾尔，芙宁娜，神里绫华，迪卢克
所有命令均需要@猫猫触发"""
        send_group_message_word(group_id, home)
        save_group_message(group_id, my_name, home)
    elif str(command).startswith("/王石"):
        stone_stat = ch_stone(command)
        send_group_message_word(group_id, stone_stat)
        save_group_message(group_id, my_name, stone_stat)
    elif str(command).startswith("/猫猫画画2"):
        pic_word = str(command)[len("/猫猫画画2"):]
        current_time = time.time()
        # 判断距离上次请求图片的时间间隔是否小于60秒（1分钟）
        if current_time - last_request_time >= 0:
            send_group_message_word(group_id, "图片正在生成中...")
            last_request_time = current_time
            send_group_message_pic(group_id, user_id, story_start2(AI_lora_getpic_prompt(pic_word)))
    elif str(command).startswith("/整点瑟瑟2"):
        pic_word = str(command)[len("/整点瑟瑟2"):]
        current_time = time.time()
        # 判断距离上次请求图片的时间间隔是否小于60秒（1分钟）
        if group_id == 637612718:
            send_group_message_word(group_id, "本群该功能已被禁用!")
        else:
            if current_time - last_request_time >= 0:
                send_group_message_word(group_id, "图片正在生成中...")
                last_request_time = current_time
                message_id = send_group_message_pic(group_id, user_id, story_start2("nsfw,best quality,masterpiece," + AI_lora_getpic_prompt(pic_word)))[
                    'data']['message_id']
                time.sleep(18)
                del_group_message(message_id)
    elif str(command).startswith("/猫猫画画"):
        pic_word = str(command)[len("/猫猫画画"):]
        current_time = time.time()
        # 判断距离上次请求图片的时间间隔是否小于60秒（1分钟）
        if current_time - last_request_time >= 0:
            send_group_message_word(group_id, "图片正在生成中...")
            last_request_time = current_time
            send_group_message_pic(group_id, user_id, story_start(AI_lora_getpic_prompt(pic_word)))
    elif str(command).startswith("/整点瑟瑟"):
        pic_word = str(command)[len("/整点瑟瑟"):]
        current_time = time.time()
        # 判断距离上次请求图片的时间间隔是否小于60秒（1分钟）
        if group_id == 637612718:
            send_group_message_word(group_id, "本群该功能已被禁用!")
        else:
            if current_time - last_request_time >= 0:
                send_group_message_word(group_id, "图片正在生成中...")
                last_request_time = current_time
                message_id = send_group_message_pic(group_id, user_id, story_start("nsfw,"+AI_lora_getpic_prompt(pic_word)))['data']['message_id']
                time.sleep(10)
                del_group_message(message_id)
    elif str(command).startswith("/语音合成"):
        send_group_message_word(group_id, "语音合成开始")
        name = "xxx"
        text = "xxx"
        if str(command).startswith("/语音合成诺艾尔"):
            name = "诺艾尔"
            text = str(command)[len("/语音合成诺艾尔"):]
        elif str(command).startswith("/语音合成芙宁娜"):
            name = "芙宁娜"
            text = str(command)[len("/语音合成芙宁娜"):]
        elif str(command).startswith("/语音合成神里绫华"):
            name = "神里绫华"
            text = str(command)[len("/语音合成神里绫华"):]
        elif str(command).startswith("/语音合成迪卢克"):
            name = "迪卢克"
            text = str(command)[len("/语音合成迪卢克"):]
        result = get_free_tts(name, text)
        if result.startswith("https"):
            send_group_vioce(group_id, result)
        else:
            send_group_message_word(group_id, result)

    else:
        error = "抱歉没有找到你所输入的命令！如有关于命令的问题请@猫猫输入/菜单进行查询\n示例：\n@猫猫 /菜单"
        send_group_message_word(group_id, error)
        save_group_message(group_id, my_name, error)

def ch_stone(name):
    """
    根据关键词搜索王石，返回具体介绍
    """
    word = name.lstrip("/王石")
    for i in range(len(stones)):
        if stones.loc[i, 'name'] == word:
            stone_up_on = stones.loc[i, 'up_on']
            if stone_up_on == "Nan":
                stone_up_on = ""
            else:
                stone_up_on = "\n可强化:" + stone_up_on
            return stones.loc[i, 'type'] +" " + stones.loc[i, 'name'] + stone_up_on + stones.loc[i, 'stat']
    return "这里似乎没有你想找的东西喵~\n如果要查找王石，请使用/王石查询正确的王石名称，示例：\n@猫猫 /王石科隆老大\n如果不清楚王石全名，可以在[掌上波多姆]中进行模糊搜索喵"

def change2setu(raw_message,group_id,user_id, model=1):
    """
    进入变瑟图命令
    """
    message = str(raw_message)
    # 提取 QQ 号
    pattern = r'\[CQ:at,qq=(\d+),name=[^]]+\]'
    match = re.search(pattern, message)
    if match:
        qq = match.group(1)
        try:
            img_url = f'https://q1.qlogo.cn/g?b=qq&nk={qq}&s=640'
            # 创建 user 文件夹
            on_file_path = os.path.join('user', qq)
            if not os.path.exists(on_file_path):
                os.makedirs(on_file_path)
            # 下载图片
            img_response = requests.get(img_url)
            img_response.raise_for_status()
            file_path = os.path.join('user', qq, f'{qq}.jpg')
            with open(file_path, 'wb') as file:
                file.write(img_response.content)
            print(f'图片已保存到 {file_path}')
            send_group_message_word(group_id, "开始变身！")
            message_id = send_group_message_pic(group_id, user_id, story_start_p2p_sese(qq, model))
            print(message_id)
            time.sleep(10)
            del_group_message(message_id['data']['message_id'])
        except Exception as e:
            print(f'发生未知错误: {e}')
            send_group_message_word(group_id, "变身失败！")
    else:
        print('未找到 QQ 号')

def change2girl(raw_message,group_id,user_id, model=1):
    """
    进入变美少女命令
    """
    message = str(raw_message)
    # 提取 QQ 号
    pattern = r'\[CQ:at,qq=(\d+),name=[^]]+\]'
    match = re.search(pattern, message)
    if match:
        qq = match.group(1)
        try:
            img_url = f'https://q1.qlogo.cn/g?b=qq&nk={qq}&s=640'
            # 创建 user 文件夹
            on_file_path = os.path.join('user', qq)
            if not os.path.exists(on_file_path):
                os.makedirs(on_file_path)
            # 下载图片
            img_response = requests.get(img_url)
            img_response.raise_for_status()
            file_path = os.path.join('user', qq, f'{qq}.jpg')
            with open(file_path, 'wb') as file:
                file.write(img_response.content)
            print(f'图片已保存到 {file_path}')
            send_group_message_word(group_id, "开始变身！")
            send_group_message_pic(group_id, user_id, story_start_p2p(qq,model))
        except Exception as e:
            print(f'发生未知错误: {e}')
            send_group_message_word(group_id, "变身失败！")
    else:
        print('未找到 QQ 号')


@app.route('/stat', methods=['GET'])
def stat_get():
    return {"heartbeat": last_heartbeat_time}

@app.route('/cat', methods=['POST'])
def chat():
    global last_heartbeat_time
    data = request.json
    data_type = data['post_type']
    # 管理员模式，目前只有开启主机功能
    if 'message_type' in data:
        if data['message_type'] == 'private':
            # 进入管理员模式
            if data['raw_message'] == "原神启动" and data['user_id'] == 3494084267:
                send_magic_packet("22-43-4D-05-34-B2")
                send_private_message_word(data['user_id'], "主机3B0IJ2U已启动！")
    print(data)
    # 进入群聊模式
    if data_type == "message" and data['message_type'] == "group":
        group_id = data['group_id']
        user_id = data['user_id']
        message = data['message']
        raw_message = data['raw_message']
        time_now = data['time']
        user_name = data['sender']['nickname']

        # 首先判断消息类型，如果为图片类型则存入url，否则存入csv文件中
        if message[0].get('type')=='image':
            image_data = message[0].get('data', {})
            url = image_data.get('url')
            save_group_message(group_id, user_name, f"图片内容：{url}")
        else:
            save_group_message(group_id, user_name, raw_message)

        # 然后根据命令优先级进入AI对话模式，在这里写入AI模块
        if str(raw_message).startswith("变美少女[CQ:at"):
            change2girl(raw_message, group_id, user_id)
        elif str(raw_message).startswith("变美少女2[CQ:at"):
            change2setu(raw_message, group_id, user_id, 2)
        elif str(raw_message).startswith("变瑟图[CQ:at"):
            if group_id == 637612718:
                send_group_message_word(group_id,"本群该功能已被禁用!")
            else:
                change2setu(raw_message, group_id, user_id)
        elif str(raw_message).startswith("变瑟图2[CQ:at"):
            if group_id == 637612718:
                send_group_message_word(group_id, "本群该功能已被禁用!")
            else:
                change2setu(raw_message, group_id, user_id, 2)
        elif str(raw_message) == "看不懂中文":
            print("开始日语语音合成")
            ja_text = get_ja_trans(group_id)
            bs64 = get_gpu_tts(ja_text)
            if bs64 == "获取失败":
                print(bs64)
                send_group_message_word(group_id, "转写失败")
            else:
                print("发送语音开始！")
                send_group_vioce(group_id, "base64://"+bs64)
        elif my_name in raw_message:
            on_my_message(group_id, user_id, user_name, raw_message, message)
            print(
                f"收到message!群号：{group_id}，发送人：{user_id}，时间{time_now}\n"
                f"消息内容：{message}\n原始消息内容：{raw_message}")
        else:
            print(f"群聊{group_id}已记录新消息！")

    # 进入提示消息模式
    elif data_type == "notice":
        notice_type = data['notice_type']
        group_id = data['group_id']
        if notice_type == "group_increase":
            user_id = data['user_id']
            print(f"群成员增加！群号：{group_id},加入者：{user_id}")
            requests.post('http://127.0.0.1:3000/send_group_msg', json={
                'group_id': group_id,
                'message': [{
                    "type": "at",
                    "data": {
                        "qq": f"{user_id}"
                    }
                }, {
                    'type': 'text',
                    'data': {
                        'text': "欢迎新成员加入群聊喵~"
                    }
                }]
            })
        elif notice_type == "notify" and data['target_id'] == data['self_id']:
            user_id = data['user_id']
            print(f"我被{user_id}戳了一下！")
            if get_week()=="星期四":
                # 触发疯狂星期四文案
                # 生成一个0到99之间的随机整数,控制特殊文案概率为40%
                rand_num = random.randint(0, 99)
                print(rand_num)
                send_group_message_word(group_id,get_v50()) if rand_num < 30 else send_group_message_word(group_id,"₍^ >ヮ<^₎")
            else:
                send_group_message_word(group_id,"₍^ >ヮ<^₎")
            send_group_poke(group_id, user_id)
    elif data_type == "meta_event":
        if data["meta_event_type"] == "heartbeat" and data["status"]['online'] == True:
            last_heartbeat_time = data['time']
    return ""


# 允许的图片文件扩展名
ALLOWED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}


@app.route('/pic', methods=['GET'])
def pic_send():
    # 获取URL参数中的path
    image_path = request.args.get('path')

    # 检查path参数是否存在
    if not image_path:
        abort(400, description="缺少path参数")

    # 检查文件是否存在
    if not os.path.exists(image_path):
        abort(404, description="图片文件不存在")

    # 检查文件是否为图片类型
    file_ext = os.path.splitext(image_path)[1].lower()
    if file_ext not in ALLOWED_EXTENSIONS:
        abort(400, description="不支持的文件类型，仅允许图片文件")

    # 检查是否为文件（不是目录）
    if not os.path.isfile(image_path):
        abort(400, description="指定路径不是文件")

    try:
        # 返回图片文件
        return send_file(image_path)
    except Exception as e:
        abort(500, description=f"服务器错误：{str(e)}")

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=3100)