教程示例：构建IPC设备的智能语义检索系统

import argparse
import alibabacloud_oss_v2 as oss
from alibabacloud_oss_v2.models import ListObjectsRequest, PutObjectTaggingRequest, Tagging, TagSet, Tag, PutObjectRequest
import os

def upload_video(client, bucket, video_config):
    """上传视频"""
    try:
        # 检查文件是否存在
        if not os.path.exists(video_config['path']):
            print(f'文件不存在: {video_config["path"]}')
            return False

        # 创建一个用于上传文件的对象
        uploader = client.uploader()

        # 执行上传请求 - 正确传递filepath参数
        result = uploader.upload_file(
            filepath=video_config['path'],  # 添加filepath参数
            request=PutObjectRequest(
                bucket=bucket,
                key=video_config['key']
            )
        )

        print(f'上传 {video_config["key"]} 成功:')
        print(f'status code: {result.status_code},'
              f' request id: {result.request_id}')
        return True
    except Exception as e:
        print(f'上传 {video_config["key"]} 失败: {str(e)}')
        return False

def main():
    # OSS配置
    args = argparse.Namespace(
        region='cn-beijing',
        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 = [
        {
            'path': '<Your-Path>/videoa.mp4',
            'key': 'videoa.mp4'
        },
        {
            'path': '<Your-Path>/videob.mp4',
            'key': 'videob.mp4'
        },
        {
            'path': '<Your-Path>/videoc.mp4',
            'key': 'videoc.mp4'
        }
    ]

    # 上传所有视频
    for video in videos:
        print(f"\n开始上传视频: {video['key']}")
        upload_video(
            client=client,
            bucket=args.bucket,
            video_config=video
        )

if __name__ == "__main__":
    main()

# -*- 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>']
    # 添加 MediaTypes 标签，这里硬编码为 video
    xml_parts.append('<MediaTypes><MediaType>video</MediaType></MediaTypes>')

    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-videos-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):
        print(f"\n文件 {i}:")

        # 提取并打印各种属性
        uri_element = file.find('URI')
        uri = uri_element.text if uri_element is not None else 'N/A'
        print(f"  URI: {uri}")

        key_element = file.find('Filename')
        key = key_element.text if key_element is not None else 'N/A'
        print(f"  文件名: {key}")

        size_element = file.find('Size')
        size = size_element.text if size_element is not None else 'N/A'
        print(f"  大小: {size}")

        modified_time_element = file.find('FileModifiedTime')
        modified_time = modified_time_element.text if modified_time_element is not None else 'N/A'
        print(f"  修改时间: {modified_time}")

        content_type_element = file.find('ContentType')
        content_type = content_type_element.text if content_type_element is not None else 'N/A'
        print(f"  ContentType: {content_type}")

        media_type_element = file.find('MediaType')
        media_type = media_type_element.text if media_type_element is not None else 'N/A'
        print(f"  MediaType: {media_type}")

        # 可以根据需要添加更多属性的提取和打印，例如 ImageHeight, ImageWidth, OSSStorageClass 等
        # image_height_element = file.find('ImageHeight')
        # image_height = image_height_element.text if image_height_element is not None else 'N/A'
        # print(f"  ImageHeight: {image_height}")

        # 生成预签名URL (如果需要的话)
        if key != 'N/A': # 只有文件名存在时才生成URL
             try:
                pre_url = client.presign(
                    oss.GetObjectRequest(
                        bucket=args.bucket,  # 指定存储空间名称
                        key=key,        # 指定对象键名
                    )
                )
                print(f"  文件地址 (预签名URL): {pre_url.url}")
             except Exception as e:
                 print(f"  生成预签名URL失败: {e}")


        print("-" * 20) # 分隔符
        
if __name__ == "__main__":
    semantic_search()

import argparse
import alibabacloud_oss_v2 as oss
from alibabacloud_oss_v2.models import PutObjectRequest
import os

