Smart Subtitle Access Tutorial

{
    "InputInfo": {
        "Type": "COS", 
        "CosInputInfo": {
            "Bucket": "facedetectioncos-125*****11", 
            "Region": "ap-guangzhou", 
            "Object": "/video/123.mp4"
        }
    }, 
    "ScheduleId": 12345, //Replace it with a custom orchestration ID. 12345 is a sample code and has no practical significance.
    "Action": "ProcessMedia", 
    "Version": "2019-06-12"
}

{
    "MediaProcessTask": {
        "TranscodeTaskSet": [
            {
                "Definition": 100040, //Transcoding template ID. It should be replaced with the transcoding template you need.
                "OverrideParameter": { //Overwriting parameters that are used for flexibly overwriting some parameters in the transcoding template.
                    "SubtitleTemplate": { //Subtitle application configuration.
                        "Path": "https://test-1234567.cos.ap-nanjing.myqcloud.com/mps_autotest/subtitle/1.vtt", 
                        "StreamIndex": 2, 
                        "FontType": "simkai.ttf", 
                        "FontSize": "10px", 
                        "FontColor": "0xFFFFFF", 
                        "FontAlpha": 0.9
                    }
                }
            }
        ]
    }, 
    "InputInfo": { //Input information.
        "Type": "URL", 
        "UrlInputInfo": {
            "Url": "https://test-1234567.cos.ap-nanjing.myqcloud.com/mps_autotest/subtitle/123.mkv"
        }
    }, 
    "OutputStorage": { //Output bucket.
        "Type": "COS", 
        "CosOutputStorage": {
            "Bucket": "test-1234567", 
            "Region": "ap-nanjing"
        }
    }, 
    "OutputDir": "/mps_autotest/output2/", //Output path.
    "Action": "ProcessMedia", 
    "Version": "2019-06-12"
}

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
import struct
import time
import os
import signal
import sys
import hashlib
import hmac
import random
from urllib.parse import urlencode, urlunsplit, quote
import websockets
import asyncio
import logging
import json

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AudioPacket:
    def __init__(self, format=1, is_end=False, timestamp=0, audio_src_id="123456", ext_data=b'', data=b''):
        self.format = format
        self.is_end = is_end
        self.timestamp = timestamp
        self.audio_src_id = audio_src_id
        self.ext_data = ext_data
        self.data = data

    def marshal(self):
        """Serialize audio packet to binary format"""
        header = struct.pack(
            '>BBQH', 
            self.format, 
            1 if self.is_end else 0,
            self.timestamp,
            len(self.audio_src_id)
        )
        audio_src_bytes = self.audio_src_id.encode('utf-8')
        ext_len = struct.pack('>H', len(self.ext_data))
        return header + audio_src_bytes + ext_len + self.ext_data + self.data

def sha256hex(s):
    """Calculate SHA256 hex digest"""
    if isinstance(s, str):
        s = s.encode('utf-8')
    return hashlib.sha256(s).hexdigest()

def hmacsha256(s, key):
    """Calculate HMAC-SHA256"""
    if isinstance(s, str):
        s = s.encode('utf-8')
    if isinstance(key, str):
        key = key.encode('utf-8')
    return hmac.new(key, s, hashlib.sha256).digest()

def generate_random_number(digits):
    """Generate random number with specified digits"""
    low = 10 ** (digits - 1)
    high = (10 ** digits) - 1
    return random.randint(low, high)

def generate_url_v3(args):
    """Generate WebSocket URL with TC3-HMAC-SHA256 signature"""
    query_params = {}
    if args.dstLang:
        query_params["transSrc"] = args.lang
        query_params["transDst"] = args.dstLang
    else:
        query_params["asrDst"] = args.lang
    
    query_params["fragmentNotify"] = "1" if args.frame else "0"
    query_params["timeoutSec"] = str(args.timeout)
    
    timestamp = int(time.time())
    expire_timestamp = timestamp + 3600
    
    query_params["timeStamp"] = str(timestamp)
    query_params["expired"] = str(expire_timestamp)
    query_params["secretId"] = args.secretId
    query_params["nonce"] = str(generate_random_number(10))
    
    # Sort keys and build canonical query string
    sorted_keys = sorted(query_params.keys())
    canonical_query = "&".join(
        ["{}={}".format(k, quote(query_params[k], safe='')) 
         for k in sorted_keys]
    )
    
    # Build canonical request
    path = "/wss/v1/{}".format(args.appid)
    http_method = "post"
    canonical_uri = path
    canonical_headers = "content-type:application/json; charset=utf-8\nhost:{}\n".format(args.addr)
    signed_headers = "content-type;host"
    
    canonical_request = "{}\n{}\n{}\n{}\n{}\n".format(
        http_method,
        canonical_uri,
        canonical_query,
        canonical_headers,
        signed_headers,
    )
    
    # Build string to sign
    date = time.strftime("%Y-%m-%d", time.gmtime(timestamp))
    credential_scope = "{}/mps/tc3_request".format(date)
    hashed_canonical = sha256hex(canonical_request)
    
    algorithm = "TC3-HMAC-SHA256"
    string_to_sign = "{}\n{}\n{}\n{}".format(
        algorithm,
        timestamp,
        credential_scope,
        hashed_canonical
    )
    
    # Calculate signature
    secret_date = hmacsha256(date, "TC3" + args.secretKey)
    secret_service = hmacsha256("mps", secret_date)
    secret_signing = hmacsha256("tc3_request", secret_service)
    signature = hmac.new(
        secret_signing, 
        string_to_sign.encode('utf-8'), 
        hashlib.sha256
    ).hexdigest()
    
    # Add signature to query params
    query_params["signature"] = signature
    
    # Build final URL
    scheme = "wss" if args.ssl else "ws"
    url = urlunsplit((
        scheme,
        args.addr,
        path,
        urlencode(query_params),
        ""
    ))
    return url

