文章目录

1、事件起因2、相关回复3、七牛云简介3.1 主要服务3.2 七牛云的优势3.3 常见应用场景3.4 总结

4、数美简介4.1 核心服务4.2 技术优势4.3 常见应用场景4.4 总结

5、我能做什么5.1. 七牛云图片上传5.2. 调用数美审核图片5.3 步骤详解5.3.1 安装依赖库5.3.2 上传图片到七牛云5.3.3 调用数美内容审核 API

5.4 代码示例5.5 代码说明5.6 审核结果处理5.7 提示

6、项目demo实现6.1 功能介绍6.2 核心功能6.3 优化与并发6.4 测试过程6.4.1 处理数据【130+张图】6.4.2 合规检测结果【部分】

6.5 上传图片耗时优化【图片压缩】6.5.1 上传图片耗时优化前【部分】6.5.2 上传图片耗时优化后【部分】6.5.3 上传图片耗时优化前后对比

1、事件起因

2、相关回复

3、七牛云简介

七牛云（Qiniu Cloud） 是一家中国的领先云服务提供商，成立于 2011 年，专注于为开发者和企业提供一站式的数据管理和服务解决方案。七牛云提供的服务涵盖对象存储、数据处理、内容分发加速、视频云、数据分析等多个领域，帮助用户应对海量数据存储和处理的需求。

3.1 主要服务

对象存储（Kodo）

七牛云的对象存储服务（Kodo）是其核心产品之一。它允许用户上传和存储海量数据，如图片、视频、文档等，并提供高效的访问控制和数据管理工具。数据可以通过 HTTP/HTTPS 协议访问，适用于静态网站托管、应用数据存储、备份等场景。

内容分发网络（CDN）

七牛云的 CDN 服务提供全球范围内的内容分发加速，帮助企业将静态和动态内容快速、安全地传递给全球用户。CDN 网络具备智能调度和缓存功能，可以有效提高网站和应用的响应速度，降低延迟。

数据处理

七牛云提供了一系列数据处理服务，包括图像处理（如缩放、裁剪、格式转换）、视频处理（如转码、拼接、截图）、文件转换（如 PDF 转换）等。用户可以通过 API 接口实时调用这些功能，满足各种复杂的业务需求。

视频云

七牛云的视频云服务支持音视频的直播、点播和处理。它提供从媒体上传、转码、存储、分发到播放的一站式解决方案，并具备直播转码、录制、实时剪辑、AI 智能识别等功能，适合视频媒体、教育、娱乐等行业。

AI 和数据分析

七牛云还为企业提供了丰富的 AI 和数据分析服务，例如图像识别、自然语言处理、大数据分析、物联网数据处理等。结合七牛的云存储和计算能力，用户可以轻松地进行海量数据的智能处理和分析。

安全与合规

七牛云提供了全面的安全防护服务，包括 DDoS 防护、Web 应用防火墙（WAF）、数据加密、访问控制等功能，确保用户数据的安全性和隐私性。此外，七牛云遵循国际化的合规标准，帮助企业应对全球各地的合规需求。

3.2 七牛云的优势

高可用性与可靠性

七牛云的数据存储和服务拥有多重备份机制，确保数据的高可用性和持久性。其 CDN 网络覆盖全球，保证服务的可靠交付。

强大的 API 支持

七牛云的所有服务都提供丰富的 API 接口，开发者可以方便地集成其服务到自己的应用或系统中，实现自动化管理和数据处理。

灵活的扩展性

七牛云支持弹性扩展，无论是个人开发者的小型项目，还是企业级的大型应用，都可以根据需求动态调整资源，适应业务的发展。

完善的开发者支持

七牛云为开发者提供了详细的文档、SDK、示例代码和技术支持，帮助他们快速上手并高效使用云服务。七牛云支持多种编程语言，包括 Python、Java、Go、PHP、Node.js 等。

3.3 常见应用场景

媒体与娱乐：视频直播与点播、视频编辑与发布、图片处理与存储。电商与零售：商品图片存储与处理、促销视频分发、海量用户数据管理。在线教育：课件视频的录制、存储与分发，在线课堂直播、数据分析等。智能设备：物联网数据的收集、存储与分析，智能设备的远程控制与管理。

3.4 总结

