构建IPC设备的智能语义检索系统

# -*- coding: utf-8 -*-
import base64
import argparse
import alibabacloud_oss_v2 as oss


def video_processor():
    # OSS配置
    args = argparse.Namespace(
        region='cn-beijing',
        target_bucket='ipc-frames-oss-metaquery-demo',    
        bucket='ipc-videos-oss-metaquery-demo',     
        endpoint='https://oss-cn-beijing.aliyuncs.com'
    )
    
    # 配置OSS客户端
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    cfg.endpoint = args.endpoint
    client = oss.Client(cfg)

    # 视频文件列表 替换为您需要处理的视频
    videos = [('视频A.mp4', '2025-04-07'), 
              ('视频B.mp4', '2025-04-07'), 
              ('视频C.mp4', '2025-04-07')]
    
    for video, date in videos:
        # 将截取的图片分日期放入目标文件夹
        target_key = f"{date}/{video}/{{index}}.{{autoext}}"
        
        # 填入配置好的视频样式
        style = 'style/ipc-snap'
        
        # 对目标存储空间名和文件名进行URL安全的Base64编码
        target_bucket_base64 = base64.urlsafe_b64encode(args.target_bucket.encode()).decode()
        target_key_base64 = base64.urlsafe_b64encode(target_key.encode()).decode()
        
        # 构建处理指令，包含样式和保存位置
        process = f"{style}|sys/saveas,b_{target_bucket_base64},o_{target_key_base64}"
        
        try:
            # 发送异步处理请求
            result = client.async_process_object(oss.AsyncProcessObjectRequest(
                bucket=args.bucket,
                key=video,
                process=process
            ))
            
            print(f"Status Code: {result.status_code}")
            print(f"RequestId: {result.request_id}")
            print(f"TaskId: {result.task_id}")
            print(f"EventId: {result.event_id}")
            print(f"ProcessRequestId: {result.process_request_id}")
            
        except Exception as e:
            print(f"Error: {e}")

if __name__ == "__main__":
    video_processor()

# -*- 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='ipc-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()

import argparse
import alibabacloud_oss_v2 as oss
from alibabacloud_oss_v2.models import GetObjectTaggingRequest, PutObjectTaggingRequest, Tagging, TagSet, Tag

def list_and_process_frames(client, bucket, prefix, tags):
    """列出并处理指定前缀下的所有文件"""
    try:
        # 创建列举对象请求
        list_request = oss.models.ListObjectsRequest(
            bucket=bucket,
            prefix=prefix
        )
        
        # 获取对象列表并处理
        result = client.list_objects(list_request)
        if result.contents:
            for obj in result.contents:
                apply_tags_to_frame(client, bucket, obj.key, tags)
                    
    except Exception as e:
        print(f"处理文件夹失败: {str(e)}")

def apply_tags_to_frame(client, bucket, frame_key, tags):
    """为截帧图片添加标签"""
    try:
        # 构建标签对象
        tagging = Tagging(
            version=1,
            tag_set=TagSet(tags=tags)
        )
        
        # 创建更新标签请求
        put_tag_request = PutObjectTaggingRequest(
            bucket=bucket,
            key=frame_key,
            tagging=tagging
        )
        
        # 更新对象标签
        result = client.put_object_tagging(put_tag_request)
        
        # 将标签转换为字符串以便打印
        tags_str = '&'.join([f"{tag.key}={tag.value}" for tag in tags])
        print(f"成功为图片 {frame_key} 添加标签: {tags_str}")
        return True
    except Exception as e:
        print(f"为图片添加标签失败: {str(e)}")
        return False

