R524809
70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
"""为历史图片素材补齐真实宽高，不改写已有图片文件。"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import mimetypes
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import or_, select
|
|
|
|
SERVER_DIR = Path(__file__).resolve().parents[1]
|
|
if str(SERVER_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SERVER_DIR))
|
|
|
|
from db import get_session_factory
|
|
from config import get_settings
|
|
from models import ProductAsset
|
|
from services.storage import local_path, normalize_image_bytes, read_local_bytes
|
|
|
|
|
|
async def backfill(dry_run: bool = False) -> tuple[int, list[str]]:
|
|
"""返回成功回填数和最多十条失败摘要。"""
|
|
updated = 0
|
|
failures: list[str] = []
|
|
async with get_session_factory()() as db:
|
|
assets = (await db.scalars(
|
|
select(ProductAsset).where(
|
|
ProductAsset.type == "img",
|
|
or_(ProductAsset.width.is_(None), ProductAsset.height.is_(None)),
|
|
)
|
|
)).all()
|
|
for asset in assets:
|
|
url = asset.original_url or asset.stored_url
|
|
path = local_path(url or "")
|
|
if path is None:
|
|
if len(failures) < 10:
|
|
failures.append(f"{asset.id}: 本地原图不存在，跳过")
|
|
continue
|
|
try:
|
|
data = read_local_bytes(path)
|
|
image = normalize_image_bytes(
|
|
data,
|
|
max_bytes=max(len(data), get_settings().image_download_max_bytes),
|
|
)
|
|
asset.width = image.width
|
|
asset.height = image.height
|
|
updated += 1
|
|
except Exception as exc: # noqa: BLE001
|
|
if len(failures) < 10:
|
|
failures.append(f"{asset.id}: {exc}")
|
|
if not dry_run:
|
|
await db.commit()
|
|
return updated, failures
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--dry-run", action="store_true", help="只检查，不写数据库")
|
|
args = parser.parse_args()
|
|
updated, failures = asyncio.run(backfill(dry_run=args.dry_run))
|
|
mode = "将回填" if args.dry_run else "已回填"
|
|
print(f"{mode} {updated} 条图片素材的宽高")
|
|
for failure in failures:
|
|
print(f"跳过：{failure}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|