shanxia2.0/backend/app/crud/oss.py

"""统一附件 sys_oss CRUD"""
from __future__ import annotations

from typing import Optional

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.crm_models import SysOss


def create_oss(
    db: Session,
    *,
    tenant_no: str,
    url: str,
    original_name: str,
    file_suffix: str = "",
    file_size: int = 0,
    file_type: str = "",
    biz_type: str = "",
    biz_id: str = "",
    attach_type: str = "attachment",
    sort_order: int = 0,
    user_id: Optional[str] = None,
    ext1: Optional[str] = None,
) -> SysOss:
    """写入 sys_oss 附件记录"""
    stored_name = url.rsplit("/", 1)[-1] if url else original_name
    obj = SysOss(
        id=generate_uuid(),
        tenant_no=tenant_no,
        biz_type=biz_type or None,
        biz_id=biz_id or None,
        file_name=stored_name or original_name,
        original_name=original_name,
        file_suffix=file_suffix,
        url=url,
        file_size=file_size,
        file_type=file_type,
        attach_type=attach_type,
        sort_order=sort_order,
        service="minio",
        ext1=ext1,
        create_by=user_id,
        create_dept_id="",
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def get_oss_by_url(db: Session, tenant_no: str, url: str) -> Optional[SysOss]:
    """按 MinIO 相对路径查询附件"""
    if not url:
        return None
    return (
        db.query(SysOss)
        .filter(SysOss.url == url, SysOss.tenant_no == tenant_no, SysOss.del_flag == 0)
        .first()
    )