def frame_tags():
    # OSS配置
    args = argparse.Namespace(
        region='cn-beijing',   
        frame_bucket='ipc-frames-oss-metaquery-demo',     
        endpoint='https://oss-cn-beijing.aliyuncs.com'
    )
    
    # 配置OSS客户端
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    cfg.endpoint = args.endpoint
    client = oss.Client(cfg)

    # 配置您的存储帧路径前缀
    videos = [
        {
            'frame_prefix': '2025-04-07/视频A.mp4/' 
        },    
        {
            'frame_prefix': '2025-04-07/视频B.mp4/'
        },
        {
            'frame_prefix': '2025-04-07/视频C.mp4/'
        }
    ]

    # 处理每个视频的截帧
    for video in videos:


        # 从 frame_prefix 中提取视频名称部分，例如 '2025-04-07/视频A.mp4/' -> '视频A.mp4'
        prefix_parts = video['frame_prefix'].strip('/').split('/')
        video_filename = prefix_parts[-1] if prefix_parts else '' # 取最后一个非空部分，例如 '视频A.mp4'

        # 去掉文件扩展名，例如 '视频A.mp4' -> '视频A'
        video_name_base = video_filename.split('.')[0] if '.' in video_filename else video_filename

        print(f"\n开始处理视频 {video_filename} 的截帧") # 打印原始提取的文件名

        tags = [
            # 使用去除扩展名后的视频名称的最后一个字符设置摄像机标识
            Tag(key='camera', value=f'camera-{video_name_base[-1].lower()}' if video_name_base else 'camera-unknown'),
            # Tag(key='category', value='video_monitoring')       # 添加实际业务分类信息
        ]

        # 处理该视频的所有截帧
        list_and_process_frames(client, args.frame_bucket, video['frame_prefix'], tags)

        print(f"完成视频 {video_filename} 的截帧处理\n") # 打印原始提取的文件名

if __name__ == "__main__":
    frame_tags()

# -*- coding: utf-8 -*-
import argparse
import alibabacloud_oss_v2 as oss
# 解析XML响应
import xml.etree.ElementTree as ET
import json
from datetime import datetime
# --- 新增导入 ---
from alibabacloud_oss_v2.models import GetObjectTaggingRequest
# --- 新增导入结束 ---

def get_inputs():
    """获取用户的语义查询（必选）和筛选条件（可选）"""
    # 1. 获取必选的语义查询
    print(" 请输入语义描述（示例：停着车的院子）")
    query = input("> 语义关键词: ").strip()
    while not query:
        print(" 语义关键词不能为空！")
        query = input("> 语义关键词: ").strip()

    conditions = [] # 初始化筛选条件列表
    target_tag = None # --- 新增：用于存储目标标签 ---

    # 2. 获取可选的标签筛选
    print("\n (可选)请输入标签筛选条件（将用于客户端过滤，示例：camera=camera-a，按回车跳过）") # 修改提示
    tag_input = input("> 标签 (格式 key=value): ").strip()
    if tag_input:
        if '=' in tag_input:
            # --- 修改：不再添加到 conditions，而是存到 target_tag ---
            # conditions.append({
            #     'field': 'Tags',
            #     'value': tag_input,
            #     'op': 'eq'
            # })
            target_tag = tag_input
            print(f" 信息：标签 '{target_tag}' 将在获取结果后于客户端进行过滤。")
            # --- 修改结束 ---
        else:
             print(" 标签格式不正确，已忽略。请使用 'key=value' 格式。")


    # 3. 获取可选的路径前缀筛选
    print("\n (可选)请输入路径前缀筛选（示例：YYYY-MM-DD，按回车跳过）")
    date_prefix = input("> 路径前缀: ").strip()
    if date_prefix:
        conditions.append({
            'field': 'Filename',
            'value': date_prefix,
            'op': 'prefix'
        })

    # --- 修改：返回 target_tag ---
    return query, conditions, target_tag
    # --- 修改结束 ---

def build_metaquery_xml(query, conditions):
    """构建符合OSS规范的MetaQuery XML (包含语义查询和可选筛选)"""
    # 始终包含语义查询部分
    xml_parts = [f'<Query>{query}</Query>']

    # 添加可选的筛选条件 - 在语义模式下，仅支持部分字段（如此处的 Filename）
    for cond in conditions:
        # 在 semantic 模式下跳过 Tags 字段的 SimpleQuery 构建 (虽然我们已经不在 conditions 里加 Tags 了，但保留以防万一)
        if cond['field'] == 'Tags':
            # print(f" 信息：由于处于语义搜索模式，标签筛选条件 '{cond['value']}' 将被忽略。") # 这行不再需要
            continue

        json_query = json.dumps({
            "Field": cond['field'],
            "Value": cond['value'],
            "Operation": cond['op']
        }, ensure_ascii=False)
        xml_parts.append(f'<SimpleQuery>{json_query}</SimpleQuery>')

    # 组合成完整的 MetaQuery XML
    meta_query_xml = f'''<MetaQuery>
    {"".join(xml_parts)}
</MetaQuery>'''
    return meta_query_xml

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

