Spaces:

Cnass
/

mangaocr

Sleeping

mangaocr / utils.py

Upload 9 files

184793b verified 3 months ago

9.12 kB

"""
utils.py
Projedeki tum modullerin ortak kullandigi yardimci fonksiyonlar:
loglama kurulumu, SHA256 hashleme, otomatik dosya numaralandirma,
ve kucuk goruntu islemleri.
"""

from __future__ import annotations

import hashlib
import logging
import sys
import zipfile
from pathlib import Path
from typing import List, Optional, Tuple

from PIL import Image
import numpy as np

import config


class ZipTooLargeError(ValueError):
    """Yuklenen zip, izin verilen sayfa sayisi sinirini asarsa firlatilir."""


# ============================================================
# LOGLAMA
# ============================================================

def get_logger(name: str) -> logging.Logger:
    """
    Verilen isimle yapilandirilmis bir logger dondurur.
    Hem konsola (stdout) hem de config.LOG_FILE dosyasina yazar.
    Ayni isimle tekrar cagrildiginda handler'lari coklamaz.
    """
    logger = logging.getLogger(name)
    if logger.handlers:
        return logger

    logger.setLevel(config.LOG_LEVEL)
    formatter = logging.Formatter(config.LOG_FORMAT)

    stream_handler = logging.StreamHandler(sys.stdout)
    stream_handler.setFormatter(formatter)
    logger.addHandler(stream_handler)

    try:
        file_handler = logging.FileHandler(config.LOG_FILE, encoding="utf-8")
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)
    except OSError:
        # Dosya sistemine yazilamiyorsa (salt-okunur ortam vb.) sessizce
        # yalnizca konsol loglamasina devam edilir.
        pass

    logger.propagate = False
    return logger


log = get_logger("utils")


# ============================================================
# HASH / DUPLICATE KONTROLU
# ============================================================

def compute_file_sha256(path: Path) -> str:
    """Bir dosyanin SHA256 hash'ini hesaplar (hex string olarak)."""
    sha256 = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            sha256.update(chunk)
    return sha256.hexdigest()


def compute_bytes_sha256(data: bytes) -> str:
    """Bellekteki bir byte dizisinin SHA256 hash'ini hesaplar."""
    return hashlib.sha256(data).hexdigest()


def compute_image_sha256(image: Image.Image) -> str:
    """
    Bir PIL Image nesnesinin piksel icerigine dayali SHA256 hash'ini
    hesaplar. Ayni gorsel farkli formatlarda kaydedilse bile (PNG/JPEG)
    piksel verisi ayni oldugu surece ayni hash uretilir.
    """
    normalized = image.convert("RGB")
    return compute_bytes_sha256(normalized.tobytes())


# ============================================================
# DOSYA ADLANDIRMA
# ============================================================

def format_sample_filename(index: int) -> str:
    """
    Ornek indeksini config.DATASET.filename_digits basamakli, sifirla
    doldurulmus dosya adina cevirir. Ornek: 245 -> '000245.png'
    """
    digits = config.DATASET.filename_digits
    ext = config.DATASET.filename_ext
    return f"{index:0{digits}d}{ext}"


def format_sample_id(index: int) -> str:
    """Commit mesajlarinda kullanilan sifirla doldurulmus ornek kimligi."""
    digits = config.DATASET.filename_digits
    return f"{index:0{digits}d}"


# ============================================================
# GORUNTU YARDIMCILARI
# ============================================================

def safe_open_image(path: Path) -> Optional[Image.Image]:
    """
    Bir goruntu dosyasini guvenli sekilde acar. Bozuk/desteklenmeyen
    dosyalarda None doner ve hatayi loglar.
    """
    try:
        image = Image.open(path)
        image.load()
        return image.convert("RGB")
    except Exception as exc:  # noqa: BLE001 - kullanici girdisi guvenilmez
        log.error("Goruntu acilamadi: %s (%s)", path, exc)
        return None


