151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
"""Umwandlung Zwischenablage-Inhalt <-> Klartext-Payload (+ Meta).
|
|
|
|
text : UTF-8
|
|
image : PNG
|
|
files : ZIP mit den kopierten Dateien/Ordnern auf oberster Ebene
|
|
"""
|
|
import hashlib
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import time
|
|
import zipfile
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from .config import cache_dir
|
|
|
|
|
|
@dataclass
|
|
class ClipContent:
|
|
kind: str # text | image | files
|
|
text: str = ""
|
|
png: bytes = b""
|
|
paths: list[str] = field(default_factory=list)
|
|
fp: str = "" # Fingerabdruck zur Schleifen-Vermeidung
|
|
|
|
@staticmethod
|
|
def fingerprint(kind: str, data: bytes) -> str:
|
|
return kind + ":" + hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def text_fp(text: str) -> str:
|
|
return ClipContent.fingerprint("text", text.encode("utf-8"))
|
|
|
|
|
|
def files_fp(paths: list[str]) -> str:
|
|
norm = sorted(os.path.normcase(os.path.abspath(p)) for p in paths)
|
|
return ClipContent.fingerprint("files", "\n".join(norm).encode("utf-8"))
|
|
|
|
|
|
def human(n: float) -> str:
|
|
for unit in ("B", "KB", "MB", "GB"):
|
|
if n < 1024 or unit == "GB":
|
|
return f"{n:.0f} {unit}" if unit == "B" else f"{n:.1f} {unit}"
|
|
n /= 1024
|
|
return f"{n:.1f} GB"
|
|
|
|
|
|
def files_size(paths: list[str], limit: int) -> int:
|
|
"""Summe der Dateigroessen; bricht ab, sobald limit ueberschritten ist."""
|
|
total = 0
|
|
for p in paths:
|
|
p = Path(p)
|
|
if p.is_file():
|
|
total += p.stat().st_size
|
|
else:
|
|
for root, _dirs, files in os.walk(p):
|
|
for name in files:
|
|
try:
|
|
total += os.path.getsize(os.path.join(root, name))
|
|
except OSError:
|
|
pass
|
|
if total > limit:
|
|
return total
|
|
if total > limit:
|
|
return total
|
|
return total
|
|
|
|
|
|
def build(content: ClipContent, device_name: str) -> tuple[dict, Path]:
|
|
"""Schreibt die Payload in eine temporaere Datei. Rueckgabe: (meta, pfad)."""
|
|
fd, tmp = tempfile.mkstemp(prefix="netclip-", suffix=".plain")
|
|
tmp = Path(tmp)
|
|
meta = {"v": 1, "kind": content.kind, "device": device_name}
|
|
with os.fdopen(fd, "wb") as f:
|
|
if content.kind == "text":
|
|
data = content.text.encode("utf-8")
|
|
f.write(data)
|
|
meta.update(preview=content.text[:300], chars=len(content.text))
|
|
elif content.kind == "image":
|
|
f.write(content.png)
|
|
meta.update(mime="image/png", preview=f"Bild ({human(len(content.png))})")
|
|
else:
|
|
names, count = [], 0
|
|
with zipfile.ZipFile(f, "w", zipfile.ZIP_DEFLATED, compresslevel=1) as z:
|
|
for p in content.paths:
|
|
p = Path(p)
|
|
names.append(p.name)
|
|
if p.is_dir():
|
|
z.write(p, p.name + "/")
|
|
for root, dirs, files in os.walk(p):
|
|
rel_root = Path(p.name) / Path(root).relative_to(p)
|
|
for d in dirs:
|
|
z.write(Path(root) / d, (rel_root / d).as_posix() + "/")
|
|
for name in files:
|
|
try:
|
|
z.write(Path(root) / name, (rel_root / name).as_posix())
|
|
count += 1
|
|
except OSError:
|
|
pass # z.B. gesperrte Datei unter Windows
|
|
elif p.is_file():
|
|
z.write(p, p.name)
|
|
count += 1
|
|
meta.update(names=names, count=count, preview=", ".join(names)[:300])
|
|
meta["size"] = tmp.stat().st_size
|
|
return meta, tmp
|
|
|
|
|
|
def received_dir() -> Path:
|
|
return cache_dir() / "received"
|
|
|
|
|
|
def materialize(kind: str, plain: Path, clip_id: str) -> ClipContent:
|
|
"""Erzeugt aus einer entschluesselten Payload den Zwischenablage-Inhalt.
|
|
Dateien werden nach <cache>/received/<clip_id>/ entpackt."""
|
|
if kind == "text":
|
|
return ClipContent("text", text=plain.read_text("utf-8"))
|
|
if kind == "image":
|
|
return ClipContent("image", png=plain.read_bytes())
|
|
target = received_dir() / clip_id
|
|
if not target.exists():
|
|
part = target.with_name(clip_id + ".part")
|
|
shutil.rmtree(part, ignore_errors=True)
|
|
part.mkdir(parents=True)
|
|
with zipfile.ZipFile(plain) as z:
|
|
for info in z.infolist():
|
|
# Pfad-Traversal verhindern
|
|
dest = (part / info.filename).resolve()
|
|
if part.resolve() not in dest.parents and dest != part.resolve():
|
|
continue
|
|
z.extract(info, part)
|
|
if not info.is_dir():
|
|
mtime = time.mktime(info.date_time + (0, 0, -1))
|
|
os.utime(dest, (mtime, mtime))
|
|
os.replace(part, target)
|
|
paths = sorted(str(p) for p in target.iterdir())
|
|
return ClipContent("files", paths=paths)
|
|
|
|
|
|
def cleanup_received(days: int):
|
|
base = received_dir()
|
|
if not base.exists():
|
|
return
|
|
cutoff = time.time() - days * 86400
|
|
for d in base.iterdir():
|
|
try:
|
|
if d.stat().st_mtime < cutoff:
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
except OSError:
|
|
pass
|