dsh/test_oss_upload.py

#!/usr/bin/env python3
"""
OSS 上传测试脚本
测试本地图片上传到 MinIO
"""

import os
import sys
import cv2
import tempfile
import logging
import numpy as np

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from dual_camera_system.oss_uploader import OSSUploader


def create_test_image(width=640, height=480, text="Test Image"):
    """创建测试图片"""
    # 创建白色背景
    img = np.ones((height, width, 3), dtype=np.uint8) * 255

    # 画几个矩形
    cv2.rectangle(img, (50, 50), (width-50, height-50), (200, 200, 200), 2)
    cv2.rectangle(img, (100, 100), (width-100, height-100), (150, 150, 150), -1)

    # 添加文字
    cv2.putText(img, text, (width//4, height//2),
                cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)

    return img


def main():
    # 创建临时测试图片
    with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as tmp:
        temp_path = tmp.name

    try:
        # 生成测试图片
        img = create_test_image(640, 480, "OSS Test")
        cv2.imwrite(temp_path, img)
        logger.info(f"测试图片已创建: {temp_path}")

        # 初始化 OSS 上传器
        logger.info("初始化 OSS 上传器...")
        uploader = OSSUploader()

        if not uploader.enabled:
            logger.error("OSS 上传器未启用，请检查配置")
            return 1

        # 启动上传器
        logger.info("启动上传器...")
        uploader.start()

        # 测试上传
        test_key = "test/test_image.jpg"
        batch_id = "test_batch_001"

        logger.info(f"开始上传测试图片: {temp_path}")
        logger.info(f"OSS Key: {test_key}")

        # 同步上传测试
        result = uploader.upload_image_sync(
            local_path=temp_path,
            oss_key=test_key,
            batch_id=batch_id,
            image_type='test'
        )

        logger.info(f"上传结果: success={result.success}")
        logger.info(f"OSS URL: {result.oss_url}")

        if result.error:
            logger.error(f"错误信息: {result.error}")

        if result.success:
            logger.info("✅ OSS 上传测试成功!")
        else:
            logger.error("❌ OSS 上传测试失败!")
            return 1

    except Exception as e:
        logger.exception(f"测试异常: {e}")
        return 1
    finally:
        # 清理临时文件
        if os.path.exists(temp_path):
            os.remove(temp_path)
            logger.info(f"临时文件已清理: {temp_path}")

    return 0


if __name__ == '__main__':
    sys.exit(main())