AiOpsWechatiPad/common/utils.py

import io
import os
import uuid
import requests
from urllib.parse import urlparse
from PIL import Image
from common.log import logger
import oss2,time,json
from urllib.parse import urlparse, unquote
from voice.ali.ali_voice import AliVoice
from voice import audio_convert


import cv2
import os
import tempfile
from moviepy import VideoFileClip

from common import redis_helper

from datetime import datetime

def fsize(file):
    if isinstance(file, io.BytesIO):
        return file.getbuffer().nbytes
    elif isinstance(file, str):
        return os.path.getsize(file)
    elif hasattr(file, "seek") and hasattr(file, "tell"):
        pos = file.tell()
        file.seek(0, os.SEEK_END)
        size = file.tell()
        file.seek(pos)
        return size
    else:
        raise TypeError("Unsupported type")


def compress_imgfile(file, max_size):
    if fsize(file) <= max_size:
        return file
    file.seek(0)
    img = Image.open(file)
    rgb_image = img.convert("RGB")
    quality = 95
    while True:
        out_buf = io.BytesIO()
        rgb_image.save(out_buf, "JPEG", quality=quality)
        if fsize(out_buf) <= max_size:
            return out_buf
        quality -= 5


def split_string_by_utf8_length(string, max_length, max_split=0):
    encoded = string.encode("utf-8")
    start, end = 0, 0
    result = []
    while end < len(encoded):
        if max_split > 0 and len(result) >= max_split:
            result.append(encoded[start:].decode("utf-8"))
            break
        end = min(start + max_length, len(encoded))
        # 如果当前字节不是 UTF-8 编码的开始字节，则向前查找直到找到开始字节为止
        while end < len(encoded) and (encoded[end] & 0b11000000) == 0b10000000:
            end -= 1
        result.append(encoded[start:end].decode("utf-8"))
        start = end
    return result


def get_path_suffix(path):
    path = urlparse(path).path
    return os.path.splitext(path)[-1].lstrip('.')


def convert_webp_to_png(webp_image):
    from PIL import Image
    try:
        webp_image.seek(0)
        img = Image.open(webp_image).convert("RGBA")
        png_image = io.BytesIO()
        img.save(png_image, format="PNG")
        png_image.seek(0)
        return png_image
    except Exception as e:
        logger.error(f"Failed to convert WEBP to PNG: {e}")
        raise


def generate_timestamp():
    # 获取当前时间
    now = datetime.now()
    # 格式化时间字符串为 'yyyyMMddHHmmssSS'
    timestamp = now.strftime('%Y%m%d%H%M%S%f')[:-4]
    return timestamp

def at_extract_content(text):
    # 找到最后一个空格的索引
    last_space_index = text.rfind(" ")
    if last_space_index != -1:
        # 返回空格后面的内容
        return text[last_space_index + 1:]
    return ""

def audio_extract_content(text):
    result = text.split('\n', 1)[1]
    return result


def save_to_local_from_url(url):
    '''
    从url保存到本地tmp目录
    '''

    parsed_url = urlparse(url)
     # 从 URL 提取文件名
    filename = os.path.basename(parsed_url.path)
    # tmp_dir = os.path(__file__)  # 获取系统临时目录
    # print(tmp_dir)
    tmp_file_path = os.path.join(os.getcwd(),'tmp', filename)  # 拼接完整路径

    # 检查是否存在同名文件
    if os.path.exists(tmp_file_path):
        logger.info(f"文件已存在，将覆盖：{tmp_file_path}")

    # 下载文件并保存到临时目录
    response = requests.get(url, stream=True)
    with open(tmp_file_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=1024):
            if chunk:  # 检查是否有内容
                f.write(chunk)

    return tmp_file_path


