"""AI 商品信息辅助填写 — 店铺数据 + AnySearch 联网搜索 + AI 生成"""
import json
import logging
import httpx
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel

logger = logging.getLogger(__name__)
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func

from app.api.deps import get_db, get_admin_user
from app.core.models.user import User
from app.core.models.tenant_settings import TenantSettings
from app.core.models.product import Product, ProductVariant
from app.core.models.category import Category
from app.core.ai_utils import call_ai, resolve_ai_extra, AI_PROVIDER_DEFAULTS as _AI_PROVIDER_DEFAULTS
from app.core.services.ai_quota import consume_ai_quota

router = APIRouter(prefix="/admin/ai", tags=["AI 辅助"])

_ANYSEARCH_ENDPOINT = "https://api.anysearch.com/mcp"

_SEARXNG_INSTANCES = [
    "https://searx.be",
    "https://search.bus-hit.me",
    "https://searxng.site",
    "https://search.inetol.net",
]


def _searxng_bases() -> list[str]:
    """有效 SearXNG 实例列表：配置的内置实例优先，公共实例兜底。"""
    from app.config import settings
    bases = []
    if getattr(settings, "SEARXNG_URL", ""):
        bases.append(settings.SEARXNG_URL.rstrip("/"))
    bases.extend(_SEARXNG_INSTANCES)
    return bases


class ProductAssistRequest(BaseModel):
    name: str
    keywords: str = ""
    category: str = ""
    category_id: int | None = None
    lang: str = "zh"


class ProductAssistResult(BaseModel):
    description: str
    meta_title: str
    meta_description: str
    tags: list[str]
    search_snippets: list[str]
    store_context_used: bool = False
    ai_description: str = ""


# ── 解析 AnySearch markdown 结果 ─────────────────────────────────────
def _parse_anysearch_markdown(text: str, max_results: int) -> list[dict]:
    """将 AnySearch 返回的 markdown 格式解析为结构化结果列表"""
    import re
    results: list[dict] = []
    # 按 ### 标题分块，每块是一条搜索结果
    blocks = re.split(r"(?=^###\s)", text, flags=re.MULTILINE)
    for block in blocks:
        block = block.strip()
        if not block.startswith("###"):
            continue
        title_match = re.match(r"###\s*\d*\.?\s*(.*)", block.split("\n")[0])
        title = title_match.group(1).strip() if title_match else ""
        url_match = re.search(r"\*\*URL\*\*:\s*(https?://\S+)", block)
        href = url_match.group(1).strip() if url_match else ""
        body_lines = []
        for line in block.split("\n")[1:]:
            line = line.strip()
            if line.startswith("- **URL**") or line.startswith("**URL**"):
                continue
            if line.startswith("- "):
                line = line[2:]
            if line.startswith("**") and "**:" in line:
                line = re.sub(r"\*\*[^*]+\*\*:\s*", "", line)
            if line:
                body_lines.append(line)
        body = " ".join(body_lines)[:300]
        if title or body:
            results.append({"title": title, "body": body, "href": href})
        if len(results) >= max_results:
            break
    return results


# ── 联网搜索（AnySearch API）──────────────────────────────────────────
async def _search_anysearch(query: str, max_results: int = 5, api_key: str = "") -> list[dict]:
    headers = {"Content-Type": "application/json"}
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"

    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {
            "name": "search",
            "arguments": {"query": query, "max_results": min(max_results, 10)},
        },
    }

    try:
        async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
            resp = await client.post(_ANYSEARCH_ENDPOINT, json=payload, headers=headers)
            if resp.status_code == 200:
                data = resp.json()
                content_list = (
                    data.get("result", {}).get("content", [])
                    if "result" in data
                    else []
                )
                text = ""
                for item in content_list:
                    if item.get("type") == "text":
                        text += item.get("text", "")
                if text:
                    print(f"[AnySearch] 原始文本前200字: {text[:200]}")
                    # 尝试 JSON 解析
                    try:
                        parsed = json.loads(text)
                        if isinstance(parsed, list):
                            results = [
                                {
                                    "title": r.get("title", ""),
                                    "body": r.get("snippet", r.get("content", r.get("body", ""))),
                                    "href": r.get("url", r.get("href", "")),
                                }
                                for r in parsed[:max_results]
                            ]
                            if results:
                                print(f"[AnySearch] 返回 JSON 格式，{len(results)} 条结果")
                                return results
                    except json.JSONDecodeError:
                        pass
                    # 尝试 markdown 解析
                    results = _parse_anysearch_markdown(text, max_results)
                    if results:
                        print(f"[AnySearch] 返回 Markdown 格式，{len(results)} 条结果")
                        return results
                    # 兜底：按行切分
                    lines = [l.strip() for l in text.split("\n") if l.strip() and not l.startswith("#")]
                    print(f"[AnySearch] 按行切分，{len(lines[:max_results])} 行")
                    return [{"title": l[:80], "body": l, "href": ""} for l in lines[:max_results]]
            else:
                print(f"[AnySearch] HTTP {resp.status_code}")
    except Exception as e:
        print(f"[AnySearch] 请求异常: {e}")
    return []


