From d7c978614b22b255a72a3bf3ba4fa75865dc7818 Mon Sep 17 00:00:00 2001 From: Alone <675061370@qq.com> Date: Mon, 23 Sep 2024 08:55:36 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=E5=85=A8=E7=BD=91=E6=90=9C?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/controller/Other.php | 236 +++++++++++++++ app/common.php | 191 +++++++++++- chat_channel.py | 549 +++++++++++++++++++++++++++++++++++ 3 files changed, 975 insertions(+), 1 deletion(-) create mode 100644 app/api/controller/Other.php create mode 100644 chat_channel.py diff --git a/app/api/controller/Other.php b/app/api/controller/Other.php new file mode 100644 index 0000000..cadbb6e --- /dev/null +++ b/app/api/controller/Other.php @@ -0,0 +1,236 @@ +model = new SourceModel(); + $this->cookie = Config('qfshop.quark_cookie'); + } + + /** + * 全网搜索 该接口仅用于微信自动回复 + * + * @return void + */ + public function all_search() + { + $searchdata = input('post.'); + if (empty($searchdata['title'])) { + return jerr("请输入要看的内容"); + } + $title = $searchdata['title']; + + + $map[] = ['status', '=', 1]; + $map[] = ['is_delete', '=', 0]; + $map[] = ['is_time', '=', 1]; + $map[] = ['title|description', 'like', '%' . trim($title) . '%']; + + $urls = $this->model->where($map)->field('source_id as id, title, url')->order('update_time', 'desc')->limit(5)->select()->toArray(); + if (!empty($urls)) { + + // 获取所有需要更新的ID + $ids = []; + foreach ($urls as $item) { + $ids[] = $item['id']; + } + + // 更新数据库中的 update_time 字段 + if (!empty($ids)) { + $this->model->whereIn('source_id', $ids)->update(['update_time' => time()]); + } + + return jok('临时资源获取成功',$urls); + } + + $searchList = []; //查询的结果集 + $datas = []; //最终数据 + $num_total = 2; //最多想要几条结果 + $num_success = 0; + + // 处理第2个源 + foreach (source2($title) as $value) { + if ($num_success >= $num_total) { + break; // 有效结果数量已达到，则跳出循环 + } + // 如果 URL 不存在则新增 $value + if (!$this->urlExists($searchList, $value['url'])) { + $searchList[] = $value; + $this->processUrl($value, $num_success, $datas); + } + } + + + + // 处理第4个源 + if ($num_success < $num_total) { + foreach (source4($title) as $value) { + if ($num_success >= $num_total) { + break; // 有效结果数量已达到，则跳出循环 + } + // 如果 URL 不存在则新增 $value + if (!$this->urlExists($searchList, $value['url'])) { + $searchList[] = $value; + $this->processUrl($value, $num_success, $datas); + } + } + } + + + // 处理第3个源 + if ($num_success < $num_total) { + foreach (source3($title) as $value) { + if ($num_success >= $num_total) { + break; // 有效结果数量已达到，则跳出循环 + } + // 如果 URL 不存在则新增 $value + if (!$this->urlExists($searchList, $value['url'])) { + $searchList[] = $value; + $this->processUrl($value, $num_success, $datas); + } + } + } + + + // 处理第1个源 第一个源放最后 + if ($num_success < $num_total) { + foreach (source1($title) as $value) { + if ($num_success >= $num_total) { + break; // 有效结果数量已达到，则跳出循环 + } + // 如果 URL 不存在则新增 $value + if (!$this->urlExists($searchList, $value['url'])) { + $searchList[] = $value; + $this->processUrl($value, $num_success, $datas); + } + } + } + + return jok('临时资源获取成功',$datas); + } + + // 检查 URL 是否已存在（忽略查询参数） + public function urlExists($searchList, $urlToCheck) { + // 解析待检查的 URL + $parsedUrlToCheck = parse_url($urlToCheck); + + foreach ($searchList as $item) { + $parsedUrl = parse_url($item['url']); + + // 比较 scheme, host 和 path + if ($parsedUrlToCheck['scheme'] === $parsedUrl['scheme'] && + $parsedUrlToCheck['host'] === $parsedUrl['host'] && + $parsedUrlToCheck['path'] === $parsedUrl['path']) { + return true; + } + } + + return false; + } + + /** + * 临时资源转存 + * + * @return void + */ + public function processUrl($value, &$num_success, &$datas) + { + $substring = strstr($value['url'], 's/'); + if ($substring === false) { + return; // 模拟 continue 行为 + } + + $pwd_id = substr($substring, 2); // 去除 's/' 部分 + + $urlData = array( + 'cookie' => $this->cookie, + 'url' => $value['url'], + 'expired_type' => 2, + 'to_pdir_fid' => '', //存入目标文件 + 'ad_fid' => '', //分享时带上这个文件 + ); + $res = curlHelper(Request::domain()."/api/open/transfer", "POST", $urlData)['body']; + $res = json_decode($res, true); + + if($res['code'] !== 200){ + return; // 模拟 continue 行为 + } + + $patterns = '/^\d+\./'; + $title = preg_replace($patterns, '', $value['title']); + // 添加资源到系统中 + $data["title"] =$title; + $data["url"] =$res['data']['share_url']; + $data["is_type"] = determineIsType($data["url"]); + $data["fid"] =$res['data']['fid']??''; + $data["is_time"] = 1; + $data["update_time"] = time(); + $data["create_time"] = time(); + $this->model->insertGetId($data); + $datas[] =$data; + $num_success++; + } + + + /** + * 30分钟后清除临时资源 + * + * @return void + */ + public function delete_search() + { + // 搜索条件 + $map[] = ['is_time', '=', 1]; + $map[] = ['update_time', '<=', time() - (30 * 60)]; + $abc = $this->model->where($map)->select(); + + + $this->model->where($map)->chunk(100, function ($order) { + foreach ($order as $value) { + $deles = $value->toArray(); + $filelist = []; + $filelist[] = $deles['fid']; + + + $urlData = array( + 'action_type' => 2, + 'exclude_fids' => [], + 'filelist' => $filelist, + ); + $urlHeader = array( + 'Accept: application/json, text/plain, */*', + 'Accept-Language: zh-CN,zh;q=0.9', + 'content-type: application/json;charset=UTF-8', + 'sec-ch-ua: "Chromium";v="122", "Not(A:Brand";v="24", "Google Chrome";v="122"', + 'sec-ch-ua-mobile: ?0', + 'sec-ch-ua-platform: "Windows"', + 'sec-fetch-dest: empty', + 'sec-fetch-mode: cors', + 'sec-fetch-site: same-site', + 'Referer: https://pan.quark.cn/', + 'Referrer-Policy: strict-origin-when-cross-origin', + 'user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', + 'cookie: '.$this->cookie, + ); + $res = curlHelper("https://drive-pc.quark.cn/1/clouddrive/file/delete?pr=ucpro&fr=pc&uc_param_str=", "POST", json_encode($urlData), $urlHeader)['body']; + $res = json_decode($res, true); + if($res['status'] == 200){ + $this->model->where('fid', $deles['fid'])->delete(); + } + } + }); + + return jok('临时资源删除成功',$abc); + } + +} diff --git a/app/common.php b/app/common.php index f722b12..e92d045 100644 --- a/app/common.php +++ b/app/common.php @@ -787,4 +787,193 @@ function determineIsType($url) { // 默认值是夸克网盘，返回 0 return 0; } -} \ No newline at end of file +} + + +/** + * 网络资源搜索源一 + * @return array + */ + + function source1($title) + { + $d = []; + + return $d; + } + + /** + * 网络资源搜索源二(5条线路) + * 每个线路只取第一个 + * @return array + */ + function source2($title) + { + $urlDefault = "http://s.kkkob.com"; //http://s.kkkob.com + $url2 = []; + $res = curlHelper($urlDefault."/v/api/getToken", "GET")['body']; + $res = json_decode($res, true); + $token = $res['token'] ?? ''; + if(empty($token)){ + return $url2; + } + + $urlData = array( + 'name' => $title, + 'token' => $token + ); + $urlHeader = array('Content-Type: application/json'); + // 定义正则表达式模式 + $pattern = '/https:\/\/pan\.quark\.cn\/[^\s]*/'; + + //线路2 + $res = curlHelper($urlDefault."/v/api/getJuzi", "POST", json_encode($urlData), $urlHeader)['body']; + $res = json_decode($res, true); + if (!empty($res['list'] ?? [])) { + foreach ($res['list'] as $key => $value) { + if(preg_match($pattern, $value['answer'], $matches)){ + // 匹配成功，$matches[0] 包含了匹配到的链接 + $link = $matches[0]; + $url2[] = [ + 'title' => preg_replace('/\s*[\(（]?(夸克)?[\)）]?\s*/u', '', $value['question']), + 'url' => $link + ]; + break; + } + } + } + + if(!empty($url2)){ + return $url2; + } + + //线路4 + $res = curlHelper($urlDefault."/v/api/getXiaoyu", "POST", json_encode($urlData), $urlHeader)['body']; + $res = json_decode($res, true); + if (!empty($res['list'] ?? [])) { + foreach ($res['list'] as $key => $value) { + if(preg_match($pattern, $value['answer'], $matches)){ + // 匹配成功，$matches[0] 包含了匹配到的链接 + $link = $matches[0]; + $url2[] = [ + 'title' => preg_replace('/\s*[\(（]?(夸克)?[\)）]?\s*/u', '', $value['question']), + 'url' => $link + ]; + break; + } + } + } + + + // //线路1 + // $res = curlHelper($urlDefault."/v/api/search", "POST", json_encode($urlData), $urlHeader)['body']; + // $res = json_decode($res, true); + // if (!empty($res['list'] ?? [])) { + // $item = $res['list'][0]; + // // 使用正则表达式进行匹配 + // if (preg_match($pattern, $item['answer'], $matches)) { + // // 匹配成功，$matches[0] 包含了匹配到的链接 + // $link = $matches[0]; + // $url2[] = [ + // 'title' => '①'.preg_replace('/\s*[\(（]?(夸克)?[\)）]?\s*/u', '', $item['question']), + // 'url' => $link + // ]; + // } + // } + + return $url2; + } + + + /** + * 网络资源搜索源三 + * @return array + */ + function source3($title) + { + $url3 = []; + $url = 'https://www.qileso.com/tag/quark?s='.$title; + $dom = getDom($url); + $finder = new DomXPath($dom); + // 查询class值为list-group post-list mt-3的元素 + $nodes =$finder->query("//*[contains(concat(' ', normalize-space(@class), ' '), ' list-group post-list mt-3 ')]//a"); + + if ($nodes->length > 0) { + $firstNode =$nodes->item(0); + $href = $firstNode->getAttribute('href'); + + $dom = getDom($href); + $finder = new DomXPath($dom); + + // 查询包含特定前缀的href属性的所有标签 + $nodes =$finder->query("//@*[starts-with(., 'https://pan.quark.cn/s/')]"); + if ($nodes->length > 0) { + $firstNode =$nodes->item(0); + $value = $firstNode->value; + + // 查询元素 + $nodes =$finder->query("/html/head/title"); + + if ($nodes->length > 0) { + $titleNode =$nodes->item(0); + $titleText =$titleNode->textContent; + // 去掉 " - 奇乐搜" 部分 + $title = preg_replace('/ - 奇乐搜|网盘|夸克/', '', $titleText); + } + + $url3[] = [ + 'title' => '②'.$title, + 'url' => $value + ]; + } + } + + return $url3; + } + + + /** + * 网络资源搜索源四 + * @return array + */ + function source4($title) + { + $url = 'https://www.pansearch.me/search?keyword='.urlencode($title).'&pan=quark'; + $dom = getDom($url); + $finder = new DomXPath($dom); + + // 使用 XPath 查询选择具有特定类名的元素 + $nodes =$finder->query('//div[contains(concat(" ", normalize-space(@class), " "), " whitespace-pre-wrap ") and contains(concat(" ", normalize-space(@class), " "), " break-all ")]'); + + $results = []; + foreach ($nodes as $node) { + // 获取元素的内容，包括其子元素 + $content = $node->textContent; + + // Initialize an associative array to store parsed data + $parsedItem = [ + 'title' => '', + 'url' => '' + ]; + + // Use regular expressions to extract the title and url + if (preg_match('/名称：(.*?)\n\n描述：/s', $content, $titleMatch)) { + $parsedItem['title'] = '「推荐」'.trim($titleMatch[1]); + } + + if (preg_match('/链接：(https:\/\/pan\.quark\.cn\/s\/[a-zA-Z0-9]+)/', $content, $urlMatch)) { + $parsedItem['url'] = trim($urlMatch[1]); + } + + if ($parsedItem['title'] && $parsedItem['url']) { + $results[] = $parsedItem; + } + + if (count($results) >= 3) { + break; + } + } + + return $results; + } + \ No newline at end of file diff --git a/chat_channel.py b/chat_channel.py new file mode 100644 index 0000000..cbf9b79 --- /dev/null +++ b/chat_channel.py @@ -0,0 +1,549 @@ +import os +import re +import threading +import time +import requests +from asyncio import CancelledError +from concurrent.futures import Future, ThreadPoolExecutor + +from bridge.context import * +from bridge.reply import * +from channel.channel import Channel +from common.dequeue import Dequeue +from common import memory +from plugins import * + +try: + from voice.audio_convert import any_to_wav +except Exception as e: + pass + +handler_pool = ThreadPoolExecutor(max_workers=8) # 处理消息的线程池 + + +# 抽象类, 它包含了与消息通道无关的通用处理逻辑 +class ChatChannel(Channel): + name = None # 登录的用户名 + user_id = None # 登录的用户id + futures = {} # 记录每个session_id提交到线程池的future对象, 用于重置会话时把没执行的future取消掉，正在执行的不会被取消 + sessions = {} # 用于控制并发，每个session_id同时只能有一个context在处理 + lock = threading.Lock() # 用于控制对sessions的访问 + + def __init__(self): + _thread = threading.Thread(target=self.consume) + _thread.setDaemon(True) + _thread.start() + + # 根据消息构造context，消息内容相关的触发项写在这里 + def _compose_context(self, ctype: ContextType, content, **kwargs): + context = Context(ctype, content) + context.kwargs = kwargs + # context首次传入时，origin_ctype是None, + # 引入的起因是：当输入语音时，会嵌套生成两个context，第一步语音转文本，第二步通过文本生成文字回复。 + # origin_ctype用于第二步文本回复时，判断是否需要匹配前缀，如果是私聊的语音，就不需要匹配前缀 + if "origin_ctype" not in context: + context["origin_ctype"] = ctype + # context首次传入时，receiver是None，根据类型设置receiver + first_in = "receiver" not in context + # 群名匹配过程，设置session_id和receiver + if first_in: # context首次传入时，receiver是None，根据类型设置receiver + config = conf() + cmsg = context["msg"] + user_data = conf().get_user_data(cmsg.from_user_id) + context["openai_api_key"] = user_data.get("openai_api_key") + context["gpt_model"] = user_data.get("gpt_model") + if context.get("isgroup", False): + group_name = cmsg.other_user_nickname + group_id = cmsg.other_user_id + + group_name_white_list = config.get("group_name_white_list", []) + group_name_keyword_white_list = config.get("group_name_keyword_white_list", []) + if any( + [ + group_name in group_name_white_list, + "ALL_GROUP" in group_name_white_list, + check_contain(group_name, group_name_keyword_white_list), + ] + ): + group_chat_in_one_session = conf().get("group_chat_in_one_session", []) + session_id = cmsg.actual_user_id + if any( + [ + group_name in group_chat_in_one_session, + "ALL_GROUP" in group_chat_in_one_session, + ] + ): + session_id = group_id + else: + logger.debug(f"No need reply, groupName not in whitelist, group_name={group_name}") + return None + context["session_id"] = session_id + context["receiver"] = group_id + else: + context["session_id"] = cmsg.other_user_id + context["receiver"] = cmsg.other_user_id + e_context = PluginManager().emit_event(EventContext(Event.ON_RECEIVE_MESSAGE, {"channel": self, "context": context})) + context = e_context["context"] + if e_context.is_pass() or context is None: + return context + if cmsg.from_user_id == self.user_id and not config.get("trigger_by_self", True): + logger.debug("[chat_channel]self message skipped") + return None + + # 消息内容匹配过程，并处理content + if ctype == ContextType.TEXT: + if first_in and "」\n- - - - - - -" in content: # 初次匹配 过滤引用消息 + logger.debug(content) + logger.debug("[chat_channel]reference query skipped") + return None + + pattern = f"@{re.escape(self.name)}(\u2005|\u0020)" + content_search = re.sub(pattern, r"", content) + if isinstance(context["msg"].at_list, list): + for at in context["msg"].at_list: + pattern = f"@{re.escape(at)}(\u2005|\u0020)" + content_search = re.sub(pattern, r"", content_search) + if content_search == content and context["msg"].self_display_name: + # 前缀移除后没有变化，使用群昵称再次移除 + pattern = f"@{re.escape(context['msg'].self_display_name)}(\u2005|\u0020)" + content_search = re.sub(pattern, r"", content) + + + # 去除字符串开头和结尾的所有空格字符 + content_search = content_search.strip() + # logger.info("[来消息了] content={}, content_search={}".format(content, content_search)) + content_search = process_string(content_search) + + + nick_name_black_list = conf().get("nick_name_black_list", []) + if context.get("isgroup", False): # 群聊 + + if any(content_search.startswith(prefix) for prefix in ["搜剧", "搜", "全网搜"]) and not content_search.startswith("搜索"): + content_search = process_string2(content_search) + user_nickname = context['msg'].actual_user_nickname + reply_text = f"@{user_nickname}" + + + contentSearch = remove_prefix(content_search, ["搜剧", "搜", "全网搜"]).strip() + + def perform_search(): + # 初次搜索 + response_data = search_question(contentSearch) if not content_search.startswith("全网搜") else [] + if not response_data: + # 通知用户深入搜索 + reply_text2 = f"@{user_nickname}\n正在深入搜索，请稍等..." + self._send_reply(context, Reply(ReplyType.TEXT, reply_text2)) + + # 启动线程进行第二次搜索 + def perform_second_search(): + response_data = search_alone(contentSearch) + send_final_reply(response_data, reply_text, context) + + second_search_thread = threading.Thread(target=perform_second_search) + second_search_thread.start() + else: + # 如果第一次搜索找到结果，发送最终回复 + send_final_reply(response_data, reply_text, context) + + def send_final_reply(response_data, reply_text, context): + is_times = 0 + if not response_data: + reply_text_final = f"{reply_text}\n未找到，可换个关键词尝试哦~" + reply_text_final += "\n⚠️宁少写，不多写、错写~" + reply_text_final += "\n--------------------" + reply_text_final += "\n可访问以下链接提交资源需求" + reply_text_final += "\nhttps://pan.xinyuedh.com" + # reply_text_final += "\n--------------------" + # reply_text_final += "\nGPT小助手分享" + # reply_text_final += "\n--------------------" + # reply_text_final += "\nhttps://chat.xinyuedh.com" + else: + reply_text_final = f"{reply_text}\n--------------------" + for item in response_data: + if item.get('is_time') == 1: + reply_text_final += f"\n 🌐️ {item.get('title', '未知标题')}" + is_times += 1 + else: + reply_text_final += f"\n{item.get('title', '未知标题')}" + reply_text_final += f"\n{item.get('url', '未知URL')}" + reply_text_final += "\n--------------------" + + if is_times > 0: + reply_text_final += "\n 🌐️资源来源网络，30分钟后删除" + reply_text_final += "\n--------------------" + else: + reply_text_final += "\n 不是短剧？请尝试：全网搜XX" + reply_text_final += "\n--------------------" + + reply_text_final += "\n欢迎观看！如果喜欢可以喊你的朋友一起来哦" + + reply = Reply(ReplyType.TEXT, reply_text_final) + self._send_reply(context, reply) + + + # 启动线程执行第一次搜索 + first_search_thread = threading.Thread(target=perform_search) + first_search_thread.start() + return None + + # 校验关键字 + match_prefix = check_prefix(content, conf().get("group_chat_prefix")) + match_contain = check_contain(content, conf().get("group_chat_keyword")) + flag = False + if context["msg"].to_user_id != context["msg"].actual_user_id: + if match_prefix is not None or match_contain is not None: + flag = True + if match_prefix: + content = content.replace(match_prefix, "", 1).strip() + if context["msg"].is_at: + nick_name = context["msg"].actual_user_nickname + if nick_name and nick_name in nick_name_black_list: + # 黑名单过滤 + logger.warning(f"[chat_channel] Nickname {nick_name} in In BlackList, ignore") + return None + + logger.info("[chat_channel]receive group at") + if not conf().get("group_at_off", False): + flag = True + pattern = f"@{re.escape(self.name)}(\u2005|\u0020)" + subtract_res = re.sub(pattern, r"", content) + if subtract_res.startswith("画"): + subtract_res = "生成图片要求如下：\n" + subtract_res[1:] + + if isinstance(context["msg"].at_list, list): + for at in context["msg"].at_list: + pattern = f"@{re.escape(at)}(\u2005|\u0020)" + subtract_res = re.sub(pattern, r"", subtract_res) + if subtract_res == content and context["msg"].self_display_name: + # 前缀移除后没有变化，使用群昵称再次移除 + pattern = f"@{re.escape(context['msg'].self_display_name)}(\u2005|\u0020)" + subtract_res = re.sub(pattern, r"", content) + content = subtract_res + if not flag: + if context["origin_ctype"] == ContextType.VOICE: + logger.info("[chat_channel]receive group voice, but checkprefix didn't match") + return None + else: # 单聊 + nick_name = context["msg"].from_user_nickname + if nick_name and nick_name in nick_name_black_list: + # 黑名单过滤 + logger.warning(f"[chat_channel] Nickname '{nick_name}' in In BlackList, ignore") + return None + + match_prefix = check_prefix(content, conf().get("single_chat_prefix", [""])) + if match_prefix is not None: # 判断如果匹配到自定义前缀，则返回过滤掉前缀+空格后的内容 + content = content.replace(match_prefix, "", 1).strip() + elif context["origin_ctype"] == ContextType.VOICE: # 如果源消息是私聊的语音消息，允许不匹配前缀，放宽条件 + pass + else: + return None + content = content.strip() + img_match_prefix = check_prefix(content, conf().get("image_create_prefix",[""])) + if img_match_prefix: + content = content.replace(img_match_prefix, "", 1) + context.type = ContextType.IMAGE_CREATE + else: + context.type = ContextType.TEXT + context.content = content.strip() + if "desire_rtype" not in context and conf().get("always_reply_voice") and ReplyType.VOICE not in self.NOT_SUPPORT_REPLYTYPE: + context["desire_rtype"] = ReplyType.VOICE + elif context.type == ContextType.VOICE: + if "desire_rtype" not in context and conf().get("voice_reply_voice") and ReplyType.VOICE not in self.NOT_SUPPORT_REPLYTYPE: + context["desire_rtype"] = ReplyType.VOICE + return context + + def _handle(self, context: Context): + if context is None or not context.content: + return + logger.debug("[chat_channel] ready to handle context: {}".format(context)) + # reply的构建步骤 + reply = self._generate_reply(context) + + logger.debug("[chat_channel] ready to decorate reply: {}".format(reply)) + + # reply的包装步骤 + if reply and reply.content: + reply = self._decorate_reply(context, reply) + + # reply的发送步骤 + self._send_reply(context, reply) + + def _generate_reply(self, context: Context, reply: Reply = Reply()) -> Reply: + e_context = PluginManager().emit_event( + EventContext( + Event.ON_HANDLE_CONTEXT, + {"channel": self, "context": context, "reply": reply}, + ) + ) + reply = e_context["reply"] + if not e_context.is_pass(): + logger.debug("[chat_channel] ready to handle context: type={}, content={}".format(context.type, context.content)) + if context.type == ContextType.TEXT or context.type == ContextType.IMAGE_CREATE: # 文字和图片消息 + context["channel"] = e_context["channel"] + reply = super().build_reply_content(context.content, context) + elif context.type == ContextType.VOICE: # 语音消息 + cmsg = context["msg"] + cmsg.prepare() + file_path = context.content + wav_path = os.path.splitext(file_path)[0] + ".wav" + try: + any_to_wav(file_path, wav_path) + except Exception as e: # 转换失败，直接使用mp3，对于某些api，mp3也可以识别 + logger.warning("[chat_channel]any to wav error, use raw path. " + str(e)) + wav_path = file_path + # 语音识别 + reply = super().build_voice_to_text(wav_path) + # 删除临时文件 + try: + os.remove(file_path) + if wav_path != file_path: + os.remove(wav_path) + except Exception as e: + pass + # logger.warning("[chat_channel]delete temp file error: " + str(e)) + + if reply.type == ReplyType.TEXT: + new_context = self._compose_context(ContextType.TEXT, reply.content, **context.kwargs) + if new_context: + reply = self._generate_reply(new_context) + else: + return + elif context.type == ContextType.IMAGE: # 图片消息，当前仅做下载保存到本地的逻辑 + memory.USER_IMAGE_CACHE[context["session_id"]] = { + "path": context.content, + "msg": context.get("msg") + } + elif context.type == ContextType.SHARING: # 分享信息，当前无默认逻辑 + pass + elif context.type == ContextType.FUNCTION or context.type == ContextType.FILE: # 文件消息及函数调用等，当前无默认逻辑 + pass + else: + logger.warning("[chat_channel] unknown context type: {}".format(context.type)) + return + return reply + + def _decorate_reply(self, context: Context, reply: Reply) -> Reply: + if reply and reply.type: + e_context = PluginManager().emit_event( + EventContext( + Event.ON_DECORATE_REPLY, + {"channel": self, "context": context, "reply": reply}, + ) + ) + reply = e_context["reply"] + desire_rtype = context.get("desire_rtype") + if not e_context.is_pass() and reply and reply.type: + if reply.type in self.NOT_SUPPORT_REPLYTYPE: + logger.error("[chat_channel]reply type not support: " + str(reply.type)) + reply.type = ReplyType.ERROR + reply.content = "不支持发送的消息类型: " + str(reply.type) + + if reply.type == ReplyType.TEXT: + reply_text = reply.content + if desire_rtype == ReplyType.VOICE and ReplyType.VOICE not in self.NOT_SUPPORT_REPLYTYPE: + reply = super().build_text_to_voice(reply.content) + return self._decorate_reply(context, reply) + if context.get("isgroup", False): + if not context.get("no_need_at", False): + reply_text = "@" + context["msg"].actual_user_nickname + "\n" + reply_text.strip() + reply_text = conf().get("group_chat_reply_prefix", "") + reply_text + conf().get("group_chat_reply_suffix", "") + else: + reply_text = conf().get("single_chat_reply_prefix", "") + reply_text + conf().get("single_chat_reply_suffix", "") + reply.content = reply_text + elif reply.type == ReplyType.ERROR or reply.type == ReplyType.INFO: + reply.content = "[" + str(reply.type) + "]\n" + reply.content + elif reply.type == ReplyType.IMAGE_URL or reply.type == ReplyType.VOICE or reply.type == ReplyType.IMAGE or reply.type == ReplyType.FILE or reply.type == ReplyType.VIDEO or reply.type == ReplyType.VIDEO_URL: + pass + else: + logger.error("[chat_channel] unknown reply type: {}".format(reply.type)) + return + if desire_rtype and desire_rtype != reply.type and reply.type not in [ReplyType.ERROR, ReplyType.INFO]: + logger.warning("[chat_channel] desire_rtype: {}, but reply type: {}".format(context.get("desire_rtype"), reply.type)) + return reply + + def _send_reply(self, context: Context, reply: Reply): + if reply and reply.type: + e_context = PluginManager().emit_event( + EventContext( + Event.ON_SEND_REPLY, + {"channel": self, "context": context, "reply": reply}, + ) + ) + reply = e_context["reply"] + if not e_context.is_pass() and reply and reply.type: + logger.debug("[chat_channel] ready to send reply: {}, context: {}".format(reply, context)) + self._send(reply, context) + + def _send(self, reply: Reply, context: Context, retry_cnt=0): + try: + self.send(reply, context) + except Exception as e: + logger.error("[chat_channel] sendMsg error: {}".format(str(e))) + if isinstance(e, NotImplementedError): + return + logger.exception(e) + if retry_cnt < 2: + time.sleep(3 + 3 * retry_cnt) + self._send(reply, context, retry_cnt + 1) + + def _success_callback(self, session_id, **kwargs): # 线程正常结束时的回调函数 + logger.debug("Worker return success, session_id = {}".format(session_id)) + + def _fail_callback(self, session_id, exception, **kwargs): # 线程异常结束时的回调函数 + logger.exception("Worker return exception: {}".format(exception)) + + def _thread_pool_callback(self, session_id, **kwargs): + def func(worker: Future): + try: + worker_exception = worker.exception() + if worker_exception: + self._fail_callback(session_id, exception=worker_exception, **kwargs) + else: + self._success_callback(session_id, **kwargs) + except CancelledError as e: + logger.info("Worker cancelled, session_id = {}".format(session_id)) + except Exception as e: + logger.exception("Worker raise exception: {}".format(e)) + with self.lock: + self.sessions[session_id][1].release() + + return func + + def produce(self, context: Context): + session_id = context["session_id"] + with self.lock: + if session_id not in self.sessions: + self.sessions[session_id] = [ + Dequeue(), + threading.BoundedSemaphore(conf().get("concurrency_in_session", 4)), + ] + if context.type == ContextType.TEXT and context.content.startswith("#"): + self.sessions[session_id][0].putleft(context) # 优先处理管理命令 + else: + self.sessions[session_id][0].put(context) + + # 消费者函数，单独线程，用于从消息队列中取出消息并处理 + def consume(self): + while True: + with self.lock: + session_ids = list(self.sessions.keys()) + for session_id in session_ids: + context_queue, semaphore = self.sessions[session_id] + if semaphore.acquire(blocking=False): # 等线程处理完毕才能删除 + if not context_queue.empty(): + context = context_queue.get() + logger.debug("[chat_channel] consume context: {}".format(context)) + future: Future = handler_pool.submit(self._handle, context) + future.add_done_callback(self._thread_pool_callback(session_id, context=context)) + if session_id not in self.futures: + self.futures[session_id] = [] + self.futures[session_id].append(future) + elif semaphore._initial_value == semaphore._value + 1: # 除了当前，没有任务再申请到信号量，说明所有任务都处理完毕 + self.futures[session_id] = [t for t in self.futures[session_id] if not t.done()] + assert len(self.futures[session_id]) == 0, "thread pool error" + del self.sessions[session_id] + else: + semaphore.release() + time.sleep(0.1) + + # 取消session_id对应的所有任务，只能取消排队的消息和已提交线程池但未执行的任务 + def cancel_session(self, session_id): + with self.lock: + if session_id in self.sessions: + for future in self.futures[session_id]: + future.cancel() + cnt = self.sessions[session_id][0].qsize() + if cnt > 0: + logger.info("Cancel {} messages in session {}".format(cnt, session_id)) + self.sessions[session_id][0] = Dequeue() + + def cancel_all_session(self): + with self.lock: + for session_id in self.sessions: + for future in self.futures[session_id]: + future.cancel() + cnt = self.sessions[session_id][0].qsize() + if cnt > 0: + logger.info("Cancel {} messages in session {}".format(cnt, session_id)) + self.sessions[session_id][0] = Dequeue() + + +def check_prefix(content, prefix_list): + if not prefix_list: + return None + for prefix in prefix_list: + if content.startswith(prefix): + return prefix + return None + + +def check_contain(content, keyword_list): + if not keyword_list: + return None + for ky in keyword_list: + if content.find(ky) != -1: + return True + return None + + +def remove_prefix(content, prefixes): + for prefix in prefixes: + if content.startswith(prefix): + return content[len(prefix):].strip() + return content.strip() + + + +def process_string(s): + # 判断是否以@开头并且包含"搜"字 + if s.startswith('@') and '搜' in s: + # 找到"搜"字的位置 + index = s.index('搜') + # 去除"搜"字前面的内容 + return s[index:] + else: + return s + +def process_string2(s): + # 判断是否包含@ + if '@' in s: + # 找到@字符的位置 + index = s.index('@') + # 删除包含@在内后面的所有字符 + return s[:index] + else: + return s + + + +def search_question(question): + url = 'https://pan.xinyuedh.com/api/search' + params = { + 'is_time': '1', + 'page_no': '1', + 'page_size': '5', + 'title': question + } + try: + response = requests.get(url, params=params) + response.raise_for_status() # 检查请求是否成功 + responseData = response.json().get('data', {}).get('items', []) + return responseData + except requests.exceptions.RequestException as e: + print(f"Error fetching data: {e}") + return [] + +def search_alone(question): + url = 'https://pan.xinyuedh.com/api/other/all_search' + payload = { + 'title': question + } + try: + response = requests.post(url, json=payload) + response.raise_for_status() + responseData = response.json().get('data', []) + return responseData + except requests.exceptions.RequestException as e: + print(f"Error fetching data: {e}") + return [] +