七牛云作为一家综合性的云服务平台，通过其强大的数据存储和处理能力、全球内容分发加速以及丰富的开发者支持，成为众多企业和开发者的首选。它广泛应用于媒体、教育、电商、金融等行业，帮助企业应对大规模数据处理、存储及内容分发的挑战。

4、数美简介

数美是一家专注于内容安全与合规审核技术的公司，成立于中国，致力于为各类企业提供智能化的内容审核和风控服务。数美主要通过人工智能技术为互联网平台、社交媒体、电商、广告等领域提供专业的内容审核解决方案，确保平台的内容安全、合规，并维护良好的用户体验。

4.1 核心服务

内容安全审核

数美通过先进的机器学习和自然语言处理技术，提供对图片、视频、文本、音频等多媒体内容的审核服务，识别并过滤非法、不良或敏感内容，如暴力、色情、涉政、仇恨言论等。其审核系统可以实时处理大量内容，确保内容的及时性和安全性。

AI 智能识别

数美的信息审核系统基于深度学习和计算机视觉等 AI 技术，能够识别图片和视频中的违法元素，文本中的敏感词汇，甚至音频中的敏感语句。它结合大规模训练数据和不断迭代的算法，有效提升内容审核的精度和覆盖面。

内容风控

除了审核不良内容，数美还提供内容风控服务，帮助企业监控和应对内容风险。通过数据分析和智能风控引擎，预警潜在风险内容，并提供相应的合规建议，帮助企业降低运营风险。

定制化服务

数美为不同行业提供定制化的内容审核解决方案。无论是社交媒体平台、电商网站、广告投放平台，还是教育、游戏等垂直领域，数美都能够根据行业特点和法规要求，定制适合的内容审核策略与风控模型。

人工与 AI 结合

在数美的内容审核服务中，AI 自动审核和人工复核相结合。对于复杂或边界模糊的内容，AI 系统将自动筛选出可能存在问题的内容，并交由人工审核员进行进一步复核，从而保证审核的准确性和可靠性。

4.2 技术优势

高效的内容审核引擎

数美依靠其高效的算法和分布式架构，能够支持海量数据的实时处理，适用于内容高频更新的大型互联网平台。同时，审核系统可根据不同内容类别进行分类处理，提升审核的效率与精度。

多媒体审核覆盖

数美不仅支持对文本内容的审核，还包括图片、视频和音频等多媒体内容的审核，适用于各类场景，如社交平台的用户生成内容（UGC）审核、电商平台的商品图片审核、广告内容的合规性审核等。

高准确率与低误判率

得益于深度学习模型的不断优化与海量数据的训练，数美的 AI 审核系统在内容审核中的识别准确率非常高，同时误判率低。这对需要大规模审核内容的平台尤为重要，有效降低人工审核成本的同时保证合规性。

灵活的部署方式

数美的信息审核服务可以根据客户的需求灵活部署，支持公有云、私有云、以及本地化部署。企业可以根据自身需求和合规要求，选择适合的部署方案。

4.3 常见应用场景

社交媒体：社交平台需要对用户上传的文字、图片、视频等内容进行严格审核，以防止不良内容的传播。数美的审核系统帮助这些平台实现内容合规和社区安全管理。

电商平台：商品描述、图片等内容审核是电商平台面临的挑战。数美可以帮助电商平台审核商品是否包含违规内容，确保广告合规性及用户体验。

在线广告：广告审核是一个敏感领域，广告内容需要符合国家法律法规和道德标准。数美帮助广告投放平台在大规模投放前快速审核广告的合规性。

视频与直播平台：视频与直播平台面临大量实时内容的审核需求。数美的审核系统能够实时检测和过滤违规内容，确保平台的合规运营。

4.4 总结

科技作为内容安全和合规领域的专家，依靠先进的人工智能技术，提供全面的内容审核和风险控制服务。通过与平台的无缝集成，数美帮助企业在保障内容合规和提升用户体验的同时，减少违规内容带来的法律和声誉风险。

5、我能做什么

当然其他内容格式也是大同小异，比如文本，音频，视频==，不局限于图片

5.1. 七牛云图片上传

首先使用七牛云 SDK 将图片上传到存储空间中。你可以参考之前的七牛上传代码，将图片上传至七牛云。

