shanxia2.0/backend/scripts/migration/minio_url.py

"""MinIO 附件 URL 规范化（数据迁移共用）。

背景
----
CRM / 地磅 / 报销与 shanxia2.0 共用同一 MinIO（bucket 默认 ``ruoyi``），
物理对象无需复制。旧 CRM ``sys_oss.url`` 存的是带域名的绝对 URL（含历史
``47.99.164.66:20004``、``121.43.228.102:9000`` 等），门户 ``sys_oss.url`` 与
业务表字段存 **相对 object key**（如 ``2025/10/08/xxx.png``）。

旧表 ``sys_oss.file_name`` 通常已是 object key，与 ``url`` 中的路径一致，
解析失败时可优先回退 ``file_name``。

报销 ``bx_attachment.file_path`` 本身即为相对路径（如 ``bx/2026-05-19/...``），
可直接写入目标库。
"""
from __future__ import annotations

from typing import Iterable, Optional
from urllib.parse import unquote, urlparse


DEFAULT_BUCKET = "ruoyi"

# 历史绝对 URL 前缀（域名/网关变更）；匹配时去掉前缀得到 object key
LEGACY_URL_PREFIXES = (
    "http://47.99.164.66:20004/ruoyi/",
    "https://47.99.164.66:20004/ruoyi/",
    "http://121.43.228.102:9000/ruoyi/",
    "https://121.43.228.102:9000/ruoyi/",
    "http://www.chengyantek.cn:20004/ruoyi/",
    "https://www.chengyantek.cn:20004/ruoyi/",
    "http://www.chengyantek.cn/ruoyi/",
    "https://www.chengyantek.cn/ruoyi/",
    "http://www.chengyantek.cn/zkxw/",
    "https://www.chengyantek.cn/zkxw/",
)


def strip_url_query(raw: str) -> str:
    """去掉预签名 query（``?X-Amz-...``）。"""
    value = (raw or "").strip()
    if not value:
        return ""
    return value.split("?", 1)[0].strip()


def _strip_legacy_prefix(url: str) -> Optional[str]:
    normalized = strip_url_query(url)
    for prefix in LEGACY_URL_PREFIXES:
        if normalized.startswith(prefix):
            return normalized[len(prefix) :].lstrip("/")
    return None


def extract_minio_object_key(
    raw_url: Optional[str],
    *,
    bucket: str = DEFAULT_BUCKET,
    file_name_fallback: Optional[str] = None,
) -> str:
    """将绝对/相对 URL 规范为 MinIO object key（不含 bucket）。"""
    url = strip_url_query(raw_url or "")
    if not url:
        fallback = (file_name_fallback or "").strip().lstrip("/")
        return fallback

    legacy = _strip_legacy_prefix(url)
    if legacy:
        return legacy

    if url.startswith("http://") or url.startswith("https://"):
        parsed = urlparse(url)
        path = unquote((parsed.path or "").lstrip("/"))
        bucket_prefix = f"{bucket}/"
        if path.startswith(bucket_prefix):
            return path[len(bucket_prefix) :]
        # 少数 URL 路径不含 bucket 段
        return path

    return url.lstrip("/")


def normalize_oss_storage_url(
    url: Optional[str],
    file_name: Optional[str] = None,
    *,
    bucket: str = DEFAULT_BUCKET,
) -> str:
    """迁移 CRM ``sys_oss`` 行：``url`` + ``file_name`` → 门户相对路径。"""
    from_url = extract_minio_object_key(url, bucket=bucket)
    if from_url:
        return from_url
    return (file_name or "").strip().lstrip("/")


def join_relative_urls(urls: Iterable[str]) -> str:
    """逗号拼接相对路径（去空、去重保序）。"""
    seen = set()
    parts: list[str] = []
    for item in urls:
        key = (item or "").strip().lstrip("/")
        if not key or key in seen:
            continue
        seen.add(key)
        parts.append(key)
    return ",".join(parts)


def resolve_oss_id_csv(
    oss_id_csv: Optional[str],
    oss_id_to_relative_url: dict[str, str],
) -> str:
    """业务表 OSS ID 串 → 相对 URL 串（用于客户证照、跟进附件等）。"""
    if not oss_id_csv:
        return ""
    urls: list[str] = []
    for token in str(oss_id_csv).split(","):
        oss_id = token.strip()
        if not oss_id:
            continue
        mapped = oss_id_to_relative_url.get(oss_id)
        if mapped:
            urls.append(mapped)
    return join_relative_urls(urls)


if __name__ == "__main__":
    samples = [
        (
            "http://47.99.164.66:20004/ruoyi/2025/10/08/5ac0966ab3a74eadab571ce0bb3b441a.png",
            "2025/10/08/5ac0966ab3a74eadab571ce0bb3b441a.png",
        ),
        (
            "http://121.43.228.102:9000/ruoyi/2025/11/01/0048348b3d6c439f99db454c8c05e59a.gif?X-Amz-Algorithm=AWS4",
            "2025/11/01/0048348b3d6c439f99db454c8c05e59a.gif",
        ),
        ("2025/10/17/d601338282a44cf092bf68afe1aef239.jpeg", "2025/10/17/d601338282a44cf092bf68afe1aef239.jpeg"),
        ("bx/2026-05-19/4ca4ffbeb7fc4e22913936d0860b2e2d.png", "bx/2026-05-19/4ca4ffbeb7fc4e22913936d0860b2e2d.png"),
    ]
    for raw, expected in samples:
        got = extract_minio_object_key(raw)
        assert got == expected, f"{raw!r} -> {got!r}, want {expected!r}"
    print("minio_url self-check ok")