# ── 联网搜索（SearXNG 保底）─────────────────────────────────────────
async def _search_searxng(query: str, max_results: int = 5) -> list[dict]:
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; SMEStore/1.0)",
        "Accept": "application/json",
    }
    params = {"q": query, "format": "json", "categories": "general", "language": "auto"}

    for base in _searxng_bases():
        try:
            async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client:
                resp = await client.get(f"{base}/search", params=params, headers=headers)
                if resp.status_code == 200:
                    data = resp.json()
                    results = [
                        {"title": r.get("title", ""), "body": r.get("content", ""), "href": r.get("url", "")}
                        for r in data.get("results", [])[:max_results]
                    ]
                    if results:
                        return results
        except Exception:
            continue
    return []


# 明显不该作为商品图的来源：矢量图标、素材站、图库站
_IMG_BAD_HOSTS = (
    "shutterstock", "istockphoto", "gettyimages", "dreamstime", "alamy",
    "freepik", "vecteezy", "flaticon", "iconfinder", "123rf", "depositphotos",
    "stock.adobe", "pngtree", "vectorstock", "clipart",
)


def _is_bad_image_url(url: str) -> bool:
    """过滤掉 SVG 矢量图与素材/图标站，避免选图网格里混入非商品图。"""
    low = url.lower()
    if low.split("?")[0].endswith(".svg"):
        return True
    return any(h in low for h in _IMG_BAD_HOSTS)


# ── Open Food Facts 直连（条形码 → 正品图）──────────────────────────
async def _fetch_off_image(barcode: str) -> list[str]:
    """用 EAN/UPC 条码直查 Open Food Facts，命中则返回正品图（正面优先）。
    仅覆盖食品/饮料，非食品或未收录时返回空列表。"""
    if not (barcode.isdigit() and 8 <= len(barcode) <= 14):
        return []
    url = f"https://world.openfoodfacts.org/api/v2/product/{barcode}.json"
    try:
        async with httpx.AsyncClient(timeout=8, follow_redirects=True) as client:
            resp = await client.get(url, headers={"User-Agent": "SMEStore/1.0 (product image lookup)"})
            if resp.status_code != 200:
                return []
            product = resp.json().get("product") or {}
    except Exception as e:
        print(f"[OFF] 查询异常: {e}")
        return []
    # 正面图优先，其次包装图；营养表/成分表对选主图无用，忽略
    imgs = []
    for key in ("image_front_url", "image_url", "image_packaging_url"):
        u = product.get(key)
        if u and u.startswith("http") and u not in imgs:
            imgs.append(u)
    if imgs:
        print(f"[OFF] {barcode} 命中 {len(imgs)} 张正品图")
    return imgs