5.2. 调用数美审核图片

上传完成后，通过调用数美的内容审核 API，对图片进行合规性审核。数美的审核服务通常会包括对色情、暴力、政治敏感内容等的检测。

5.3 步骤详解

5.3.1 安装依赖库

确保安装了七牛云的 SDK 和 HTTP 请求库（如 <code>requests）。

pip install qiniu requests

5.3.2 上传图片到七牛云

使用 put_file 函数，将图片上传到七牛云。

5.3.3 调用数美内容审核 API

假设你已经获得了数美的 API 密钥，可以使用 requests 库来调用审核接口。一般审核的步骤如下：

获取图片的 URL（存储在七牛云上）。使用 HTTP 请求发送图片 URL 给数美审核 API。获取审核结果，判断图片是否合规。

5.4 代码示例

import os

import requests

from qiniu import Auth, put_file

from PIL import Image

from io import BytesIO

# 上传图片到七牛云

def upload_file_to_qiniu(file_path, bucket_name, file_key):

# 七牛云配置

access_key = 'your_access_key'

secret_key = 'your_secret_key'

q = Auth(access_key, secret_key)

# 获取上传Token

token = q.upload_token(bucket_name, file_key, 3600)

# 上传文件

ret, info = put_file(token, file_key, file_path)

# 检查上传结果

if info.status_code == 200:

return f"http://{ bucket_name}.your_domain.com/{ file_key}"

else:

print(f"上传失败: { info}")

return None

# 调用数美审核API

def review_image_with_meisu(image_url):

api_url = "https://api.meisuapi.com/v1/image/review"

headers = {

"Authorization": "Bearer your_meisu_api_key", # 替换为你的数美API密钥

"Content-Type": "application/json"

}

payload = {

"url": image_url,

"type": "porn,violence,politics" # 可根据需要调整审核类型

}

response = requests.post(api_url, json=payload, headers=headers)

if response.status_code == 200:

review_data = response.json()

if review_data["code"] == 0:

return review_data["data"]

else:

print(f"审核失败: { review_data}")

return None

else:

print(f"HTTP 请求失败: { response.status_code}")

return None

# 示例调用

bucket_name = "your_bucket_name"

file_path = "path_to_your_image.jpg"

file_key = os.path.basename(file_path)

# 上传图片到七牛云

image_url = upload_file_to_qiniu(file_path, bucket_name, file_key)

if image_url:

print(f"图片已上传到七牛云: { image_url}")

# 调用数美进行内容审核

review_result = review_image_with_meisu(image_url)

if review_result:

print("审核结果: ", review_result)

else:

print("审核失败")

else:

print("图片上传失败")

5.5 代码说明

upload_file_to_qiniu：上传图片到七牛云，返回图片的 URL。review_image_with_meisu：将图片 URL 发送给数美的审核接口，审核内容类型可以根据需求定制（例如色情、暴力、政治等）。API 密钥管理：请确保使用安全的方式存储和管理七牛云和数美的 API 密钥。

5.6 审核结果处理

审核结果通常会返回一组包含审核状态、内容类别以及得分的 JSON 数据。你可以根据这些结果来判断图片是否合规，并采取相应的措施（如阻止不合规图片的使用，或者将不合规图片移动到专门的文件夹中）。

5.7 提示

七牛和数美的 API 请求都有速率限制，因此在大量图片审核时，请注意控制请求频率，或者结合多线程、异步处理进行优化。图片 URL 必须是公开的才能让数美的审核服务访问到，因此需要确保图片 URL 没有访问权限问题。

通过将图片存储与内容审核服务结合，你可以构建一个自动化的图片合规检测系统。

6、项目demo实现

import requests

import json

import os

import shutil

import time

import pandas as pd

from qiniu import put_file

from concurrent.futures import ThreadPoolExecutor

from PIL import Image

from io import BytesIO

# 定义一个空的列表，用于存储每个图片的处理时间数据

timing_data = []

def compress_image(file_path, max_size=256 * 1024, quality=50):

"""

压缩图片，保持文件大小不超过 max_size（字节）。

使用 Pillow 库进行压缩。

"""

# 提前检查文件大小，避免不必要的操作

file_size = os.path.getsize(file_path)

