from __future__ import annotations

import hashlib
import io
import json
import mimetypes
import re
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urljoin, urlsplit

from bs4 import BeautifulSoup, Tag
from PIL import Image, ImageDraw, ImageFont, UnidentifiedImageError

MAX_IMAGE_PIXELS = 40_000_000
DERIVATIVE_TARGETS = {"card": 256, "detail": 1024}
WATERBODY_MEDIA_ROLES = frozenset({
    "waterbody_cover", "waterbody_map", "waterbody_depth_map", "waterbody_screenshot",
})
TACKLE_MEDIA_ROLES = frozenset({
    "tackle_card", "tackle_detail", "rig_diagram", "tackle_screenshot",
})
MEDIA_ROLES_BY_ENTITY = {
    "waterbody": WATERBODY_MEDIA_ROLES,
    "tackle": TACKLE_MEDIA_ROLES,
}


@dataclass(frozen=True, slots=True)
class MediaCandidate:
    source_page: str
    asset_url: str
    entity_type: str
    label: str | None
    external_id: str | None


def _label(node: Tag) -> str | None:
    value = node.get("alt") or node.get("title")
    if isinstance(value, str) and value.strip():
        return " ".join(value.split())[:300]
    parent = node.find_parent(["article", "figure", "a", "section"])
    if parent:
        text = " ".join(parent.get_text(" ", strip=True).split())
        return text[:300] or None
    return None


def _kind(url: str, label: str | None, context: str) -> str:
    path = urlsplit(url).path.casefold()
    value = f"{label or ''} {context}".casefold()
    if "/flags/" in path or "/themes/" in path or path.endswith(("/logo.png", "/logo.svg")):
        return "reference"
    # A structural tackle path is stronger evidence than words in a product
    # name (for example, bait named "Краб и рыба").
    if re.search(r"(?:^|[/_-])(?:bait|lure|rig|rod|reel|hook|tackle|köder|rute|rolle)(?:[/_.-]|$)", path):
        return "tackle"
    if re.search(r"/(?:fish|fishes|fische|species)/", path) or re.search(r"/(?:fish|species)[_-]", path) or "/media/fische/" in value or any(word in value for word in ("рыба ", "fish: ", "fisch: ")):
        return "fish"
    if re.search(r"/(?:maps?|levels?|lakes?|waterbodies)/", path) or "/media/levels/" in value:
        return "waterbody"
    return "reference"


def _external_id(node: Tag, url: str) -> str | None:
    link = node.find_parent("a", href=True)
    if link:
        return str(link.get("href")).rstrip("/").rsplit("/", 1)[-1]
    parts = urlsplit(url)
    if parts.hostname == "oss.rf4db.com" and parts.path.startswith("/game/"):
        return Path(parts.path).stem
    return None


def extract_media_candidates(html: str, *, source_page: str) -> list[MediaCandidate]:
    soup = BeautifulSoup(html, "html.parser")
    result: dict[str, MediaCandidate] = {}
    for node in soup.select("img[src], img[data-src], source[srcset]"):
        raw = node.get("src") or node.get("data-src") or node.get("srcset")
        if not isinstance(raw, str):
            continue
        raw = raw.split(",", 1)[0].strip().split(" ", 1)[0]
        url = urljoin(source_page, raw)
        if urlsplit(url).scheme != "https":
            continue
        label = _label(node)
        context = " ".join(node.parent.get_text(" ", strip=True).split())[:500] if node.parent else ""
        external_id = _external_id(node, url)
        result.setdefault(url, MediaCandidate(source_page, url, _kind(url, label, f"{source_page} {context}"), label, external_id))
    for match in re.finditer(r"url\((['\"]?)(https://[^)'\"]+)\1\)", html, re.I):
        url = match.group(2)
        result.setdefault(url, MediaCandidate(source_page, url, _kind(url, None, source_page), None, None))
    return list(result.values())


