Object Storage Service:Create an intelligent semantic indexing system for IP cameras

# -*- coding: utf-8 -*-
import base64
import argparse
import alibabacloud_oss_v2 as oss


def video_processor():
    # Configure basic OSS parameters.
    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'
    )
    
    # Configure an OSSClient instance.
    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)

    # Specify the video objects that you want to process.
    videos = [('Video A.mp4', '2025-04-07'), 
              ('Video B.mp4', '2025-04-07'), 
              ('Video C.mp4', '2025-04-07')]
    
    for video, date in videos:
        # Store the captured images in the destination directory by date.
        target_key = f"{date}/{video}/{{index}}.{{autoext}}"
        
        # Specify the configured style for frame capture.
        style = 'style/ipc-snap'
        
        # Base64-encode the destination bucket name and object name to make sure that they are URL-safe.
        target_bucket_base64 = base64.urlsafe_b64encode(args.target_bucket.encode()).decode()
        target_key_base64 = base64.urlsafe_b64encode(target_key.encode()).decode()
        
        # Create a process instruction that includes the style and destination path.
        process = f"{style}|sys/saveas,b_{target_bucket_base64},o_{target_key_base64}"
        
        try:
            # Send an asynchronous request.
            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
# Parse the response in the XML format.
import xml.etree.ElementTree as ET
import json 
from datetime import datetime 

def get_search_conditions():
    """Obtain the multiple conditions specified by the user."""
    print("Specify a semantic description (Example: yard with a car)")
    query = input("> Semantic keywords: ").strip()
    while not query:
        print("Semantic keywords cannot be empty!")
        query = input("> Semantic keywords: ").strip()
    return query  

def build_metaquery_xml(query):
    """Create an XML MetaQuery request that complies with OSS specifications"""
    xml_parts = [f'<Query>{query}</Query>']

    meta_query_xml = f'''<MetaQuery>
    {"".join(xml_parts)}
</MetaQuery>'''
    return meta_query_xml # Encode the results to a UTF-8 byte stream.


def format_result(key, pre_url):
    """Format a search result"""
    return f"""Object path: {key}
 Object URL: {pre_url}
-----------------------"""

def semantic_search():
    # Assign values to the command line parameters.
    args = argparse.Namespace(
        region='cn-beijing', # Specify the region in which the bucket is located.
        bucket='ipc-frames-oss-metaquery-demo', # Specify the name of the bucket.
        endpoint='https://oss-cn-beijing.aliyuncs.com', # Specify the endpoint of the region in which the bucket is located. If you do not require it, leave the parameter empty or delete the parameter.
    )

    # Initialize the OSSClient instance.
    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)

    # Obtain user configurations.
    query=get_search_conditions()# Query search conditions.
    # Create a request.
    try:
        # Create the XML request body.
        data_str=build_metaquery_xml (query) # Do not pass conditions.

        # Specify the operation inputs.
        req = oss.OperationInput(
            op_name='DoMetaQuery', # Specify the name of the custom operation.
            method='POST', # The HTTP method.
            parameters ={# The query parameters.
                'metaQuery': '',
                'mode': 'semantic',
                'comp': 'query',
            },
            headers=None, # Optional. The custom request headers.
            body=data_str.encode("utf-8"), # The request body, which is encoded as a UTF-8 byte stream.
            bucket=args.bucket,          # The name of the destination bucket.
        )

        # Call the operation.
        resp = client.invoke_operation(req)

    except oss.exceptions.ServiceError as e:
        print(f "Server error:{e.message}")
        return
    
    root = ET.fromstring(resp.http_response.content.decode('utf-8'))
     # Find all File elements.
    files = root.findall('.//File')
    
    print(f"\n A total of {len(objects)} match the conditions:")
    


    for i, file in enumerate(files, 1):
        # Query the name of the object.
        key_element = file.find('Filename')
        if key_element is None:
            continue # Skips the current loop iteration if no object name is returned.
        key = key_element.text
        
        
        # Generate a presigned URL.
        pre_url = client.presign(
            oss.GetObjectRequest(
                bucket=args.bucket,  # Specify the name of the bucket.
                key=key, # Specify the name of the object.
            )
        )
        
        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):
    """List and process all objects whose names contain a specific prefix"""
    try:
        # Create a request to list objects.
        list_request = oss.models.ListObjectsRequest(
            bucket=bucket,
            prefix=prefix
        )
        
        # Obtain and process the objects.
        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 "Failed to process the directory: {str(e)}")

def apply_tags_to_frame(client, bucket, frame_key, tags):
    """Add tags to a captured frame"""
    try:
        # Specify a tag.
        tagging = Tagging(
            version=1,
            tag_set=TagSet(tags=tags)
        )
        
        # Create a request to update tags.
        put_tag_request = PutObjectTaggingRequest(
            bucket=bucket,
            key=frame_key,
            tagging=tagging
        )
        
        # Update object tags.
        result = client.put_object_tagging(put_tag_request)
        
        # Convert the tags to a string for display.
        tags_str = '&'.join([f"{tag.key}={tag.value}" for tag in tags])
        print(f "Successfully add tags to the image named {frame_key}: {tags_str}")
        return True
    except Exception as e:
        print(f "Failed to add tags to the image: {str(e)}")
        return False

