119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
"""微博商品图片下载与内嵌服务。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
from urllib.parse import urlsplit
|
||
|
||
import requests
|
||
from loguru import logger
|
||
|
||
from core.weibo import normalize_weibo_image_url
|
||
|
||
from .file_service import sniff_content_type
|
||
|
||
IMAGE_CONCURRENCY = 6
|
||
IMAGE_TIMEOUT_SECONDS = 3.0
|
||
MAX_IMAGE_BYTES = 5 << 20
|
||
_IMAGE_CONTENT_TYPES = frozenset({"image/jpeg", "image/png", "image/webp"})
|
||
_IMAGE_HEADERS = {
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/156.0.0.0 Safari/537.36"
|
||
),
|
||
"Referer": "https://vipclub.e.weibo.com/",
|
||
"Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8",
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class WeiboImageLoadResult:
|
||
total: int
|
||
loaded: int
|
||
|
||
@property
|
||
def failed(self) -> int:
|
||
return max(0, self.total - self.loaded)
|
||
|
||
|
||
def fetch_weibo_image_data_url(raw_url: str) -> str:
|
||
"""下载一张微博图片，校验后转成浏览器可直接显示的 data URL。"""
|
||
|
||
image_url = normalize_weibo_image_url(raw_url)
|
||
parsed = urlsplit(image_url)
|
||
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
||
raise ValueError("图片地址必须是 HTTP(S) URL")
|
||
|
||
with requests.get(
|
||
image_url,
|
||
headers=_IMAGE_HEADERS,
|
||
timeout=IMAGE_TIMEOUT_SECONDS,
|
||
stream=True,
|
||
) as response:
|
||
if not 200 <= response.status_code < 300:
|
||
raise ValueError(f"图片 HTTP 状态异常: {response.status_code}")
|
||
declared_size = response.headers.get("Content-Length", "").strip()
|
||
if declared_size.isdigit() and int(declared_size) > MAX_IMAGE_BYTES:
|
||
raise ValueError("图片超过 5 MiB")
|
||
|
||
data = bytearray()
|
||
for chunk in response.iter_content(chunk_size=64 * 1024):
|
||
if not chunk:
|
||
continue
|
||
data.extend(chunk)
|
||
if len(data) > MAX_IMAGE_BYTES:
|
||
raise ValueError("图片超过 5 MiB")
|
||
|
||
if not data:
|
||
raise ValueError("图片内容为空")
|
||
content_type = sniff_content_type(bytes(data))
|
||
if content_type not in _IMAGE_CONTENT_TYPES:
|
||
raise ValueError(f"响应不是支持的图片类型: {content_type or 'unknown'}")
|
||
encoded = base64.b64encode(data).decode("ascii")
|
||
return f"data:{content_type};base64,{encoded}"
|
||
|
||
|
||
def embed_weibo_goods_images(
|
||
goods: list[dict[str, Any]],
|
||
*,
|
||
account_id: int | None = None,
|
||
) -> WeiboImageLoadResult:
|
||
"""并发把商品列表图片转成 data URL；单张失败时保留 HTTPS 兜底地址。"""
|
||
|
||
total = len(goods)
|
||
if total == 0:
|
||
logger.info("微博商品图片加载完成：图片 0/0")
|
||
return WeiboImageLoadResult(total=0, loaded=0)
|
||
|
||
loaded = 0
|
||
jobs: dict[Any, int] = {}
|
||
workers = min(IMAGE_CONCURRENCY, total)
|
||
with ThreadPoolExecutor(max_workers=workers) as executor:
|
||
for index, item in enumerate(goods):
|
||
image_url = normalize_weibo_image_url(item.get("goods_image"))
|
||
item["goods_image"] = image_url
|
||
if not image_url:
|
||
continue
|
||
if image_url.startswith("data:image/"):
|
||
loaded += 1
|
||
continue
|
||
jobs[executor.submit(fetch_weibo_image_data_url, image_url)] = index
|
||
|
||
for future in as_completed(jobs):
|
||
index = jobs[future]
|
||
try:
|
||
goods[index]["goods_image"] = future.result()
|
||
except Exception as exc: # noqa: BLE001 - 单张图片失败不能影响商品列表
|
||
logger.debug(f"微博商品图片下载失败：{type(exc).__name__}: {exc}")
|
||
else:
|
||
loaded += 1
|
||
|
||
result = WeiboImageLoadResult(total=total, loaded=loaded)
|
||
context = f"（account_id={account_id}）" if account_id is not None else ""
|
||
logger.info(f"微博商品图片加载完成：图片 {result.loaded}/{result.total}{context}")
|
||
return result
|