def upload_oss(
    access_key_id, 
    access_key_secret, 
    endpoint, 
    bucket_name, 
    file_source, 
    prefix, 
    expiration_days=7
):
    """
    上传文件到阿里云OSS并设置生命周期规则，同时返回文件的公共访问地址。
    
    :param access_key_id: 阿里云AccessKey ID
    :param access_key_secret: 阿里云AccessKey Secret
    :param endpoint: OSS区域对应的Endpoint
    :param bucket_name: OSS中的Bucket名称
    :param file_source: 本地文件路径或HTTP链接
    :param prefix: 设置规则应用的前缀为文件所在目录
    :param expiration_days: 文件保存天数，默认7天后删除
    :return: 文件的公共访问地址
    """
    
    # 创建Bucket实例
    auth = oss2.Auth(access_key_id, access_key_secret)
    bucket = oss2.Bucket(auth, endpoint, bucket_name)

    ### 1. 设置生命周期规则 ###
    rule_id = f'delete_after_{expiration_days}_days'  # 规则ID
    # prefix = oss_file_name.split('/')[0] + '/'  # 设置规则应用的前缀为文件所在目录
    
    
    # 定义生命周期规则
    rule = oss2.models.LifecycleRule(rule_id, prefix, status=oss2.models.LifecycleRule.ENABLED,
                                     expiration=oss2.models.LifecycleExpiration(days=expiration_days))

    # 设置Bucket的生命周期
    lifecycle = oss2.models.BucketLifecycle([rule])
    bucket.put_bucket_lifecycle(lifecycle)

    print(f"已设置生命周期规则：文件将在{expiration_days}天后自动删除")

    ### 2. 判断文件来源并上传到OSS ###
    if file_source.startswith('http://') or file_source.startswith('https://'):
        # HTTP 链接，先下载文件
        try:
            response = requests.get(file_source, stream=True)
            response.raise_for_status()
            parsed_url = urlparse(file_source)
            # 提取路径部分并解码
            path = unquote(parsed_url.path)
            # 获取路径的最后一部分作为文件名
            filename = path.split('/')[-1]
            oss_file_name=prefix+'/'+ filename
            bucket.put_object(oss_file_name, response.content)
            print(f"文件从 HTTP 链接上传成功：{file_source}")
        except requests.exceptions.RequestException as e:
            print(f"从 HTTP 链接下载文件失败: {e}")
            return None
    else:
        # 本地文件路径
        try:
            filename=os.path.basename(file_source)
            oss_file_name=prefix+'/'+ filename
            bucket.put_object_from_file(oss_file_name, file_source)
            print(f"文件从本地路径上传成功：{file_source}")
        except oss2.exceptions.OssError as e:
            print(f"从本地路径上传文件失败: {e}")
            return None

    ### 3. 构建公共访问URL ###
    file_url = f"http://{bucket_name}.{endpoint.replace('http://', '')}/{oss_file_name}"

    print(f"文件上传成功，公共访问地址：{file_url}")
    
    return file_url

def generate_guid_no_dashes():
    """
    生成一个无分隔符的 GUID
    :return: 返回生成的无分隔符 GUID 字符串
    """
    return str(uuid.uuid4()).replace('-', '')

def dialogue_message(wxid_from:str,wxid_to:str,wx_content:list,is_ai:bool=False):
    """
    构造消息的 JSON 数据
    :param contents: list，包含多个消息内容，每个内容为字典，如：
                     [{"type": "text", "text": "AAAAAAA"},
                      {"type": "image_url", "image_url": {"url": "https://AAAAA.jpg"}},
                      {"type":"file","file_url":{"url":"https://AAAAA.pdf"}}
                      ]
    :return: JSON 字符串
    """

    # 获取当前时间戳，精确到毫秒
    current_timestamp = int(time.time() * 1000)

    # 获取当前时间，格式化为 "YYYY-MM-DD HH:MM:SS"
    current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    
    # 构造 JSON 数据
    data = {
        "message_id": str(current_timestamp),
        "topic": "topic.ai.ops.wx",
        "time": current_time,
        "data": {
            "msg_type": "dialogue",
            "is_ai":is_ai,
            "content": {
                "wxid_from": wxid_from,
                "wxid_to": wxid_to,
                "wx_content":wx_content
            }
        }
    }

    return json.dumps(data, separators=(',', ':'), ensure_ascii=False)

def kafka_base_message(content: dict)->dict: 
    # 获取当前时间戳，精确到毫秒
    current_timestamp = int(time.time() * 1000)

    # 获取当前时间，格式化为 "YYYY-MM-DD HH:MM:SS"
    current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    
    # 构造 JSON 数据
    data = {
        "message_id": str(current_timestamp),
        "topic": "topic.ai.ops.wx",
        "time": current_time,
        "data": {
            "msg_type": "login-qrcode",
            "content": content
        }
    }
    return data


def login_qrcode_message(token_id: str,agent_tel:str,qr_code_img_base64:str,qr_code_url:list)->str: 
        """
        构造消息的 JSON 数据
        :param contents: list，包含多个消息内容，每个内容为字典，如：
                        {
                            "tel":"18029274615",
                            "token_id":"f828cb3c-1039-489f-b9ae-7494d1778a15",
                            "qr_code_urls":["url1","url2","url3","url4",],
                            "qr_code_img_base64":"aaaaaaaaaaaaaa"
                            }
        :return: JSON 字符串
        """
        content = {
            "tel":agent_tel,
            "token_id":token_id,
            "qr_code_urls":qr_code_url,
            "qr_code_img_base64":qr_code_img_base64
        }
        data=kafka_base_message(content)
        return json.dumps(data, separators=(',', ':'), ensure_ascii=False)