def merge_manifest(path: Path, candidates: list[MediaCandidate]) -> dict:
    current = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {"version": 1, "assets": []}
    assets = {item["asset_url"]: item for item in current.get("assets", [])}
    seen_at = datetime.now(timezone.utc).isoformat()
    for candidate in candidates:
        item = assets.get(candidate.asset_url, {})
        source_pages = set(item.get("source_pages", []))
        if item.get("source_page"):
            source_pages.add(item["source_page"])
        source_pages.add(candidate.source_page)
        item.update(asdict(candidate))
        item["source_pages"] = sorted(source_pages)
        item.setdefault("status", "queued")
        item.setdefault("first_seen_at", seen_at)
        item["last_seen_at"] = seen_at
        assets[candidate.asset_url] = item
    output = {"version": 1, "updated_at": seen_at, "assets": sorted(assets.values(), key=lambda item: item["asset_url"])}
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(output, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return output


def normalize_entity_label(value: str) -> str:
    """Normalize conservative Russian catalog spelling variants for matching."""
    return " ".join(value.casefold().replace("ё", "е").split())


def reconcile_queued_duplicates(path: Path) -> dict:
    """Quarantine queued label duplicates while preserving their provenance."""
    manifest = json.loads(path.read_text(encoding="utf-8"))
    saved: dict[tuple[str, str], dict] = {}
    for item in manifest.get("assets", []):
        parts = urlsplit(str(item.get("asset_url") or ""))
        if not item.get("external_id") and parts.hostname == "oss.rf4db.com" and parts.path.startswith("/game/"):
            item["external_id"] = Path(parts.path).stem
        label = item.get("label")
        if item.get("status") not in {"stored", "approved"} or not isinstance(label, str):
            continue
        saved.setdefault((str(item.get("entity_type")), normalize_entity_label(label)), item)
    duplicates = 0
    for item in manifest.get("assets", []):
        label = item.get("label")
        if item.get("status") != "queued" or not isinstance(label, str):
            continue
        match = saved.get((str(item.get("entity_type")), normalize_entity_label(label)))
        if match and match.get("asset_url") != item.get("asset_url"):
            item.update({"status": "duplicate", "duplicate_of": match["asset_url"]})
            duplicates += 1
    manifest["updated_at"] = datetime.now(timezone.utc).isoformat()
    path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return {"duplicates": duplicates, "queued": sum(item.get("status") == "queued" for item in manifest.get("assets", []))}


def queue_quality_upgrades(path: Path, *, minimum_dimension: int = 256) -> dict:
    """Queue alternatives to low-resolution published fish without unpublishing them."""
    manifest = json.loads(path.read_text(encoding="utf-8"))
    assets = manifest.get("assets", [])
    low_urls = {
        item.get("asset_url")
        for item in assets
        if item.get("status") == "approved"
        and item.get("entity_type") == "fish"
        and min(int(item.get("width") or 0), int(item.get("height") or 0)) < minimum_dimension
    }
    queued = 0
    for item in assets:
        if (
            item.get("status") == "duplicate"
            and item.get("entity_type") == "fish"
            and item.get("duplicate_of") in low_urls
        ):
            item["status"] = "upgrade_queued"
            queued += 1
    manifest["updated_at"] = datetime.now(timezone.utc).isoformat()
    path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return {
        "minimum_dimension": minimum_dimension,
        "low_resolution_published": len(low_urls),
        "upgrade_queued": queued,
    }


def approve_stored_assets(path: Path, *, note: str) -> dict:
    """Publish every stored asset after an explicit owner-level approval."""
    manifest = json.loads(path.read_text(encoding="utf-8"))
    reviewed_at = datetime.now(timezone.utc).isoformat()
    approved = 0
    for item in manifest.get("assets", []):
        if item.get("status") != "stored":
            continue
        stable_id = item.get("external_id") or item.get("sha256") or hashlib.sha256(item["asset_url"].encode()).hexdigest()
        item.update({
            "status": "approved",
            "entity_key": f"{item.get('entity_type', 'reference')}:{stable_id}",
            "reviewed_at": reviewed_at,
            "review_note": note,
        })
        approved += 1
    manifest["updated_at"] = reviewed_at
    path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return {"approved": approved}


def publish_quality_upgrades(path: Path, *, note: str, minimum_dimension: int = 256) -> dict:
    """Atomically promote every reviewed quality candidate and retain its fallback."""
    manifest = json.loads(path.read_text(encoding="utf-8"))
    assets = manifest.get("assets", [])
    by_url = {item.get("asset_url"): item for item in assets}
    candidates = [item for item in assets if item.get("status") == "upgrade_stored"]

    # Validate the complete set before changing any public mapping.
    for candidate in candidates:
        fallback = by_url.get(candidate.get("duplicate_of"))
        if not fallback or fallback.get("status") != "approved":
            raise ValueError(f"{candidate.get('asset_url')}: approved fallback is missing")
        if not fallback.get("entity_key"):
            raise ValueError(f"{candidate.get('asset_url')}: fallback has no entity_key")
        if min(int(candidate.get("width") or 0), int(candidate.get("height") or 0)) < minimum_dimension:
            raise ValueError(f"{candidate.get('asset_url')}: candidate is below {minimum_dimension}px")
        if not candidate.get("sha256") or not candidate.get("local_path"):
            raise ValueError(f"{candidate.get('asset_url')}: stored candidate metadata is incomplete")

    reviewed_at = datetime.now(timezone.utc).isoformat()
    for candidate in candidates:
        fallback = by_url[candidate["duplicate_of"]]
        candidate.update({
            "status": "approved",
            "entity_key": fallback["entity_key"],
            "supersedes": fallback["asset_url"],
            "reviewed_at": reviewed_at,
            "review_note": note,
        })
        fallback.update({
            "status": "superseded",
            "replaced_by": candidate["asset_url"],
            "reviewed_at": reviewed_at,
            "review_note": note,
        })

    manifest["updated_at"] = reviewed_at
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    temporary.replace(path)
    return {"published": len(candidates), "retained_fallbacks": len(candidates)}


def rollback_quality_upgrade(path: Path, *, asset_url: str, note: str) -> dict:
    """Restore the superseded fallback for one explicitly selected upgrade."""
    manifest = json.loads(path.read_text(encoding="utf-8"))
    assets = manifest.get("assets", [])
    candidate = next((item for item in assets if item.get("asset_url") == asset_url), None)
    if not candidate or candidate.get("status") != "approved" or not candidate.get("supersedes"):
        raise ValueError("rollback requires an approved upgrade with a supersedes link")
    fallback = next((item for item in assets if item.get("asset_url") == candidate["supersedes"]), None)
    if not fallback or fallback.get("status") != "superseded" or fallback.get("replaced_by") != asset_url:
        raise ValueError("rollback requires the matching superseded fallback")

    reviewed_at = datetime.now(timezone.utc).isoformat()
    candidate.update({"status": "upgrade_stored", "reviewed_at": reviewed_at, "review_note": note})
    candidate.pop("supersedes", None)
    fallback.update({"status": "approved", "reviewed_at": reviewed_at, "review_note": note})
    fallback.pop("replaced_by", None)
    manifest["updated_at"] = reviewed_at
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    temporary.replace(path)
    return {"rolled_back": asset_url, "restored": fallback["asset_url"]}


def reclassify_manifest(path: Path) -> dict:
    manifest = json.loads(path.read_text(encoding="utf-8"))
    for item in manifest.get("assets", []):
        item.setdefault("source_pages", [item["source_page"]])
        item["entity_type"] = _kind(item["asset_url"], item.get("label"), item.get("source_page", ""))
    path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return manifest


def review_asset(
    path: Path, *, asset_url: str, decision: str, entity_type: str | None = None,
    entity_key: str | None = None, note: str | None = None,
    media_role: str | None = None,
) -> dict:
    if decision not in {"approved", "rejected"}:
        raise ValueError("decision must be approved or rejected")
    manifest = json.loads(path.read_text(encoding="utf-8"))
    item = next((asset for asset in manifest.get("assets", []) if asset["asset_url"] == asset_url), None)
    if item is None:
        raise ValueError("asset URL is not present in manifest")
    if media_role is not None and media_role not in MEDIA_ROLES_BY_ENTITY.get(entity_type or "", set()):
        raise ValueError(f"media role is only valid for {entity_type}; use a known {entity_type} role")
    if decision == "approved":
        if item.get("status") != "stored":
            raise ValueError("only a stored asset can be approved")
        if entity_type not in {"fish", "waterbody", "tackle", "reference"} or not entity_key:
            raise ValueError("approved asset requires entity type and canonical key")
        item.update({"entity_type": entity_type, "entity_key": entity_key})
    if media_role is not None:
        item["media_role"] = media_role
    item.update({
        "status": decision,
        "reviewed_at": datetime.now(timezone.utc).isoformat(),
        "review_note": note or None,
    })
    path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return item


def audit_media_catalog(root: Path) -> dict:
    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
    statuses: dict[str, int] = {}
    issues: list[str] = []
    referenced: set[str] = set()
    for item in manifest.get("assets", []):
        status = str(item.get("status", "unknown"))
        statuses[status] = statuses.get(status, 0) + 1
        local_path = item.get("local_path")
        if status in {"stored", "upgrade_stored", "approved", "superseded"}:
            if not isinstance(local_path, str):
                issues.append(f"{item['asset_url']}: {status} asset has no local_path")
                continue
            referenced.add(local_path)
            target = root / local_path
            if not target.is_file():
                issues.append(f"{item['asset_url']}: local file is missing")
                continue
            body = target.read_bytes()
            if hashlib.sha256(body).hexdigest() != item.get("sha256"):
                issues.append(f"{item['asset_url']}: SHA-256 mismatch")
            try:
                width, height, mime = inspect_image(body)
                if (width, height, mime) != (item.get("width"), item.get("height"), item.get("content_type")):
                    issues.append(f"{item['asset_url']}: image metadata mismatch")
            except ValueError as exc:
                issues.append(f"{item['asset_url']}: {exc}")
        for variant in item.get("derivatives", []):
            variant_path = variant.get("local_path")
            if not isinstance(variant_path, str):
                issues.append(f"{item['asset_url']}: derivative has no local_path")
                continue
            referenced.add(variant_path)
            target = root / variant_path
            if not target.is_file():
                issues.append(f"{item['asset_url']}: derivative file is missing")
                continue
            body = target.read_bytes()
            if hashlib.sha256(body).hexdigest() != variant.get("sha256"):
                issues.append(f"{item['asset_url']}: derivative SHA-256 mismatch")
            try:
                width, height, mime = inspect_image(body)
                if (width, height, mime) != (variant.get("width"), variant.get("height"), variant.get("content_type")):
                    issues.append(f"{item['asset_url']}: derivative image metadata mismatch")
                if len(body) != variant.get("bytes"):
                    issues.append(f"{item['asset_url']}: derivative byte count mismatch")
            except ValueError as exc:
                issues.append(f"{item['asset_url']}: derivative {exc}")
        if status == "approved" and (not item.get("entity_key") or item.get("entity_type") not in {"fish", "waterbody", "tackle", "reference"}):
            issues.append(f"{item['asset_url']}: approved asset has no valid canonical mapping")
        if item.get("media_role") not in {None, *WATERBODY_MEDIA_ROLES, *TACKLE_MEDIA_ROLES}:
            issues.append(f"{item['asset_url']}: unknown media role")
        elif item.get("media_role") is not None and item.get("media_role") not in MEDIA_ROLES_BY_ENTITY.get(item.get("entity_type"), set()):
            issues.append(f"{item['asset_url']}: media role does not match entity type")
    files_root = root / "files"
    orphaned = sorted(str(path.relative_to(root)) for path in files_root.rglob("*") if path.is_file() and str(path.relative_to(root)) not in referenced) if files_root.exists() else []
    return {"total": sum(statuses.values()), "statuses": statuses, "issues": issues, "orphaned_files": orphaned}


def media_coverage(root: Path) -> dict:
    """Compare quarantined and approved media with the versioned catalog baseline."""
    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
    baseline = json.loads((root / "catalog-baseline.json").read_text(encoding="utf-8"))
    result: dict[str, dict] = {}
    for entity_type, target in baseline["entities"].items():
        candidates = [item for item in manifest.get("assets", []) if item.get("entity_type") == entity_type]
        candidate_labels = {
            normalize_entity_label(item["label"])
            for item in candidates
            if isinstance(item.get("label"), str) and item["label"].strip()
        }
        approved_keys = {item.get("entity_key") for item in candidates if item.get("status") == "approved" and item.get("entity_key")}
        expected = target.get("count")
        result[entity_type] = {
            "expected": expected,
            "candidate_assets": len(candidates),
            "unique_candidates": len(candidate_labels),
            "unidentified_assets": sum(1 for item in candidates if not isinstance(item.get("label"), str) or not item["label"].strip()),
            "approved": len(approved_keys),
            "candidate_gap": max(expected - len(candidate_labels), 0) if isinstance(expected, int) else None,
            "approved_gap": max(expected - len(approved_keys), 0) if isinstance(expected, int) else None,
        }
    return {"baseline_date": baseline["verified_at"], "entities": result}


def media_quality_report(root: Path, *, minimum_dimension: int = 256, display_dimension: int = 180) -> dict:
    """Report published raster quality and known higher-quality source alternatives offline."""
    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
    assets = manifest.get("assets", [])
    approved = [item for item in assets if item.get("status") == "approved" and item.get("entity_type") == "fish"]
    low = [item for item in approved if min(int(item.get("width") or 0), int(item.get("height") or 0)) < minimum_dimension]
    upsampled = [item for item in approved if min(int(item.get("width") or 0), int(item.get("height") or 0)) < display_dimension]
    low_urls = {item.get("asset_url") for item in low}
    alternatives = [
        item for item in assets
        if item.get("entity_type") == "fish"
        and item.get("status") in {"duplicate", "queued", "upgrade_queued", "upgrade_stored"}
        and (item.get("duplicate_of") in low_urls or item.get("status") == "queued")
    ]

    by_source: dict[str, dict[str, int]] = {}
    for item in approved:
        host = urlsplit(str(item.get("asset_url") or "")).hostname or "unknown"
        summary = by_source.setdefault(host, {"published": 0, "below_minimum": 0, "upsampled_in_cards": 0})
        summary["published"] += 1
        summary["below_minimum"] += item in low
        summary["upsampled_in_cards"] += item in upsampled

    return {
        "entity_type": "fish",
        "minimum_dimension": minimum_dimension,
        "display_dimension": display_dimension,
        "published": len(approved),
        "below_minimum": len(low),
        "upsampled_in_cards": len(upsampled),
        "known_alternative_urls": len(alternatives),
        "by_source": by_source,
        "examples": [
            {"label": item.get("label"), "width": item.get("width"), "height": item.get("height"), "asset_url": item.get("asset_url")}
            for item in sorted(low, key=lambda row: str(row.get("label") or ""))[:20]
        ],
    }


def compare_quality_upgrades(root: Path, *, minimum_dimension: int = 256) -> dict:
    """Compare stored upgrade candidates with their published fallbacks offline."""
    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
    assets = manifest.get("assets", [])
    by_url = {item.get("asset_url"): item for item in assets}
    comparisons: list[dict] = []
    issues: list[str] = []
    for candidate in assets:
        if candidate.get("status") != "upgrade_stored":
            continue
        fallback = by_url.get(candidate.get("duplicate_of"))
        if not fallback or fallback.get("status") != "approved":
            issues.append(f"{candidate.get('asset_url')}: published fallback is missing")
            continue
        candidate_path = root / str(candidate.get("local_path") or "")
        fallback_path = root / str(fallback.get("local_path") or "")
        if not candidate_path.is_file() or not fallback_path.is_file():
            issues.append(f"{candidate.get('asset_url')}: comparison file is missing")
            continue
        with Image.open(candidate_path) as image:
            candidate_alpha = image.mode in {"LA", "RGBA"} or "transparency" in image.info
        with Image.open(fallback_path) as image:
            fallback_alpha = image.mode in {"LA", "RGBA"} or "transparency" in image.info
        candidate_width, candidate_height = int(candidate["width"]), int(candidate["height"])
        fallback_width, fallback_height = int(fallback["width"]), int(fallback["height"])
        candidate_ratio = candidate_width / candidate_height
        fallback_ratio = fallback_width / fallback_height
        comparisons.append({
            "label": candidate.get("label"),
            "candidate_url": candidate.get("asset_url"),
            "fallback_url": fallback.get("asset_url"),
            "candidate": {
                "width": candidate_width, "height": candidate_height,
                "bytes": candidate.get("bytes"), "content_type": candidate.get("content_type"),
                "alpha": candidate_alpha, "aspect_ratio": round(candidate_ratio, 4),
            },
            "fallback": {
                "width": fallback_width, "height": fallback_height,
                "bytes": fallback.get("bytes"), "content_type": fallback.get("content_type"),
                "alpha": fallback_alpha, "aspect_ratio": round(fallback_ratio, 4),
            },
            "meets_minimum": min(candidate_width, candidate_height) >= minimum_dimension,
            "aspect_ratio_matches": abs(candidate_ratio - fallback_ratio) < 0.01,
        })
    return {
        "minimum_dimension": minimum_dimension,
        "compared": len(comparisons),
        "meets_minimum": sum(item["meets_minimum"] for item in comparisons),
        "aspect_ratio_matches": sum(item["aspect_ratio_matches"] for item in comparisons),
        "issues": issues,
        "comparisons": comparisons,
    }


def generate_quality_contact_sheets(
    root: Path, output_dir: Path, *, batch_size: int = 20,
) -> dict:
    """Create offline old/selected contact sheets for manually reviewed upgrades."""
    if batch_size < 1:
        raise ValueError("batch_size must be positive")
    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
    assets = manifest.get("assets", [])
    by_url = {item.get("asset_url"): item for item in assets}
    pairs = []
    issues: list[str] = []
    for selected in assets:
        if selected.get("status") != "approved" or not selected.get("supersedes"):
            continue
        old = by_url.get(selected["supersedes"])
        if not old:
            issues.append(f"{selected.get('asset_url')}: superseded asset is missing")
            continue
        old_path, selected_path = root / str(old.get("local_path") or ""), root / str(selected.get("local_path") or "")
        if not old_path.is_file() or not selected_path.is_file():
            issues.append(f"{selected.get('asset_url')}: contact-sheet image is missing")
            continue
        pairs.append((old, selected))

    output_dir.mkdir(parents=True, exist_ok=True)
    sheets = []
    cell_width, cell_height, image_size = 320, 220, 180
    font_paths = (
        Path("/usr/share/fonts/Fonts/DejaVuSans.ttf"),
        Path("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"),
        Path("/usr/share/fonts/dejavu/DejaVuSans.ttf"),
    )
    font_path = next((path for path in font_paths if path.is_file()), None)
    font = ImageFont.truetype(font_path, 12) if font_path else ImageFont.load_default()
    for sheet_index in range(0, len(pairs), batch_size):
        chunk = pairs[sheet_index:sheet_index + batch_size]
        canvas = Image.new("RGB", (cell_width * 2, cell_height * len(chunk)), "white")
        draw = ImageDraw.Draw(canvas)
        metadata = []
        for row, (old, selected) in enumerate(chunk):
            y = row * cell_height
            for column, item in enumerate((old, selected)):
                path = root / str(item["local_path"])
                with Image.open(path) as image:
                    preview = image.convert("RGBA")
                    preview.thumbnail((image_size, image_size), Image.Resampling.LANCZOS)
                    x = column * cell_width + (image_size - preview.width) // 2
                    canvas.paste(preview, (x, y + 4), preview if preview.mode == "RGBA" else None)
                title = "СТАРЫЙ" if column == 0 else "ВЫБРАННЫЙ"
                label = str(item.get("label") or "Без подписи")[:34]
                source = urlsplit(str(item.get("source_page") or "")).hostname or "unknown"
                draw.text((column * cell_width + 190, y + 8), f"{title}: {label}", fill="black", font=font)
                draw.text((column * cell_width + 190, y + 30), f"{item.get('width')}×{item.get('height')} · {source}", fill="black", font=font)
                draw.text((column * cell_width + 190, y + 52), str(item.get("asset_url") or "")[:42], fill="gray", font=font)
            draw.line((0, y + cell_height - 1, cell_width * 2, y + cell_height - 1), fill="#cccccc")
            metadata.append({"label": selected.get("label"), "old_url": old.get("asset_url"), "selected_url": selected.get("asset_url")})
        image_path = output_dir / f"quality-upgrades-{sheet_index // batch_size + 1:03d}.png"
        json_path = image_path.with_suffix(".json")
        canvas.save(image_path, format="PNG", optimize=True)
        json_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        sheets.append(str(image_path))
    return {"pairs": len(pairs), "sheets": sheets, "issues": issues}


def inspect_image(body: bytes) -> tuple[int, int, str]:
    try:
        with Image.open(io.BytesIO(body)) as image:
            width, height = image.size
            image_format = image.format
            image.verify()
    except (UnidentifiedImageError, OSError, ValueError) as exc:
        raise ValueError("asset is not a valid raster image") from exc
    if width < 1 or height < 1 or width * height > MAX_IMAGE_PIXELS:
        raise ValueError("asset dimensions are outside safe limits")
    mime = Image.MIME.get(image_format or "")
    if mime not in {"image/jpeg", "image/png", "image/webp", "image/gif", "image/avif"}:
        raise ValueError(f"unsupported image format: {image_format}")
    return width, height, mime


def store_asset(root: Path, body: bytes, *, content_type: str, source_url: str) -> tuple[str, str, int, int, str]:
    width, height, detected_mime = inspect_image(body)
    declared_mime = content_type.split(";", 1)[0]
    if declared_mime != detected_mime:
        raise ValueError(f"image MIME mismatch: declared {declared_mime}, detected {detected_mime}")
    digest = hashlib.sha256(body).hexdigest()
    extension = mimetypes.guess_extension(detected_mime) or Path(urlsplit(source_url).path).suffix
    extension = ".jpg" if extension == ".jpe" else extension
    target = root / "files" / digest[:2] / f"{digest}{extension}"
    target.parent.mkdir(parents=True, exist_ok=True)
    if not target.exists():
        target.write_bytes(body)
    return digest, str(target.relative_to(root)), width, height, detected_mime


def generate_media_derivatives(
    root: Path, *, targets: dict[str, int] | None = None,
) -> dict:
    """Generate deterministic WebP/AVIF derivatives for approved local assets.

    Derivatives are content-addressed like originals and never upscale a source
    image.  The approved original remains the fallback and is not rewritten.
    """
    manifest_path = root / "manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    selected_targets = targets or DERIVATIVE_TARGETS
    generated = skipped = 0
    issues: list[str] = []

    for item in manifest.get("assets", []):
        if item.get("status") != "approved" or not item.get("local_path"):
            continue
        source_path = root / str(item["local_path"])
        if not source_path.is_file():
            issues.append(f"{item.get('asset_url')}: derivative source is missing")
            continue
        try:
            with Image.open(source_path) as source:
                source.load()
                source_width, source_height = source.size
                source_format = source.format
                if source_format not in {"JPEG", "PNG", "WEBP", "GIF"}:
                    raise ValueError(f"unsupported source format: {source_format}")
                variants: list[dict] = []
                for role, target_width in selected_targets.items():
                    width = min(source_width, int(target_width))
                    height = max(1, round(source_height * width / source_width))
                    resized = source if width == source_width else source.resize((width, height), Image.Resampling.LANCZOS)
                    for image_format, mime, extension, save_options in (
                        ("WEBP", "image/webp", ".webp", {"quality": 85, "method": 6}),
                        ("AVIF", "image/avif", ".avif", {"quality": 80}),
                    ):
                        output = io.BytesIO()
                        if resized.mode not in {"RGB", "RGBA", "L", "LA"}:
                            converted = resized.convert("RGBA" if "A" in resized.mode else "RGB")
                        else:
                            converted = resized
                        converted.save(output, format=image_format, **save_options)
                        body = output.getvalue()
                        digest = hashlib.sha256(body).hexdigest()
                        target = root / "files" / digest[:2] / f"{digest}{extension}"
                        target.parent.mkdir(parents=True, exist_ok=True)
                        if not target.exists():
                            target.write_bytes(body)
                            generated += 1
                        else:
                            skipped += 1
                        variants.append({
                            "role": role, "format": image_format.lower(), "sha256": digest,
                            "local_path": str(target.relative_to(root)), "content_type": mime,
                            "bytes": len(body), "width": width, "height": height,
                        })
                item["derivatives"] = variants
        except (OSError, ValueError) as exc:
            issues.append(f"{item.get('asset_url')}: {exc}")

    manifest["updated_at"] = datetime.now(timezone.utc).isoformat()
    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return {"approved_assets": sum(item.get("status") == "approved" for item in manifest.get("assets", [])), "generated": generated, "already_present": skipped, "issues": issues}