# -*- coding: utf-8 -*- """ 썸네일 슬롯 한정 픽셀(헤더) 룰 스코어러 — Pillow/numpy 의존성 없이 구현. 배경 (zzz/ado2-thumbnail-selection 파일럿 대비 축소 이식): 기존 태그 매칭(calculate_image_slot_score_multi)은 의미 태그 교집합만 보고 실제 화질·프레이밍을 보지 않는다. 파일럿(수원화성, 2026-07-20)에서 검증된 6축 스코어러 중 "실질 가치는 최고 컷 찾기보다 명백히 나쁜 컷의 자동 배제" (참고: references/scoring-logic.md)라는 결론에 따라, 하드 리젝트에 직결되는 두 축(해상도 업스케일 배율, 종횡비)만 이식한다. 이 두 축은 실제 픽셀 색상이 필요 없고 원본 가로/세로 픽셀 수만 알면 되므로, 이미지를 전체 디코드하지 않고 파일 헤더 몇십 KB만 읽어(HTTP Range) 포맷별 고정 위치에서 크기를 파싱한다(JPEG는 SOF 마커 스캔). Pillow의 픽셀 디코드가 없으므로 서버(Standard_B2als_v2, 2 vCPU 버스터블) CPU 부담이 사실상 0에 가깝다. 텍스트 안전영역 밀도·명암 대비(원 스코어러의 safe_area/contrast 축)는 실제 픽셀 값이 있어야 계산되는 축이라 이 방식으로는 커버하지 못한다 — 의도적으로 범위 밖으로 남겨둔 트레이드오프. """ from __future__ import annotations import asyncio import struct import httpx from app.utils.logger import get_logger logger = get_logger("thumbnail_fitness") # ── 커버 스펙 (thumbnail-composition 렌더 기준) ────────────────────────── TARGET_H = 1920 # 9:16 출력 캔버스 기준 세로 px _CROP_RATIO = 9 / 16 # tight_crop(fit=cover) 목표 종횡비 REJECT_UPSCALE = 2.5 # 9:16 크롭 후 업스케일 배율 초과 → 배제 (OG 1200×630 ≈ 3.05배) REJECT_ASPECT = 2.2 # 원본 종횡비(w/h) 초과 극단 가로형 → 배제 _RANGE_BYTES = 131072 # 헤더만 읽기 위한 최대 다운로드 크기 (128KB, SOF 스캔 여유분) _DOWNLOAD_TIMEOUT = 8.0 _DEFAULT_CONCURRENCY = 4 # 헤더만 받으므로 CPU 부담 없음 — I/O 대기 기준으로 넉넉히 # ── 포맷별 헤더 파싱 (순수 struct, 전체 디코드 없음) ────────────────────── def parse_image_size(data: bytes) -> tuple[int, int] | None: """이미지 바이트(파일 앞부분)에서 원본 (width, height)를 추출합니다. 지원: JPEG(SOF 마커 스캔), PNG(IHDR), GIF(고정 오프셋), WebP(VP8X/VP8L). 단순 손실 WebP(VP8)는 비트 단위 파싱이 필요해 범위 밖 — 실패 시 None (호출자가 중립 처리, 배제하지 않음). """ if len(data) < 12: return None if data[:2] == b"\xff\xd8": return _parse_jpeg_size(data) if data[:8] == b"\x89PNG\r\n\x1a\n": return _parse_png_size(data) if data[:6] in (b"GIF87a", b"GIF89a"): return _parse_gif_size(data) if data[:4] == b"RIFF" and data[8:12] == b"WEBP": return _parse_webp_size(data) return None def _parse_png_size(data: bytes) -> tuple[int, int] | None: if len(data) < 24: return None width, height = struct.unpack(">II", data[16:24]) return width, height def _parse_gif_size(data: bytes) -> tuple[int, int] | None: if len(data) < 10: return None width, height = struct.unpack(" tuple[int, int] | None: n = len(data) pos = 2 # SOI(0xFFD8) 이후부터 마커 스캔 while pos < n: if data[pos] != 0xFF: pos += 1 continue # 0xFF 뒤 연속된 fill byte(0xFF) 스킵 후 실제 마커 바이트 탐색 marker_pos = pos while marker_pos < n and data[marker_pos] == 0xFF: marker_pos += 1 if marker_pos >= n: return None marker = data[marker_pos] pos = marker_pos + 1 if marker in _JPEG_NO_LENGTH_MARKERS: continue if pos + 2 > n: return None seg_len = struct.unpack(">H", data[pos:pos + 2])[0] if marker in _JPEG_SOF_MARKERS: if pos + 7 > n: return None # SOF가 헤더 범위 밖에 있음 — 파싱 실패(중립 처리) height, width = struct.unpack(">HH", data[pos + 3:pos + 7]) return width, height pos += seg_len return None def _parse_webp_size(data: bytes) -> tuple[int, int] | None: if len(data) < 30: return None fourcc = data[12:16] if fourcc == b"VP8X": width = 1 + (data[24] | (data[25] << 8) | (data[26] << 16)) height = 1 + (data[27] | (data[28] << 8) | (data[29] << 16)) return width, height if fourcc == b"VP8L": if len(data) < 25 or data[20] != 0x2F: return None b0, b1, b2, b3 = data[21:25] bits = b0 | (b1 << 8) | (b2 << 16) | (b3 << 24) width = (bits & 0x3FFF) + 1 height = ((bits >> 14) & 0x3FFF) + 1 return width, height # 단순 손실 WebP(VP8) — 비트 단위 파싱 범위 밖, 중립 처리 return None # ── 점수화 (실제 픽셀 없이 w,h만으로 산출) ───────────────────────────── def score_pixel_fitness(width: int, height: int) -> dict: """해상도(업스케일 배율)·종횡비만으로 썸네일 적합도를 판정합니다. Returns: {"score": 0.0~1.0 (배율, 태그 점수에 곱해 쓴다), "reject": str} reject가 비어있지 않으면 하드 배제 대상. """ if width <= 0 or height <= 0: return {"score": 0.0, "reject": "invalid_dimensions"} ratio = width / height if ratio > REJECT_ASPECT: return {"score": 0.0, "reject": f"극단 가로형({ratio:.2f}:1)"} cropped_h = height if ratio > _CROP_RATIO else int(width / _CROP_RATIO) if cropped_h <= 0: return {"score": 0.0, "reject": "invalid_dimensions"} upscale = TARGET_H / cropped_h if upscale > REJECT_UPSCALE: return {"score": 0.0, "reject": f"저해상도(업스케일 {upscale:.2f}x)"} soft_score = 1.0 if upscale <= 1.0 else max(0.0, 1.0 - (upscale - 1.0) / 2.0) return {"score": round(soft_score, 4), "reject": ""} # ── 네트워크: 헤더 바이트만 받기 (HTTP Range, 미지원 서버도 조기 종료로 방어) ── async def _fetch_header_bytes( client: httpx.AsyncClient, url: str, max_bytes: int = _RANGE_BYTES, timeout: float = _DOWNLOAD_TIMEOUT ) -> bytes | None: """이미지 URL에서 앞부분 max_bytes만 받아옵니다. Range 헤더를 무시하고 전체를 돌려주는 서버가 있어도, 스트리밍을 max_bytes 수신 즉시 중단해 실제 다운로드량을 상한선 이내로 강제한다. """ try: async with client.stream( "GET", url, headers={"Range": f"bytes=0-{max_bytes - 1}"}, timeout=timeout ) as response: if response.status_code not in (200, 206): return None buf = bytearray() async for chunk in response.aiter_bytes(): buf.extend(chunk) if len(buf) >= max_bytes: break return bytes(buf[:max_bytes]) except Exception as e: logger.warning(f"[thumbnail_fitness] 헤더 다운로드 실패: {url} - {e}") return None async def score_pool_thumbnail_fitness( pool: list[dict], client: httpx.AsyncClient, concurrency: int = _DEFAULT_CONCURRENCY, total_timeout: float = 20.0, ) -> dict[str, dict]: """이미지 풀(URL 중복 제거)의 썸네일 픽셀 적합도를 병렬로 계산합니다. 다운로드/파싱 실패 항목은 결과 dict에서 아예 빠진다 — 호출자가 "데이터 없음 = 판정 보류(중립)"로 처리하도록 유도(하드 배제로 오판하지 않기 위함). 전체 배치에 total_timeout 상한을 걸어, 네트워크 불량 시 이 부가 기능이 영상 생성 사전 준비 전체를 지연시키지 않도록 한다(초과 시 빈 dict). Returns: {image_url: {"score": float, "reject": str}} — 성공한 URL만 포함. """ semaphore = asyncio.Semaphore(concurrency) urls = list({item["image_url"] for item in pool if item.get("image_url")}) async def _score_one(url: str) -> tuple[str, dict | None]: async with semaphore: header = await _fetch_header_bytes(client, url) if header is None: return url, None size = parse_image_size(header) if size is None: logger.warning(f"[thumbnail_fitness] 이미지 크기 파싱 실패(포맷 미지원/헤더 부족): {url}") return url, None width, height = size return url, score_pixel_fitness(width, height) try: results = await asyncio.wait_for( asyncio.gather(*[_score_one(u) for u in urls]), timeout=total_timeout, ) except asyncio.TimeoutError: logger.warning( f"[thumbnail_fitness] 전체 스코어링 타임아웃({total_timeout}s, urls={len(urls)}) — " "픽셀 적합도 없이(태그 점수만) 진행" ) return {} return {url: fitness for url, fitness in results if fitness is not None}