Object Storage Service:チュートリアル：IPC デバイス向けスマートな意味検索システムの構築

タグフィルターによる検索

意味検索では、説明に一致するすべての動画が返されます。結果をさらに絞り込むには（例：特定のカメラで撮影された裏庭の映像のみを取得）、初期検索結果が返された後にタグフィルターを適用します。

以下の例では、意味検索とクライアント側でのタグフィルターを組み合わせています：

# -*- 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。スキップする場合は Enter を押してください）")
    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:
        # 意味モードでは 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',  # ご利用のエンドポイントに置き換え
    )

    # 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 データ）の構築
        # --- Tags を含まない conditions のみを渡す ---
        data_str = build_metaquery_xml(query, conditions)

        # 操作入力の定義
        req = oss.OperationInput(
            op_name='DoMetaQuery',
            method='POST',
            parameters={
                'metaQuery': '',
                'mode': '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 ステータスコード: {e.status_code}")
        print(f"   - エラーコード: {e.error_code}")
        print(f"   - リクエスト 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() # トレースバックを完全に出力
        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}。スキップします。")
                    # --- デバッグ用にトレースバックを追加 ---
                    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()

特定のカメラで撮影された裏庭の映像をフィルターするには、プログラムを実行し、以下の内容を入力します。

1. 意味説明： a yard with a parked car

2. タグフィルター： camera=camera-a

この例では、VideoA と VideoC の両方に説明に一致するシーンが含まれています。タグフィルターにより、VideoA（camera-a というタグが付与）のみが残り、最終結果には 1 件のファイルが含まれます。

出力例：

DoMetaQuery リクエストを送信中...

リクエストが成功しました。HTTP ステータスコード: 200

OSS から 2 件の初期一致結果が得られました。クライアント側タグフィルターを開始します...

[1] ファイル 'VideoA.mp4' がすべての条件に一致しました:
 ファイルアドレス: https://ipc-videos-oss-metaquery-demo.oss-cn-beijing.aliyuncs.com/VideoA.mp4?x-oss-signature-version=OSS4-HMAC-SHA256&x-oss-date=20250526T054908Z&x-oss-expires=900&x-oss-credential=LTAI********************%2Fcn-beijing%2Foss%2Faliyun_v4_request&x-oss-signature=01bbf29790763d8e0f177d4cb0469cb00ae1c69d565219edb3866f75110b37ab
 ファイルパス: VideoA.mp4
-----------------------

クライアント側フィルター処理が完了しました。最終的な一致結果は 1 件です。