def extract_manual_crop_from_editor_composite(composite: Image.Image) -> Optional[Image.Image]:
    """
    Gradio ImageEditor'in crop araciyla uretilen 'composite' ciktisindan
    kullanicinin sectigi dikdortgen bolgeyi cikarir.
    ImageEditor'in crop transform'u, secilen alan disini seffaf
    (alpha=0) yapar ama canvas boyutunu degistirmez. Bu fonksiyon,
    alpha kanalinda opak olan piksellerin bounding box'ini bulup o
    bolgeyi kirpar ve RGB bir goruntu olarak dondurur.
    Kullanici hic kirpma yapmadiysa (tum canvas opak/RGB ise) veya
    goruntude alpha kanali yoksa, goruntunun tamami RGB olarak
    dondurulur (kirpma islenmemis kabul edilir).
    Returns:
        Kirpilmis RGB PIL Image, veya secili bolge bulunamazsa None.
    """
    if composite is None:
        return None

    if composite.mode != "RGBA":
        # Alpha kanali yok; tum goruntu secilmis kabul edilir.
        return composite.convert("RGB")

    alpha = np.array(composite.split()[-1])
    ys, xs = np.where(alpha > 0)

    if ys.size == 0 or xs.size == 0:
        return None

    x_min, x_max = int(xs.min()), int(xs.max()) + 1
    y_min, y_max = int(ys.min()), int(ys.max()) + 1

    cropped = composite.crop((x_min, y_min, x_max, y_max))
    return cropped.convert("RGB")


def clamp(value: float, low: float, high: float) -> float:
    """Bir degeri [low, high] araligina sikistirir."""
    return max(low, min(high, value))


# ============================================================
# ZIP TOPLU YUKLEME
# ============================================================

# Zip icinde manga sayfasi olarak kabul edilen dosya uzantilari.
_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp"}


