till
7c23c3f4c7
- 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
94 lines
2.7 KiB
Python
94 lines
2.7 KiB
Python
"""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)
|