151 lines
6.2 KiB
Python
151 lines
6.2 KiB
Python
"""② split — 상세페이지 롱이미지를 섹션 조각으로 자르고 태깅한다.
|
|
|
|
공지·이벤트·스틸·시놉시스·캐스트·스케줄이 한 장(세로 수천~1만px)에 세로로 붙어 있어
|
|
영상 씬으로 쓰려면 잘라야 한다.
|
|
|
|
경계는 대개 한 색으로 균일한 가로 띠라, 행마다 표준편차를 재서 균일한 행이 gap_min 이상
|
|
이어지면 경계로 본다. 검정 배경처럼 배경 자체가 균일하면 구분이 약해지므로 그 띠 안에서
|
|
좌우 끝 색이 바뀌는 지점을 우선한다.
|
|
"""
|
|
import asyncio
|
|
import io
|
|
from collections.abc import Sequence
|
|
|
|
import numpy as np
|
|
from PIL import Image
|
|
|
|
from answers.detail_section_tag_answer import DetailSectionTagAnswer
|
|
from models.detail_section import UNTAGGED, DetailSection, DetailSplitResult
|
|
from settings import settings
|
|
from utils.common_llm import StructuredLLM
|
|
from utils.image import to_data_uri
|
|
from utils.prompt import load_prompt
|
|
|
|
Image.MAX_IMAGE_PIXELS = None
|
|
|
|
UNIFORM_STD_THRESHOLD = 6.0 # 행 내 RGB 표준편차가 이 아래면 '한 색'으로 본다
|
|
DEFAULT_GAP_MIN = 40 # 경계로 볼 균일 띠 최소 높이
|
|
DEFAULT_MIN_HEIGHT = 260 # 조각 최소 높이. 미만이면 앞 조각에 병합
|
|
COLOUR_SHIFT_THRESHOLD = 30 # 균일 띠 안에서 배경색이 바뀌었다고 볼 차이
|
|
TRIM_KEEP_PX = 12 # 조각 위아래에 남길 여백
|
|
|
|
SHEET_TILE_W = 180
|
|
SHEET_TILE_MAX_H = 900
|
|
SHEET_MAX_COLS = 14
|
|
SHEET_GAP = 8
|
|
TAG_IMAGE_MAX_SIZE = (360, 1200)
|
|
|
|
DETAIL_SECTION_TAG_PROMPT = load_prompt("detail_section_tag")
|
|
|
|
detail_section_llm = StructuredLLM("gpt-4o", settings.chatgpt_api_key)
|
|
|
|
|
|
def uniform_rows(pixels: np.ndarray, threshold: float = UNIFORM_STD_THRESHOLD) -> np.ndarray:
|
|
"""각 행이 한 색에 가까운지."""
|
|
return pixels.std(axis=1).max(axis=1) <= threshold
|
|
|
|
|
|
def split_bounds(image: Image.Image, gap_min: int = DEFAULT_GAP_MIN,
|
|
min_height: int = DEFAULT_MIN_HEIGHT) -> list[tuple[int, int]]:
|
|
pixels = np.asarray(image.convert("RGB")).astype(np.float32)
|
|
height = pixels.shape[0]
|
|
uniform = uniform_rows(pixels)
|
|
|
|
cuts = []
|
|
y = 0
|
|
while y < height:
|
|
if not uniform[y]:
|
|
y += 1
|
|
continue
|
|
band_start = y
|
|
while y < height and uniform[y]:
|
|
y += 1
|
|
if y - band_start < gap_min:
|
|
continue
|
|
# 경계는 여백 한가운데. 여백 안에서 배경색이 바뀌면 그 지점을 우선한다
|
|
edge = pixels[band_start:y, 0, :]
|
|
shift = np.abs(np.diff(edge, axis=0)).sum(axis=1)
|
|
peak = int(shift.argmax())
|
|
cuts.append(band_start + peak + 1 if shift[peak] > COLOUR_SHIFT_THRESHOLD
|
|
else (band_start + y) // 2)
|
|
|
|
bounds = []
|
|
previous = 0
|
|
for cut in cuts + [height]:
|
|
if cut - previous >= min_height:
|
|
bounds.append((previous, cut))
|
|
previous = cut
|
|
if previous < height:
|
|
# 남은 꼬리는 마지막 조각에 붙인다 — 얇은 구분선 때문에 쪼개지는 것을 막는다
|
|
if bounds:
|
|
bounds[-1] = (bounds[-1][0], height)
|
|
else:
|
|
bounds.append((0, height))
|
|
return bounds
|
|
|
|
|
|
def trim_uniform(image: Image.Image, keep: int = TRIM_KEEP_PX) -> Image.Image:
|
|
"""조각 위아래의 균일 여백을 keep px만 남기고 걷어낸다."""
|
|
uniform = uniform_rows(np.asarray(image.convert("RGB")).astype(np.float32))
|
|
content = np.where(~uniform)[0]
|
|
if len(content) == 0:
|
|
return image
|
|
top = max(0, content[0] - keep)
|
|
bottom = min(image.height, content[-1] + 1 + keep)
|
|
return image.crop((0, top, image.width, bottom))
|
|
|
|
|
|
def contact_sheet(sections: Sequence[DetailSection]) -> Image.Image:
|
|
"""조각을 폭 180으로 줄여 타일로 붙인다. 긴 조각은 900px 단위로 끊는다."""
|
|
tiles = []
|
|
for section in sections:
|
|
scaled = section.image.resize(
|
|
(SHEET_TILE_W, max(1, round(section.image.height * SHEET_TILE_W
|
|
/ section.image.width))))
|
|
for top in range(0, scaled.height, SHEET_TILE_MAX_H):
|
|
tiles.append(scaled.crop((0, top, SHEET_TILE_W,
|
|
min(top + SHEET_TILE_MAX_H, scaled.height))))
|
|
if not tiles:
|
|
return Image.new("RGB", (SHEET_TILE_W, SHEET_TILE_MAX_H), "white")
|
|
|
|
cols = min(len(tiles), SHEET_MAX_COLS)
|
|
rows = (len(tiles) + cols - 1) // cols
|
|
sheet = Image.new("RGB", (cols * (SHEET_TILE_W + SHEET_GAP),
|
|
rows * (SHEET_TILE_MAX_H + SHEET_GAP)), "white")
|
|
for order, tile in enumerate(tiles):
|
|
sheet.paste(tile, ((order % cols) * (SHEET_TILE_W + SHEET_GAP),
|
|
(order // cols) * (SHEET_TILE_MAX_H + SHEET_GAP)))
|
|
return sheet
|
|
|
|
|
|
async def tag_section(section: DetailSection) -> str:
|
|
"""태깅 실패가 분할을 막지 않도록 실패는 untagged로 흘린다."""
|
|
try:
|
|
answer = await detail_section_llm.ask_with_images(
|
|
DetailSectionTagAnswer, DETAIL_SECTION_TAG_PROMPT,
|
|
[("조각:", to_data_uri(section.image, TAG_IMAGE_MAX_SIZE))], detail="low")
|
|
except Exception:
|
|
return UNTAGGED
|
|
return answer.tag
|
|
|
|
|
|
async def split_details(details: Sequence[bytes | Image.Image], *, tag: bool = True,
|
|
gap_min: int = DEFAULT_GAP_MIN,
|
|
min_height: int = DEFAULT_MIN_HEIGHT) -> DetailSplitResult:
|
|
sections: list[DetailSection] = []
|
|
for order, detail in enumerate(details, 1):
|
|
image = (Image.open(io.BytesIO(detail)) if isinstance(detail, bytes)
|
|
else detail).convert("RGB")
|
|
source = f"detail_{order:02d}"
|
|
for index, (y0, y1) in enumerate(split_bounds(image, gap_min, min_height), 1):
|
|
piece = trim_uniform(image.crop((0, y0, image.width, y1)))
|
|
sections.append(DetailSection(source=source, index=index, y0=y0, y1=y1,
|
|
image=piece))
|
|
|
|
if tag:
|
|
tags = await asyncio.gather(*(tag_section(section) for section in sections))
|
|
for section, resolved in zip(sections, tags, strict=True):
|
|
section.tag = resolved
|
|
|
|
return DetailSplitResult(sections=sections, sheet=contact_sheet(sections))
|