Object Storage Service:IP カメラのインテリジェントなセマンティックインデックスシステムを作成する

# -*- 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'
    )
    
    # OSSClient インスタンスを構成します。
    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 = [('Video A.mp4', '2025-04-07'), 
              ('Video B.mp4', '2025-04-07'), 
              ('Video C.mp4', '2025-04-07')]
    
    for video, date in videos:
        # キャプチャされた画像を日付別に宛先ディレクトリに保存します。
        target_key = f"{date}/{video}/{{index}}.{{autoext}}"
        
        # 構成されたフレームキャプチャのスタイルを指定します。
        style = 'style/ipc-snap'
        
        # 宛先バケット名とオブジェクト名を Base64 エンコードして、URL セーフであることを確認します。
        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("セマンティックな説明を指定してください (例: yard with a car)")
    query = input("> セマンティックキーワード: ").strip()
    while not query:
        print("セマンティックキーワードは空にできません!")
        query = input("> セマンティックキーワード: ").strip()
    return query  

def build_metaquery_xml(query):
    """OSS 仕様に準拠した XML MetaQuery リクエストを作成します"""
    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}
 オブジェクト URL: {pre_url}
-----------------------"""

def semantic_search():
    # コマンドラインパラメーターに値を割り当てます。
    args = argparse.Namespace(
        region='cn-beijing', # バケットが配置されているリージョンを指定します。
        bucket='ipc-frames-oss-metaquery-demo', # バケットの名前を指定します。
        endpoint='https://oss-cn-beijing.aliyuncs.com', # バケットが配置されているリージョンのエンドポイントを指定します。 必要ない場合は、パラメーターを空のままにするか、パラメーターを削除します。
    )

    # OSSClient インスタンスを初期化します。
    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,          # 宛先バケットの名前。
        )

        # 操作を呼び出します。
        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(objects)} 件が条件に一致します:")
    


    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'
    )
    
    # OSSClient インスタンスを設定します。
    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/Video A.mp4/' 
        },    
        {
            'frame_prefix': '2025-04-07/Video B.mp4/'
        },
        {
            'frame_prefix': '2025-04-07/Video C.mp4/'
        }
    ]

    # 各ビデオのフレームを処理します。
    for video in videos:


        # frame_prefix からビデオ名を抽出します。たとえば、 '2025-04-07/Video A.mp4/' から 'Video A.mp4' を抽出します。
        prefix_parts = video['frame_prefix'].strip('/').split('/')
        video_filename=prefix_parts [-1] if prefix_parts else '' # 空でない最後の部分を取得します。例: Video A.mp4。

        # オブジェクト名の拡張子を削除します。たとえば、 'Video A.mp4' のオブジェクト名の拡張子を削除して 'Video 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"\n {video_filename} という名前のビデオのフレームの処理を終了します") # 抽出された元のオブジェクト名を表示します。

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。スキップするには Enter キーを押します。）") # プロンプトを変更します。
    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。スキップするには Enter キーを押します。）")
    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 仕様に準拠した XML MetaQuery リクエストを作成します（セマンティック記述とオプションのフィルタ条件を含む）。"""
    # 常にセマンティッククエリ条件を含めます。
    xml_parts = [f'<Query>{query}</Query>']

    # オプションのフィルタリング条件を追加します。セマンティックモードでは、この例のように Filename など、特定のフィールドのみがサポートされます。
    for cond in conditions:
        # セマンティックモードでは、Tags フィールドの SimpleQuery 構築をスキップします。（タグは conditions に追加されなくなりましたが、念のため残しておきます）。
        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>')

    """上記の構成により、XML MetaQuery リクエストが作成されます"""
    meta_query_xml = f'''<MetaQuery>
    {"".join(xml_parts)}
</MetaQuery>'''
    return meta_query_xml

def format_result(key, pre_url):
    """検索結果をフォーマットします"""
    return f""" オブジェクト URL: {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', # バケットが配置されているリージョンのエンドポイントを指定します。
    )

    # OSSClient インスタンスを初期化します。
    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 のみ渡します。---
        data_str = build_metaquery_xml(query, conditions)
        
        print(f"\n XML MetaQuery リクエストを生成します:\n{data_str}")

        # 操作入力パラメータを指定します。
        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)

                    # --- 変更：.tag の代わりに .tags を使用します。---
                    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 # 最終結果に 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" エラー: 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'
    )
    
    # OSSClient インスタンスを設定します。
    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 = [('Video A.mp4', '2025-04-07'), 
              ('Video B.mp4', '2025-04-07'), 
              ('Video C.mp4', '2025-04-07')]
    
    for video, date in videos:
        # キャプチャされたイメージを日付別に宛先ディレクトリに保存します。
        target_key = f"{date}/{video}/{{index}}.{{autoext}}"
        
        # フレームキャプチャ用に設定されたスタイルを指定します。
        style = 'style/ipc-snap'
        
        # 宛先バケット名とオブジェクト名を Base64 エンコードして、URL セーフであることを確認します。
        target_bucket_base64 = base64.urlsafe_b64encode(args.target_bucket.encode()).decode()
        target_key_base64 = base64.urlsafe_b64encode(target_key.encode()).decode()
        
        # スタイルと宛先パスを含む処理命令を作成します。 test-topic という名前の 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()