Files

55 lines
1.7 KiB
Python

"""图片代理:绕过源站防盗链,供前端 <img> 预览与生图参考使用。
参考 laowang.putumiao.shop 的 /api/proxy-image?url=... 形式。
"""
from __future__ import annotations
from urllib.parse import urlparse
from fastapi import APIRouter, HTTPException, Query, Response
from services import storage
router = APIRouter(prefix="/api", tags=["proxy"])
# 域名片段 → 防盗链所需 Referer
_REFERER_BY_DOMAIN: list[tuple[str, str]] = [
("alicdn.com", "https://www.taobao.com"),
("taobao.com", "https://www.taobao.com"),
("tmall.com", "https://www.tmall.com"),
("1688.com", "https://www.1688.com"),
("ozon.ru", "https://www.ozon.ru"),
("ozon.kz", "https://www.ozon.ru"),
("ozon.by", "https://www.ozon.ru"),
("ozonusercontent.com", "https://www.ozon.ru"),
]
def guess_referer(url: str) -> str | None:
host = (urlparse(url).hostname or "").lower()
for frag, referer in _REFERER_BY_DOMAIN:
if frag in host:
return referer
return None
@router.get("/proxy-image")
async def proxy_image(url: str = Query(..., description="源站图片 URL")):
scheme = urlparse(url).scheme
if scheme not in ("http", "https"):
raise HTTPException(status_code=400, detail="仅支持 http/https URL")
try:
data, ctype = await storage.download_bytes(url, referer=guess_referer(url))
except Exception as exc: # noqa: BLE001
raise HTTPException(status_code=502, detail=f"图片拉取失败: {exc}") from exc
if not ctype.startswith("image/"):
ctype = "image/jpeg"
return Response(
content=data,
media_type=ctype,
headers={
"Cache-Control": "public, max-age=86400",
"Access-Control-Allow-Origin": "*",
},
)