如何通过Python SDK实现Azure Media Service视频内容的下载交付

该需求完全可以通过Python调用Azure Media Service的SDK或REST API实现，下面给出具体可落地的实现方案：

前置准备 #

• 先在Azure Portal创建Media Service账户，获取订阅ID、资源组名、AMS账户名、租户ID、客户端ID、客户端密钥等核心认证信息
• 若使用SDK方案，需提前安装依赖包：pip install azure-mgmt-media azure-identity requests
• 若使用REST API方案，只需安装基础依赖：pip install requests

方案1：使用Azure Python SDK实现 #

实现逻辑：先完成Azure身份认证，再获取目标视频资产的下载定位器，生成SAS下载链接后写入本地即可，示例代码如下：

from azure.identity import ClientSecretCredential
from azure.mgmt.media import AzureMediaServices
import requests

# 以下参数替换为自身账户对应值
TENANT_ID = "你的租户ID"
CLIENT_ID = "你的客户端ID"
CLIENT_SECRET = "你的客户端密钥"
SUBSCRIPTION_ID = "你的订阅ID"
RESOURCE_GROUP_NAME = "你的资源组名"
AMS_ACCOUNT_NAME = "你的AMS账户名"
TARGET_ASSET_NAME = "待下载的视频资产名称"
LOCAL_SAVE_PATH = "./本地存储的视频文件名.mp4"

# 初始化认证和AMS客户端
credential = ClientSecretCredential(TENANT_ID, CLIENT_ID, CLIENT_SECRET)
ams_client = AzureMediaServices(credential, SUBSCRIPTION_ID)

# 获取资产关联的下载定位器，无现成定位器可调用create接口临时创建
locators = ams_client.assets.list_streaming_locators(
    resource_group_name=RESOURCE_GROUP_NAME,
    account_name=AMS_ACCOUNT_NAME,
    asset_name=TARGET_ASSET_NAME
)
download_locator = next(l for l in locators.streaming_locators if l.streaming_locator_type == "Download")

# 生成视频下载链接
paths = ams_client.streaming_locators.list_paths(
    resource_group_name=RESOURCE_GROUP_NAME,
    account_name=AMS_ACCOUNT_NAME,
    streaming_locator_name=download_locator.name
)
download_url = paths.download_paths[0].paths[0]

# 流式下载到本地
resp = requests.get(download_url, stream=True)
with open(LOCAL_SAVE_PATH, "wb") as f:
    for chunk in resp.iter_content(chunk_size=1024*1024):
        if chunk:
            f.write(chunk)

print("视频下载完成")

方案2：使用Azure REST API实现 #

无需安装AMS专属SDK，直接调用HTTP接口实现，示例代码如下：

import requests

# 以下参数替换为自身账户对应值
TENANT_ID = "你的租户ID"
CLIENT_ID = "你的客户端ID"
CLIENT_SECRET = "你的客户端密钥"
SUBSCRIPTION_ID = "你的订阅ID"
RESOURCE_GROUP_NAME = "你的资源组名"
AMS_ACCOUNT_NAME = "你的AMS账户名"
TARGET_ASSET_NAME = "待下载的视频资产名称"
LOCAL_SAVE_PATH = "./本地存储的视频文件名.mp4"

# 第一步：获取Azure接口访问令牌
token_url = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/token"
token_data = {
    "grant_type": "client_credentials",
    "client_id": CLIENT_ID,
    "client_secret": CLIENT_SECRET,
    "resource": "https://management.azure.com"
}
token_resp = requests.post(token_url, data=token_data)
access_token = token_resp.json()["access_token"]
req_headers = {"Authorization": f"Bearer {access_token}"}

# 第二步：获取目标资产的下载定位器
list_locator_url = f"https://management.azure.com/subscriptions/{SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP_NAME}/providers/Microsoft.Media/mediaServices/{AMS_ACCOUNT_NAME}/assets/{TARGET_ASSET_NAME}/listStreamingLocators?api-version=2023-01-01"
locator_resp = requests.post(list_locator_url, headers=req_headers)
download_locator = next(l for l in locator_resp.json()["streamingLocators"] if l["streamingLocatorType"] == "Download")

# 第三步：获取视频下载路径
list_path_url = f"https://management.azure.com/subscriptions/{SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP_NAME}/providers/Microsoft.Media/mediaServices/{AMS_ACCOUNT_NAME}/streamingLocators/{download_locator['name']}/listPaths?api-version=2023-01-01"
path_resp = requests.post(list_path_url, headers=req_headers)
download_url = path_resp.json()["downloadPaths"][0]["paths"][0]

# 第四步：下载到本地
resp = requests.get(download_url, stream=True)
with open(LOCAL_SAVE_PATH, "wb") as f:
    for chunk in resp.iter_content(chunk_size=1024*1024):
        if chunk:
            f.write(chunk)

常见注意事项 #

• 下载定位器默认有过期时间，创建时可自定义过期时长，最长支持设置为1年
• 若目标视频资产配置了加密，需先完成解密流程才能正常下载播放
• 大文件下载建议自行添加断点续传逻辑，避免网络波动导致下载失败

内容的提问来源于stack exchange，提问作者youpiyoful