Object Storage Service：教程樣本：構建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()