# ── 图片搜索（SearXNG images 类别）─────────────────────────────────
async def _search_images(query: str, max_results: int = 10) -> list[str]:
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; SMEStore/1.0)",
        "Accept": "application/json",
    }
    params = {"q": query, "format": "json", "categories": "images", "language": "auto"}

    bases = _searxng_bases()
    print(f"[图片搜索] 待尝试实例: {bases}")
    for base in bases:
        try:
            async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client:
                resp = await client.get(f"{base}/search", params=params, headers=headers)
                if resp.status_code == 200:
                    data = resp.json()
                    urls = []
                    for r in data.get("results", []):
                        url = r.get("img_src") or r.get("url", "")
                        if url and url.startswith("http") and url not in urls and not _is_bad_image_url(url):
                            urls.append(url)
                            if len(urls) >= max_results:
                                break
                    if urls:
                        print(f"[图片搜索] {base} 返回 {len(urls)} 张")
                        return urls
                    print(f"[图片搜索] {base} 200 但无图片结果")
                else:
                    print(f"[图片搜索] {base} HTTP {resp.status_code}")
        except Exception as e:
            print(f"[图片搜索] {base} 异常: {e}")
            continue
    print("[图片搜索] 所有实例均无结果")
    return []


# ── 统一搜索入口：有 AnySearch key 则优先用，否则走 SearXNG ──────────
async def _search_web(query: str, max_results: int = 5, api_key: str = "") -> list[dict]:
    if api_key:
        results = await _search_anysearch(query, max_results, api_key)
        if results:
            return results
        print("[搜索] AnySearch 无结果，回退 SearXNG")
    else:
        print("[搜索] 未配置 AnySearch key，使用 SearXNG")
    results = await _search_searxng(query, max_results)
    if results:
        print(f"[SearXNG] 返回 {len(results)} 条结果")
    else:
        print("[搜索] AnySearch 和 SearXNG 均无结果")
    return results


# ── 读取店铺数据 ──────────────────────────────────────────────────────
async def _get_store_context(
    db: AsyncSession,
    tenant_id: int,
    category_id: int | None,
    store_name: str,
) -> dict:
    """查询本店分类列表、同类已有商品、热门商品"""

    # 所有分类
    cats_result = await db.execute(
        select(Category.name).where(Category.tenant_id == tenant_id).order_by(Category.name)
    )
    all_categories = [r[0] for r in cats_result.fetchall()]

    # 同类已有商品（最多 8 条）
    same_cat_products: list[str] = []
    if category_id:
        prod_result = await db.execute(
            select(Product.name)
            .where(Product.tenant_id == tenant_id, Product.category_id == category_id, Product.status == "active")
            .limit(8)
        )
        same_cat_products = [r[0] for r in prod_result.fetchall()]

    # 热销商品（按销量排序，最多 5 条）
    featured_result = await db.execute(
        select(Product.name)
        .where(Product.tenant_id == tenant_id, Product.status == "active")
        .order_by(Product.sales_count.desc())
        .limit(5)
    )
    featured_products = [r[0] for r in featured_result.fetchall()]

    # 商品总数
    count_result = await db.execute(
        select(func.count()).where(Product.tenant_id == tenant_id, Product.status == "active")
    )
    total_products = count_result.scalar() or 0

    return {
        "store_name": store_name,
        "all_categories": all_categories,
        "same_cat_products": same_cat_products,
        "featured_products": featured_products,
        "total_products": total_products,
    }


# ── 调用 AI（委托给 ai_utils.call_ai）────────────────────────────────
async def _call_ai(prompt: str, extra: dict) -> str:
    return await call_ai(prompt, extra, timeout=300, max_tokens=4096)


def _parse_ai_json(raw: str, fallback_name: str, fallback_category: str) -> dict:
    """解析 AI 返回的 JSON，容错处理 markdown 代码块和前后多余文字"""
    text = raw.strip()
    # 去掉 ```json ... ``` 包裹
    if "```" in text:
        parts = text.split("```")
        for part in parts:
            part = part.strip()
            if part.startswith("json"):
                part = part[4:].strip()
            if part.startswith("{"):
                text = part
                break
    # 提取第一个 { 到最后一个 } 之间的内容（忽略 JSON 前后的说明文字）
    start = text.find("{")
    end = text.rfind("}")
    if start != -1 and end > start:
        text = text[start:end + 1]
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {
            "description": "",
            "meta_title": fallback_name,
            "meta_description": "",
            "tags": [fallback_category] if fallback_category else [],
            "ai_description": "",
        }


