Object Storage Service：Tutorial: Membangun sistem pencarian semantik cerdas untuk perangkat IPC

Pencarian dengan filter tag

Pencarian semantik mengembalikan semua video yang sesuai dengan deskripsi. Untuk mempersempit hasil lebih lanjut—misalnya, hanya mengambil rekaman halaman belakang dari kamera tertentu—terapkan filter tag setelah hasil pencarian awal dikembalikan.

Contoh berikut menggabungkan pencarian semantik dengan pemfilteran tag di sisi klien:

# -*- coding: utf-8 -*-
import argparse
import alibabacloud_oss_v2 as oss
# Parse the XML response
import xml.etree.ElementTree as ET
import json
from datetime import datetime
# --- New import ---
from alibabacloud_oss_v2.models import GetObjectTaggingRequest
# --- End of new import ---

def get_inputs():
    """Get the user's semantic query (required) and filter conditions (optional)"""
    # 1. Get the required semantic query
    print(" Enter a semantic description (example: a yard with a parked car)")
    query = input("> Semantic keywords: ").strip()
    while not query:
        print(" Semantic keywords cannot be empty!")
        query = input("> Semantic keywords: ").strip()

    conditions = [] # Initialize the list of filter conditions
    target_tag = None # --- Used to store the target tag ---

    # 2. Get the optional tag filter
    print("\n (Optional) Enter tag filter conditions (used for client-side filtering, example: camera=camera-a, press Enter to skip)")
    tag_input = input("> Tag (format 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" Info: The tag '{target_tag}' will be used for client-side filtering after getting the results.")
        else:
             print(" Incorrect tag format, ignored. Use the 'key=value' format.")

    # --- Return target_tag ---
    return query, conditions, target_tag

def build_metaquery_xml(query, conditions):
    """Build a MetaQuery XML that complies with OSS specifications (includes semantic query and optional filters)"""
    # Always include the semantic query part
    xml_parts = [f'<Query>{query}</Query>']

    # Add the media type restriction
    xml_parts.append('<MediaTypes><MediaType>video</MediaType></MediaTypes>')

    # Add optional filter conditions - in semantic mode, only some fields are supported (such as Filename here)
    for cond in conditions:
        # In semantic mode, skip building SimpleQuery for the Tags field (although we no longer add Tags to conditions, this is kept just in case)
        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>')

    # Combine into the complete MetaQuery XML
    meta_query_xml = f'''<MetaQuery>
    {"".join(xml_parts)}
</MetaQuery>'''
    return meta_query_xml

def format_result(key, pre_url):
    """Format a single search result"""
    return f""" File address: {pre_url}
 File path: {key}
-----------------------"""

def perform_search():
    # Assign values directly to command line arguments
    args = argparse.Namespace(
        region='cn-beijing',  # Replace with your region
        bucket='ipc-videos-oss-metaquery-demo',  # Replace with your bucket name
        endpoint='https://oss-cn-beijing.aliyuncs.com',  # Replace with your endpoint
    )

    # Initialize the OSS client
    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)

    # Get user input (semantic query + optional conditions + target tag)
    query, conditions, target_tag = get_inputs()

    # Build the request
    try:
        # Construct the request body (XML data)
        # --- Pass only conditions (without Tags) ---
        data_str = build_metaquery_xml(query, conditions)

        # Define the operation input
        req = oss.OperationInput(
            op_name='DoMetaQuery',
            method='POST',
            parameters={
                'metaQuery': '',
                'mode': 'semantic', # <-- Re-add semantic mode
                'comp': 'query',
            },
            headers=None,
            body=data_str.encode("utf-8"),
            bucket=args.bucket,
        )

        # Call the generic interface to execute the operation
        print("\n Sending DoMetaQuery request...")
        resp = client.invoke_operation(req)
        print(f"\n Request successful, HTTP status code: {resp.http_response.status_code}")
    except oss.exceptions.ServiceError as e:
        print(f" Server error (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: # Add catch for client errors
        print(f" Client or network error (ClientError): {e.message}")
        return
    except Exception as e: # Catch other possible exceptions
        print(f" Unknown error: {e}")
        import traceback
        traceback.print_exc() # Print the full traceback
        return


    # Parse and process the results...
    final_results_count = 0 # --- New: counter to tally the final matching results ---
    try:
        root = ET.fromstring(resp.http_response.content.decode('utf-8'))
        # Find all File elements
        files = root.findall('.//File')

        print(f"\n Got {len(files)} initial matching results from OSS, starting client-side tag filtering...")

        if not files:
            # Check for NextContinuationToken, paging may be needed
            next_token_elem = root.find('.//NextContinuationToken')
            if next_token_elem is not None and next_token_elem.text:
                print(" Note: There may be more results. The current implementation does not handle paging.")
            print("\n No initial matching results.")
            return # Return early if there are no files


        for i, file in enumerate(files, 1):
            # Get the file name
            key_element = file.find('Filename')
            if key_element is None:
                print(f" Warning: The {i}-th initial result is missing the Filename field, skipped.")
                continue
            key = key_element.text

            # --- Client-side tag filtering ---
            if target_tag:
                try:
                    tagging_req = GetObjectTaggingRequest(bucket=args.bucket, key=key)
                    tagging_resp = client.get_object_tagging(tagging_req)
                    # Check if the returned tag set contains the target tag
                    tag_found = False
                    target_k, target_v = target_tag.split('=', 1)

                    if tagging_resp.tag_set and tagging_resp.tag_set.tags: # Use .tags
                        for tag in tagging_resp.tag_set.tags: # Use .tags

                            if tag.key == target_k and tag.value == target_v:
                                tag_found = True
                                break
                    if not tag_found:
                        continue # Tag does not match, skip this file
                except oss.exceptions.ServiceError as tag_err:
                     if tag_err.status_code == 404 and tag_err.error_code == 'NoSuchTagSet':
                         continue
                     else:
                        print(f" Warning: Error getting tags for file '{key}': {tag_err.error_code} - {tag_err.message}, skipped.")
                        continue
                except Exception as tag_e:
                    print(f" Warning: Unknown error while getting or processing tags for file '{key}': {tag_e}, skipped.")
                    # --- Add traceback for debugging ---
                    import traceback
                    traceback.print_exc()
                    # --- End of addition ---
                    continue
            # --- End of client-side tag filtering ---

            # --- If it passed the tag filter (or no tag filter was set), process and print the result ---
            final_results_count += 1 # Increment the final result counter
            print(f"\n[{final_results_count}] File '{key}' matches all conditions:") # Print the final result number

            # Generate a signed 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" Warning: Error generating signed URL for file '{key}': {presign_e}")
                print(format_result(key, "[Could not generate URL]"))
        # --- Print final statistics after the loop ---
        print(f"\n Client-side filtering complete, found {final_results_count} final matching results.")

    except ET.ParseError as xml_e:
        print(f" Error: Error parsing OSS response XML - {xml_e}")
    except Exception as parse_e:
        print(f" Error: Unexpected error while processing results - {parse_e}")


if __name__ == "__main__":
    perform_search()

Untuk memfilter video halaman belakang dari kamera tertentu, jalankan program dan masukkan hal berikut:

1. Deskripsi semantik: a yard with a parked car

2. Filter tag: camera=camera-a

Dalam contoh ini, baik VideoA maupun VideoC berisi adegan yang sesuai dengan deskripsi tersebut. Filter tag hanya menyimpan VideoA (ditandai camera-a), sehingga hasil akhir hanya berisi satu file.

Contoh output:

Sending DoMetaQuery request...

Request successful, HTTP status code: 200

Got 2 initial matching results from OSS, starting client-side tag filtering...

[1] File 'VideoA.mp4' matches all conditions:
 File address: 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
 File path: VideoA.mp4
-----------------------

 Client-side filtering complete, found 1 final matching result.