441 lines
15 KiB
Python
441 lines
15 KiB
Python
"""Single-file import pipeline.
|
|
|
|
The Importer is the synchronous core of FC-2a's import path. A Celery task
|
|
(tasks/import_file.py) instantiates one per job and calls import_one().
|
|
|
|
The service is intentionally not async — Celery tasks are run in a
|
|
synchronous worker process, and the DB session is a sync session passed in
|
|
by the task. The Quart side uses an async session via the same models.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import shutil
|
|
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
from pathlib import Path
|
|
|
|
from PIL import Image
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models import (
|
|
Artist,
|
|
ImageProvenance,
|
|
ImageRecord,
|
|
ImportSettings,
|
|
Post,
|
|
Source,
|
|
)
|
|
from ..utils.paths import derive_subdir, derive_top_level_artist, hash_suffixed_name
|
|
from ..utils.phash import compute_phash, find_similar
|
|
from ..utils.sidecar import find_sidecar, parse_sidecar
|
|
from ..utils.slug import slugify
|
|
from .thumbnailer import Thumbnailer
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class SkipReason(StrEnum):
|
|
too_small = "too_small"
|
|
too_transparent = "too_transparent"
|
|
single_color = "single_color"
|
|
duplicate_hash = "duplicate_hash"
|
|
duplicate_phash = "duplicate_phash"
|
|
invalid_image = "invalid_image"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ImportResult:
|
|
status: str # 'imported' | 'skipped' | 'failed' | 'superseded'
|
|
image_id: int | None = None
|
|
skip_reason: SkipReason | None = None
|
|
error: str | None = None
|
|
|
|
|
|
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tif", ".tiff"}
|
|
VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v", ".wmv", ".flv"}
|
|
ALL_EXTS = IMAGE_EXTS | VIDEO_EXTS
|
|
|
|
|
|
def is_supported(path: Path) -> bool:
|
|
return path.suffix.lower() in ALL_EXTS
|
|
|
|
|
|
def is_video(path: Path) -> bool:
|
|
return path.suffix.lower() in VIDEO_EXTS
|
|
|
|
|
|
def _mime_for(path: Path) -> str:
|
|
suffix = path.suffix.lower()
|
|
image_mimes = {
|
|
".png": "image/png",
|
|
".jpg": "image/jpeg",
|
|
".jpeg": "image/jpeg",
|
|
".gif": "image/gif",
|
|
".webp": "image/webp",
|
|
".bmp": "image/bmp",
|
|
".tif": "image/tiff",
|
|
".tiff": "image/tiff",
|
|
}
|
|
video_mimes = {
|
|
".mp4": "video/mp4",
|
|
".mov": "video/quicktime",
|
|
".avi": "video/x-msvideo",
|
|
".mkv": "video/x-matroska",
|
|
".webm": "video/webm",
|
|
".m4v": "video/x-m4v",
|
|
".wmv": "video/x-ms-wmv",
|
|
".flv": "video/x-flv",
|
|
}
|
|
return image_mimes.get(suffix) or video_mimes.get(suffix) or "application/octet-stream"
|
|
|
|
|
|
def _sha256_of(path: Path, chunk_size: int = 1 << 20) -> str:
|
|
h = hashlib.sha256()
|
|
with path.open("rb") as fp:
|
|
while True:
|
|
block = fp.read(chunk_size)
|
|
if not block:
|
|
break
|
|
h.update(block)
|
|
return h.hexdigest()
|
|
|
|
|
|
class Importer:
|
|
"""Imports one file at a time. Constructed per-task by the Celery task."""
|
|
|
|
def __init__(
|
|
self,
|
|
session: Session,
|
|
images_root: Path,
|
|
import_root: Path,
|
|
thumbnailer: Thumbnailer,
|
|
settings: ImportSettings,
|
|
):
|
|
self.session = session
|
|
self.images_root = images_root
|
|
self.import_root = import_root
|
|
self.thumbnailer = thumbnailer
|
|
self.settings = settings
|
|
|
|
def import_one(self, source: Path) -> ImportResult:
|
|
"""The full pipeline for a single file.
|
|
|
|
Returns an ImportResult; callers (the Celery task) are responsible for
|
|
translating that into ImportTask.status and ImageRecord linkage.
|
|
"""
|
|
if not is_supported(source):
|
|
return ImportResult(
|
|
status="skipped", skip_reason=SkipReason.invalid_image,
|
|
error=f"unsupported extension {source.suffix}",
|
|
)
|
|
|
|
# Compute file dimensions (images only) and apply filters.
|
|
width = height = None
|
|
has_alpha = False
|
|
if not is_video(source):
|
|
try:
|
|
with Image.open(source) as im:
|
|
im.verify()
|
|
with Image.open(source) as im:
|
|
width, height = im.size
|
|
has_alpha = im.mode in ("RGBA", "LA") or (
|
|
im.mode == "P" and "transparency" in im.info
|
|
)
|
|
except Exception as exc:
|
|
return ImportResult(
|
|
status="skipped", skip_reason=SkipReason.invalid_image, error=str(exc),
|
|
)
|
|
|
|
if (self.settings.min_width and width < self.settings.min_width) or (
|
|
self.settings.min_height and height < self.settings.min_height
|
|
):
|
|
return ImportResult(
|
|
status="skipped", skip_reason=SkipReason.too_small,
|
|
error=f"{width}x{height} below min {self.settings.min_width}x{self.settings.min_height}",
|
|
)
|
|
|
|
if self.settings.skip_transparent and has_alpha:
|
|
pct = self._transparency_pct(source)
|
|
if pct >= self.settings.transparency_threshold:
|
|
return ImportResult(
|
|
status="skipped", skip_reason=SkipReason.too_transparent,
|
|
error=f"{pct:.2%} transparent",
|
|
)
|
|
|
|
# Hash dedup (exact).
|
|
sha = _sha256_of(source)
|
|
existing_stmt = select(ImageRecord).where(ImageRecord.sha256 == sha)
|
|
existing = self.session.execute(existing_stmt).scalar_one_or_none()
|
|
if existing:
|
|
return ImportResult(
|
|
status="skipped", skip_reason=SkipReason.duplicate_hash,
|
|
image_id=existing.id, error="sha256 already present",
|
|
)
|
|
|
|
# Perceptual near-dup (images only; videos keep phash NULL).
|
|
phash = None
|
|
if not is_video(source):
|
|
with Image.open(source) as im:
|
|
phash = compute_phash(im)
|
|
if phash is not None:
|
|
cand_rows = self.session.execute(
|
|
select(
|
|
ImageRecord.phash,
|
|
ImageRecord.width,
|
|
ImageRecord.height,
|
|
ImageRecord.id,
|
|
).where(ImageRecord.phash.is_not(None))
|
|
).all()
|
|
candidates = [
|
|
(c.phash, c.width or 0, c.height or 0, c.id)
|
|
for c in cand_rows
|
|
]
|
|
rel, match_id = find_similar(
|
|
phash, width or 0, height or 0,
|
|
candidates, self.settings.phash_threshold,
|
|
)
|
|
if rel == "larger_exists":
|
|
return ImportResult(
|
|
status="skipped",
|
|
skip_reason=SkipReason.duplicate_phash,
|
|
image_id=match_id,
|
|
error="perceptual near-duplicate of larger existing image",
|
|
)
|
|
if rel == "smaller_exists":
|
|
target = self.session.get(ImageRecord, match_id)
|
|
self._supersede(target, source, sha, phash, width, height)
|
|
return ImportResult(
|
|
status="superseded", image_id=match_id
|
|
)
|
|
|
|
# Destination path.
|
|
subdir = derive_subdir(source, self.import_root)
|
|
dest_dir = self.images_root / subdir if subdir else self.images_root
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest_name = hash_suffixed_name(source.stem, sha, source.suffix)
|
|
dest = dest_dir / dest_name
|
|
|
|
# Atomic copy: write to .partial, rename. Avoids half-imported files on crash.
|
|
partial = dest.with_suffix(dest.suffix + ".partial")
|
|
shutil.copy2(source, partial)
|
|
partial.rename(dest)
|
|
|
|
record = ImageRecord(
|
|
path=str(dest),
|
|
sha256=sha,
|
|
phash=phash,
|
|
size_bytes=dest.stat().st_size,
|
|
mime=_mime_for(source),
|
|
width=width,
|
|
height=height,
|
|
origin="imported_filesystem",
|
|
integrity_status="unknown",
|
|
)
|
|
self.session.add(record)
|
|
self.session.flush()
|
|
|
|
# Folder→artist auto-tag.
|
|
artist = None
|
|
artist_name = derive_top_level_artist(source, self.import_root)
|
|
if artist_name:
|
|
artist = self._attach_artist(record, artist_name)
|
|
|
|
# Sidecar provenance (best-effort; never fails the import).
|
|
self._apply_sidecar(record, source, artist)
|
|
|
|
# Thumbnail is queued separately by the calling task; the importer
|
|
# does not generate thumbnails inline so the import queue stays moving.
|
|
|
|
self.session.commit()
|
|
return ImportResult(status="imported", image_id=record.id)
|
|
|
|
def _upsert_artist(self, name: str) -> Artist:
|
|
slug = slugify(name)
|
|
artist = self.session.execute(
|
|
select(Artist).where(Artist.slug == slug)
|
|
).scalar_one_or_none()
|
|
if artist is None:
|
|
artist = Artist(name=name, slug=slug, is_subscription=False)
|
|
self.session.add(artist)
|
|
self.session.flush()
|
|
return artist
|
|
|
|
def _attach_artist(self, record: ImageRecord, artist_name: str) -> Artist:
|
|
"""Upsert the Artist and set it as the image's canonical artist.
|
|
Artist-kind tags were retired in FC-2d-vii-c — the Artist row is
|
|
the single source of truth. Returns the Artist (sidecar
|
|
provenance attaches its Source to it)."""
|
|
artist = self._upsert_artist(artist_name)
|
|
if record.artist_id is None:
|
|
record.artist_id = artist.id
|
|
self.session.flush()
|
|
return artist
|
|
|
|
@staticmethod
|
|
def _sidecar_artist_name(data: dict) -> str | None:
|
|
for k in ("artist", "author_name"):
|
|
v = data.get(k)
|
|
if isinstance(v, str) and v.strip():
|
|
return v.strip()
|
|
for k in ("user", "creator"):
|
|
v = data.get(k)
|
|
if isinstance(v, str) and v.strip():
|
|
return v.strip()
|
|
if isinstance(v, dict):
|
|
for sub in ("name", "full_name", "vanity", "account"):
|
|
sv = v.get(sub)
|
|
if isinstance(sv, str) and sv.strip():
|
|
return sv.strip()
|
|
return None
|
|
|
|
def _apply_sidecar(
|
|
self, record: ImageRecord, source: Path, artist: Artist | None
|
|
) -> None:
|
|
sc = find_sidecar(source)
|
|
if sc is None:
|
|
return
|
|
try:
|
|
data = json.loads(sc.read_text("utf-8"))
|
|
if not isinstance(data, dict):
|
|
raise ValueError("sidecar JSON is not an object")
|
|
except Exception as exc:
|
|
log.warning("sidecar parse failed for %s: %s", sc, exc)
|
|
return
|
|
|
|
sd = parse_sidecar(data)
|
|
|
|
if artist is None:
|
|
name = self._sidecar_artist_name(data)
|
|
if name:
|
|
artist = self._upsert_artist(name)
|
|
if artist is None:
|
|
log.warning(
|
|
"sidecar present for %s but no artist; skipping provenance",
|
|
source,
|
|
)
|
|
return
|
|
|
|
if record.artist_id is None:
|
|
record.artist_id = artist.id
|
|
|
|
platform = sd.platform or "unknown"
|
|
url = sd.post_url or f"sidecar:{platform}"
|
|
src = self.session.execute(
|
|
select(Source).where(
|
|
Source.artist_id == artist.id,
|
|
Source.platform == platform,
|
|
Source.url == url,
|
|
)
|
|
).scalar_one_or_none()
|
|
if src is None:
|
|
src = Source(artist_id=artist.id, platform=platform, url=url)
|
|
self.session.add(src)
|
|
self.session.flush()
|
|
|
|
epid = sd.external_post_id or sc.stem
|
|
post = self.session.execute(
|
|
select(Post).where(
|
|
Post.source_id == src.id,
|
|
Post.external_post_id == epid,
|
|
)
|
|
).scalar_one_or_none()
|
|
if post is None:
|
|
post = Post(source_id=src.id, external_post_id=epid)
|
|
self.session.add(post)
|
|
self.session.flush()
|
|
if sd.post_url is not None:
|
|
post.post_url = sd.post_url
|
|
if sd.post_title is not None:
|
|
post.post_title = sd.post_title
|
|
if sd.post_date is not None:
|
|
post.post_date = sd.post_date
|
|
if sd.description is not None:
|
|
post.description = sd.description
|
|
if sd.attachment_count is not None:
|
|
post.attachment_count = sd.attachment_count
|
|
post.raw_metadata = sd.raw
|
|
|
|
exists = self.session.execute(
|
|
select(ImageProvenance.id).where(
|
|
ImageProvenance.image_record_id == record.id,
|
|
ImageProvenance.post_id == post.id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if exists is None:
|
|
self.session.add(
|
|
ImageProvenance(
|
|
image_record_id=record.id,
|
|
post_id=post.id,
|
|
source_id=src.id,
|
|
captured_metadata=sd.raw,
|
|
)
|
|
)
|
|
if record.primary_post_id is None:
|
|
record.primary_post_id = post.id
|
|
self.session.flush()
|
|
|
|
def _supersede(
|
|
self, existing: ImageRecord, source: Path, sha: str,
|
|
phash: str, width: int | None, height: int | None,
|
|
) -> None:
|
|
"""Replace `existing`'s file with the larger `source`, keeping the
|
|
row id (so tags/series/curation stay attached). ML is cleared so
|
|
the import task re-derives it on the new pixels."""
|
|
subdir = derive_subdir(source, self.import_root)
|
|
dest_dir = self.images_root / subdir if subdir else self.images_root
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest = dest_dir / hash_suffixed_name(source.stem, sha, source.suffix)
|
|
partial = dest.with_suffix(dest.suffix + ".partial")
|
|
shutil.copy2(source, partial)
|
|
partial.rename(dest)
|
|
|
|
old_path = existing.path
|
|
old_thumb = existing.thumbnail_path
|
|
|
|
existing.path = str(dest)
|
|
existing.sha256 = sha
|
|
existing.phash = phash
|
|
existing.size_bytes = dest.stat().st_size
|
|
existing.mime = _mime_for(source)
|
|
existing.width = width
|
|
existing.height = height
|
|
existing.thumbnail_path = None
|
|
existing.integrity_status = "unknown"
|
|
existing.tagger_predictions = None
|
|
existing.tagger_model_version = None
|
|
existing.siglip_embedding = None
|
|
existing.siglip_model_version = None
|
|
existing.centroid_scores = None
|
|
# created_at intentionally preserved; updated_at auto-bumps.
|
|
self.session.flush()
|
|
self.session.commit()
|
|
|
|
for stale in (old_path, old_thumb):
|
|
if not stale:
|
|
continue
|
|
try:
|
|
p = Path(stale)
|
|
if p.exists():
|
|
p.unlink()
|
|
except Exception:
|
|
# Benign orphan; the DB swap already committed. Don't undo it.
|
|
pass
|
|
|
|
def _transparency_pct(self, source: Path) -> float:
|
|
"""Fraction of fully-transparent pixels in the image. 0.0 if no alpha."""
|
|
with Image.open(source) as im:
|
|
if im.mode not in ("RGBA", "LA") and not (
|
|
im.mode == "P" and "transparency" in im.info
|
|
):
|
|
return 0.0
|
|
if im.mode != "RGBA":
|
|
im = im.convert("RGBA")
|
|
alpha = im.getchannel("A")
|
|
histogram = alpha.histogram()
|
|
transparent = histogram[0]
|
|
total = sum(histogram)
|
|
return transparent / total if total else 0.0
|