Klarbildbot v1.0 — AI-Bildaufbereitung (Freistellen + Upscaling) als Telegram-Bot

- Telegram-Bot (Polling) mit Inline-Menue: Klarbild x2/x4, Freistellen, kombiniert - Upscaling via OpenCV dnn_superres (FSRCNN default, EDSR optional), gekachelt - Freistellen via rembg (isnet-general-use, Alpha-Matting) - Picdrop-Batch via SFTP (/picdrop) - Health-Server (aiohttp) fuer Coolify, Zugriffsschutz via ALLOWED_USER_IDS - Dockerfile backt Modelle (SR von raw.githubusercontent, rembg von HF-Mirror) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XNQ8ghPfzAfsyVYd6HgFb6
2026-07-23 10:14:27 +00:00
commit 7c23c3f4c7
13 changed files with 1001 additions and 0 deletions
@@ -0,0 +1,93 @@
+"""Picdrop-Zugriff via SFTP: Galerien/Ordner listen, Bilder ziehen & hochladen."""
+from __future__ import annotations
+
+import posixpath
+import stat
+from contextlib import contextmanager
+from typing import Iterator
+
+import paramiko
+
+from . import config
+
+IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".tif", ".tiff", ".bmp"}
+
+
+@contextmanager
+def _sftp() -> Iterator[paramiko.SFTPClient]:
+    transport = paramiko.Transport((config.PICDROP_HOST, config.PICDROP_PORT))
+    transport.banner_timeout = 30
+    transport.connect(username=config.PICDROP_USER, password=config.PICDROP_PASSWORD)
+    try:
+        sftp = paramiko.SFTPClient.from_transport(transport)
+        sftp.get_channel().settimeout(60)
+        yield sftp
+    finally:
+        transport.close()
+
+
+def list_dir(path: str = "/") -> list[dict]:
+    """Eintraege eines Verzeichnisses (Name, ob Ordner, Groesse)."""
+    with _sftp() as sftp:
+        out = []
+        for attr in sftp.listdir_attr(path):
+            is_dir = stat.S_ISDIR(attr.st_mode)
+            out.append({
+                "name": attr.filename,
+                "is_dir": is_dir,
+                "size": attr.st_size,
+                "path": posixpath.join(path, attr.filename),
+            })
+        out.sort(key=lambda e: (not e["is_dir"], e["name"].lower()))
+        return out
+
+
+def list_images(path: str) -> list[dict]:
+    return [
+        e for e in list_dir(path)
+        if not e["is_dir"] and posixpath.splitext(e["name"])[1].lower() in IMAGE_EXTS
+    ]
+
+
+def find_gallery(name_substr: str, root: str = "/", max_depth: int = 3) -> list[str]:
+    """Sucht Ordner, deren Name den Teilstring enthaelt (case-insensitive)."""
+    needle = name_substr.lower()
+    matches: list[str] = []
+
+    def walk(path: str, depth: int) -> None:
+        if depth > max_depth:
+            return
+        try:
+            entries = list_dir(path)
+        except OSError:
+            return
+        for e in entries:
+            if e["is_dir"]:
+                if needle in e["name"].lower():
+                    matches.append(e["path"])
+                walk(e["path"], depth + 1)
+
+    walk(root, 0)
+    return matches
+
+
+def download(remote_path: str) -> bytes:
+    with _sftp() as sftp:
+        with sftp.open(remote_path, "rb") as fh:
+            fh.prefetch()
+            return fh.read()
+
+
+def upload(remote_path: str, data: bytes) -> None:
+    with _sftp() as sftp:
+        # Zielordner sicherstellen
+        d = posixpath.dirname(remote_path)
+        parts, cur = d.strip("/").split("/"), ""
+        for p in parts:
+            cur = cur + "/" + p
+            try:
+                sftp.stat(cur)
+            except FileNotFoundError:
+                sftp.mkdir(cur)
+        with sftp.open(remote_path, "wb") as fh:
+            fh.write(data)