def frame_tags():
    # Configure basic OSS parameters.
    args = argparse.Namespace(
        region='cn-beijing',   
        frame_bucket='ipc-frames-oss-metaquery-demo',     
        endpoint='https://oss-cn-beijing.aliyuncs.com'
    )
    
    # Configure an OSSClient instance.
    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)

    # Specify the prefix contained in the name of the stored frames.
    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/'
        }
    ]

    # Process the frames of each video.
    for video in videos:


        # Extract the video name from frame_prefix. For example, extract 'Video A.mp4' from '2025-04-07/Video A.mp4/'.
        prefix_parts = video['frame_prefix'].strip('/').split('/')
        video_filename=prefix_parts [-1] if prefix_parts else '' # Obtain the last non-empty part. Example: Video A.mp4.

        # Remove the object name extension. For example, remove the object name extension of 'Video A.mp4' to obtain 'Video A'.
        video_name_base = video_filename.split('.')[0] if '.' in video_filename else video_filename

        print(f"\n Start processing the frames of the video named {video_filename}") # Display the original extracted object name.

        tags = [
            # Specify the camera name by using the last word of the video name with the extension removed.
            Tag(key='camera', value=f'camera-{video_name_base[-1].lower()}' if video_name_base else 'camera-unknown'),
            # Tag(key='category', value='video_monitoring') # Add a tag to the frame by using the actual business classification information.
        ]

        # Process all frames of the video.
        list_and_process_frames(client, args.frame_bucket, video['frame_prefix'], tags)

        print(f"\n Finish processing the frames of the video named {video_filename}") # Display the original extracted object name.

if __name__ == "__main__":
    frame_tags()

# -*- coding: utf-8 -*-
import argparse
import alibabacloud_oss_v2 as oss
# Parse the response in the XML format.
import xml.etree.ElementTree as ET
import json
from datetime import datetime
# --- Import required resources---
from alibabacloud_oss_v2.models import GetObjectTaggingRequest
# --- Import completed---

def get_inputs():
    """Obtain semantic descriptions (required) and filter conditions (optional)"""
    # 1. Obtain required semantic descriptions.
    print("Specify a semantic description (Example: yard with a car)")
    query = input("> Semantic keywords: ").strip()
    while not query:
        print("Semantic keywords cannot be empty!")
        query = input("> Semantic keywords: ").strip()

    conditions = [] # Initialize the filter conditions.
    target_tag=None# --- New: used to store destination tags.---

    # 2. (Optional) Obtain tag filtering conditions.
    print("\n  Specify the tag filtering conditions used for client filtering. (Example: camera=camera-a. Press Enter to skip.)") # Modify the prompt.
    tag_input=input ("> Tag (Format: key=value): ").strip()
    if tag_input:
        if '=' in tag_input:
            # --- Modify: no longer add to conditions, but save to target_tag ---
            # conditions.append({
            #     'field': 'Tags',
            #     'value': tag_input,
            #     'op': 'eq'
            # })
            target_tag = tag_input
            print(f "Information: The tag '{target_tag}' will be filtered on the client after the result is obtained. ")
            # --- Modification complete---
        else:
             print("The tag format is invalid and has been ignored. Format: 'key=value'.")


    # 3. (Optional) Obtain path prefix filtering conditions.
    print("\n (Optional) Enter the path prefix filter conditions. (Example: YYYY-MM-DD. Press Enter to skip.)")
    date_prefix=input ("> Path prefix: ").strip()
    if date_prefix:
        conditions.append({
            'field': 'Filename',
            'value': date_prefix,
            'op': 'prefix'
        })

    # --- Modify: return target_tag ---
    return query, conditions, target_tag
    # --- Modification complete---

def build_metaquery_xml(query, conditions):
    """Create an XML MetaQuery request that complies with OSS specifications (including semantic descriptions and optional filter conditions)."""
    # Always include the semantic query condition.
    xml_parts = [f'<Query>{query}</Query>']

    # Add optional filtering conditions. In semantic mode, only specific fields are supported, such as Filename in this example.
    for cond in conditions:
        # Skip the SimpleQuery construction of the Tags field in semantic mode. Although tags are no longer added in conditions, retaining this as a precaution).
        if cond['field'] == 'Tags':
            # print(f "Information: The tag filtering condition '{cond['value']}' is ignored in semantic search mode. ") # This line is no longer needed.
            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>')

    """The preceding configurations create an XML MetaQuery request"""
    meta_query_xml = f'''<MetaQuery>
    {"".join(xml_parts)}
</MetaQuery>'''
    return meta_query_xml

def format_result(key, pre_url):
    """Format a search result"""
    return f""" Object URL: {pre_url}
 Object path: {key}
-----------------------"""