if file_size <= max_size:

# 如果文件大小已经小于 max_size，直接返回原路径

return file_path

# 打开图片

img = Image.open(file_path)

# 检查是否是 RGBA 模式，如果是则转换为 RGB 模式

if img.mode == 'RGBA':

img = img.convert('RGB')

output = BytesIO()

# 初始压缩

img.save(output, format="JPEG", quality=quality)code>

output_size = output.tell()

# 调整质量以继续压缩，直到文件大小小于 max_size

while output_size > max_size and quality > 10:

quality -= 5

output = BytesIO() # 重置 BytesIO 对象

img.save(output, format="JPEG", quality=quality)code>

output_size = output.tell()

# 保存压缩后的图片到临时路径

temp_compressed_path = file_path + ".compressed.jpg"

with open(temp_compressed_path, "wb") as f:

f.write(output.getvalue())

return temp_compressed_path

def get_upload_token(fileName="test.txt"):code>

url = "替换实际七牛URL"

headers = { 'Content-Type': 'application/json'}

data = {

"appKey": "app-screenshot-review",

"fileName": fileName

}

try:

# 发送 POST 请求

response = requests.post(url, headers=headers, data=json.dumps(data))

response.raise_for_status() # 会自动抛出 HTTPError 异常

# 解析响应

response_data = response.json()

data = response_data.get('data', { })

cached_token = data.get('token')

cached_key = data.get('resourceName')

if cached_token and cached_key:

return cached_token, cached_key

else:

print("Error: Upload token or key is missing in the response.")

return "", ""

except requests.RequestException as e:

print(f"RequestException: { e}")

return "", ""

except json.JSONDecodeError:

print("Error: Failed to decode JSON response.")

return "", ""

def upload_file(upload_token, key, filePath):

try:

# 上传文件

ret, info = put_file(up_token=upload_token, key=key, file_path=filePath, version='v2')code>

# 检查上传是否成功

if info.status_code == 200:

# 获取文件 URL

file_url = ret.get('data', { }).get('url')

if file_url:

# 将 URL 从 HTTPS 转换为 HTTP

return file_url.replace("https://", "http://")

else:

print("Error: 'url' not found in the response data.")

return ""

else:

print(f"Error: Upload failed with status code { info.status_code}.")

return ""

except Exception as e:

print(f"Exception during file upload: { e}")

return ""

def check_pic_pass(file_url):

url = "替换实际数美URL"

# 构造请求参数

data = {

"account": "servertest",

"packageName": "com.xxx.xxx",

"imageUrl": file_url

}

try:

response = requests.post(url, data=data)

if response.status_code == 200:

response_data = response.json()

print(f"Response data: { response_data}")

reviewMsg = response_data.get('data', { }).get('reviewMsg')

return reviewMsg

else:

print("Failed to call API. Status code:%s", response.status_code)

return ""

except Exception as e:

print("Exception:", e)

return ""

def process_single_image(file_path, error_folder):

filename = os.path.basename(file_path)

# 记录开始时间

start_time = time.time()

# 记录压缩图片的开始时间

compress_start_time = time.time()

# 压缩大于1MB的图片

file_path = compress_image(file_path)

# 记录压缩图片的结束时间并打印耗时

compress_end_time = time.time()

compress_time = compress_end_time - compress_start_time

print(f"Step 0 - 压缩图片'{ filename}': { compress_time:.2f} 秒")

# 1、获取上传 Token 和 Key

token_start_time = time.time()

token, key = get_upload_token(file_path)

token_end_time = time.time()

token_time = token_end_time - token_start_time

print(f"Step 1 - 获取上传 Token 和 Key'{ filename}': { token_time:.2f} 秒")

# 2、上传图片

upload_start_time = time.time()

file_url = upload_file(upload_token=token, key=key, filePath=file_path)

upload_end_time = time.time()

upload_time = upload_end_time - upload_start_time

print(f"Step 2 - 上传图片'{ filename}': { upload_time:.2f} 秒")

# 3、检查图片

check_start_time = time.time()

review_msg = check_pic_pass(file_url)

check_end_time = time.time()

check_time = check_end_time - check_start_time

print(f"Step 3 - 检查图片'{ filename}': { check_time:.2f} 秒")