def extract_images_from_zip(zip_path: Path, output_dir: Path) -> List[Tuple[str, Path]]:
    """
    Bir zip arsivi icindeki tum gecerli goruntu dosyalarini acar ve
    HER BIRINI HEMEN DISKE (output_dir altina) yazar.
    Onemli tasarim karari: fonksiyon PIL Image nesnelerini bellekte
    tutup dondurmez, bunun yerine diske yazip (dosya_adi, disk_yolu)
    ciftleri dondurur. Bunun nedeni: 50+ sayfalik bir manga zip'inde
    tum sayfalari ayni anda PIL Image olarak bellekte tutmak (ozellikle
    bu nesneler bir Gradio State icinde saklanip her etkilesimde
    serialize ediliyorsa) RAM tuketimini ve network payload boyutunu
    asiri buyutup zaman asimi/baglanti hatalarina yol acar. Diske
    yazmak, bellek kullanimini sabit (bir seferde tek goruntu) tutar.
    Sonuc, zip icindeki dosya adina gore (dogal/alfabetik) sirali
    (dosya_adi, disk_yolu) ciftleri listesi olarak doner; boylece
    manga sayfalari genellikle beklenen okuma sirasinda islenir
    (orn. 001.png, 002.png, ...).
    Guvenlik limitleri (config.ZIP_UPLOAD):
        - Zip icindeki sayfa sayisi max_pages'i asarsa ZipTooLargeError.
        - Tek bir goruntunun piksel boyutu asiri buyukse (olasi bir
          "zip bomb" / bozuk dosya senaryosu) o sayfa atlanir.
    Bozuk/acilamayan tekil dosyalar atlanir ve loglanir; tüm arsiv
    basarisiz olmaz.
    Raises:
        zipfile.BadZipFile: Dosya gecerli bir zip arsivi degilse.
        ZipTooLargeError: Sayfa sayisi izin verilen sinirin uzerindeyse.
    """
    output_dir.mkdir(parents=True, exist_ok=True)
    results: List[Tuple[str, Path]] = []

    with zipfile.ZipFile(zip_path, "r") as zf:
        names = sorted(
            n
            for n in zf.namelist()
            if not n.endswith("/")  # dizin girdilerini atla
            and not n.split("/")[-1].startswith(".")  # gizli/sistem dosyalarini atla
            and Path(n).suffix.lower() in _IMAGE_EXTENSIONS
        )

        if len(names) > config.ZIP_UPLOAD.max_pages:
            raise ZipTooLargeError(
                f"Zip icinde {len(names)} sayfa var; "
                f"tek seferde en fazla {config.ZIP_UPLOAD.max_pages} sayfa desteklenir. "
                f"Lutfen zip'i daha kucuk parcalara bolup ayri ayri yukleyin."
            )

        for i, name in enumerate(names, start=1):
            try:
                with zf.open(name) as f:
                    image = Image.open(f)
                    image.load()
                    image = image.convert("RGB")
            except Exception as exc:  # noqa: BLE001 - zip icerigi guvenilmez
                log.error("Zip icindeki goruntu acilamadi: %s (%s)", name, exc)
                continue

            width, height = image.size
            if width * height > config.ZIP_UPLOAD.max_page_pixels:
                log.error(
                    "Sayfa cok buyuk, atlaniyor: %s (%dx%d)", name, width, height
                )
                continue

            # Sayfayi diske yaz; orijinal formattan bagimsiz olarak PNG
            # kullanilir (tutarlilik + segmentasyon/OCR icin guvenilir
            # decode). Dosya adi cakismalarini onlemek icin sira
            # numarasi ile benzersizlestirilir.
            safe_name = f"{i:04d}_{Path(name).stem}.png"
            out_path = output_dir / safe_name
            try:
                image.save(out_path, format="PNG")
            except OSError as exc:
                log.error("Sayfa diske yazilamadi, atlaniyor: %s (%s)", name, exc)
                continue

            results.append((Path(name).name, out_path))

    return results

	"""
	utils.py

	Projedeki tum modullerin ortak kullandigi yardimci fonksiyonlar:
	loglama kurulumu, SHA256 hashleme, otomatik dosya numaralandirma,
	ve kucuk goruntu islemleri.
	"""

	from __future__ import annotations

	import hashlib
	import logging
	import sys
	import zipfile
	from pathlib import Path
	from typing import List, Optional, Tuple

	from PIL import Image
	import numpy as np

	import config


	class ZipTooLargeError(ValueError):
	"""Yuklenen zip, izin verilen sayfa sayisi sinirini asarsa firlatilir."""


	# ============================================================
	# LOGLAMA
	# ============================================================

	def get_logger(name: str) -> logging.Logger:
	"""
	Verilen isimle yapilandirilmis bir logger dondurur.

	Hem konsola (stdout) hem de config.LOG_FILE dosyasina yazar.
	Ayni isimle tekrar cagrildiginda handler'lari coklamaz.
	"""
	logger = logging.getLogger(name)
	if logger.handlers:
	return logger

	logger.setLevel(config.LOG_LEVEL)
	formatter = logging.Formatter(config.LOG_FORMAT)

	stream_handler = logging.StreamHandler(sys.stdout)
	stream_handler.setFormatter(formatter)
	logger.addHandler(stream_handler)

	try:
	file_handler = logging.FileHandler(config.LOG_FILE, encoding="utf-8")
	file_handler.setFormatter(formatter)
	logger.addHandler(file_handler)
	except OSError:
	# Dosya sistemine yazilamiyorsa (salt-okunur ortam vb.) sessizce
	# yalnizca konsol loglamasina devam edilir.
	pass

	logger.propagate = False
	return logger


	log = get_logger("utils")


	# ============================================================
	# HASH / DUPLICATE KONTROLU
	# ============================================================

	def compute_file_sha256(path: Path) -> str:
	"""Bir dosyanin SHA256 hash'ini hesaplar (hex string olarak)."""
	sha256 = hashlib.sha256()
	with open(path, "rb") as f:
	for chunk in iter(lambda: f.read(65536), b""):
	sha256.update(chunk)
	return sha256.hexdigest()


	def compute_bytes_sha256(data: bytes) -> str:
	"""Bellekteki bir byte dizisinin SHA256 hash'ini hesaplar."""
	return hashlib.sha256(data).hexdigest()


	def compute_image_sha256(image: Image.Image) -> str:
	"""
	Bir PIL Image nesnesinin piksel icerigine dayali SHA256 hash'ini
	hesaplar. Ayni gorsel farkli formatlarda kaydedilse bile (PNG/JPEG)
	piksel verisi ayni oldugu surece ayni hash uretilir.
	"""
	normalized = image.convert("RGB")
	return compute_bytes_sha256(normalized.tobytes())


	# ============================================================
	# DOSYA ADLANDIRMA
	# ============================================================

	def format_sample_filename(index: int) -> str:
	"""
	Ornek indeksini config.DATASET.filename_digits basamakli, sifirla
	doldurulmus dosya adina cevirir. Ornek: 245 -> '000245.png'
	"""
	digits = config.DATASET.filename_digits
	ext = config.DATASET.filename_ext
	return f"{index:0{digits}d}{ext}"


	def format_sample_id(index: int) -> str:
	"""Commit mesajlarinda kullanilan sifirla doldurulmus ornek kimligi."""
	digits = config.DATASET.filename_digits
	return f"{index:0{digits}d}"


	# ============================================================
	# GORUNTU YARDIMCILARI
	# ============================================================

	def safe_open_image(path: Path) -> Optional[Image.Image]:
	"""
	Bir goruntu dosyasini guvenli sekilde acar. Bozuk/desteklenmeyen
	dosyalarda None doner ve hatayi loglar.
	"""
	try:
	image = Image.open(path)
	image.load()
	return image.convert("RGB")
	except Exception as exc: # noqa: BLE001 - kullanici girdisi guvenilmez
	log.error("Goruntu acilamadi: %s (%s)", path, exc)
	return None


	def extract_manual_crop_from_editor_composite(composite: Image.Image) -> Optional[Image.Image]:
	"""
	Gradio ImageEditor'in crop araciyla uretilen 'composite' ciktisindan
	kullanicinin sectigi dikdortgen bolgeyi cikarir.

	ImageEditor'in crop transform'u, secilen alan disini seffaf
	(alpha=0) yapar ama canvas boyutunu degistirmez. Bu fonksiyon,
	alpha kanalinda opak olan piksellerin bounding box'ini bulup o
	bolgeyi kirpar ve RGB bir goruntu olarak dondurur.

	Kullanici hic kirpma yapmadiysa (tum canvas opak/RGB ise) veya
	goruntude alpha kanali yoksa, goruntunun tamami RGB olarak
	dondurulur (kirpma islenmemis kabul edilir).

	Returns:
	Kirpilmis RGB PIL Image, veya secili bolge bulunamazsa None.
	"""
	if composite is None:
	return None

	if composite.mode != "RGBA":
	# Alpha kanali yok; tum goruntu secilmis kabul edilir.
	return composite.convert("RGB")

	alpha = np.array(composite.split()[-1])
	ys, xs = np.where(alpha > 0)

	if ys.size == 0 or xs.size == 0:
	return None

	x_min, x_max = int(xs.min()), int(xs.max()) + 1
	y_min, y_max = int(ys.min()), int(ys.max()) + 1

	cropped = composite.crop((x_min, y_min, x_max, y_max))
	return cropped.convert("RGB")


	def clamp(value: float, low: float, high: float) -> float:
	"""Bir degeri [low, high] araligina sikistirir."""
	return max(low, min(high, value))


	# ============================================================
	# ZIP TOPLU YUKLEME
	# ============================================================

	# Zip icinde manga sayfasi olarak kabul edilen dosya uzantilari.
	_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".bmp"}


	def extract_images_from_zip(zip_path: Path, output_dir: Path) -> List[Tuple[str, Path]]:
	"""
	Bir zip arsivi icindeki tum gecerli goruntu dosyalarini acar ve
	HER BIRINI HEMEN DISKE (output_dir altina) yazar.

	Onemli tasarim karari: fonksiyon PIL Image nesnelerini bellekte
	tutup dondurmez, bunun yerine diske yazip (dosya_adi, disk_yolu)
	ciftleri dondurur. Bunun nedeni: 50+ sayfalik bir manga zip'inde
	tum sayfalari ayni anda PIL Image olarak bellekte tutmak (ozellikle
	bu nesneler bir Gradio State icinde saklanip her etkilesimde
	serialize ediliyorsa) RAM tuketimini ve network payload boyutunu
	asiri buyutup zaman asimi/baglanti hatalarina yol acar. Diske
	yazmak, bellek kullanimini sabit (bir seferde tek goruntu) tutar.

	Sonuc, zip icindeki dosya adina gore (dogal/alfabetik) sirali
	(dosya_adi, disk_yolu) ciftleri listesi olarak doner; boylece
	manga sayfalari genellikle beklenen okuma sirasinda islenir
	(orn. 001.png, 002.png, ...).

	Guvenlik limitleri (config.ZIP_UPLOAD):
	- Zip icindeki sayfa sayisi max_pages'i asarsa ZipTooLargeError.
	- Tek bir goruntunun piksel boyutu asiri buyukse (olasi bir
	"zip bomb" / bozuk dosya senaryosu) o sayfa atlanir.

	Bozuk/acilamayan tekil dosyalar atlanir ve loglanir; tüm arsiv
	basarisiz olmaz.

	Raises:
	zipfile.BadZipFile: Dosya gecerli bir zip arsivi degilse.
	ZipTooLargeError: Sayfa sayisi izin verilen sinirin uzerindeyse.
	"""
	output_dir.mkdir(parents=True, exist_ok=True)
	results: List[Tuple[str, Path]] = []

	with zipfile.ZipFile(zip_path, "r") as zf:
	names = sorted(
	n
	for n in zf.namelist()
	if not n.endswith("/") # dizin girdilerini atla
	and not n.split("/")[-1].startswith(".") # gizli/sistem dosyalarini atla
	and Path(n).suffix.lower() in _IMAGE_EXTENSIONS
	)

	if len(names) > config.ZIP_UPLOAD.max_pages:
	raise ZipTooLargeError(
	f"Zip icinde {len(names)} sayfa var; "
	f"tek seferde en fazla {config.ZIP_UPLOAD.max_pages} sayfa desteklenir. "
	f"Lutfen zip'i daha kucuk parcalara bolup ayri ayri yukleyin."
	)

	for i, name in enumerate(names, start=1):
	try:
	with zf.open(name) as f:
	image = Image.open(f)
	image.load()
	image = image.convert("RGB")
	except Exception as exc: # noqa: BLE001 - zip icerigi guvenilmez
	log.error("Zip icindeki goruntu acilamadi: %s (%s)", name, exc)
	continue

	width, height = image.size
	if width * height > config.ZIP_UPLOAD.max_page_pixels:
	log.error(
	"Sayfa cok buyuk, atlaniyor: %s (%dx%d)", name, width, height
	)
	continue

	# Sayfayi diske yaz; orijinal formattan bagimsiz olarak PNG
	# kullanilir (tutarlilik + segmentasyon/OCR icin guvenilir
	# decode). Dosya adi cakismalarini onlemek icin sira
	# numarasi ile benzersizlestirilir.
	safe_name = f"{i:04d}_{Path(name).stem}.png"
	out_path = output_dir / safe_name
	try:
	image.save(out_path, format="PNG")
	except OSError as exc:
	log.error("Sayfa diske yazilamadi, atlaniyor: %s (%s)", name, exc)
	continue

	results.append((Path(name).name, out_path))

	return results