def upload_video_with_tags(client, bucket, video_config):
    """上传视频并携带标签"""
    try:
        # 检查文件是否存在
        if not os.path.exists(video_config['path']):
            print(f'文件不存在: {video_config["path"]}')
            return False

        # 创建一个用于上传文件的对象
        uploader = client.uploader()

        # 直接使用字符串形式的tagging（键值都为简单字符串时可不编码）
        tagging_str = f"camera={video_config['camera_tag']}"

        # 执行上传请求 - 直接携带标签
        result = uploader.upload_file(
            filepath=video_config['path'],
            request=PutObjectRequest(
                bucket=bucket,
                key=video_config['key'],
                tagging=tagging_str
            )
        )

        print(f'上传 {video_config["key"]} 成功:')
        print(f'status code: {result.status_code},'
              f' request id: {result.request_id}')
        print(f'已添加标签: {tagging_str}')
        return True
    except Exception as e:
        print(f'上传 {video_config["key"]} 失败: {str(e)}')
        return False


def main():
    # OSS配置
    args = argparse.Namespace(
        region='cn-beijing',
        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 = [
        {
            'path': '<Your-Path>/videoa.mp4',
            'key': 'videoa.mp4',
            'camera_tag': 'camera-a'
        },
        {
            'path': '<Your-Path>/videob.mp4',
            'key': 'videob.mp4',
            'camera_tag': 'camera-b'
        },
        {
            'path': '<Your-Path>/videoc.mp4',
            'key': 'videoc.mp4',
            'camera_tag': 'camera-c'
        }
    ]

    # 上传所有视频并添加标签
    for video in videos:
        print(f"\n开始上传视频: {video['key']}")
        upload_video_with_tags(
            client=client,
            bucket=args.bucket,
            video_config=video
        )


if __name__ == "__main__":
    main()

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


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-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 = [
        'videoa.mp4',    
        'videob.mp4',
        'videoc.mp4'
    ]

    # 处理每个视频
    for video_filename in videos:
        print(f"\n开始处理视频 {video_filename} 的打标") # 打印原始提取的文件名

        # 去掉文件扩展名，例如 'videoa.mp4' -> 'videoa'
        # 示例代码为提取视频文件最后一个字符作为相机标识，仅供参考，您可以根据您的实际业务需求添加标签
        video_name_base = video_filename.split('.')[0] if '.' in video_filename else 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')       # 添加实际业务分类信息
        ]

        # 为视频打标
        apply_tags_to_frame(client, args.frame_bucket, video_filename, 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.append({
            #     'field': 'Tags',
            #     'value': tag_input,
            #     'op': 'eq'
            # })
            target_tag = tag_input
            print(f" 信息：标签 '{target_tag}' 将在获取结果后于客户端进行过滤。")
        else:
             print(" 标签格式不正确，已忽略。请使用 'key=value' 格式。")

    # --- 返回 target_tag ---
    return query, conditions, target_tag

def build_metaquery_xml(query, conditions):
    """构建符合OSS规范的MetaQuery XML (包含语义查询和可选筛选)"""
    # 始终包含语义查询部分
    xml_parts = [f'<Query>{query}</Query>']
    
    # 添加媒体类型限制
    xml_parts.append('<MediaTypes><MediaType>video</MediaType></MediaTypes>')
    
    # 添加可选的筛选条件 - 在语义模式下，仅支持部分字段（如此处的 Filename）
    for cond in conditions:
        # 在 semantic 模式下跳过 Tags 字段的 SimpleQuery 构建 (虽然我们已经不在 conditions 里加 Tags 了，但保留以防万一)
        if cond['field'] == 'Tags':
            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-videos-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, conditions, target_tag = get_inputs()

    # 构建请求
    try:
        # 构造请求体（XML 数据）
        # --- 只传入 conditions (不含Tags) ---
        data_str = build_metaquery_xml(query, conditions)

        # 定义操作输入
        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)

                    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()