Flask+AWS部署API遇TypeError及Lambda触发问题求助

1. 解决TypeError: can't concat NoneType to bytes #

这个错误的核心是yield语句中使用的frame变量为None，无法和字节类型拼接。触发场景包括：视频流读取失败导致image为None、cv2.imencode()执行失败返回无效结果、外层异常捕获直接pass隐藏初始化错误导致frame未被赋值。

修改后的代码：

@app.route('/<id>')
def yolo_v4(id):
    frame = None
    try:
        kinesis1 = vid_stream.get_hls_url(id)
        cap = cv2.VideoCapture(kinesis1)
        # 提前检查流是否打开成功
        if not cap.isOpened():
            raise ValueError(f"无法打开流 {id}")

        previous = time()
        yolo_v4.delta = 0
        while True:
            success, image = cap.read()
            if not success:
                break
                
            current = time()
            yolo_v4.delta += current - previous
            previous = current

            try:
                create_alert(image)
            except Exception as e:
                print("create_alert函数异常:", e)

            # 确保帧编码成功再处理
            ret, jpeg = cv2.imencode('.jpg', image)
            if ret:
                frame = jpeg.tobytes()
                # 移到循环内yield，只输出有效帧
                yield b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + frame + b'\r\n\r\n'
            else:
                print(f"流 {id} 帧编码失败")
                continue

    except Exception as e:
        print(f"处理流 {id} 时出错:", e)
        # 返回错误提示
        yield b'--frame\r\nContent-Type: text/plain\r\n\r\n处理流失败\r\n\r\n'

    # 处理无有效帧的情况
    if frame is None:
        yield b'--frame\r\nContent-Type: text/plain\r\n\r\n未处理到有效帧\r\n\r\n'

关键修改点：

• 初始化frame变量，避免未定义报错
• 检查视频流是否成功打开，提前抛出错误
• 将yield移至循环内部，仅输出编码成功的有效帧
• 替换外层pass为错误打印，便于定位问题

2. 解决移除yield后的"The function either returned None or ended without a return statement" #

Flask路由函数必须返回有效的HTTP响应（字符串、JSON、Response对象等），移除yield后函数变为普通函数，无return语句会默认返回None，触发该错误。

修改后的代码（无需流式输出时）：

from flask import jsonify

@app.route('/<id>')
def yolo_v4(id):
    try:
        kinesis1 = vid_stream.get_hls_url(id)
        cap = cv2.VideoCapture(kinesis1)
        if not cap.isOpened():
            return jsonify({"error": f"无法打开流 {id}"}), 400

        previous = time()
        yolo_v4.delta = 0
        while True:
            success, image = cap.read()
            if not success:
                break
                
            current = time()
            yolo_v4.delta += current - previous
            previous = current

            try:
                create_alert(image)
            except Exception as e:
                print("create_alert函数异常:", e)

        cap.release()
        return jsonify({"status": "success", "message": f"流 {id} 处理完成"}), 200

    except Exception as e:
        print(f"处理流 {id} 时出错:", e)
        return jsonify({"error": str(e)}), 500

关键修改点：

• 导入jsonify返回标准化JSON响应
• 所有分支（成功、流打开失败、异常）都返回明确的响应和状态码
• 处理完流后调用cap.release()释放资源

Lambda调用优化 #

Lambda直接返回urllib3的HTTPResponse对象会导致序列化问题，建议修改为返回结构化数据：

def lambda_handler(event, context):
    print("PPE Event", event)
    streamName = event['deviceId']
    print("StreamName", streamName)
    
    http = urllib3.PoolManager()
    logger.debug("test")
    
    try:
        response = http.request("GET", f"http://ec2-102.somenumber.ap-south-1.compute.amazonaws.com:8080/{streamName}")
        print(response.status)
        return {
            "statusCode": response.status,
            "body": response.data.decode('utf-8'),
            "headers": dict(response.headers)
        }
    except Exception as e:
        logger.error(f"调用Flask API失败: {e}")
        return {
            "statusCode": 500,
            "body": f"API调用错误: {str(e)}"
        }

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