构建m3u8视频的智能语义检索系统

# -*- coding: utf-8 -*-
import argparse
import alibabacloud_oss_v2 as oss
# 解析XML响应
import xml.etree.ElementTree as ET
import json 
from datetime import datetime 

def get_search_conditions():
    """获取用户的多条件输入"""
    print("请输入语义描述（示例：停着车的院子）")
    query = input("> 语义关键词: ").strip()
    while not query:
        print("语义关键词不能为空！")
        query = input("> 语义关键词: ").strip()
    return query  

def build_metaquery_xml(query):
    """构建符合OSS规范的MetaQuery XML"""
    xml_parts = [f'<Query>{query}</Query>']

    meta_query_xml = f'''<MetaQuery>
    {"".join(xml_parts)}
</MetaQuery>'''
    return meta_query_xml  # 编码为UTF-8字节流


def format_result(key, pre_url):
    """格式化单个搜索结果"""
    return f""" 文件路径：{key}
 文件地址：{pre_url}
-----------------------"""

def semantic_search():
    # 直接赋值给命令行参数
    args = argparse.Namespace(
        region='cn-beijing',  # 替换为你的区域
        bucket='m3u8-frames-oss-metaquery-demo',  # 替换为你的存储空间名称
        endpoint='https://oss-cn-beijing.aliyuncs.com',  # 替换为你的endpoint，如果不需要可以留空或删除
    )

    # 初始化OSS客户端
    credentials = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials
    cfg.region = args.region
    if args.endpoint:
        cfg.endpoint = args.endpoint
    client = oss.Client(cfg)

    # 获取用户输入
    query = get_search_conditions()  # 只获取查询
    # 构建请求
    try:
        # 构造请求体（XML 数据）
        data_str = build_metaquery_xml(query)  # 不再传递条件

        # 定义操作输入
        req = oss.OperationInput(
            op_name='DoMetaQuery',  # 自定义操作名称
            method='POST',            # HTTP 方法
            parameters={              # 查询参数
                'metaQuery': '',
                'mode': 'semantic',
                'comp': 'query',
            },
            headers=None,             # 自定义请求头（可选）
            body=data_str.encode("utf-8"),  # 请求体（编码为 UTF-8 字节流）
            bucket=args.bucket,       # 目标 Bucket 名称
        )

        # 调用泛化接口执行操作
        resp = client.invoke_operation(req)

    except oss.exceptions.ServiceError as e:
        print(f" 服务端错误：{e.message}")
        return
    
    root = ET.fromstring(resp.http_response.content.decode('utf-8'))
     # 查找所有File元素
    files = root.findall('.//File')
    
    print(f"\n 共找到 {len(files)} 个匹配结果：")
    


    for i, file in enumerate(files, 1):
        # 获取文件名
        key_element = file.find('Filename')
        if key_element is None:
            continue  # 如果没有找到文件名，则跳过当前循环迭代
        key = key_element.text
        
        
        # 生成预签名URL
        pre_url = client.presign(
            oss.GetObjectRequest(
                bucket=args.bucket,  # 指定存储空间名称
                key=key,        # 指定对象键名
            )
        )
        
        print(format_result(key, pre_url.url))
        
    

if __name__ == "__main__":
    semantic_search()