# ── 主接口 ────────────────────────────────────────────────────────────
@router.post("/product-assist", response_model=ProductAssistResult, summary="AI 辅助填写商品信息")
async def product_assist(
    body: ProductAssistRequest,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_admin_user),
):
    """
    1. 读取本店铺数据（分类、同类商品、精选商品）
    2. 联网搜索商品真实信息（SearXNG）
    3. 组合成富上下文 Prompt 发给 AI
    4. 返回：描述、SEO 标题、SEO 描述、标签
    """
    # ── 读 AI 配置 ────────────────────────────────────────────────────
    result = await db.execute(
        select(TenantSettings).where(TenantSettings.tenant_id == current_user.tenant_id)
    )
    s = result.scalar_one_or_none()
    extra = s.extra or {} if s else {}
    extra = await resolve_ai_extra(db, current_user.tenant_id, extra)
    store_name = s.store_name if s else "本店"

    if not extra.get("ai_enabled"):
        raise HTTPException(status_code=400, detail="AI 功能未启用，请先在系统设置 → AI 配置中开启")

    await consume_ai_quota(db, current_user.tenant_id)

    # ── 并行：搜索 + 店铺数据 ─────────────────────────────────────────
    import asyncio
    search_query = f"{body.name} {body.keywords} {body.category}".strip()
    anysearch_key = extra.get("anysearch_api_key", "")
    search_results, store_ctx = await asyncio.gather(
        _search_web(search_query, max_results=5, api_key=anysearch_key),
        _get_store_context(db, current_user.tenant_id, body.category_id, store_name),
    )

    # ── 组装搜索上下文 ────────────────────────────────────────────────
    snippets = [f"【{r['title']}】{r['body']}" for r in search_results if r.get("body")][:5]
    search_context = "\n".join(snippets) if snippets else "（暂无搜索结果，请根据商品名称和店铺风格自行生成）"

    # ── 组装店铺上下文 ────────────────────────────────────────────────
    store_context_parts = [f"店铺名称：{store_ctx['store_name']}"]
    if store_ctx["all_categories"]:
        store_context_parts.append(f"本店分类：{', '.join(store_ctx['all_categories'])}")
    if store_ctx["same_cat_products"]:
        store_context_parts.append(f"同类已有商品：{', '.join(store_ctx['same_cat_products'])}")
    if store_ctx["featured_products"]:
        store_context_parts.append(f"本店热销商品：{', '.join(store_ctx['featured_products'])}")
    store_context_parts.append(f"在售商品总数：{store_ctx['total_products']} 件")
    store_context = "\n".join(store_context_parts)
    store_context_used = bool(store_ctx["all_categories"] or store_ctx["same_cat_products"])

    # ── 构建 Prompt ───────────────────────────────────────────────────
    lang_hint = "用中文回复" if body.lang == "zh" else "Reply in English"
    prompt = f"""你是 {store_ctx['store_name']} 的资深电商运营专家，精通商品文案撰写、SEO 优化和客服知识库建设。
请仔细阅读以下商品信息和搜索参考资料，为该商品生成完整、专业、详细的电商文案。

⚠️ 核心原则：只写搜索资料中能确认的事实，不要编造任何具体数据。
对于无法确认的信息（如成分表、保质期、热量、产地等），不要猜测，直接写"详情请咨询在线客服"。
宁可少写也不要写错，错误信息会损害店铺信誉。

═══ 新商品信息 ═══
名称：{body.name}
关键词：{body.keywords or "无"}
分类：{body.category or "未指定"}

═══ 本店铺数据（用于保持风格一致） ═══
{store_context}

═══ 网络搜索参考资料（请充分利用这些真实信息） ═══
{search_context}

═══ 输出要求 ═══
请严格只返回以下 JSON（不要任何额外文字、解释或 markdown）：
{{
  "description": "400-600字详细商品描述。要求：①开头一句话概括商品核心卖点；②分段介绍产品特色（如原料/工艺/口感/功效/设计亮点等，只写搜索资料中能确认的信息）；③使用场景和适用人群（如送礼、自用、聚会、日常等）；④品质保障或品牌背景（如有搜索到的信息）。用自然流畅的语言，避免堆砌关键词，适合直接作为商品详情页展示。对于无法确认的具体参数不要编造。",
  "meta_title": "SEO标题，50字以内。格式建议：品牌名+商品名+核心卖点+规格。示例：果子熟了 栀栀乌龙970ml*12瓶 无糖茶饮 清香回甘",
  "meta_description": "SEO摘要，120-160字。用一段完整的话概括商品最大亮点、适用场景和购买理由，吸引用户从搜索结果中点击进来。",
  "tags": ["标签1", "标签2", "标签3", "标签4", "标签5", "标签6", "标签7", "标签8"],
  "ai_description": "供AI客服读取和搜索的知识库文本（不在前台展示），要求200-350字，格式如下：\\n第一行：8~15个搜索关键词和同义词（空格分隔），覆盖商品名、品牌名、品类名、口语化叫法、英文名等，越全面用户问询时命中率越高。例如：栀栀乌龙 果子熟了 乌龙茶 无糖茶 瓶装茶 茶饮料 0糖 低卡 解腻\\n第二行起分要点详写：\\n· 商品基本信息（规格/容量等，只写搜索资料中明确提到的数据）\\n· 口感/外观/使用体验描述\\n· 适用场景（办公、运动、聚餐、送礼等）\\n· 与同类商品的差异和优势\\n· 常见客户问题预判及回答要点（对于无法确认的具体数据如保质期、成分、热量等，统一回答"详情请咨询在线客服"，不要编造数字）\\n· 搭配推荐或购买建议"
}}

{lang_hint}，只返回 JSON。"""

    raw = await _call_ai(prompt, extra)
    data = _parse_ai_json(raw, body.name, body.category)

    return ProductAssistResult(
        description=data.get("description", ""),
        meta_title=data.get("meta_title", ""),
        meta_description=data.get("meta_description", ""),
        tags=data.get("tags", []),
        search_snippets=[r["title"] for r in search_results],
        store_context_used=store_context_used,
        ai_description=data.get("ai_description", ""),
    )