def perform_search():
    # Assign values to the command line parameters.
    args = argparse.Namespace(
        region='cn-beijing', # Specify the region in which the bucket is located.
        bucket='ipc-frames-oss-metaquery-demo', # Specify the name of the bucket.
        endpoint='https://oss-cn-beijing.aliyuncs.com', # Specify the endpoint of the region in which the bucket is located.
    )

    # Initialize the OSSClient instance.
    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)

    # Obtain user configurations (semantic description, filter conditions, and destination tags).
    # --- Modify: receive target_tag ---
    query, conditions, target_tag = get_inputs()
    # --- Modification complete---

    # Create a request.
    try:
        # Create the XML request body.
        # --- Pass only conditions. ---
        data_str = build_metaquery_xml(query, conditions)
        
        print(f"\n Generate an XML MetaQuery request:\n{data_str}")

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

        # Call the operation.
        print("\n Send a DoMetaQuery request.")
        resp = client.invoke_operation(req)
        print(f"\n Request succeeded. HTTP status code: {resp.http_response.status_code}")
    # --- Modify exceptions ---
    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: # Capture client errors.
        print(f "Client or network error (ClientError):{e.message}")
        return
    except Exception as e: # Capture other possible exceptions.
        print(f "Unknown error: {e}")
        import traceback
        traceback.print_exc() # Display the complete traceback.
        return

    # --- Modification complete---

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

        print(f"\n Obtain {len(files)} preliminary matching results from OSS and start tag filtering...")

        if not files:
            # Check the NextContinuationToken parameter.
            next_token_elem = root.find('.//NextContinuationToken')
            if next_token_elem is not None and next_token_elem.text:
                print("Note: More results may be displayed in other pages.")
            print("\n No results found.")
            return # You are notified in advance if no objects are found.


        for i, file in enumerate(files, 1):
            # Query the name of the object.
            key_element = file.find('Filename')
            if key_element is None:
                print(f "Warning: The Filename field is not configured for the {i}th preliminary result. Skip the result.")
                continue
            key = key_element.text

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

                    # --- Modify: use .tags instead of .tag.---
                    if tagging_resp.tag_set and tagging_resp.tag_set.tags: # Use .tags.
                        for tag in tagging_resp.tag_set.tags: # Use .tags.
                    # --- Modification complete---
                            if tag.key == target_k and tag.value == target_v:
                                tag_found = True
                                break
                    if not tag_found:
                        continue # The tag does not match. Skip this object.
                except oss.exceptions.ServiceError as tag_err:
                     if tag_err.status_code == 404 and tag_err.error_code == 'NoSuchTagSet':
                         continue
                     else:
                        print(f "Warning: An error occurred while obtaining the tag of the object named '{key}': {tag_err.error_code} - {tag_err.message}. Skip the error.")
                        continue
                except Exception as tag_e:
                    print(f "Warning: An unknown error occurred while obtaining or processing the tag of the object named '{key}': {tag_e}. Skip the error.")
                    # --- Add traceback to facilitate debugging---
                    import traceback
                    traceback.print_exc()
                    # --- Add complete---
                    continue
            # --- Client tag filtering completed---

            # --- If the tag filter conditions are passed or not specified, process and display the result.---
            final_results_count += 1 # Add 1 to the final result.
            print(f"\n[{final_results_count}] file '{key}' meets all conditions:") # Display the final result.

            # Generate a presigned 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: An error occurred while generating the presigned URL of the object named '{key}': {presign_e}")
                print(format_result(key, "[Failed to generate the URL]"))
        # --- Display the final statistics after the loop ends.---
        print(f"\n Client filtering complete. {final_results_count} results found.")

    except ET.ParseError as xml_e:
        print(f" Error: An error occurred while parsing the response in the XML format - {xml_e}")
    except Exception as parse_e:
        print(f" Error: An unexpected error occurred while processing results- {parse_e}")


if __name__ == "__main__":
    perform_search()

# -*- coding: utf-8 -*-
import base64
import argparse
import alibabacloud_oss_v2 as oss


def video_processor():
    # Configure basic OSS parameters.
    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'
    )
    
    # Configure an OSSClient instance.
    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)

    # Specify the video objects that you want to process.
    videos = [('Video A.mp4', '2025-04-07'), 
              ('Video B.mp4', '2025-04-07'), 
              ('Video C.mp4', '2025-04-07')]
    
    for video, date in videos:
        # Store the captured images in the destination directory by date.
        target_key = f"{date}/{video}/{{index}}.{{autoext}}"
        
        # Specify the configured style for frame capture.
        style = 'style/ipc-snap'
        
        # Base64-encode the destination bucket name and object name to make sure that they are URL-safe.
        target_bucket_base64 = base64.urlsafe_b64encode(args.target_bucket.encode()).decode()
        target_key_base64 = base64.urlsafe_b64encode(target_key.encode()).decode()
        
        # Create a process instruction that includes the style and destination path. Send the conversion result notification to the SMQ with the topic named test-topic. 
        process = f"{style}|sys/saveas,b_{target_bucket_base64},o_{target_key_base64}/notify,topic_dGVzdC10b3BpYw"
        
        try:
            # Send an asynchronous request.
            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()