robot_ai/handlers/xunfei/intent_handler.py

import threading

from config.config.dify_config import difyconfig
from handlers.dify.recognize_intention import chat_with_dify
import hashlib
import pymysql
from utils.pc2_requests import _send_qa_task, _send_led_color_task
from utils.tts_client import play_text_async
import json5
import random
import json
from utils.logger import logger
import requests
import paramiko
from sshtunnel import SSHTunnelForwarder
from config.config.settings import config
from strategies.vision.qwenv import process_base64_image
from strategies.action.execute_actions import ActionExecutor
from utils.logger import logger
"""
意图处理模块
"""


class IntentHandler:
    """意图处理类"""

    def __init__(self):
        self.detected_intent = None
        self.tts_text = ""
        self.cbm_semantic_processed = False  # 标记是否已处理视觉内容
        self.executor = ActionExecutor()
        self.vision_dealing_words = [
            "我将从多个角度分析",  # 日常聊天标配，简单自然
            "我将从多维视角分析",  # 像说"我瞅瞅相关内容"，带着行动感
            "我将从不同层面分析",  # 随口的礼貌，特生活化
            "我将从多方面评估",  # 直白展现正在动脑筋
            "我将从多个维度剖析",  # 比"稍等下"正式点，又不刻意
        ]

    def play_vision_phrase(self, type="thinking"):
        """播放语气助词"""
        phrase = random.choice(self.vision_dealing_words)
        logger.info(f"播放视觉语气助词: {phrase}")
        play_text_async(phrase, use_cache=True)

    def handle_intent_result(self, data: dict) -> None:
        """
        处理意图识别结果

        Args:
            data: 意图数据
        """
        try:
            try:
                # 异步调用 _send_led_color_task
                led_thread = threading.Thread(
                    target=_send_led_color_task,
                    args=(self, "BREATH", "BLUE"),
                    daemon=True
                )
                led_thread.start()
            except Exception as e:
                logger.debug(f"[LED] LED控制失败，不影响意图处理: {e}")

            # logger.info('意图识别data', json.dumps(data, ensure_ascii=False, indent=2))
            text_value = data.get('content', {}).get(
                'result', {}).get('cbm_semantic', {}).get('text')
            if text_value:

                intent = json.loads(text_value)
                rc = intent.get('rc', -1)
                if rc == 0:
                    category = intent.get('category', "")
                    logger.info(f"技能结果: {category}")
                    # 排除默认技能
                    if category == "IFLYTEK.mapU":
                        logger.info(f"排除默认技能: {category}")
                        return
                    # 视觉意图
                    elif category.split('.')[-1] == "vision":
                        ###################
                        # 使用dify识别视觉意图
                        logger.info(f"使用dify识别视觉意图")
                        logger.info(
                            f"difyconfig.get_vision_switch(): {difyconfig.get_vision_switch()}")
                        if difyconfig.get_vision_switch():
                            # 判断是否是视觉意图
                            iat_txt = intent.get('text', "")
                            res = chat_with_dify(
                                question=iat_txt, image_path='')
                            if res != '视觉意图':
                                return
                    elif category.split('.')[-1] == "wake_up":
                        semantics_list = intent.get("semantic", [])
                        if semantics_list:
                            semantics = semantics_list[0]
                            skill_purpose = semantics.get('intent')
                            if skill_purpose == "WAKEUP":
                                self.cbm_semantic_processed = False
                            elif skill_purpose == "RESET_WAKEUP":
                                self.cbm_semantic_processed = True
                        ###################

                    iat_txt = intent.get('text', "")
                    logger.info(f"意图识别文本: {iat_txt}")
                    answer = intent.get('answer', {}).get('text', "")
                    logger.info(f"意图识别答案: {answer}")
                    if answer:
                        try:
                            # 异步调用 _send_qa_task
                            logger.info("发送QA接口===========> [QA]")
                            if category.split('.')[-1] != "vision":
                                qa_thread = threading.Thread(
                                    target=_send_qa_task,
                                    args=(self, {"result": answer}),
                                    daemon=True
                                )
                                qa_thread.start()
                        except Exception as e:
                            logger.error(f"QA接口请求失败: {e}")
                        if not self.cbm_semantic_processed:
                            self.cbm_semantic_processed = True
                            try:
                                play_text_async(answer, use_cache=True)
                                logger.info(f"[意图] 答案播放请求已发送")
                            except Exception as e:
                                logger.info(f"[意图] 播放请求失败: {e}")
                                import traceback
                                traceback.print_exc()

                    semantics_list = intent.get("semantic", [])
                    if semantics_list:
                        semantics = semantics_list[0]
                        skill_purpose = semantics.get('intent')
                        # 处理具体意图
                        self._handle_detected_intent(
                            category, iat_txt, skill_purpose)
                    else:
                        logger.warning("未获取到语义信息")
                        # 处理具体意图，但不传递skill_purpose
                        self._handle_detected_intent(
                            category, iat_txt, None)
            else:
                logger.warning("未获取到意图文本")

        except Exception as e:
            logger.error(f"意图处理异常: {e}")

    def _handle_detected_intent(self, intent: str, iat_txt: str, purpose) -> None:
        """
        处理检测到的意图
        Args:
            intent: 意图名称
            iat_txt: 原IAT内容
        """
        # 读取yaml配置文件
        config_data = config.get_config()
        xunfei_config = config_data.get('xunfei', {})
        categories = xunfei_config.get('category', [])

        if intent in categories:
            self.executor.execute_actions(intent, purpose)

        # 视觉调用
        elif intent.split('.')[-1] == "vision":
            # 在识别开始时播放语气助词
            logger.info(f"开始播放语气词")
            self.play_vision_phrase(type="vision")
            logger.info(f"检测到 [{intent}] 意图, 执行视觉相关")
            # 标记已处理视觉内容，避免后续NLP重复播放
            self.cbm_semantic_processed = True
            answer_text = '摄像头服务异常'
            try:
                camera_url = config._config_data.get(
                    'camera_url', 'http://127.0.0.1:34550/camera_base64'
                )
                headers = {"Content-Type": "application/json"}
                data = {}  # matches the -d '{}' from curl

                response = requests.get(camera_url, headers=headers, json=data)

                resp_data = json5.loads(response.text)
                # 处理响应
                if resp_data.get('code') == 200:
                    base64_pic = resp_data.get('data').get('base64_content')
                    answer_text = process_base64_image(
                        base64_pic, question=iat_txt)
                    logger.info(f"视觉调用结果: {answer_text}")
                    play_text_async(answer_text)
                else:
                    logger.error(f"摄像头服务异常: {str(resp_data.get('code'))}")
                try:
                    # 异步调用 _send_qa_task
                    logger.info("发送QA接口===========> [QA]")
                    qa_thread = threading.Thread(
                        target=_send_qa_task,
                        args=(self, {"result": answer_text}),
                        daemon=True
                    )
                    qa_thread.start()
                except Exception as e:
                    logger.error(f"QA接口请求失败: {e}")
            except Exception as e:
                logger.error(f"摄像头服务异常: {str(e)}")
        elif intent.split('.')[-1] == "BASE_STATION":
            try:
                match purpose:
                    case "Base_Stations":
                        sum1, sum2 = self.sql_query()
                        play_text_async(f"实时客流为{sum1}人次，累计客流为{sum2}人次")
            except requests.exceptions.RequestException as e:
                logger.error(f"区域基站客流查询API调用失败: {e}")
        elif intent.split('.')[-1] == "Game_Guess":
            try:
                match purpose:
                    case "Ready":
                        pass
                    case "Start_Game":
                        try:
                            response = requests.post(url="http://192.168.123.164:9002/game",
                                                     json={"target": "game", "cmd": "start_game", "id": "cq"})
                            result = response.json()
                            if result and result["code"] == 0:
                                play_text_async("游戏开始")
                            else:
                                play_text_async("游戏开始失败, 请稍后重试")
                        except requests.exceptions.RequestException as e:
                            logger.error(f"猜拳游戏异常(开始): {e}")
                    case "Stop_Game":
                        try:
                            response = requests.post(url="http://192.168.123.164:9002/game",
                                                     json={"target": "game", "cmd": "stop_game", "id": "cq"})
                            result = response.json()
                            if result and result["code"] == 0:
                                play_text_async("游戏结束")
                            else:
                                play_text_async("游戏结束失败, 请稍后重试")
                        except requests.exceptions.RequestException as e:
                            logger.error(f"猜拳游戏异常(结束): {e}")
            except requests.exceptions.RequestException as e:
                logger.error(f"猜拳游戏执行异常: {e}")
        elif intent.split('.')[-1] == "switch_model":
            if difyconfig.get_models_switch():
                # 切换模型
                model_dify_type = difyconfig.get_model_dify_type(purpose)
                if model_dify_type:
                    difyconfig.set_current_model(model_dify_type)
        else:
            logger.info(f"检测到未知意图: {intent}")

    def get_detected_intent(self) -> str:
        """获取检测到的意图"""
        return self.detected_intent or ""

    def get_tts_text(self) -> str:
        """获取TTS文本"""
        return self.tts_text or ""

    def is_cbm_semantic_processed(self) -> bool:
        """检查是否已处理cbm_semantic"""
        return self.cbm_semantic_processed

    def reset_cbm_semantic_processed(self) -> None:
        """重置cbm_semantic处理标记"""
        self.cbm_semantic_processed = False

    def generate_md5(self, userId: str, key: str, full_identifier: str) -> str:
        """MD5加密"""
        combined = f"{userId}{key}{full_identifier}"
        md5_hash = hashlib.md5(combined.encode('utf-8')).hexdigest()
        return md5_hash.upper()

    def sql_query(self):
        try:
            with SSHTunnelForwarder(
                    (config._config_data.get('ssh_host'),
                     config._config_data.get('ssh_port')),
                    ssh_username=config._config_data.get('ssh_user'),
                    ssh_password=config._config_data.get('ssh_pass'),
                    remote_bind_address=(
                        config._config_data.get('db_host'),
                        config._config_data.get('db_port')
                    )
            ) as tunnel:
                logger.info(
                    f"SSH 隧道已建立，开始更新数据库: {config._config_data.get('table')}")
                conn = pymysql.connect(
                    host=config._config_data.get('db_host'),
                    port=tunnel.local_bind_port,
                    user=config._config_data.get('db_user'),
                    password=config._config_data.get('db_pass'),
                    db=config._config_data.get('db_name'),
                    autocommit=True
                )
                try:
                    with conn.cursor() as cur:
                        cur.execute("SELECT 1;")
                        result = cur.fetchone()
                        if result and result[0] == 1:
                            logger.info("数据库连接成功！")
                        else:
                            logger.error("数据库连接失败！")
                    select_sql = f"""
                        SELECT flow_sum1, flow_sum2 
                        FROM {config._config_data.get('table')} 
                        WHERE identity = 'Current'
                    """
                    with conn.cursor() as cur:
                        cur.execute(select_sql)
                        row = cur.fetchone()
                        if row:
                            # flow_sum1, flow_sum2 = row
                            return row
                        else:
                            logger.warning("未查询到数据！")
                finally:
                    conn.close()
                    logger.info("数据库连接已关闭")
        except Exception as e:
            logger.error(f"更新数据库失败: {e}")