def perform_search():
    # 直接赋值给命令行参数
    args = argparse.Namespace(
        region='cn-beijing',  # 替换为你的区域
        bucket='ipc-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)

    # 获取用户输入 (语义查询 + 可选条件 + 目标标签)
    # --- 修改：接收 target_tag ---
    query, conditions, target_tag = get_inputs()
    # --- 修改结束 ---

    # 构建请求
    try:
        # 构造请求体（XML 数据）
        # --- 只传入 conditions (不含Tags) ---
        data_str = build_metaquery_xml(query, conditions)
        
        print(f"\n 生成的 MetaQuery XML:\n{data_str}")

        # 定义操作输入
        req = oss.OperationInput(
            op_name='DoMetaQuery',
            method='POST',
            parameters={
                'metaQuery': '',
                'mode': 'semantic', # <-- 重新加入 semantic 模式
                'comp': 'query',
            },
            headers=None,
            body=data_str.encode("utf-8"),
            bucket=args.bucket,
        )

        # 调用泛化接口执行操作
        print("\n 发送 DoMetaQuery 请求...")
        resp = client.invoke_operation(req)
        print(f"\n 请求成功，HTTP 状态码: {resp.http_response.status_code}")
    # --- 修改异常类型 ---
    except oss.exceptions.ServiceError as e:
        print(f" 服务端错误 (ServiceError)：{e.message}")
        print(f"   - HTTP Status Code: {e.status_code}")
        print(f"   - Error Code: {e.error_code}")
        print(f"   - Request ID: {e.request_id}")
        return
    except oss.exceptions.ClientError as e: # 添加对客户端错误的捕获
        print(f" 客户端或网络错误 (ClientError)：{e.message}")
        return
    except Exception as e: # 捕获其他可能的异常
        print(f" 未知错误：{e}")
        import traceback
        traceback.print_exc() # 打印完整的 traceback
        return

    # --- 修改结束 ---

    # 解析和处理结果...
    final_results_count = 0 # --- 新增：计数器，用于统计最终符合条件的结果 ---
    try:
        root = ET.fromstring(resp.http_response.content.decode('utf-8'))
        # 查找所有File元素
        files = root.findall('.//File')

        print(f"\n 从OSS获取到 {len(files)} 个初步匹配结果，开始进行客户端标签过滤...")

        if not files:
            # 检查是否有 NextContinuationToken，可能需要分页
            next_token_elem = root.find('.//NextContinuationToken')
            if next_token_elem is not None and next_token_elem.text:
                print(" 提示：可能还有更多结果，当前实现未处理分页。")
            print("\n 没有初步匹配结果。")
            return # 没有文件时提前返回


        for i, file in enumerate(files, 1):
            # 获取文件名
            key_element = file.find('Filename')
            if key_element is None:
                print(f" 警告：第 {i} 个初步结果缺少 Filename 字段，已跳过。")
                continue
            key = key_element.text

            # --- 客户端标签过滤 ---
            if target_tag:
                try:
                    tagging_req = GetObjectTaggingRequest(bucket=args.bucket, key=key)
                    tagging_resp = client.get_object_tagging(tagging_req)
                    # 检查返回的标签集是否包含目标标签
                    tag_found = False
                    target_k, target_v = target_tag.split('=', 1)

                    # --- 修改：使用 .tags 而不是 .tag ---
                    if tagging_resp.tag_set and tagging_resp.tag_set.tags: # 使用 .tags
                        for tag in tagging_resp.tag_set.tags: # 使用 .tags
                    # --- 修改结束 ---
                            if tag.key == target_k and tag.value == target_v:
                                tag_found = True
                                break
                    if not tag_found:
                        continue # 标签不匹配，跳过此文件
                except oss.exceptions.ServiceError as tag_err:
                     if tag_err.status_code == 404 and tag_err.error_code == 'NoSuchTagSet':
                         continue
                     else:
                        print(f" 警告：获取文件 '{key}' 的标签时出错: {tag_err.error_code} - {tag_err.message}，已跳过。")
                        continue
                except Exception as tag_e:
                    print(f" 警告：获取或处理文件 '{key}' 的标签时发生未知错误: {tag_e}，已跳过。")
                    # --- 添加 traceback 方便调试 ---
                    import traceback
                    traceback.print_exc()
                    # --- 添加结束 ---
                    continue
            # --- 客户端标签过滤结束 ---

            # --- 如果通过了标签过滤（或者没有设置标签过滤），则处理并打印结果 ---
            final_results_count += 1 # 增加最终结果计数
            print(f"\n[{final_results_count}] 文件 '{key}' 符合所有条件:") # 打印最终结果编号

            # 生成预签名URL
            try:
                pre_url = client.presign(
                    oss.GetObjectRequest(
                        bucket=args.bucket,
                        key=key,
                    )
                )
                print(format_result(key, pre_url.url))
            except Exception as presign_e:
                print(f" 警告：为文件 '{key}' 生成预签名URL时出错: {presign_e}")
                print(format_result(key, "[无法生成URL]"))
        # --- 循环结束后打印最终统计 ---
        print(f"\n 客户端过滤完成，共找到 {final_results_count} 个最终匹配结果。")

    except ET.ParseError as xml_e:
        print(f" 错误：解析OSS响应XML时出错 - {xml_e}")
    except Exception as parse_e:
        print(f" 错误：处理结果时发生意外错误 - {parse_e}")


if __name__ == "__main__":
    perform_search()

# -*- coding: utf-8 -*-
import base64
import argparse
import alibabacloud_oss_v2 as oss


def video_processor():
    # OSS配置
    args = argparse.Namespace(
        region='cn-beijing',
        target_bucket='ipc-frames-oss-metaquery-demo',    
        bucket='ipc-videos-oss-metaquery-demo',     
        endpoint='https://oss-cn-beijing.aliyuncs.com'
    )
    
    # 配置OSS客户端
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    cfg.endpoint = args.endpoint
    client = oss.Client(cfg)

    # 视频文件列表 替换为您需要处理的视频
    videos = [('视频A.mp4', '2025-04-07'), 
              ('视频B.mp4', '2025-04-07'), 
              ('视频C.mp4', '2025-04-07')]
    
    for video, date in videos:
        # 将截取的图片分日期放入目标文件夹
        target_key = f"{date}/{video}/{{index}}.{{autoext}}"
        
        # 填入配置好的视频样式
        style = 'style/ipc-snap'
        
        # 对目标存储空间名和文件名进行URL安全的Base64编码
        target_bucket_base64 = base64.urlsafe_b64encode(args.target_bucket.encode()).decode()
        target_key_base64 = base64.urlsafe_b64encode(target_key.encode()).decode()
        
        # 构建处理指令，包含样式和保存位置，最后将转换结果通知发送到主题名称为test-topic的SMQ。
        process = f"{style}|sys/saveas,b_{target_bucket_base64},o_{target_key_base64}/notify,topic_dGVzdC10b3BpYw"
        
        try:
            # 发送异步处理请求
            result = client.async_process_object(oss.AsyncProcessObjectRequest(
                bucket=args.bucket,
                key=video,
                process=process
            ))
            
            print(f"Status Code: {result.status_code}")
            print(f"RequestId: {result.request_id}")
            print(f"TaskId: {result.task_id}")
            print(f"EventId: {result.event_id}")
            print(f"ProcessRequestId: {result.process_request_id}")
            
        except Exception as e:
            print(f"Error: {e}")

if __name__ == "__main__":
    video_processor()