# ── SKU/条形码 AI 查询 ─────────────────────────────────────────────────

class SkuLookupRequest(BaseModel):
    sku: str
    force_ai: bool = False  # True=跳过本地命中，强制联网 AI 查询（用于本地信息不完善时）


class SkuLookupResult(BaseModel):
    name: str = ""
    brand: str = ""
    category: str = ""
    description: str = ""
    specifications: str = ""
    price_hint: str = ""
    search_snippets: list[str] = []
    images: list[str] = []


@router.post("/sku-lookup", response_model=SkuLookupResult, summary="通过 SKU/条形码查询商品信息")
async def sku_lookup(
    body: SkuLookupRequest,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_admin_user),
):
    if not body.sku or not body.sku.strip():
        raise HTTPException(status_code=400, detail="请输入 SKU 或条形码")

    result = await db.execute(
        select(TenantSettings).where(TenantSettings.tenant_id == current_user.tenant_id)
    )
    s = result.scalar_one_or_none()
    extra = s.extra or {} if s else {}
    extra = await resolve_ai_extra(db, current_user.tenant_id, extra)

    if not extra.get("ai_enabled"):
        raise HTTPException(status_code=400, detail="AI 功能未启用，请先在系统设置 → AI 配置中开启")

    anysearch_key = extra.get("anysearch_api_key", "")
    sku_val = body.sku.strip()

    # 先查本地：Product.sku → ProductVariant.sku → ProductVariant.barcode
    local_product = (await db.execute(
        select(Product).where(Product.tenant_id == current_user.tenant_id, Product.sku == sku_val)
    )).scalar_one_or_none()

    if not local_product:
        variant = (await db.execute(
            select(ProductVariant).where(
                ProductVariant.tenant_id == current_user.tenant_id,
                (ProductVariant.sku == sku_val) | (ProductVariant.barcode == sku_val),
            )
        )).scalars().first()
        if variant:
            local_product = (await db.execute(
                select(Product).where(
                    Product.id == variant.product_id,
                    Product.tenant_id == current_user.tenant_id,
                )
            )).scalar_one_or_none()

    if local_product and not body.force_ai:
        import re
        plain_desc = re.sub(r"<[^>]+>", "", local_product.description or "").strip()
        from app.core.models.product import ProductImage
        imgs = (await db.execute(
            select(ProductImage)
            .where(ProductImage.product_id == local_product.id)
            .order_by(ProductImage.sort_order)
        )).scalars().all()
        local_images = [(i.webp_url or i.url) for i in imgs if (i.webp_url or i.url)]
        return SkuLookupResult(
            name=local_product.name,
            brand="",
            category="",
            description=plain_desc,
            specifications="",
            price_hint=str(local_product.base_price) if local_product.base_price else "",
            search_snippets=["（本地已有商品）"],
            images=local_images,
        )

    # 本地未命中，消耗 AI 额度后联网搜索
    await consume_ai_quota(db, current_user.tenant_id)

    # 文字搜索（裸码 + 带上下文词各一轮）与 OFF 条码直查并行；
    # 图片搜索放到 AI 识别出商品名之后再做，避免用裸码搜出无关图。
    import asyncio
    r1, r2, off_images = await asyncio.gather(
        _search_web(sku_val, max_results=5, api_key=anysearch_key),
        _search_web(f"{sku_val} barcode product 商品", max_results=5, api_key=anysearch_key),
        _fetch_off_image(sku_val),
    )
    seen_titles = set()
    search_results = []
    for r in r1 + r2:
        if r.get("title") and r["title"] not in seen_titles:
            seen_titles.add(r["title"])
            search_results.append(r)

    snippets = [f"【{r['title']}】{r['body']}" for r in search_results if r.get("body")][:8]
    search_context = "\n".join(snippets) if snippets else "（未找到相关信息）"

    prompt = f"""你是一个商品信息提取专家。用户提供了一个商品条形码/SKU编号，我已经用这个编号搜索了互联网。
请从搜索结果中提取该商品的结构化信息。

⚠️ 核心原则：只提取搜索结果中明确提到的信息，不要猜测或编造。无法确认的字段留空字符串。

═══ 条形码/SKU ═══
{body.sku.strip()}

═══ 搜索结果 ═══
{search_context}

═══ 输出要求 ═══
严格只返回以下 JSON（不要任何额外文字）：
{{
  "name": "商品名称（中文优先，如有英文名可附在后面）",
  "brand": "品牌名（如：Comvita、雀巢等，未知则留空）",
  "category": "商品分类建议（如：保健品、零食饮料、日用品等）",
  "description": "50-150字商品简介，概括主要特点",
  "specifications": "规格信息（如：40粒/盒、500ml、250g等）",
  "price_hint": "参考价格（如搜索结果中有价格信息则填写，格式如 29.90 NZD，无则留空）"
}}"""

    raw = await _call_ai(prompt, extra)
    data = _parse_ai_json(raw, "", "")

    # 用 AI 提取出的品牌+商品名+规格搜图（不含裸码，避免条码数字带出无关图）；
    # OFF 正品图打底排最前，名字搜图去重后补足。多给候选（16 张）供用户挑，
    # 前端限制最多选「剩余可用数(6−已有)」，故不会超过保存接口的 10 张上限。
    name = data.get("name", "")
    brand = data.get("brand", "")
    img_query = " ".join(t for t in (brand, name, data.get("specifications", "")) if t).strip()
    name_images = await _search_images(img_query, max_results=16) if img_query else []
    images = list(off_images)
    for u in name_images:
        if u not in images:
            images.append(u)
    images = images[:16]

    return SkuLookupResult(
        name=name,
        brand=brand,
        category=data.get("category", ""),
        description=data.get("description", ""),
        specifications=data.get("specifications", ""),
        price_hint=data.get("price_hint", ""),
        search_snippets=[r["title"] for r in search_results if r.get("title")],
        images=images,
    )


