Serverless架构下的OSS应用：函数计算FC自动处理图片/视频转码（演示水印添加+缩略图生成流水线）

# 创建原始存储桶
aliyun oss mb oss://origin-bucket --region cn-hangzhou

# 创建结果存储桶
aliyun oss mb oss://processed-bucket --region cn-hangzhou

# 配置事件通知规则
aliyun oss putbucketnotification oss://origin-bucket 
--callback /path/to/notification.json

{
  "TopicConfiguration": {
    "Topic": "fc-trigger",
    "Events": ["oss:ObjectCreated:*"],
    "Filter": {
      "Key": {
        "FilterRules": [
          {"Name": "prefix", "Value": "uploads/"},
          {"Name": "suffix", "Value": ".jpg|.png|.mp4"}
        ]
      }
    }
  }
}

# template.yml
ROSTemplateFormatVersion: '2015-09-01'
Resources:
  media-process-service:
    Type: 'Aliyun::Serverless::Service'
    Properties:
      Description: '媒体处理服务'
      Policies: 
        - AliyunOSSFullAccess
      LogConfig:
        Project: media-process-log
        Logstore: fc-log
        
  image-processor:
    Type: 'Aliyun::Serverless::Function'
    Properties:
      Handler: index.image_handler
      Runtime: python3.9
      CodeUri: ./image-process/
      Timeout: 600
      MemorySize: 3072
      EnvironmentVariables:
        TARGET_BUCKET: processed-bucket
        WATERMARK_PATH: oss://config-bucket/watermark.png
        
  video-processor:
    Type: 'Aliyun::Serverless::Function'
    Properties:
      Handler: index.video_handler
      Runtime: python3.9
      CodeUri: ./video-process/
      Timeout: 900
      MemorySize: 4096
      EnvironmentVariables:
        TARGET_BUCKET: processed-bucket
        FFMPEG_PATH: /code/ffmpeg

from PIL import Image, ImageOps, ImageSequence
import oss2
import io

def add_watermark(image, watermark_path):
    """添加水印核心逻辑"""
    # 获取水印图片
    auth = oss2.Auth(os.getenv('OSS_KEY'), os.getenv('OSS_SECRET'))
    bucket = oss2.Bucket(auth, os.getenv('OSS_ENDPOINT'), 'config-bucket')
    watermark = Image.open(io.BytesIO(bucket.get_object(watermark_path).read()))
    
    # 计算水印位置（右下角偏移10px）
    position = (
        image.width - watermark.width - 10, 
        image.height - watermark.height - 10
    )
    
    # 处理透明通道
    if image.mode != 'RGBA':
        image = image.convert('RGBA')
    watermark = watermark.convert('RGBA')
    
    # 创建透明图层合并
    composite = Image.new('RGBA', image.size)
    composite.paste(image, (0,0))
    composite.paste(watermark, position, watermark)
    
    return composite.convert(image.mode)

def process_image(object_key):
    """图片处理主函数"""
    # 下载原始图片
    auth = oss2.Auth(os.getenv('OSS_KEY'), os.getenv('OSS_SECRET'))
    src_bucket = oss2.Bucket(auth, os.getenv('OSS_ENDPOINT'), 'origin-bucket')
    image_data = src_bucket.get_object(object_key).read()
    
    # 打开图片（支持动图）
    original = Image.open(io.BytesIO(image_data))
    processed_frames = []
    
    # 帧处理（动图需逐帧处理）
    for frame in ImageSequence.Iterator(original):
        # 添加水印
        watermarked = add_watermark(frame.copy(), os.getenv('WATERMARK_PATH'))
        
        # 生成缩略图
        thumbnail = watermarked.copy()
        thumbnail.thumbnail((300, 300), Image.LANCZOS)
        processed_frames.append(thumbnail)
    
    # 保存处理结果
    output = io.BytesIO()
    if len(processed_frames) > 1:
        processed_frames[0].save(
            output, 
            format=original.format,
            save_all=True,
            append_images=processed_frames[1:],
            duration=original.info.get('duration', 100),
            loop=original.info.get('loop', 0)
        )
    else:
        watermarked.save(output, format=original.format)
    
    # 上传结果
    dest_bucket = oss2.Bucket(auth, os.getenv('OSS_ENDPOINT'), os.getenv('TARGET_BUCKET'))
    dest_bucket.put_object(f'watermarked/{object_key}', output.getvalue())

def generate_thumbnails(image, object_key):
    """生成多规格缩略图"""
    thumbnail_specs = [
        {'suffix': '_large', 'size': (1024, 768)},
        {'suffix': '_medium', 'size': (640, 480)},
        {'suffix': '_small', 'size': (320, 240)}
    ]
    
    for spec in thumbnail_specs:
        # 创建缩略图
        thumb = image.copy()
        thumb.thumbnail(spec['size'], Image.LANCZOS)
        
        # 格式转换优化
        if thumb.mode == 'RGBA' and image.format != 'PNG':
            thumb = thumb.convert('RGB')
        
        # 保存并上传
        output = io.BytesIO()
        thumb.save(output, format='JPEG', quality=85, optimize=True)
        dest_bucket.put_object(
            f"thumbnails/{object_key.replace('.', spec['suffix'] + '.')}", 
            output.getvalue()
        )

# Dockerfile for FC with FFmpeg
FROM python:3.9-slim

