3218485270
f493bde8a9
- 缓存热点批量流程（有采集缓存不触发 Google） - 简报不足直接从采集缓存生成（轻量补齐） - 三图合成（模特/印花/底图）+ 底图压缩 <2MB - 热点去重→风格去重自动切换 + 不适合类目 review 兜底 - 透明背景（background=transparent）+ 提示词清洗（敏感词/背景描述） - 任务前 basemap 校验 + 模板国家校验 + 模特任务级分配
98 lines
4.2 KiB
Python
98 lines
4.2 KiB
Python
"""阿里云 OSS 图床：图片压缩（3:4、≥1340×1785、<2MB）+ 上传。
|
||
|
||
- compress_for_oss(image_path, out_path)：
|
||
中心裁剪到 3:4 → 缩放/放大到 1340×1785 → JPEG quality 迭代压到 <2MB。
|
||
- upload_to_oss(cfg, local_path, object_key)：
|
||
用 oss2 上传到 config.oss 指定的 bucket，返回可访问 URL。
|
||
- build_oss_key(country, timestamp, code, rand4)：
|
||
key = {国家}/{时间戳}/{货号}_{4位随机}.jpg（货号=前缀+3位计数，000 起最多 999）
|
||
- oss 配置（config.yaml oss 段）：
|
||
oss_bucket / oss_endpoint / oss_key_id / oss_key_secret（或环境变量 OSS_ACCESS_KEY_ID / OSS_ACCESS_KEY_SECRET）
|
||
"""
|
||
import io
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
from PIL import Image
|
||
|
||
# 目标规格：3:4 宽高比，最小 1340×1785，文件 < 2MB
|
||
TARGET_W, TARGET_H = 1340, 1785
|
||
MAX_BYTES = 2 * 1024 * 1024
|
||
|
||
|
||
def compress_for_oss(image_path: str, out_path: str, max_bytes: int = MAX_BYTES) -> str:
|
||
"""压缩图片到 3:4 / ≥1340×1785 / <2MB（JPEG）。返回输出路径。"""
|
||
with Image.open(image_path) as im:
|
||
im = im.convert("RGB")
|
||
|
||
# 1) 中心裁剪到 3:4
|
||
w, h = im.size
|
||
if w / h > 3 / 4: # 太宽 → 裁左右
|
||
new_w = int(h * 3 / 4)
|
||
x0 = (w - new_w) // 2
|
||
im = im.crop((x0, 0, x0 + new_w, h))
|
||
elif w / h < 3 / 4: # 太高 → 裁上下
|
||
new_h = int(w * 4 / 3)
|
||
y0 = (h - new_h) // 2
|
||
im = im.crop((0, y0, w, y0 + new_h))
|
||
|
||
# 2) 缩放到目标尺寸（≥1340×1785，正好 3:4）
|
||
im = im.resize((TARGET_W, TARGET_H), Image.LANCZOS)
|
||
|
||
# 3) JPEG quality 迭代，保证 <2MB
|
||
out = Path(out_path)
|
||
out.parent.mkdir(parents=True, exist_ok=True)
|
||
for quality in (92, 85, 78, 70, 62, 55, 48, 40):
|
||
buf = io.BytesIO()
|
||
im.save(buf, "JPEG", quality=quality, optimize=True, progressive=True)
|
||
if buf.tell() <= max_bytes:
|
||
out.write_bytes(buf.getvalue())
|
||
return str(out)
|
||
# 全部超限 → 用最低质量兜底（仍可能 >2MB，打印警告）
|
||
out.write_bytes(buf.getvalue())
|
||
print(f"[oss] 警告: {Path(image_path).name} 压缩后仍 {buf.tell()/1024/1024:.1f}MB > 2MB（质量 {quality}）")
|
||
return str(out)
|
||
|
||
|
||
def upload_to_oss(cfg: dict, local_path: str, object_key: str) -> Optional[str]:
|
||
"""上传本地文件到 OSS，返回 URL；配置缺失/失败返回 None（不中断）。"""
|
||
bucket = (cfg or {}).get("oss_bucket") or ""
|
||
endpoint = (cfg or {}).get("oss_endpoint") or ""
|
||
key_id = (cfg or {}).get("oss_key_id") or ""
|
||
key_secret = (cfg or {}).get("oss_key_secret") or ""
|
||
if not (bucket and endpoint and key_id and key_secret):
|
||
print("[oss] 配置缺失（config.oss），跳过上传")
|
||
return None
|
||
try:
|
||
import oss2
|
||
# 直连 session：忽略环境代理（挂 VPN 时代理会拦截国内 OSS）；
|
||
# oss2.Bucket 的 session 必须是 oss2.Session（内部封装 requests），设其底层 trust_env=False
|
||
_session = oss2.Session()
|
||
try:
|
||
_session.session.trust_env = False
|
||
except AttributeError:
|
||
pass
|
||
auth = oss2.Auth(key_id, key_secret)
|
||
bkt = oss2.Bucket(auth, endpoint, bucket, session=_session)
|
||
with open(local_path, "rb") as f:
|
||
bkt.put_object(object_key, f)
|
||
url = f"https://{bucket}.{endpoint}/{object_key}"
|
||
print(f"[oss] 已上传: {url}")
|
||
return url
|
||
except Exception as e: # noqa: BLE001
|
||
print(f"[oss] 上传失败 {local_path}: {e}")
|
||
return None
|
||
|
||
|
||
def build_oss_key(country: str, timestamp: str, code: str, rand4: str, ext: str = ".jpg") -> str:
|
||
"""对象 key：{国家}/{时间戳}/{货号}_{4位随机}.jpg（如 GB/20260820150945/DG000_Ab3x.jpg）。"""
|
||
safe = lambda s: "".join(c for c in (s or "") if c.isalnum() or c in "-_").strip()
|
||
return f"{safe(country)}/{safe(timestamp)}/{safe(code)}_{safe(rand4)}{ext}"
|
||
|
||
|
||
def random_code4() -> str:
|
||
"""4 位随机：大小写英文 + 数字。"""
|
||
import random
|
||
import string
|
||
return "".join(random.choices(string.ascii_letters + string.digits, k=4))
|