# ── 保存外部图片到本地存储 ─────────────────────────────────────────────

class ImageSaveRequest(BaseModel):
    urls: list[str]


class ImageSaveResult(BaseModel):
    saved: list[str] = []
    failed: list[str] = []


@router.post("/sku-image-save", response_model=ImageSaveResult, summary="下载外部图片并保存到存储")
async def sku_image_save(
    body: ImageSaveRequest,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_admin_user),
):
    if not body.urls:
        raise HTTPException(status_code=400, detail="请选择至少一张图片")
    if len(body.urls) > 10:
        raise HTTPException(status_code=400, detail="最多保存 10 张图片")

    import os
    import uuid
    import asyncio
    import ipaddress
    from urllib.parse import urlparse
    from app.api.routers.upload import _to_webp, _PIL_OK, _consume_quota_and_get_storage, _refund_quota, UPLOAD_DIR
    from app.core.services.storage_quota import mb_from_bytes

    _LOCAL_PREFIX = "/api/static/uploads/"
    _UPLOAD_ROOT = os.path.abspath(UPLOAD_DIR)

    async def _is_safe_url(url: str) -> bool:
        """校验 URL 并解析域名，拒绝解析到私有/环回/链路本地地址的目标（防 SSRF）。"""
        try:
            parsed = urlparse(url)
            if parsed.scheme not in ("http", "https"):
                return False
            host = parsed.hostname or ""
            if not host:
                return False
            loop = asyncio.get_event_loop()
            infos = await loop.getaddrinfo(host, parsed.port or (443 if parsed.scheme == "https" else 80))
            if not infos:
                return False
            for info in infos:
                ip = ipaddress.ip_address(info[4][0])
                if not ip.is_global:
                    return False
            return True
        except Exception:
            return False

    def _read_local_copy(url: str) -> tuple[bytes, str, str] | None:
        """本地存储图片（相对路径）读盘复制。限定在本租户目录内并防路径穿越。

        返回 (raw, ext, content_type)，非法/读取失败返回 None。
        """
        rel = url[len(_LOCAL_PREFIX):]
        # 只允许读本租户自己的目录，防止越权读取其他租户文件
        if not rel.startswith(f"tenant_{current_user.tenant_id}/"):
            return None
        safe_path = os.path.normpath(os.path.join(_UPLOAD_ROOT, rel))
        if not safe_path.startswith(_UPLOAD_ROOT + os.sep):
            return None
        try:
            with open(safe_path, "rb") as f:
                raw = f.read()
        except Exception:
            return None
        if len(raw) > 5 * 1024 * 1024:
            return None
        ext = os.path.splitext(safe_path)[1].lower() or ".webp"
        ct = "image/webp" if ext == ".webp" else "image/jpeg"
        return raw, ext, ct

    saved = []
    failed = []

    for url in body.urls:
        # ── 本地已有图片：读盘复制一份，保证与源商品/租户解耦 ──
        if url.startswith(_LOCAL_PREFIX):
            local = _read_local_copy(url)
            if not local:
                failed.append(url)
                continue
            raw, ext, stored_ct = local
        else:
            # ── 外网图片：SSRF 校验 + 下载 + 转 WebP ──
            if not await _is_safe_url(url):
                failed.append(url)
                continue
            try:
                # 关闭自动重定向，避免重定向到内网绕过 SSRF 校验
                async with httpx.AsyncClient(timeout=15, follow_redirects=False) as client:
                    resp = await client.get(url, headers={"User-Agent": "Mozilla/5.0"})
                    if resp.status_code != 200:
                        failed.append(url)
                        continue
                    ct = resp.headers.get("content-type", "")
                    if not ct.startswith("image/"):
                        failed.append(url)
                        continue
                    raw = resp.content
                    if len(raw) > 5 * 1024 * 1024:
                        failed.append(url)
                        continue
                    if _PIL_OK:
                        raw = _to_webp(raw)
                        ext = ".webp"
                        stored_ct = "image/webp"
                    else:
                        ext = ".jpg"
                        stored_ct = "image/jpeg"
            except Exception:
                failed.append(url)
                continue

        # 按最终大小扣配额，配额不足/保存失败时当作该图失败，不中断整体流程
        incoming_mb = mb_from_bytes(len(raw))
        try:
            storage, _ = await _consume_quota_and_get_storage(db, current_user.tenant_id, incoming_mb)
        except HTTPException:
            failed.append(url)
            continue
        try:
            filename = f"tenant_{current_user.tenant_id}/{uuid.uuid4().hex}{ext}"
            result_url = await storage.save(filename, raw, content_type=stored_ct)
            saved.append(result_url)
        except Exception:
            await _refund_quota(db, current_user.tenant_id, incoming_mb)
            failed.append(url)

    return ImageSaveResult(saved=saved, failed=failed)