# 4、处理异常图片

error_check_start_time = time.time()

if review_msg is not None and '正常' not in review_msg:

shutil.copy(file_path, os.path.join(error_folder, filename))

error_check_end_time = time.time()

error_check_time = error_check_end_time - error_check_start_time

print(f"Step 4 - 处理异常图片'{ filename}': { error_check_time:.2f} 秒")

# 删除临时压缩文件（如果存在）

if file_path.endswith(".compressed.jpg"):

os.remove(file_path)

# 计算总时间

total_time = time.time() - start_time

print(f"Total time for processing '{ filename}': { total_time:.2f} 秒")

# 将每个阶段的时间数据保存到列表中

timing_data.append({

'文件名': filename,

'压缩耗时(秒)': compress_time,

'获取Token耗时(秒)': token_time,

'上传耗时(秒)': upload_time,

'检查耗时(秒)': check_time,

'处理异常图片耗时(秒)': error_check_time,

'总耗时(秒)': total_time

})

# 返回检测结果用于后续的保存

return { '图片名': filename, '检测结果': review_msg}

# 在所有图片处理完成后，将 timing_data 保存到一个 Excel 文件

def save_timing_data_to_excel(output_excel_name="timing_data.xlsx", image_folder=None):code>

if image_folder is None:

# 如果未指定图片文件夹，使用当前工作目录

image_folder = os.getcwd()

# 生成保存路径，将 Excel 文件保存到图片文件夹中

output_excel_path = os.path.join(image_folder, output_excel_name)

# 将 timing_data 保存为 Excel 文件

df = pd.DataFrame(timing_data)

df.to_excel(output_excel_path, index=False, engine='openpyxl')code>

print(f"Timing data saved to { output_excel_path}")

def process_images_in_folder(input_folder, error_folder, result_file, max_workers=10):

if not os.path.exists(error_folder):

os.makedirs(error_folder)

data = []

images = [os.path.join(input_folder, filename) for filename in os.listdir(input_folder)

if filename.lower().endswith(('.png', '.jpg', '.jpeg'))]

# 使用 ThreadPoolExecutor 进行并发处理

with ThreadPoolExecutor(max_workers=max_workers) as executor:

futures = [executor.submit(process_single_image, file_path, error_folder) for file_path in images]

# 收集并保存结果

for future in futures:

result = future.result()

data.append(result)

df = pd.DataFrame(data)

df.to_excel(result_file, index=False)

save_timing_data_to_excel("timing_data.xlsx", input_folder)

def process_images_in_folder_(input_folder, error_folder, result_file):

if not os.path.exists(error_folder):

os.makedirs(error_folder)

data = []

# 遍历文件夹中的所有图片

for filename in os.listdir(input_folder):

if filename.lower().endswith(('.png', '.jpg',

'.jpeg')):

file_path = os.path.join(input_folder, filename)

# 1、获取上传 Token 和 Key

token, key = get_upload_token(file_path)

# 2、上传图片

file_url = upload_file(upload_token=token, key=key, filePath=file_path)

# 3、检查图片

review_msg = check_pic_pass(file_url)

# 4、保存检测结果

data.append({

'图片名': filename,

'检测结果': review_msg

})

# 如果有异常，将图片拷贝到错误文件夹

if review_msg is not None and '正常' not in review_msg:

shutil.copy(file_path, os.path.join(error_folder, filename))

# 将检测结果保存到 Excel 文件

df = pd.DataFrame(data)

df.to_excel(result_file, index=False)

# 检查指定目录下的图片，并将异常图片移动到错误目录，同时生成检测结果的 Excel 文件

if __name__ == "__main__":

# 定义输入文件夹路径，其中包含要处理的图片

input_folder = 'F:\\Work\\content-check\\result\\文本测试'

# 定义错误文件夹路径，异常图片将被移动到这里

error_folder = 'F:\\Work\\content-check\\result\\文本测试\\异常'

# 定义结果文件路径，将生成一个包含检测结果的 Excel 文件

result_file = 'F:\\Work\\content-check\\result\\文本测试\\检测结果.xlsx'

# 调用函数，对指定文件夹中的图片进行处理

process_images_in_folder(input_folder, error_folder, result_file, 8)

6.1 功能介绍