# 安装FFmpeg
RUN apt-get update && \
    apt-get install -y ffmpeg && \
    rm -rf /var/lib/apt/lists/*

# 安装Python依赖
COPY requirements.txt .
RUN pip install -r requirements.txt -t /code

# 复制函数代码
COPY . /code
WORKDIR /code

CMD ["index.video_handler"]

import subprocess
import os

def transcode_video(input_path, output_path, preset='mobile'):
    """视频转码函数"""
    # 转码预设配置
    presets = {
        'mobile': {
            'codec': 'libx264',
            'crf': 23,
            'preset': 'fast',
            'resolution': '1280x720'
        },
        'web': {
            'codec': 'libvpx-vp9',
            'crf': 31,
            'preset': 'medium',
            'resolution': '1920x1080'
        }
    }
    
    config = presets[preset]
    cmd = [
        'ffmpeg', '-i', input_path,
        '-c:v', config['codec'],
        '-crf', str(config['crf']),
        '-preset', config['preset'],
        '-s', config['resolution'],
        '-movflags', '+faststart',
        '-threads', str(os.cpu_count()),
        output_path
    ]
    
    try:
        # 执行转码命令
        process = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=True
        )
        return True
    except subprocess.CalledProcessError as e:
        logger.error(f"转码失败: {e.stderr.decode()}")
        return False

def process_video(object_key):
    """视频处理主函数"""
    # 下载原始视频
    local_input = f'/tmp/{os.path.basename(object_key)}'
    src_bucket.get_object_to_file(object_key, local_input)
    
    # 执行转码
    output_name = f"transcoded/{object_key.split('.')[0]}.mp4"
    local_output = f'/tmp/{os.path.basename(output_name)}'
    
    if transcode_video(local_input, local_output, preset='mobile'):
        # 上传转码结果
        dest_bucket.put_object_from_file(output_name, local_output)
        
        # 生成预览缩略图
        generate_video_thumbnail(local_input, object_key)
        
    # 清理临时文件
    os.remove(local_input)
    os.remove(local_output)

def generate_video_thumbnail(video_path, object_key):
    """生成视频预览图"""
    # 提取第一帧作为预览
    thumbnail_path = f'/tmp/{os.path.basename(object_key)}_thumb.jpg'
    cmd = [
        'ffmpeg', '-i', video_path,
        '-ss', '00:00:01',
        '-vframes', '1',
        '-q:v', '2',
        thumbnail_path
    ]
    
    try:
        subprocess.run(cmd, check=True)
        # 上传缩略图
        dest_bucket.put_object_from_file(
            f"previews/{object_key}.jpg", 
            thumbnail_path
        )
        return True
    except:
        return False

# 函数重试策略
ErrorHandling:
  MaximumRetryAttempts: 3
  RetryStep:
    - DelaySeconds: 1
    - DelaySeconds: 5
    - DelaySeconds: 15
  DeadLetterQueue:
    TargetArn: acs:mns:cn-hangzhou:1234567890:queues/dead-letter-queue

# 初始化函数外共享资源
auth = oss2.Auth(os.getenv('OSS_KEY'), os.getenv('OSS_SECRET'))
watermark_cache = None

def handler(event, context):
    global watermark_cache
    if not watermark_cache:
        # 首次加载水印到内存
        bucket = oss2.Bucket(auth, os.getenv('OSS_ENDPOINT'), 'config-bucket')
        watermark_cache = bucket.get_object('watermark.png').read()
    
    # 使用watermark_cache处理图片

def parallel_transcode(video_path):
    """并行分段转码"""
    # 获取视频时长
    cmd = ['ffprobe', '-i', video_path, '-show_entries', 'format=duration', '-v', 'quiet']
    result = subprocess.run(cmd, capture_output=True, text=True)
    duration = float(result.stdout.split('=')[1].strip())
    
    # 分割任务（每段30秒）
    segments = int(duration // 30) + 1
    futures = []
    
    with concurrent.futures.ThreadPoolExecutor() as executor:
        for i in range(segments):
            start = i * 30
            output = f'/tmp/segment_{i}.mp4'
            future = executor.submit(
                transcode_segment,
                video_path,
                output,
                start,
                min(30, duration - start)
            )
            futures.append(future)
    
    # 等待所有任务完成
    concurrent.futures.wait(futures)
    
    # 合并分段
    merge_segments('/tmp/segment_*.mp4', '/tmp/final.mp4')

每月成本 = 调用次数 × 每次GB-秒 × 单价
GB-秒 = 内存(GB) × 执行时间(秒)

#!/bin/bash
# 完整部署脚本
set -e

# 1. 创建OSS存储桶
aliyun oss mb oss://origin-bucket
aliyun oss mb oss://processed-bucket
aliyun oss mb oss://config-bucket

# 2. 上传配置文件
aliyun oss cp watermark.png oss://config-bucket/watermark.png

# 3. 创建函数计算服务
fun deploy --template template.yml

# 4. 配置OSS事件触发
aliyun oss putbucketnotification oss://origin-bucket --notification config/notification.json

# 5. 创建监控告警
aliyun cms CreateAlarm \
  --Name Media_Process_Error \
  --Namespace acs_fc \
  --MetricName ErrorCount \
  --Period 300 \
  --Statistics Average \
  --Threshold 10 \
  --ContactGroups [\"MediaTeam\"]