Object Storage Service:Tutorial: Build an intelligent semantic search system for IPC devices

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):
    """Upload a video"""
    try:
        # Check if the file exists
        if not os.path.exists(video_config['path']):
            print(f'File does not exist: {video_config["path"]}')
            return False

        # Create an object to upload the file
        uploader = client.uploader()

        # Execute the upload request - pass the filepath parameter correctly
        result = uploader.upload_file(
            filepath=video_config['path'],  # Add the filepath parameter
            request=PutObjectRequest(
                bucket=bucket,
                key=video_config['key']
            )
        )

        print(f'Upload {video_config["key"]} successful:')
        print(f'status code: {result.status_code},'
              f' request id: {result.request_id}')
        return True
    except Exception as e:
        print(f'Upload {video_config["key"]} failed: {str(e)}')
        return False

def main():
    # OSS configuration
    args = argparse.Namespace(
        region='cn-beijing',
        bucket='ipc-videos-oss-metaquery-demo',
        endpoint='https://oss-cn-beijing.aliyuncs.com'
    )

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

    # Video file configuration, including different tags
    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'
        }
    ]

    # Upload all videos
    for video in videos:
        print(f"\nStart uploading video: {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
# Parse the XML response
import xml.etree.ElementTree as ET
import json 
from datetime import datetime 

def get_search_conditions():
    """Get multi-conditional input from the user"""
    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()
    return query  

def build_metaquery_xml(query):
    """Build a MetaQuery XML that complies with OSS specifications"""
    xml_parts = [f'<Query>{query}</Query>']
    # Add the MediaTypes tag, hard-coded to video here
    xml_parts.append('<MediaTypes><MediaType>video</MediaType></MediaTypes>')

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


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

def semantic_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. You can leave this empty or delete it if not needed.
    )

    # 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
    query = get_search_conditions()  # Get only the query
    # Build the request
    try:
        # Construct the request body (XML data)
        data_str = build_metaquery_xml(query)

        # Define the operation input
        req = oss.OperationInput(
            op_name='DoMetaQuery',  # Custom action name
            method='POST',            # HTTP method
            parameters={              # Query parameters
                'metaQuery': '',
                'mode': 'semantic',
                'comp': 'query',
            },
            headers=None,             # Custom request headers (optional)
            body=data_str.encode("utf-8"),  # Request body (encoded as a UTF-8 byte stream)
            bucket=args.bucket,       # Target bucket name
        )

        # Call the generic interface to execute 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 Found {len(files)} matching results:")
    


    for i, file in enumerate(files, 1):
        print(f"\nFile {i}:")

        # Fetch and print various properties
        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"  File name: {key}")

        size_element = file.find('Size')
        size = size_element.text if size_element is not None else 'N/A'
        print(f"  Size: {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: {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}")

        # You can add fetching and printing for more properties as needed, such as ImageHeight, ImageWidth, OSSStorageClass, etc.
        # 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}")

        # Generate a signed URL (if needed)
        if key != 'N/A': # Generate a URL only if the file name exists
             try:
                pre_url = client.presign(
                    oss.GetObjectRequest(
                        bucket=args.bucket,  # Specify the bucket name
                        key=key,        # Specify the object key
                    )
                )
                print(f"  File address (signed URL): {pre_url.url}")
             except Exception as e:
                 print(f"  Failed to generate signed URL: {e}")


        print("-" * 20) # Separator
        
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):
    """Upload a video with tags"""
    try:
        # Check if the file exists
        if not os.path.exists(video_config['path']):
            print(f'File does not exist: {video_config["path"]}')
            return False

        # Create an object to upload the file
        uploader = client.uploader()

        # Use the tagging string directly (encoding is not needed if keys and values are simple strings)
        tagging_str = f"camera={video_config['camera_tag']}"

        # Execute the upload request - carry tags directly
        result = uploader.upload_file(
            filepath=video_config['path'],
            request=PutObjectRequest(
                bucket=bucket,
                key=video_config['key'],
                tagging=tagging_str
            )
        )

        print(f'Upload {video_config["key"]} successful:')
        print(f'status code: {result.status_code},'
              f' request id: {result.request_id}')
        print(f'Tag added: {tagging_str}')
        return True
    except Exception as e:
        print(f'Upload {video_config["key"]} failed: {str(e)}')
        return False


def main():
    # OSS configuration
    args = argparse.Namespace(
        region='cn-beijing',
        bucket='ipc-videos-oss-metaquery-demo',
        endpoint='https://oss-cn-beijing.aliyuncs.com'
    )

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

    # Video file configuration, including different tags
    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'
        }
    ]

    # Upload all videos and add tags
    for video in videos:
        print(f"\nStart uploading video: {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):
    """Add tags to a video"""
    try:
        # Build the tagging object
        tagging = Tagging(
            version=1,
            tag_set=TagSet(tags=tags)
        )
        
        # Create the request to update tags
        put_tag_request = PutObjectTaggingRequest(
            bucket=bucket,
            key=frame_key,
            tagging=tagging
        )
        
        # Update the object tags
        result = client.put_object_tagging(put_tag_request)
        
        # Convert tags to a string for printing
        tags_str = '&'.join([f"{tag.key}={tag.value}" for tag in tags])
        print(f"Successfully added tags to {frame_key}: {tags_str}")
        return True
    except Exception as e:
        print(f"Failed to add tags: {str(e)}")
        return False

def frame_tags():
    # OSS configuration
    args = argparse.Namespace(
        region='cn-beijing',   
        frame_bucket='ipc-videos-oss-metaquery-demo',     
        endpoint='https://oss-cn-beijing.aliyuncs.com'
    )
    
    # Configure the OSS client
    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)

    # Configure your video names
    videos = [
        'VideoA.mp4',    
        'VideoB.mp4',
        'VideoC.mp4'
    ]

    # Process each video
    for video_filename in videos:
        print(f"\nStart processing tags for video {video_filename}") # Print the original extracted file name

        # Remove the file name extension, for example, 'VideoA.mp4' -> 'VideoA'
        # The sample code extracts the last character of the video file name as the camera identifier. This is for reference only. You can add tags based on your actual business requirements.
        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')       # Add actual business category information
        ]

        # Add tags to the video
        apply_tags_to_frame(client, args.frame_bucket, video_filename, tags)

        print(f"Finished processing tags for video {video_filename}\n") # Print the original extracted file name

if __name__ == "__main__":
    frame_tags()

# -*- 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()