def wx_voice(text: str):
    try:
        # 将文本转换为语音
        reply_text_voice = AliVoice().textToVoice(text)
        reply_text_voice_path = os.path.join(os.getcwd(), reply_text_voice)
        
        # 转换为 Silk 格式
        reply_silk_path = os.path.splitext(reply_text_voice_path)[0] + ".silk"
        reply_silk_during = audio_convert.any_to_sil(reply_text_voice_path, reply_silk_path)

        # OSS 配置（建议将凭证存储在安全的地方）
        oss_access_key_id="LTAI5tRTG6pLhTpKACJYoPR5"
        oss_access_key_secret="E7dMzeeMxq4VQvLg7Tq7uKf3XWpYfN"
        oss_endpoint="http://oss-cn-shanghai.aliyuncs.com"
        oss_bucket_name="cow-agent"
        oss_prefix="cow"

        # 上传文件到 OSS
        file_path = reply_silk_path
        file_url = upload_oss(oss_access_key_id, oss_access_key_secret, oss_endpoint, oss_bucket_name, file_path, oss_prefix)

        # 删除临时文件
        try:
            os.remove(reply_text_voice_path)
        except FileNotFoundError:
            pass  # 如果文件未找到，跳过删除
        try:
            os.remove(reply_silk_path)
        except FileNotFoundError:
            pass  # 如果文件未找到，跳过删除

        return int(reply_silk_during), file_url
    except Exception as e:
        print(f"发生错误：{e}")
        return None, None  # 发生错误时返回 None

def get_login_info_by_wxid(wxid: str) ->dict:
    # 使用 SCAN 避免一次性返回所有的匹配键，逐步扫描
    cursor = 0
    while True:
        cursor, login_keys = redis_helper.redis_helper.client.scan(cursor, match='__AI_OPS_WX__:LOGININFO:*')
        
        # 批量获取所有键的 hash 数据
        for k in login_keys:
            r = redis_helper.redis_helper.get_hash(k)
            if r.get("wxid") == wxid:
                return k,r
        
        # 如果游标为 0，则表示扫描完成
        if cursor == 0:
            break
    
    return None,None


def download_video_and_get_thumbnail(url, thumbnail_path):
    """
    从指定URL下载MP4视频，提取首帧作为缩略图，并返回缩略图路径及视频时长。
    
    参数:
        url (str): 视频的URL地址。
        thumbnail_path (str): 缩略图的保存路径。
        
    返回:
        tuple: (缩略图路径, 视频时长（秒）)
        
    异常:
        可能抛出requests.exceptions.RequestException，cv2.error，IOError等异常。
    """
    # 创建临时目录以下载视频
    with tempfile.TemporaryDirectory() as tmp_dir:
        # 下载视频到临时文件
        video_path = os.path.join(tmp_dir, 'temp_video.mp4')
        response = requests.get(url, stream=True)
        response.raise_for_status()  # 确保请求成功
        
        with open(video_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:  # 过滤掉保持连接的空白块
                    f.write(chunk)
        
        # 提取视频首帧作为缩略图
        vidcap = cv2.VideoCapture(video_path)
        success, image = vidcap.read()
        vidcap.release()
        if not success:
            raise RuntimeError("无法读取视频的首帧，请检查视频文件是否有效。")
        
        # 确保缩略图的目录存在
        thumbnail_dir = os.path.dirname(thumbnail_path)
        if thumbnail_dir:
            os.makedirs(thumbnail_dir, exist_ok=True)
        
        # 保存缩略图
        cv2.imwrite(thumbnail_path, image)
        
        # 使用moviepy计算视频时长
        clip = VideoFileClip(video_path)
        duration = clip.duration
        clip.close()

        # OSS 配置（建议将凭证存储在安全的地方）
        oss_access_key_id="LTAI5tRTG6pLhTpKACJYoPR5"
        oss_access_key_secret="E7dMzeeMxq4VQvLg7Tq7uKf3XWpYfN"
        oss_endpoint="http://oss-cn-shanghai.aliyuncs.com"
        oss_bucket_name="cow-agent"
        oss_prefix="cow"

        # 上传文件到 OSS
        file_path = thumbnail_path
        file_url = upload_oss(oss_access_key_id, oss_access_key_secret, oss_endpoint, oss_bucket_name, file_path, oss_prefix)
    
        # 删除临时文件
        try:
            os.remove(thumbnail_path)
        except FileNotFoundError:
            pass  # 如果文件未找到，跳过删除

    return file_url, duration