async def receive_messages(websocket, stop_event):
    """Handle incoming WebSocket messages"""
    try:
        while not stop_event.is_set():
            message = await websocket.recv()
            if isinstance(message, bytes):
                try:
                    message = message.decode('utf-8')
                except UnicodeDecodeError:
                    message = str(message)
            logger.info("Received: %s", message)
    except Exception as e:
        logger.info("Connection closed: %s", e)

async def run_client():
    parser = argparse.ArgumentParser()
    parser.add_argument("--addr", default="mps.cloud.tencent.com", help="websocket service address")
    parser.add_argument("--file", default="./wx_voice.pcm", help="pcm file path")
    parser.add_argument("--appid", default="121313131", help="app id")
    parser.add_argument("--lang", default="zh", help="language")
    parser.add_argument("--dstLang", default="", help="destination language")
    parser.add_argument("--frame", action="store_true", help="enable frame notify")
    parser.add_argument("--secretId", default="123456", help="secret id")
    parser.add_argument("--secretKey", default="123456", help="secret key")
    parser.add_argument("--ssl", action="store_true", help="use SSL")
    parser.add_argument("--timeout", type=int, default=10, help="timeout seconds")
    parser.add_argument("--wait", type=int, default=700, help="wait seconds after end")
    args = parser.parse_args()

    url = generate_url_v3(args)
    logger.info("Connecting to %s", url)

    try:
        # Python 3.6 compatible websockets connection
        websocket = await websockets.connect(url, ping_timeout=5)

        # Handle initial response
        initial_msg = await websocket.recv()
        try:
            result = json.loads(initial_msg)
            if result.get("Code", 0) != 0:
                logger.error("Handshake failed: %s", result.get("Message", ""))
                return
            logger.info("TaskId %s handshake success", result.get("TaskId", ""))
        except ValueError:  # json.JSONDecodeError not available in 3.6
            logger.error("Invalid initial message")
            return

        # Setup signal handler
        loop = asyncio.get_event_loop()
        stop_event = asyncio.Event()
        loop.add_signal_handler(signal.SIGINT, stop_event.set)

        # Start receiver
        receiver_task = asyncio.ensure_future(receive_messages(websocket, stop_event))

        # Audio processing
        try:
            with open(args.file, "rb") as fd:
                PCM_DUR_MS = 40
                pcm = bytearray(PCM_DUR_MS * 32)
                pkt = AudioPacket(data=pcm)
                is_end = False
                wait_until = 0

                while not stop_event.is_set():
                    if is_end:
                        if time.time() > wait_until:
                            logger.info("Finish")
                            break
                        await asyncio.sleep(0.1)
                        continue

                    # Read PCM data
                    n = fd.readinto(pkt.data)
                    if n < len(pkt.data):
                        pkt.is_end = True
                        is_end = True
                        wait_until = time.time() + args.wait

                    # Send audio packet
                    await websocket.send(pkt.marshal())
                    logger.info("Sent ts %d", pkt.timestamp)
                    pkt.timestamp += n // 32

                    await asyncio.sleep(PCM_DUR_MS / 1000)

        except IOError:  # FileNotFoundError not available in 3.6
            logger.error("Open file error: %s", args.file)
            return

        # Cleanup
        await asyncio.wait_for(receiver_task, timeout=1)
        await websocket.close()

    except Exception as e:
        logger.error("Connection error: %s", e)
        return

if __name__ == "__main__":
    # Python 3.6 compatible asyncio runner
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(run_client())
    finally:
        loop.close()