如何使用智能媒体管理图片盲水印功能_智能媒体管理(IMM)-阿里云帮助中心

通过图片盲水印技术，您可以在不影响原图视觉效果的前提下，向图片中嵌入不可见的文字信息。使用解析盲水印功能，可以恢复出图片中的水印信息。盲水印图片经过压缩、缩放、裁剪、旋转、色彩变换等攻击后，仍然可以提取出水印内容。

功能简介

图片盲水印功能演示示意图如下：

使用限制

仅支持文字水印，水印内容最多支持256个字符。
图片格式仅支持 jpg、png、bmp、tiff 和 webp。图片的长宽最小为 80px，最大为 10000px。
图片盲水印可抵抗一定程度的攻击，可抵抗的攻击手段包括截屏、裁剪、JPEG压缩、缩放、换色、饱和度变换、色调变换、亮度变换、少量涂鸦。
图片盲水印不支持纯黑或纯白图片，以及分辨率过低（<200px * 200px，非严格参数）的图片。同时，图片的短边与长边的比例应大于1:2。

应用场景

鉴权追责：判断己方的图片资源是否被恶意攻击方盗取。
上传查重：判断用户上传的图片资源是否来自已存在的资源库。
资源防泄漏：通过流传的资源图片中嵌入的盲水印得到泄漏方的相关信息。

前提条件

已创建并获取AccessKey。具体操作，请参见创建AccessKey。
已开通OSS服务、创建存储空间并上传文件到存储空间。具体操作，请参见控制台上传文件。
已开通智能媒体管理服务。具体操作，请参见开通产品。
已通过智能媒体管理控制台创建项目。具体操作，请参见创建项目。

说明

您也可以调用API接口创建项目。具体操作，请参见CreateProject - 创建项目。

您可以调用ListProjects - 查询所有项目信息接口列出指定地域下已创建的所有项目信息。

使用方法

添加图片盲水印

调用EncodeBlindWatermark接口为图片添加盲水印。

图片信息

IMM项目名称：test-project
待检测图片的存储地址：oss://test-bucket/test-object.jpg
图片示例：

请求示例

{
  "ProjectName": "test-project",
  "SourceURI": "oss://test-bucket/test-object.jpg",
  "TargetURI": "oss://test-bucket/blindwatermark-object.jpg",
  "Content": "阿里云版权所有",
}

返回示例

{
  "RequestId": "4FACCDBE-6D15-4E02-9924-1CC5FAF2A9B3",
}

示例代码

以Python SDK为例，添加图片盲水印的完整示例代码如下。

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import os
import sys

from typing import List

from alibabacloud_imm20200930.client import Client as imm20200930Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_imm20200930 import models as imm_20200930_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> imm20200930Client:
        """
        使用AK&SK初始化账号Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        # 工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例仅供参考。
        # 建议使用更安全的 STS 方式，更多鉴权访问方式请参见：https://help.aliyun.com/document_detail/378659.html。
        config = open_api_models.Config(
            # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID。,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # Endpoint 请参考 https://api.aliyun.com/product/imm
        config.endpoint = f'imm.cn-hangzhou.aliyuncs.com'
        return imm20200930Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        encode_blind_watermark_request = imm_20200930_models.EncodeBlindWatermarkRequest(
            project_name='test-project',
            source_uri='oss://test-bucket/test-object.jpg',
            target_uri='oss://test-bucket/blindwatermark-object.jpg',
            content='阿里云版权所有'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            client.encode_blind_watermark_with_options(encode_blind_watermark_request, runtime)
        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        encode_blind_watermark_request = imm_20200930_models.EncodeBlindWatermarkRequest(
            project_name='test-project',
            source_uri='oss://test-bucket/test-object.jpg',
            target_uri='oss://test-bucket/blindwatermark-object.jpg',
            content='阿里云版权所有'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            await client.encode_blind_watermark_with_options_async(encode_blind_watermark_request, runtime)
        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])

创建解析盲水印任务

调用CreateDecodeBlindWatermarkTask接口创建解析盲水印任务，任务完成后，可通过GetDecodeBlindWatermarkResult接口或从异步消息中获取盲水印结果。

重要

任务开始执行后，任务信息只保存7天，超过7天则无法再获取。您可以通过以下几种方式及时获取任务信息：

调用GetTask - 获取任务信息或ListTasks - 查询任务列表接口获取返回的TaskId，查看任务信息。
在与智能媒体管理相同的地域下开通MNS服务，并配置订阅关系，及时获取任务信息通知，异步通知消息格式请参见异步通知消息格式。关于MNS SDK的更多信息，请参见队列使用手册。
在与智能媒体管理相同的地域下开通RocketMQ服务，并创建RocketMQ 4.0实例、Topic、Group，及时获取任务信息通知，异步通知消息格式请参见异步通知消息格式。关于RocketMQ使用的更多信息，请参见调用HTTP协议的SDK收发普通消息。
在与智能媒体管理相同的地域下开通并接入事件总线EventBridge服务，及时获取任务信息通知。更多信息，请参见智能媒体管理IMM事件。

图片信息

IMM项目名称：test-project
待检测图片的存储地址：oss://test-bucket/blindwatermark-object.jpg
图片示例：

请求示例

{
  "ProjectName": "test-project",
  "SourceURI": "oss://test-bucket/blindwatermark-object.jpg",
}

返回示例

{
  "TaskId": "DecodeBlindWatermark-23dee5ea-d890-4f82-9357-fa6ad027b2a9",
  "RequestId": "79DD63F2-B72D-0E53-97AA-8B98C1758DBF",
  "EventId": "073-1oOm1dJjEUrCnQz0oFqtEwO7rJr"
}

示例代码

以Python SDK为例，解析盲水印任务的完整示例代码如下。

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import os
import sys

from typing import List

from alibabacloud_imm20200930.client import Client as imm20200930Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_imm20200930 import models as imm_20200930_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> imm20200930Client:
        """
        使用AK&SK初始化账号Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        # 工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例仅供参考。
        # 建议使用更安全的 STS 方式，更多鉴权访问方式请参见：https://help.aliyun.com/document_detail/378659.html。
        config = open_api_models.Config(
            # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID。,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # Endpoint 请参考 https://api.aliyun.com/product/imm
        config.endpoint = f'imm.cn-hangzhou.aliyuncs.com'
        return imm20200930Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        create_decode_blind_watermark_task_request = imm_20200930_models.CreateDecodeBlindWatermarkTaskRequest(
            project_name='test-project',
            source_uri='oss://test-bucket/blindwatermark-object.jpg'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            client.create_decode_blind_watermark_task_with_options(create_decode_blind_watermark_task_request, runtime)
        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        create_decode_blind_watermark_task_request = imm_20200930_models.CreateDecodeBlindWatermarkTaskRequest(
            project_name='test-project',
            source_uri='oss://test-bucket/blindwatermark-object.jpg'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            await client.create_decode_blind_watermark_task_with_options_async(create_decode_blind_watermark_task_request, runtime)
        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])