vogelalbum/web/dex/management/commands/fetch_reference_images.py

"""Holt für jede Art ein Referenzfoto aus der deutschen Wikipedia.

Damit weiß man, wonach man Ausschau hält, bevor eine Art zum ersten Mal selbst
vor der Kamera saß. Urheber und Lizenz werden mitgespeichert und in der
Oberfläche angezeigt — die Bilder stehen fast alle unter CC-BY-SA und die
Namensnennung ist Bedingung, nicht Höflichkeit.
"""

from __future__ import annotations

import io
import json
import re
import time
import urllib.parse
import urllib.request

from django.core.files.base import ContentFile
from django.core.management.base import BaseCommand
from PIL import Image

from dex.models import Species

UA = "VogelalbumBot/1.0 (https://gaertner.onl; webmaster@gaertner.onl)"
WP_API = "https://de.wikipedia.org/w/api.php"
COMMONS_API = "https://commons.wikimedia.org/w/api.php"

MAX_WIDTH = 900
TAG_RE = re.compile(r"<[^>]+>")


def _get(url: str, params: dict) -> dict:
    full = f"{url}?{urllib.parse.urlencode(params)}"
    req = urllib.request.Request(full, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=25) as resp:
        return json.loads(resp.read())


def _strip_html(value: str) -> str:
    """Der Artist kommt als HTML-Schnipsel mit Link. Wir wollen nur den Namen."""
    text = TAG_RE.sub("", value or "")
    return " ".join(text.split())[:250]


class Command(BaseCommand):
    help = "Lädt Referenzfotos samt Lizenzangaben aus Wikipedia/Commons."

    def add_arguments(self, parser):
        parser.add_argument(
            "--force",
            action="store_true",
            help="Auch Arten neu laden, die schon ein Referenzfoto haben.",
        )
        parser.add_argument("--only", help="Nur diese Art (Slug).")
        parser.add_argument(
            "--delay",
            type=float,
            default=0.6,
            help="Pause zwischen Anfragen; Wikipedia bittet um Zurückhaltung.",
        )

    def handle(self, *args, **opts):
        qs = Species.objects.all()
        if opts["only"]:
            qs = qs.filter(slug=opts["only"])
        if not opts["force"]:
            qs = qs.filter(reference_image="")

        ok = skipped = failed = 0
        for species in qs:
            try:
                if self._fetch(species, opts["delay"]):
                    ok += 1
                    self.stdout.write(self.style.SUCCESS(f"  ✓ {species.german}"))
                else:
                    skipped += 1
                    self.stdout.write(f"  – {species.german}: kein Bild gefunden")
            except Exception as exc:
                failed += 1
                self.stdout.write(self.style.WARNING(f"  ✗ {species.german}: {exc}"))
            time.sleep(opts["delay"])

        self.stdout.write(
            self.style.SUCCESS(f"\n{ok} geladen, {skipped} ohne Bild, {failed} Fehler.")
        )

    def _lead_image(self, title: str) -> dict | None:
        page = _get(
            WP_API,
            {
                "action": "query",
                "titles": title,
                "prop": "pageimages",
                "piprop": "original|name",
                "format": "json",
                "formatversion": "2",
                "redirects": "1",
            },
        )
        pages = page.get("query", {}).get("pages", [])
        if not pages or "original" not in pages[0]:
            return None
        return pages[0]

    def _fetch(self, species: Species, delay: float) -> bool:
        # 1. Artikelbild der deutschen Wikipedia finden. Der deutsche Name ist
        #    meist das Artikellemma — außer bei Gimpel, Kleiber, Sperber, Star:
        #    das sind Begriffsklärungsseiten ohne Bild. Der wissenschaftliche
        #    Name leitet dort zuverlässig auf den Artikel weiter.
        page = self._lead_image(species.german)
        if page is None:
            time.sleep(delay)
            page = self._lead_image(species.scientific)
        if page is None:
            return False

        img_url = page["original"]["source"].split("?", 1)[0]
        file_name = page.get("pageimage")

        # 2. Lizenz und Urheber von Commons nachschlagen.
        credit = license_name = source = ""
        if file_name:
            time.sleep(delay)
            meta = _get(
                COMMONS_API,
                {
                    "action": "query",
                    "titles": f"File:{file_name}",
                    "prop": "imageinfo",
                    "iiprop": "extmetadata|url",
                    "format": "json",
                    "formatversion": "2",
                },
            )
            info = meta.get("query", {}).get("pages", [{}])[0].get("imageinfo")
            if info:
                ext = info[0].get("extmetadata", {})
                credit = _strip_html(ext.get("Artist", {}).get("value", ""))
                license_name = _strip_html(
                    ext.get("LicenseShortName", {}).get("value", "")
                )
                source = info[0].get("descriptionurl", "")

        # 3. Bild holen und auf vernünftige Größe bringen.
        req = urllib.request.Request(img_url, headers={"User-Agent": UA})
        with urllib.request.urlopen(req, timeout=45) as resp:
            raw = resp.read()

        img = Image.open(io.BytesIO(raw)).convert("RGB")
        if img.width > MAX_WIDTH:
            img = img.resize(
                (MAX_WIDTH, round(img.height * MAX_WIDTH / img.width)), Image.LANCZOS
            )
        buf = io.BytesIO()
        img.save(buf, format="JPEG", quality=85, optimize=True)

        species.reference_image.save(
            f"{species.slug}.jpg", ContentFile(buf.getvalue()), save=False
        )
        species.reference_credit = credit
        species.reference_license = license_name
        species.reference_source = source
        species.save(
            update_fields=[
                "reference_image",
                "reference_credit",
                "reference_license",
                "reference_source",
            ]
        )
        return True