前言
CSS Sprite,多个图片被整合到一个精灵图中，用户不需要下载多个文件，而是只需要下载单个文件，当需要特定的图像时，CSS引用这张雪碧图，通过偏移和定义尺寸来达到目的。CSS Sprite 具有如下优点：
- 更流畅的用户体验，因为一旦雪碧图被下载，所有使用雪碧图上面的图片的地方都会得到渲染，而不是一个文件一个文件的加载。
- 减少HTTP请求，将原本需要的多个请求合并为一个，较少服务器压力，从而较少网络堵塞。
- 减少图片的字节。多次比较，三张图片合并成一张图片之后的字节总是小于这三长图片的总和。
- 更换风格方便，只需要在一张或少张图片上修改图片的颜色或样式，维护起来更加方便。
在一些视频处理的场景求中，有对视频制作雪碧图需求，简单来说，就是对视频的按照一定的规律就行截帧，然后将截帧的图片组合成一张大的雪碧图。 本文基于对象存储服务 OSS 和函数计算 FC， 实现弹性高可用的海量视频的存储和雪碧图制作处理服务。
体验入口地址
- 10 * 10 的雪碧图
http://fcdemo.mofangdegisn.cn/vedio-process?srcdir=video&video=test_hd.mp4&w=144&h=128&offset=100&interval=20&sprite=10*10&saveas=picture&capture_num=100
- 4 * 4 竖屏视频雪碧图(不变形处理)
http://fcdemo.mofangdegisn.cn/vedio-process?srcdir=video&video=shupin.mp4&w=144&h=128&offset=100&interval=20&sprite=4*4&saveas=picture&capture_num=16&autofill=1&oringin_ratio=0.5625
该体验地址采用的函数计算的http trigger， 其中:
返回值
截图的zip包和雪碧图的url
参数意义:
-
srcdir: oss bucket 中上传视频的目录 -
video：视频的名字 -
w: 视频截帧的宽度 -
h: 视频截帧的高度 -
offset: 视频截帧从起始处，单位是秒 -
interval: 每多少秒截一次 -
sprite：雪碧图的组成，列数*行数 -
capture_num： 需要截图的总数 -
saveas: 截图zip包和雪碧图在oss 上保存的目录 -
autofill： 和oringin_ratio(原始视频的 宽度/高度 比)一起使用，当值为1时， 竖屏处理
架构设计图
利用oss的存储、单帧截图能力与函数计算的自定义处理能力，可以实现各种自定义的视频截图与图片组合操作，再也不用担心各家厂商提供的视频截图功能无法满足自定义功能的情况，同时，OSS 的海量高可靠 和 FC 的弹性高可用相结合，整体服务也就具有了按需付费、弹性扩容、数据高可靠，计算高可用，免运维等一系列优点。
代码
# -*- coding: utf-8 -*-
from wand.image import Image
from wand.color import Color
from threading import Thread
import oss2
from urllib import parse
import math
import zipfile
import io
bucket_name = 'xbt-video'
oss_region = "cn-hangzhou"
images = []
# 获取oss 视频截帧的返回图片
def video_capture(bucket, srcdir, vedio_name, t, process):
cap_pic = vedio_name + '_' + str(t) + '.png'
r = bucket.get_object(srcdir + "/" + vedio_name, process=process)
content = r.read()
images.append(content)
def handler(environ, start_response):
global images
images = []
context = environ['fc.context']
creds = context.credentials
# Required by OSS sdk
auth=oss2.StsAuth(
creds.access_key_id,
creds.access_key_secret,
creds.security_token)
endpoint = 'oss-{}-internal.aliyuncs.com'.format(oss_region)
bucket = oss2.Bucket(auth, endpoint, bucket_name)
# 解析出对应的请求参数
try:
query_string = environ.get('QUERY_STRING', '')
params = parse.parse_qs(query_string)
params = {k: v[0] for k, v in params.items()}
except Exception as e:
print(str(e))
srcdir = params['srcdir']
video = params['video']
saveas = params['saveas']
width, height, offset, interval = int(params['w']), int(params['h']), int(params.get('offset',0)), int(params.get('interval', 1))
sprite = params.get('sprite', "1*1")
rows, cols = sprite.split("*")
rows, cols = int(rows), int(cols)
capture_num = params.get('capture_num')
if not capture_num:
capture_num = rows*cols
capture_num = int(capture_num)
cap_width , cap_height = width , height
# autofill 可选参数的处理
autofill = params.get('autofill')
if autofill and int(autofill) == 1:
oringin_ratio = float(params['oringin_ratio'])
cap_width = int(height * oringin_ratio)
print("cap_info = ", cap_width , cap_height)
# 多线程调用oss视频截帧服务
ts = []
for i in range(capture_num):
t = (offset + i* interval) * 1000
process = "video/snapshot,t_{0},f_png,w_{1},h_{2},m_fast".format(t, cap_width, cap_height)
t = Thread(target=video_capture, args=(bucket, srcdir, video, t, process,))
t.start()
ts.append(t)
for t in ts:
t.join()
image_len = len(images)
print("image length = {}".format(image_len))
ret = []
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "a", zipfile.ZIP_DEFLATED, False) as zf:
for i, content in enumerate(images):
zf.writestr(video + "_{}.png".format(i), content)
zip_file = "{}/{}.zip".format(saveas, video)
bucket.put_object(zip_file, zip_buffer.getvalue())
# 生成截图的zip放入返回结果
zip_url = "https://{}.oss-{}.aliyuncs.com/".format(bucket_name, oss_region) + zip_file + "\n"
ret.append(bytes(zip_url, encoding = "utf8"))
# 雪碧图之间的间隙
cell_gap = 2
# BATCH_NUM 数量的图片生成一张 rows * cols 雪碧图
# 可以生成多张，如果capture_num > rows * cols
BATCH_NUM = rows*cols
xbt_num = int(math.ceil(image_len/float(BATCH_NUM)))
for x in range(xbt_num):
img = Image()
img.blank((width+cell_gap)*rows, (height+cell_gap)*cols, Color('black'))
begin = x * BATCH_NUM
end = begin + BATCH_NUM if (begin + BATCH_NUM) < image_len else image_len
sub_images = images[begin:end]
for i, content in enumerate(sub_images):
with Image(blob=content) as sub_img:
r = i % rows
j = int(i / rows)
if cap_width == width:
img.composite(image=sub_img,left=r*(width + cell_gap), top=j*(height + cell_gap))
else: # autofill为1时的特殊处理
jz_img = Image()
jz_img.blank(width, height, Color('blue'))
jz_img.composite(image=sub_img,left=int((width - cap_width)/2), top=0)
img.composite(image=jz_img,left=r*(width + cell_gap), top=j*(height + cell_gap))
pic_file = "{}/xbt_{}_{}.png".format(saveas, video, x)
bucket.put_object(pic_file, img.make_blob(format='png'))
pic_url = "https://{}.oss-{}.aliyuncs.com/".format(bucket_name, oss_region) + pic_file + "\n"
# 生成的雪碧图的url地址放入返回值
ret.append(bytes(pic_url, encoding = "utf8"))
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
return retfun 部署的template.yml
ROSTemplateFormatVersion: '2015-09-01'
Transform: 'Aliyun::Serverless-2018-04-03'
Resources:
xbt-demo-pro:
Type: 'Aliyun::Serverless::Log'
Properties:
Description: 'image process log pro'
fc-log:
Type: 'Aliyun::Serverless::Log::Logstore'
Properties:
TTL: 362
ShardCount: 1
xbt-service:
Type: 'Aliyun::Serverless::Service'
Properties:
Description: 'image process demo'
Policies:
- AliyunOSSFullAccess
LogConfig:
Project: 'xbt-demo-pro'
Logstore: 'fc-log'
xbt-func:
Type: 'Aliyun::Serverless::Function'
Properties:
Handler: index.handler
CodeUri: './'
Description: 'xbt-process http function'
Runtime: python3
Timeout: 60
MemorySize: 512
Events:
http-trigger:
Type: HTTP
Properties:
AuthType: ANONYMOUS
Methods: ['GET', 'POST']