#!/usr/bin/env python3
"""
═══════════════════════════════════════════════════════════════════════════════
VERKLI LETÖLTŐ — helyi kiszolgáló a Verkli Letöltés füléhez
Verzió: 1.0.0
TZlab · Tóth Zoltán
A böngésző önmagában nem tud videót letölteni. Ez a kis kiszolgáló a
saját gépen figyel, és a Verkli Letöltés füle beszélget vele. Amit
lehoz, azt a Verkli átveszi és beteszi a tárba.
Az AvideoZ (Universal Video Downloader) tapasztalataira épül:
Facebook/Dropbox URL-javítás, felirat-kezelés, emberi szünetek.
MIÉRT TÖLT LE A COLAB MEGBÍZHATÓBBAN?
Mert ott minden induláskor friss yt-dlp települ. A YouTube pár hetente
változtat, és egy pár hónapos yt-dlp elkezd hibázni. Ezért ez a
kiszolgáló induláskor kiírja a verziót, és van „frissítés" gombja.
Függőség: yt-dlp (kötelező), ffmpeg (a jó minőséghez és az MP3-hoz).
pip install -U yt-dlp
Indítás: python verkli_letolto.py
Leállítás: Ctrl+C
═══════════════════════════════════════════════════════════════════════════════
"""
import json
import mimetypes
import os
import re
import secrets
import shutil
import subprocess
import sys
import threading
import time
import uuid
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import unquote, urlparse
VERZIO = "2.4.0"
HOST = "127.0.0.1" # csak a saját gép — kívülről nem érhető el
PORT = 8765 # a beállításfájl felülírhatja
SAJAT_MAPPA = Path(__file__).resolve().parent
def build_sorszam(p: Path) -> int:
"""A felület build-száma a fájl TARTALMÁBÓL (KV_BUILD = 'Verkli b50').
A fájlnév és a módosítási dátum egyaránt félrevezethet: egy régebbi
build is lehet frissebb dátumú, ha később másolták a mappába, és a
névsor szerint a b9 nagyobb, mint a b50. A tartalom nem hazudik.
"""
try:
fej = p.read_bytes()
except OSError:
return -1
m = re.search(rb"KV_BUILD\s*=\s*'[^']*?b(\d+)", fej)
return int(m.group(1)) if m else -1
def verkli_html():
"""A legfrissebb Verkli felület a szkript mappájából: elsődlegesen a
legnagyobb build-szám, döntetlennél a frissebb fájl."""
jeloltek = [p for p in SAJAT_MAPPA.glob("verkli*.html")]
if not jeloltek:
return None
return max(jeloltek, key=lambda p: (build_sorszam(p), p.stat().st_mtime))
def olvas_beallitas():
"""A letöltő melletti verkli_beallitas.json — ha nincs, létrehozzuk.
Így a munkamappa, a kapu és a frissítési cím a .py szerkesztése nélkül
is átállítható; a fájl megmarad a következő verzió mellett is.
"""
f = SAJAT_MAPPA / "verkli_beallitas.json"
alap = {
"munkamappa": "", # üresen: a letöltő melletti Verkli_Work
"port": 8765,
"frissites_url": "https://verkli.tzlab.hu/index.html",
}
try:
if f.exists():
adat = json.loads(f.read_text(encoding="utf-8"))
alap.update({k: v for k, v in adat.items() if k in alap})
else:
f.write_text(json.dumps(alap, ensure_ascii=False, indent=2), encoding="utf-8")
except Exception:
pass # hibás fájl: alapértékkel megyünk
return alap
def irhato(mappa: Path) -> bool:
try:
mappa.mkdir(parents=True, exist_ok=True)
proba = mappa / ".verkli_proba"
proba.write_text("x", encoding="utf-8")
proba.unlink()
return True
except Exception:
return False
def munkamappa(beallitas) -> Path:
"""Hol dolgozzon a Verkli?
1. amit a beállításban megadtak;
2. különben a letöltő melletti Verkli_Work — így az egész mappa
hordozható, pendrive-ra másolva is minden együtt marad;
3. ha az nem írható (pl. rendszermappába került), a Dokumentumok alá.
"""
megadott = (beallitas.get("munkamappa") or "").strip()
if megadott:
p = Path(megadott).expanduser()
if irhato(p):
return p
print(f"! a megadott munkamappa nem írható: {p}")
melle = SAJAT_MAPPA / "Verkli_Work"
if irhato(melle):
return melle
for nev in ("Documents", "Dokumentumok"):
p = Path.home() / nev
if p.is_dir() and irhato(p / "Verkli_Work"):
print("! a letöltő mappája nem írható — a Dokumentumok alá dolgozunk")
return p / "Verkli_Work"
return Path.home() / "Verkli_Work"
BEALLITAS = olvas_beallitas()
PORT = int(BEALLITAS.get("port") or PORT)
MUNKA_DIR = munkamappa(BEALLITAS)
LETOLTES_DIR = MUNKA_DIR / "letoltesek" # amit a Letöltés fül lehoz
NYERS_DIR = MUNKA_DIR / "nyersanyag" # a véglegesítéshez feltöltött források
KESZ_DIR = MUNKA_DIR / "kesz" # a véglegesített műsorok
ATMENETI_DIR = MUNKA_DIR / "munka" # szegmensek véglegesítés közben
def mappak_beallit(uj: Path) -> bool:
"""Munkamappa váltása futás közben; a beállítás fájlba is kerül."""
global MUNKA_DIR, LETOLTES_DIR, NYERS_DIR, KESZ_DIR, ATMENETI_DIR
uj = Path(uj).expanduser()
if not irhato(uj):
return False
MUNKA_DIR = uj
LETOLTES_DIR = uj / "letoltesek"
NYERS_DIR = uj / "nyersanyag"
KESZ_DIR = uj / "kesz"
ATMENETI_DIR = uj / "munka"
for m in (LETOLTES_DIR, NYERS_DIR, KESZ_DIR, ATMENETI_DIR):
m.mkdir(parents=True, exist_ok=True)
try:
f = SAJAT_MAPPA / "verkli_beallitas.json"
adat = json.loads(f.read_text(encoding="utf-8")) if f.exists() else {}
adat["munkamappa"] = str(uj)
f.write_text(json.dumps(adat, ensure_ascii=False, indent=2), encoding="utf-8")
except Exception as e:
naplo("a beállítás mentése nem sikerült:", str(e)[:60])
naplo(LAMPA["ok"], "munkamappa:", uj)
return True
def mappa_tallozas():
"""Natív mappaválasztó a felhasználó gépén.
A böngésző mappaválasztója csak fogantyút ad, teljes útvonalat nem —
azt a bővítő nem tudná használni. Ezért itt nyitunk ablakot. A
hordozható Pythonban nincs tkinter, akkor a felület beírásra vált.
"""
try:
import tkinter
from tkinter import filedialog
gyoker = tkinter.Tk()
gyoker.withdraw()
gyoker.attributes("-topmost", True)
valasz = filedialog.askdirectory(title="Verkli munkamappa",
initialdir=str(MUNKA_DIR.parent))
gyoker.destroy()
return valasz or None
except Exception as e:
naplo("nincs mappaválasztó ablak:", str(e)[:60])
return False # False = nincs tkinter; None = a felhasználó elvetette
# ── VÉDELEM ────────────────────────────────────────────────────────────────
# A kiszolgáló csak a saját gépen figyel, DE a böngészőből bármelyik
# megnyitott weboldal is elérné. Ezért minden érdemi művelet
# munkamenet-tokenhez kötött: a token indításkor születik, és csak abba a
# felületbe kerül bele, amit ez a kiszolgáló ad ki. Idegen lap nem
# ismerheti, tehát nem indíthat letöltést és nem futtathat frissítést.
#
# Egyetlen kivétel a /api/pong: annyit árul el, hogy a letöltő fut. Erre
# a webről nyitott Verklinek van szüksége, hogy felajánlhassa az átlépést
# a helyi példányra. Semmilyen adatot nem ad ki és semmit nem indít el.
TOKEN = secrets.token_urlsafe(24)
# ── ÖNFRISSÍTÉS ────────────────────────────────────────────────────────────
# Induláskor a letöltő megnézi, van-e frissebb Verkli felület a szerveren,
# és ha igen, lehozza maga mellé. Így a felhasználónak soha nem kell
# fájlt cserélnie. Üres címmel az egész kimarad.
#
# A cím MUTATHAT KÖZVETLENÜL A FELÜLETRE (ajánlott):
# FRISSITES_URL = "https://verkli.pelda.hu/index.html"
# Ilyenkor a szerveren elég az index.html — a build számát a fájlból
# olvassuk ki (KV_BUILD), és abból lesz a helyi fájlnév.
#
# Vagy mutathat egy manifestre, ha úgy kényelmesebb:
# FRISSITES_URL = "https://verkli.pelda.hu/verzio.json"
# {"build": "Verkli b47", "fajl": "verkli_b47.html"}
FRISSITES_URL = BEALLITAS.get("frissites_url") or "" # a beállításfájlból
FRISSITES = True
# ═══════════════════════════════════════════════════════════════════════════
# KONZOL — log-biztos mód: fix szélesség, ASCII-visszaesés
# ═══════════════════════════════════════════════════════════════════════════
UTF = True
try:
if os.name == "nt":
os.system("chcp 65001 >nul 2>&1")
"▁█".encode(sys.stdout.encoding or "utf-8")
except Exception:
UTF = False
BOX = ("┌", "─", "┐", "│", "└", "┘") if UTF else ("+", "-", "+", "|", "+", "+")
LAMPA = {"ok": "🟢" if UTF else "[OK]", "nincs": "🔴" if UTF else "[!!]",
"vár": "🟡" if UTF else "[..]"}
def wlen(szoveg: str) -> int:
"""Hány cellát foglal a szöveg a terminálban?
A `len()` nem elég: az emoji (🟢) és a kelet-ázsiai karakterek KÉT
cellát foglalnak, ezért a keret jobb széle elcsúszna. A kombináló
jelek (ékezet-ráolvasztás) viszont nullát.
"""
import unicodedata
n = 0
for c in szoveg:
if unicodedata.combining(c):
continue
n += 2 if unicodedata.east_asian_width(c) in ("W", "F") else 1
return n
def wljust(szoveg: str, szeles: int) -> str:
"""Balra igazítás CELLÁRA számolva; ha nem fér, csonkoljuk."""
h = wlen(szoveg)
if h <= szeles:
return szoveg + " " * (szeles - h)
ki, n = "", 0
for c in szoveg:
cw = wlen(c)
if n + cw > szeles - 1:
break
ki += c
n += cw
return ki + "…" + " " * max(0, szeles - n - 1)
def wcenter(szoveg: str, szeles: int) -> str:
h = wlen(szoveg)
if h >= szeles:
return wljust(szoveg, szeles)
bal = (szeles - h) // 2
return " " * bal + szoveg + " " * (szeles - h - bal)
def panel(cimsor, sorok, szeles=74):
"""Keretezett panel — a szélesség CELLÁBAN értendő, nem karakterben."""
ki = [BOX[0] + BOX[1] * (szeles - 2) + BOX[2]]
ki.append(BOX[3] + wcenter(cimsor, szeles - 2) + BOX[3])
ki.append(BOX[3] + BOX[1] * (szeles - 2) + BOX[3])
for s in sorok:
ki.append(BOX[3] + " " + wljust(s, szeles - 4) + " " + BOX[3])
ki.append(BOX[4] + BOX[1] * (szeles - 2) + BOX[5])
return "\n".join(ki)
def savszint(szazalek, szeles=24):
"""Százalék-sáv; a szám mindig ott van mellette."""
tele = int(max(0, min(100, szazalek)) / 100 * szeles)
if UTF:
return "█" * tele + "░" * (szeles - tele)
return "#" * tele + "-" * (szeles - tele)
def naplo(*reszek):
print(datetime.now().strftime("[%H:%M:%S] ") + " ".join(str(r) for r in reszek), flush=True)
# ═══════════════════════════════════════════════════════════════════════════
# KÖRNYEZET
# ═══════════════════════════════════════════════════════════════════════════
def ytdlp_verzio():
try:
import yt_dlp
return getattr(yt_dlp.version, "__version__", "ismeretlen")
except Exception:
return None
def ffmpeg_ut():
"""Hol van ffmpeg? Három helyen keressük, ebben a sorrendben:
a rendszer útvonalán, a letöltő saját mappájában, és a pipből
telepíthető imageio-ffmpeg csomagban (ez utóbbi a legegyszerűbb
pótlás, ha a gépen nincs ffmpeg)."""
p = shutil.which("ffmpeg")
if p:
return p
for nev in ("ffmpeg.exe", "ffmpeg"):
c = SAJAT_MAPPA / nev
if c.is_file():
return str(c)
try:
import imageio_ffmpeg
u = imageio_ffmpeg.get_ffmpeg_exe()
if u and Path(u).is_file():
return u
except Exception:
pass
return None
def ffmpeg_van():
return ffmpeg_ut() is not None
def ytdlp_frissites():
"""pip install -U yt-dlp — ez a leggyakoribb megoldás a hibás letöltésre."""
naplo("yt-dlp frissítése…")
try:
r = subprocess.run([sys.executable, "-m", "pip", "install", "-U", "yt-dlp"],
capture_output=True, text=True, timeout=300, check=False)
ok = r.returncode == 0
naplo(("frissítve" if ok else "a frissítés nem sikerült") +
" — a hatályba lépéshez indítsd újra a letöltőt")
return ok, (r.stdout or r.stderr or "")[-400:]
except Exception as e:
return False, str(e)
# ═══════════════════════════════════════════════════════════════════════════
# URL-JAVÍTÁSOK (az AvideoZ-ből)
# ═══════════════════════════════════════════════════════════════════════════
def javit_url(url: str) -> str:
m = re.search(r"facebook\.com/reel/(\d+)", url)
if m:
return f"https://www.facebook.com/watch/?v={m.group(1)}"
if "dropbox.com" in url.lower():
url = re.sub(r"[&?]dl=0", "&dl=1", url)
if "dl=" not in url:
url += "&dl=1" if "?" in url else "?dl=1"
return url
# ═══════════════════════════════════════════════════════════════════════════
# MUNKÁK
# ═══════════════════════════════════════════════════════════════════════════
class Megszakitva(Exception):
"""A felhasználó megszakította a letöltést."""
munkak = {} # id -> dict
munkak_zar = threading.Lock()
def uj_render(terv):
mid = uuid.uuid4().hex[:12]
with munkak_zar:
munkak[mid] = {
"id": mid, "url": "", "mod": terv.get("mod", "video"), "allapot": "fut",
"szazalek": 0.0, "uzenet": "indul…", "cim": "véglegesítés",
"fajlok": [], "mappa": str(ATMENETI_DIR / mid),
"megszakit": False, "kezdet": time.time(), "terv": terv,
}
return mid
def uj_munka(url, mod):
mid = uuid.uuid4().hex[:12]
with munkak_zar:
munkak[mid] = {
"id": mid, "url": url, "mod": mod, "allapot": "fut",
"szazalek": 0.0, "uzenet": "indul…", "cim": url,
"fajlok": [], "mappa": str(LETOLTES_DIR / mid), "megszakit": False,
"kezdet": time.time(),
}
return mid
def munka(mid):
with munkak_zar:
return dict(munkak.get(mid) or {})
def frissit(mid, **mezok):
with munkak_zar:
if mid in munkak:
munkak[mid].update(mezok)
def ydl_opciok(mid, mod, kimenet):
"""A négy üzemmód beállításai. A lassítás a rate-limit ellen véd."""
kozos = {
"outtmpl": os.path.join(kimenet, "%(title).80s_%(id)s.%(ext)s"),
"restrictfilenames": True,
"noprogress": True,
"no_color": True,
"quiet": True,
"no_warnings": True,
"ignoreerrors": False,
"sleep_interval": 2,
"max_sleep_interval": 6,
"sleep_interval_requests": 1,
"progress_hooks": [lambda d: horog(mid, d)],
}
ff = ffmpeg_ut()
if ff and not shutil.which("ffmpeg"):
kozos["ffmpeg_location"] = ff # saját mappából vagy csomagból
feliratok = {
"writesubtitles": True, "writeautomaticsub": True,
"subtitleslangs": ["hu", "en"], "subtitlesformat": "srt",
}
if mod == "hang":
kozos.update({
"format": "bestaudio/best",
"postprocessors": [{"key": "FFmpegExtractAudio",
"preferredcodec": "mp3", "preferredquality": "192"}],
})
elif mod == "felirat":
kozos.update({"skip_download": True})
kozos.update(feliratok)
else:
kozos.update({"format": "bestvideo+bestaudio/best",
"merge_output_format": "mp4"})
if mod == "video_felirat":
kozos.update(feliratok)
return kozos
def horog(mid, d):
"""yt-dlp haladás → a munka állapota."""
if munka(mid).get("megszakit"):
raise Megszakitva
if d.get("status") == "downloading":
teljes = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
kesz = d.get("downloaded_bytes") or 0
szaz = (kesz / teljes * 100) if teljes else 0
frissit(mid, szazalek=round(szaz, 1),
uzenet=(d.get("_speed_str") or "").strip() or "letöltés…")
elif d.get("status") == "finished":
frissit(mid, szazalek=100.0, uzenet="feldolgozás…")
def futtat(mid):
import yt_dlp
m = munka(mid)
kimenet = m["mappa"]
os.makedirs(kimenet, exist_ok=True)
url = javit_url(m["url"])
naplo(f"indul: {url} [{m['mod']}]")
try:
with yt_dlp.YoutubeDL(ydl_opciok(mid, m["mod"], kimenet)) as ydl:
info = ydl.extract_info(url, download=True)
cim = (info or {}).get("title") or url
fajlok = []
for p in sorted(Path(kimenet).iterdir()):
if p.is_file() and not p.name.endswith((".part", ".ytdl")):
fajlok.append({"nev": p.name, "meret": p.stat().st_size})
if not fajlok:
frissit(mid, allapot="hiba", uzenet="nem keletkezett fájl")
naplo(LAMPA["nincs"], "nem keletkezett fájl:", url)
return
frissit(mid, allapot="kesz", szazalek=100.0, cim=cim, fajlok=fajlok,
uzenet=f"{len(fajlok)} fájl")
ossz = sum(f["meret"] for f in fajlok)
naplo(LAMPA["ok"], f"kész: {cim} ({len(fajlok)} fájl, {ossz/1048576:.1f} MB, "
f"{time.time()-m['kezdet']:.0f} mp)")
except Exception as e:
uz = str(e)[:200]
frissit(mid, allapot="hiba", uzenet=uz)
naplo(LAMPA["nincs"], "hiba:", uz)
if "Sign in to confirm" in uz or "HTTP Error 403" in uz:
naplo(" tipp: próbáld a yt-dlp frissítését (a Verkliben: Kapcsolat ellenőrzése → Frissítés)")
# ═══════════════════════════════════════════════════════════════════════════
# VÉGLEGESÍTÉS (a műsor egyben)
# ═══════════════════════════════════════════════════════════════════════════
#
# A böngésző nem tud videót kódolni, a bővítő viszont igen. A kliens előbb
# feltölti a szükséges forrásokat (amit már feltöltött, azt nem újra), majd
# elküldi a műsor tervét. Innen a munka:
#
# 1. minden műsorszámból külön szegmens készül, AZONOS paraméterekkel
# (1920×1080, 30 kép/mp, AAC 48 kHz sztereó) — enélkül az összefűzés
# vagy elhasal, vagy csúszik;
# 2. az átúsztatás a szegmens elejére/végére kerül fade-ként — ugyanaz,
# amit a lejátszó mutat: a kép feketén át megy, a hang le-fel úszik;
# 3. a szegmensek összefűzése már újrakódolás nélkül megy (concat).
NORMALIZALAS = ("scale=1920:1080:force_original_aspect_ratio=decrease,"
"pad=1920:1080:(ow-iw)/2:(oh-ih)/2,setsar=1,fps=30,format=yuv420p")
def ff(parancsok, mid=None, lepes=""):
"""ffmpeg futtatása; hiba esetén a stderr utolsó sorait dobjuk vissza."""
ut = ffmpeg_ut()
if not ut:
raise RuntimeError("nincs ffmpeg")
r = subprocess.run([ut, "-y", "-hide_banner", "-loglevel", "error", *parancsok],
capture_output=True, text=True, check=False)
if r.returncode != 0:
reszlet = (r.stderr or "").strip().splitlines()[-3:]
raise RuntimeError((lepes + ": " if lepes else "") + " / ".join(reszlet))
return True
def fade_szuro(be, ki, hossz):
"""Kép-fade a szegmens elejére/végére."""
d = []
if be > 0:
d.append(f"fade=t=in:st=0:d={be:.2f}")
if ki > 0:
d.append(f"fade=t=out:st={max(0, hossz - ki):.2f}:d={ki:.2f}")
return d
def afade_szuro(be, ki, hossz):
d = []
if be > 0:
d.append(f"afade=t=in:st=0:d={be:.2f}")
if ki > 0:
d.append(f"afade=t=out:st={max(0, hossz - ki):.2f}:d={ki:.2f}")
return d
def szegmens_video(szam, munka_dir, sorszam, be_fade, ki_fade):
"""Egy műsorszám videó-szegmense."""
hossz = float(szam.get("hossz") or 0)
cel = munka_dir / f"seg{sorszam:03d}.mp4"
v = szam.get("video")
h = szam.get("hang")
kepek = szam.get("kepek") or []
vf = [NORMALIZALAS] + fade_szuro(be_fade, ki_fade, hossz)
af = afade_szuro(be_fade, ki_fade, hossz)
if v:
forras = NYERS_DIR / v["fajl"]
p = ["-ss", str(v["start"]), "-to", str(v["end"]), "-i", str(forras)]
if h: # néma kép + külön hang
hf = NYERS_DIR / h["fajl"]
p += ["-ss", str(h["start"]), "-to", str(h["end"]), "-i", str(hf),
"-map", "0:v:0", "-map", "1:a:0"]
p += ["-vf", ",".join(vf)]
if af:
p += ["-af", ",".join(af)]
p += ["-c:v", "libx264", "-preset", "veryfast", "-crf", "20",
"-c:a", "aac", "-b:a", "192k", "-ar", "48000", "-ac", "2",
"-t", str(hossz), str(cel)]
ff(p, lepes=f"{sorszam}. szám (videó)")
return cel
if kepek:
# a képek külön-külön, majd egymás után; a hang megy rá
reszek = []
pontok = [k["at"] for k in kepek] + [hossz]
for i, k in enumerate(kepek):
d = max(0.2, pontok[i + 1] - k["at"])
r = munka_dir / f"seg{sorszam:03d}_k{i:02d}.mp4"
ff(["-loop", "1", "-t", f"{d:.3f}", "-i", str(NYERS_DIR / k["fajl"]),
"-vf", NORMALIZALAS, "-c:v", "libx264", "-preset", "veryfast",
"-crf", "20", "-an", str(r)], lepes=f"{sorszam}. szám ({i+1}. kép)")
reszek.append(r)
lista = munka_dir / f"seg{sorszam:03d}_kepek.txt"
lista.write_text("".join(f"file '{r.name}'\n" for r in reszek), encoding="utf-8")
nema = munka_dir / f"seg{sorszam:03d}_nema.mp4"
ff(["-f", "concat", "-safe", "0", "-i", str(lista), "-c", "copy", str(nema)],
lepes=f"{sorszam}. szám (képek összefűzése)")
p = ["-i", str(nema)]
if h:
p += ["-ss", str(h["start"]), "-to", str(h["end"]),
"-i", str(NYERS_DIR / h["fajl"])]
else: # néma diasor: csend a sávra
p += ["-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo"]
p += ["-map", "0:v:0", "-map", "1:a:0"]
if fade_szuro(be_fade, ki_fade, hossz):
p += ["-vf", ",".join(fade_szuro(be_fade, ki_fade, hossz))]
p += ["-c:v", "libx264", "-preset", "veryfast", "-crf", "20"]
else:
p += ["-c:v", "copy"]
if af:
p += ["-af", ",".join(af)]
p += ["-c:a", "aac", "-b:a", "192k", "-ar", "48000", "-ac", "2",
"-t", str(hossz), "-shortest", str(cel)]
ff(p, lepes=f"{sorszam}. szám (kép + hang)")
return cel
if h: # hang kép nélkül: fekete alap
p = ["-f", "lavfi", "-i", f"color=c=black:s=1920x1080:r=30:d={hossz:.3f}",
"-ss", str(h["start"]), "-to", str(h["end"]), "-i", str(NYERS_DIR / h["fajl"]),
"-map", "0:v:0", "-map", "1:a:0"]
if fade_szuro(be_fade, ki_fade, hossz):
p += ["-vf", ",".join(fade_szuro(be_fade, ki_fade, hossz))]
if af:
p += ["-af", ",".join(af)]
p += ["-c:v", "libx264", "-preset", "veryfast", "-crf", "20",
"-c:a", "aac", "-b:a", "192k", "-ar", "48000", "-ac", "2",
"-t", str(hossz), str(cel)]
ff(p, lepes=f"{sorszam}. szám (hang)")
return cel
raise RuntimeError(f"{sorszam}. szám: nincs forrás")
def szegmens_hang(szam, munka_dir, sorszam, be_fade, ki_fade):
"""Egy műsorszám hang-szegmense (csak hang mód)."""
hossz = float(szam.get("hossz") or 0)
cel = munka_dir / f"seg{sorszam:03d}.m4a"
forras = szam.get("hang") or szam.get("video")
if not forras:
# néma szám: csend, hogy a hossz stimmeljen
ff(["-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo", "-t", str(hossz),
"-c:a", "aac", "-b:a", "192k", str(cel)], lepes=f"{sorszam}. szám (csend)")
return cel
p = ["-ss", str(forras["start"]), "-to", str(forras["end"]),
"-i", str(NYERS_DIR / forras["fajl"]), "-vn"]
af = afade_szuro(be_fade, ki_fade, hossz)
if af:
p += ["-af", ",".join(af)]
p += ["-c:a", "aac", "-b:a", "192k", "-ar", "48000", "-ac", "2",
"-t", str(hossz), str(cel)]
ff(p, lepes=f"{sorszam}. szám (hang)")
return cel
def render_futtat(mid):
m = munka(mid)
terv = m["terv"]
mod = terv.get("mod", "video")
szamok = terv.get("szamok") or []
munka_dir = Path(m["mappa"])
munka_dir.mkdir(parents=True, exist_ok=True)
KESZ_DIR.mkdir(parents=True, exist_ok=True)
naplo(f"véglegesítés indul: {len(szamok)} szám, mód: {mod}")
t0 = time.time()
try:
szegmensek = []
for i, szam in enumerate(szamok):
if munka(mid).get("megszakit"):
raise Megszakitva
# az átúsztatás fele a szám végén, fele a következő elején —
# ugyanaz a felosztás, amit a lejátszó használ
be = 0.0
tr = szam.get("atmenet") or {}
if tr.get("tipus") == "fade":
be = float(tr.get("hossz") or 1) / 2
ki = 0.0
if i + 1 < len(szamok):
tr2 = (szamok[i + 1].get("atmenet") or {})
if tr2.get("tipus") == "fade":
ki = float(tr2.get("hossz") or 1) / 2
frissit(mid, szazalek=round(i / max(1, len(szamok)) * 90, 1),
uzenet=f"{i+1}/{len(szamok)}: {szam.get('cim','')[:40]}")
keszit = szegmens_hang if mod == "hang" else szegmens_video
szegmensek.append(keszit(szam, munka_dir, i + 1, be, ki))
frissit(mid, szazalek=92.0, uzenet="összefűzés…")
lista = munka_dir / "lista.txt"
lista.write_text("".join(f"file '{p.name}'\n" for p in szegmensek), encoding="utf-8")
bel = datetime.now().strftime("%Y%m%d_%H%M")
if mod == "hang":
nev = f"musor_{bel}.mp3"
ff(["-f", "concat", "-safe", "0", "-i", str(lista),
"-c:a", "libmp3lame", "-b:a", "192k", str(KESZ_DIR / nev)],
lepes="összefűzés")
else:
nev = f"musor_{bel}.mp4"
ff(["-f", "concat", "-safe", "0", "-i", str(lista), "-c", "copy",
"-movflags", "+faststart", str(KESZ_DIR / nev)], lepes="összefűzés")
kesz = KESZ_DIR / nev
frissit(mid, allapot="kesz", szazalek=100.0, cim=nev,
fajlok=[{"nev": nev, "meret": kesz.stat().st_size}],
uzenet=f"{kesz.stat().st_size/1048576:.0f} MB")
naplo(LAMPA["ok"], f"kész: {nev} ({kesz.stat().st_size/1048576:.1f} MB, "
f"{time.time()-t0:.0f} mp)")
naplo(" helye:", KESZ_DIR)
# a szegmensek már nem kellenek
for p in munka_dir.glob("seg*"):
p.unlink(missing_ok=True)
except Megszakitva:
frissit(mid, allapot="hiba", uzenet="megszakítva")
naplo("véglegesítés megszakítva")
except Exception as e:
frissit(mid, allapot="hiba", uzenet=str(e)[:200])
naplo(LAMPA["nincs"], "véglegesítés hiba:", str(e)[:160])
# ═══════════════════════════════════════════════════════════════════════════
# HTTP
# ═══════════════════════════════════════════════════════════════════════════
class Kezelo(BaseHTTPRequestHandler):
server_version = "VerkliLetolto/" + VERZIO
# HTTP/1.1 kell, mert a böngésző nagy feltöltésnél „Expect: 100-continue"
# fejlécet küld, és arra vár engedélyt. HTTP/1.0-ban ezt a kiszolgáló nem
# nyugtázza, ezért a feltöltés látszólag megtörtént, de a válasz elmaradt.
# HTTP/1.1-nél viszont MINDEN válaszhoz kell Content-Length — enélkül a
# böngésző a következő adagra várna.
protocol_version = "HTTP/1.1"
def log_message(self, *a):
pass # a saját naplónkat használjuk
# — segédek —
# — védelem —
def sajat_eredet(self):
"""A kérés a saját felületünkről jön-e? (Origin nélküli kérés is jó:
navigáció, curl, saját lap azonos eredetről.)"""
o = self.headers.get("Origin")
if not o:
return True
return o in (f"http://{HOST}:{PORT}", f"http://localhost:{PORT}",
f"http://[::1]:{PORT}")
def jo_token(self):
fejlecben = self.headers.get("X-Verkli-Token")
if fejlecben and secrets.compare_digest(fejlecben, TOKEN):
return True
from urllib.parse import parse_qs
q = parse_qs(urlparse(self.path).query).get("token", [""])[0]
return bool(q) and secrets.compare_digest(q, TOKEN)
def engedelyezett(self):
"""Érdemi művelethez saját eredet ÉS érvényes token kell."""
if not self.sajat_eredet():
naplo(LAMPA["nincs"], "idegen eredetű kérés elutasítva:",
self.headers.get("Origin"), self.path)
return False
if not self.jo_token():
naplo(LAMPA["nincs"], "token nélküli kérés elutasítva:", self.path)
return False
return True
def fejlec(self, kod=200, tipus="application/json; charset=utf-8", extra=None,
nyilt=False):
self.send_response(kod)
self.send_header("Content-Type", tipus)
# CORS csak a nyilvános végpontra; minden másra a saját eredet kell
o = self.headers.get("Origin")
if nyilt:
self.send_header("Access-Control-Allow-Origin", "*")
elif o and self.sajat_eredet():
self.send_header("Access-Control-Allow-Origin", o)
self.send_header("Vary", "Origin")
self.send_header("Access-Control-Allow-Headers", "Content-Type, X-Verkli-Token")
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.send_header("Cache-Control", "no-store")
for k, v in (extra or {}).items():
self.send_header(k, v)
self.end_headers()
def tiltva(self):
self.json_valasz({"hiba": "nincs jogosultság"}, 403)
def json_valasz(self, adat, kod=200):
test = json.dumps(adat, ensure_ascii=False).encode("utf-8")
self.fejlec(kod, extra={"Content-Length": str(len(test))})
self.wfile.write(test)
def do_OPTIONS(self):
nyilt = urlparse(self.path).path == "/api/pong"
self.fejlec(204, "text/plain", extra={"Content-Length": "0"}, nyilt=nyilt)
def do_GET(self):
ut = urlparse(self.path).path
# nyilvános: csak annyit mond, hogy fut — ez kell a webes Verkli
# „átlépés a helyi példányra" felajánlásához
if ut == "/api/pong":
test = json.dumps({"verkli": True, "verzio": VERZIO}).encode("utf-8")
self.fejlec(200, extra={"Content-Length": str(len(test))}, nyilt=True)
self.wfile.write(test)
return
if ut == "/api/nyersanyag":
if not self.engedelyezett():
self.tiltva()
return
NYERS_DIR.mkdir(parents=True, exist_ok=True)
self.json_valasz({"fajlok": [{"nev": p.name, "meret": p.stat().st_size}
for p in sorted(NYERS_DIR.iterdir()) if p.is_file()]})
return
if ut == "/api/allapot":
if not self.engedelyezett():
self.tiltva()
return
self.json_valasz({
"verkli": True, "verzio": VERZIO,
"ytdlp": ytdlp_verzio(), "ffmpeg": ffmpeg_van(),
"mappa": str(MUNKA_DIR),
"letoltes_mappa": str(LETOLTES_DIR),
"kesz_mappa": str(KESZ_DIR),
})
return
m = re.match(r"^/api/allapot/([0-9a-f]+)$", ut)
if m:
if not self.engedelyezett():
self.tiltva()
return
j = munka(m.group(1))
if not j:
self.json_valasz({"hiba": "nincs ilyen munka"}, 404)
return
self.json_valasz({k: j[k] for k in
("id", "allapot", "szazalek", "uzenet", "cim", "fajlok")})
return
# — a Verkli felülete a saját mappából —
if ut in ("/", "/index.html"):
f = verkli_html()
if not f:
uz = "Nincs Verkli HTML a letöltő mellett.
".encode("utf-8")
self.fejlec(404, "text/html; charset=utf-8",
{"Content-Length": str(len(uz))})
self.wfile.write(uz)
return
adat = f.read_bytes()
# A token CSAK ebbe a lapba kerül bele — így az általunk kiadott
# felület tud dolgozni, idegen lap viszont nem.
jel = b""
if jel in adat[:4000]:
belovo = ('').encode("utf-8")
adat = adat.replace(jel, jel + belovo, 1)
self.fejlec(200, "text/html; charset=utf-8",
{"Content-Length": str(len(adat))})
self.wfile.write(adat)
return
if re.match(r"^/[\w.\-]+\.(html|css|js|png|jpg|svg|ico|py|bat)$", ut):
cel = (SAJAT_MAPPA / unquote(ut.lstrip("/"))).resolve()
if str(cel).startswith(str(SAJAT_MAPPA)) and cel.is_file():
tipus = mimetypes.guess_type(cel.name)[0] or "application/octet-stream"
adat = cel.read_bytes()
self.fejlec(200, tipus, {"Content-Length": str(len(adat))})
self.wfile.write(adat)
return
m = re.match(r"^/api/fajl/([0-9a-f]+)/(.+)$", ut)
if m:
if not self.engedelyezett():
self.tiltva()
return
j = munka(m.group(1))
nev = unquote(m.group(2))
if not j:
self.json_valasz({"hiba": "nincs ilyen munka"}, 404)
return
# csak a munka saját mappájából, könyvtár-kilépés nélkül
cel = (Path(j["mappa"]) / nev).resolve()
if not cel.is_file(): # véglegesített műsor
jelolt = (KESZ_DIR / nev).resolve()
if str(jelolt).startswith(str(KESZ_DIR.resolve())) and jelolt.is_file():
cel = jelolt
if not (str(cel).startswith(str(Path(j["mappa"]).resolve()))
or str(cel).startswith(str(KESZ_DIR.resolve()))) or not cel.is_file():
self.json_valasz({"hiba": "nincs ilyen fájl"}, 404)
return
tipus = mimetypes.guess_type(cel.name)[0] or "application/octet-stream"
self.fejlec(200, tipus, {"Content-Length": str(cel.stat().st_size)})
with open(cel, "rb") as f:
shutil.copyfileobj(f, self.wfile)
return
self.json_valasz({"hiba": "ismeretlen cím"}, 404)
def do_POST(self):
ut = urlparse(self.path).path
if not self.engedelyezett(): # minden módosító művelet védett
self.tiltva()
return
hossz = int(self.headers.get("Content-Length") or 0)
# A feltöltés nyers bájtokat küld, nem JSON-t — ezt előbb kezeljük.
if ut == "/api/feltoltes":
from urllib.parse import parse_qs
nev = parse_qs(urlparse(self.path).query).get("nev", [""])[0]
nev = os.path.basename(unquote(nev))
if not nev or nev.startswith("."):
self.json_valasz({"hiba": "hibás fájlnév"}, 400)
return
NYERS_DIR.mkdir(parents=True, exist_ok=True)
cel = NYERS_DIR / nev
# Időtúllépés: ha a böngésző a bejelentett méretnél kevesebbet
# küld (közben módosult vagy elérhetetlenné vált a fájl), e
# nélkül a kiszolgáló örökre várná a maradékot, a felület pedig
# némán megállna a „Feltöltés 2/9" feliratnál.
regi = self.connection.gettimeout()
self.connection.settimeout(60)
marad = hossz
t0 = time.time()
try:
with open(cel, "wb") as f:
while marad > 0:
darab = self.rfile.read(min(1024 * 1024, marad))
if not darab:
break
f.write(darab)
marad -= len(darab)
except OSError as e:
self.connection.settimeout(regi)
naplo(LAMPA["nincs"], "feltöltés megszakadt:", nev, "—", str(e)[:60])
cel.unlink(missing_ok=True)
self.json_valasz({"hiba": f"a feltöltés megszakadt: {e}"}, 400)
return
self.connection.settimeout(regi)
kesz = cel.stat().st_size
if marad > 0 or kesz != hossz:
naplo(LAMPA["nincs"], "hiányos feltöltés:", nev,
f"({kesz}/{hossz} bájt)")
cel.unlink(missing_ok=True)
self.json_valasz({"hiba": f"hiányos feltöltés ({kesz}/{hossz} bájt)"}, 400)
return
naplo(LAMPA["ok"], "feltöltve:", nev,
f"({kesz/1048576:.1f} MB, {time.time()-t0:.1f} mp)")
self.json_valasz({"nev": nev, "meret": kesz})
return
try:
test = json.loads(self.rfile.read(hossz) or b"{}")
except Exception:
test = {}
if ut == "/api/mappa":
uj = (test.get("ut") or "").strip()
if not uj:
self.json_valasz({"hiba": "üres útvonal"}, 400)
return
if mappak_beallit(Path(uj)):
self.json_valasz({"mappa": str(MUNKA_DIR)})
else:
self.json_valasz({"hiba": "ez a mappa nem írható"}, 400)
return
if ut == "/api/mappa/tallozas":
v = mappa_tallozas()
if v is False:
self.json_valasz({"hiba": "ezen a gépen nincs mappaválasztó ablak",
"beir": True}, 501)
return
if not v:
self.json_valasz({"megse": True})
return
if mappak_beallit(Path(v)):
self.json_valasz({"mappa": str(MUNKA_DIR)})
else:
self.json_valasz({"hiba": "ez a mappa nem írható"}, 400)
return
if ut == "/api/vegleges":
if not ffmpeg_ut():
self.json_valasz({"hiba": "nincs ffmpeg"}, 500)
return
szamok = test.get("szamok") or []
if not szamok:
self.json_valasz({"hiba": "üres műsor"}, 400)
return
mid = uj_render(test)
threading.Thread(target=render_futtat, args=(mid,), daemon=True).start()
self.json_valasz({"id": mid})
return
if ut == "/api/letoltes":
url = (test.get("url") or "").strip()
mod = test.get("mod") or "video"
if not re.match(r"^https?://", url):
self.json_valasz({"hiba": "hibás cím"}, 400)
return
if not ytdlp_verzio():
self.json_valasz({"hiba": "a yt-dlp nincs telepítve"}, 500)
return
mid = uj_munka(url, mod)
threading.Thread(target=futtat, args=(mid,), daemon=True).start()
self.json_valasz({"id": mid})
return
m = re.match(r"^/api/megszakit/([0-9a-f]+)$", ut)
if m:
frissit(m.group(1), megszakit=True, allapot="hiba", uzenet="megszakítva")
naplo("megszakítva:", m.group(1))
self.json_valasz({"ok": True})
return
if ut == "/api/frissites":
ok, kimenet = ytdlp_frissites()
self.json_valasz({"ok": ok, "kimenet": kimenet, "ytdlp": ytdlp_verzio()})
return
self.json_valasz({"hiba": "ismeretlen cím"}, 404)
# ═══════════════════════════════════════════════════════════════════════════
# INDULÁS
# ═══════════════════════════════════════════════════════════════════════════
def build_szam(tartalom: bytes) -> str | None:
"""A felület build-száma a KV_BUILD konstansból (pl. 'Verkli b47')."""
m = re.search(rb"KV_BUILD\s*=\s*'([^']{1,40})'", tartalom)
return m.group(1).decode("utf-8", "ignore") if m else None
def helyi_build() -> str | None:
f = verkli_html()
if not f:
return None
return build_szam(f.read_bytes())
def frissites_ellenorzes():
"""Van-e újabb Verkli felület a szerveren? Ha igen, lehozzuk.
A cím mutathat közvetlenül a felületre (ilyenkor a build-számot a
fájlból olvassuk ki), vagy egy manifestre. Hálózat nélkül csendben
kimarad — a letöltő enélkül is működik.
"""
if not FRISSITES or not FRISSITES_URL:
return None
import urllib.request
try:
with urllib.request.urlopen(FRISSITES_URL, timeout=8) as r:
adat = r.read()
# — manifest-ág —
if FRISSITES_URL.rstrip("/").lower().endswith(".json"):
j = json.loads(adat.decode("utf-8"))
nev = j.get("fajl")
if not nev or not re.match(r"^verkli[\w.\-]*\.html$", nev):
return None
if (SAJAT_MAPPA / nev).exists():
return None
cim = FRISSITES_URL.rsplit("/", 1)[0] + "/" + nev
with urllib.request.urlopen(cim, timeout=30) as r:
adat = r.read()
# — közvetlen felület-ág: a letöltött lap MAGA a friss felület —
else:
if b"