iot-device-management-service/app/services/oss_storage.py

import uuid
from datetime import datetime, timezone
from typing import Optional
from pathlib import Path

# import oss2  # TODO: 阿里云 OSS 待配置完成后启用
from app.config import settings
from app.utils.logger import logger


class OSSStorage:
    """
    图片存储类
    TODO: 阿里云 OSS 配置完成后，注释掉 use_local_only=True 即可启用 OSS
    """

    def __init__(self, use_local_only: bool = True):
        self.use_local_only = use_local_only
        self.bucket = None

        if not self.use_local_only:
            self._init_bucket()

    def _init_bucket(self):
        """
        初始化 OSS Bucket（待配置完成后启用）
        """
        # TODO: 阿里云 OSS 配置完成后启用以下代码
        # if not settings.oss.access_key_id or not settings.oss.bucket_name:
        #     logger.warning("OSS配置不完整，将使用本地存储")
        #     self.use_local_only = True
        #     return
        #
        # try:
        #     auth = oss2.Auth(settings.oss.access_key_id, settings.oss.access_key_secret)
        #     self.bucket = oss2.Bucket(auth, settings.oss.endpoint, settings.oss.bucket_name)
        #     logger.info(f"OSS连接成功: {settings.oss.bucket_name}")
        # except Exception as e:
        #     logger.error(f"OSS连接失败: {e}")
        #     self.bucket = None
        logger.info("OSS存储已注释，启用本地存储模式")

    def upload_image(self, image_data: bytes, filename: Optional[str] = None) -> str:
        """
        上传图片（当前使用本地存储，OSS 配置完成后可启用）
        """
        if self.use_local_only or not self.bucket:
            return self._upload_local(image_data, filename)

        # TODO: 阿里云 OSS 配置完成后启用以下代码
        # if not self.bucket:
        #     return self._upload_local(image_data, filename)
        #
        # if filename is None:
        #     timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
        #     unique_id = uuid.uuid4().hex[:8]
        #     ext = ".jpg"
        #     filename = f"alerts/{timestamp}_{unique_id}{ext}"
        #
        # try:
        #     self.bucket.put_object(filename, image_data)
        #     url = f"{settings.oss.url_prefix}/{filename}"
        #     logger.info(f"图片上传OSS成功: {url}")
        #     return url
        # except Exception as e:
        #     logger.error(f"图片上传OSS失败: {e}")
        #     return self._upload_local(image_data, filename)

        return self._upload_local(image_data, filename)

    def _upload_local(self, image_data: bytes, filename: Optional[str] = None) -> str:
        """
        本地存储（当前使用）
        """
        upload_dir = Path("uploads")
        upload_dir.mkdir(exist_ok=True)

        if filename is None:
            timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
            unique_id = uuid.uuid4().hex[:8]
            filename = f"alerts/{timestamp}_{unique_id}.jpg"

        file_path = upload_dir / filename
        file_path.parent.mkdir(parents=True, exist_ok=True)

        with open(file_path, "wb") as f:
            f.write(image_data)

        local_url = f"/uploads/{filename}"
        logger.info(f"图片保存本地: {local_url}")
        return local_url

    def get_url(self, path: str) -> str:
        """
        获取图片访问URL
        """
        if path.startswith("http"):
            return path
        # TODO: 阿里云 OSS 配置完成后启用以下代码
        # if self.bucket:
        #     return f"{settings.oss.url_prefix}/{path}"
        return f"/uploads/{path}" if not path.startswith("/") else path


# 使用本地存储模式（OSS 配置完成后可修改为 OSSStorage(use_local_only=False)）
oss_storage = OSSStorage(use_local_only=True)


def get_oss_storage() -> OSSStorage:
    return oss_storage