wehub-resource-sync
90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from PIL import Image, ImageOps, UnidentifiedImageError
|
|
|
|
from utils.runtime_limits import log_memory
|
|
|
|
|
|
class DocumentConversionError(Exception):
|
|
pass
|
|
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class DocumentConversionService:
|
|
@staticmethod
|
|
def _make_png_output_path(file_path: str, output_dir: str) -> Path:
|
|
input_path = Path(file_path)
|
|
output_path = Path(output_dir) / f"{input_path.stem}.png"
|
|
|
|
try:
|
|
if output_path.resolve() == input_path.resolve():
|
|
output_path = Path(output_dir) / f"{input_path.stem}-converted.png"
|
|
except OSError:
|
|
pass
|
|
|
|
if not output_path.exists():
|
|
return output_path
|
|
|
|
index = 1
|
|
while True:
|
|
candidate = Path(output_dir) / f"{input_path.stem}-{index}.png"
|
|
if not candidate.exists():
|
|
return candidate
|
|
index += 1
|
|
|
|
def convert_image_to_png(
|
|
self,
|
|
file_path: str,
|
|
output_dir: str,
|
|
timeout_seconds: int = 180,
|
|
) -> str:
|
|
del timeout_seconds
|
|
|
|
Path(output_dir).mkdir(parents=True, exist_ok=True)
|
|
output_path = self._make_png_output_path(file_path, output_dir)
|
|
|
|
try:
|
|
LOGGER.info(
|
|
"[DocumentConversion] Image conversion start input=%s output=%s",
|
|
file_path,
|
|
output_path,
|
|
)
|
|
log_memory(LOGGER, "document_conversion.image.start", input=file_path)
|
|
|
|
with Image.open(file_path) as image:
|
|
image.seek(0)
|
|
converted = ImageOps.exif_transpose(image)
|
|
|
|
if converted.mode in ("RGBA", "LA") or (
|
|
converted.mode == "P" and "transparency" in converted.info
|
|
):
|
|
rgba = converted.convert("RGBA")
|
|
background = Image.new("RGB", rgba.size, (255, 255, 255))
|
|
background.paste(rgba, mask=rgba.getchannel("A"))
|
|
converted = background
|
|
elif converted.mode != "RGB":
|
|
converted = converted.convert("RGB")
|
|
|
|
converted.save(output_path, format="PNG")
|
|
|
|
LOGGER.info(
|
|
"[DocumentConversion] Image conversion complete input=%s output=%s",
|
|
file_path,
|
|
output_path,
|
|
)
|
|
log_memory(LOGGER, "document_conversion.image.finish", input=file_path)
|
|
return str(output_path)
|
|
except (OSError, UnidentifiedImageError) as exc:
|
|
LOGGER.error(
|
|
"[DocumentConversion] Image conversion failed input=%s error=%s",
|
|
file_path,
|
|
exc,
|
|
)
|
|
raise DocumentConversionError(
|
|
f"Image conversion failed for {os.path.basename(file_path)}: {exc}"
|
|
) from exc
|