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()
    )


def get_oss_by_id(db: Session, tenant_no: str, oss_id: str) -> Optional[SysOss]:
    """按主键 id 查询附件（兼容旧 CRM 雪花 oss_id 迁入后的记录）"""
    if not oss_id:
        return None
    return (
        db.query(SysOss)
        .filter(SysOss.id == str(oss_id).strip(), SysOss.tenant_no == tenant_no, SysOss.del_flag == 0)
        .first()
    )


def resolve_storage_path(db: Session, tenant_no: str, file_url_or_id: str) -> tuple[str, Optional[SysOss]]:
    """将 fileUrl 或旧 CRM 数字 ossId 解析为 MinIO 相对路径。

    门户新上传把相对路径写入业务字段；迁入合同附件仍存数字 ossId。
    """
    raw = (file_url_or_id or "").strip()
    if not raw:
        return "", None
    by_url = get_oss_by_url(db, tenant_no, raw)
    if by_url:
        return (by_url.url or raw), by_url
    # 纯数字 / UUID：按 sys_oss.id 查真实 object key
    if "/" not in raw and not raw.startswith("http"):
        by_id = get_oss_by_id(db, tenant_no, raw)
        if by_id and by_id.url:
            return by_id.url, by_id
    return raw, None