ad0284585b
Goon — self-hosted aggregator for adult-content scene metadata. Indexes scenes from TPDB, StashDB, and 30+ public adult tube sites. Cross-source deduplication via perceptual hash + Levenshtein distance. FastAPI backend + APScheduler worker + React Native (Expo) mobile client. FOSS, ad-free, donation-funded. See README for details.
83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
"""End-to-end status check for każdego pornapp tube'a:
|
|
1. Fetch 1 alive playback_source per origin
|
|
2. Wywołaj _resolve_via_pornapp
|
|
3. Sprawdź czy zwraca direct stream URL (stream_url after _proxify_link)
|
|
4. Test proxy fetch GET na stream URL
|
|
Wynik: tabela origin → status + reason.
|
|
"""
|
|
from __future__ import annotations
|
|
import logging
|
|
import time
|
|
import traceback
|
|
|
|
import httpx
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.playback import _resolve_via_pornapp
|
|
from app.db import session_scope
|
|
from app.models.playback_source import PlaybackSource
|
|
|
|
logging.basicConfig(level=logging.WARNING)
|
|
|
|
|
|
def main() -> None:
|
|
# Get one alive sample per origin
|
|
samples: list[tuple[str, str, str]] = []
|
|
with session_scope() as s:
|
|
origins = s.execute(
|
|
select(PlaybackSource.origin)
|
|
.where(PlaybackSource.dead_at.is_(None))
|
|
.where(PlaybackSource.origin.like("pornapp:%"))
|
|
.group_by(PlaybackSource.origin)
|
|
).scalars().all()
|
|
for origin in origins:
|
|
row = s.execute(
|
|
select(PlaybackSource.id, PlaybackSource.page_url)
|
|
.where(PlaybackSource.origin == origin)
|
|
.where(PlaybackSource.dead_at.is_(None))
|
|
.limit(1)
|
|
).first()
|
|
if row:
|
|
samples.append((origin, str(row[0]), row[1]))
|
|
|
|
print(f"{'origin':28s} {'status':10s} {'detail'}")
|
|
print("-" * 100)
|
|
for origin, pb_id, page_url in samples:
|
|
sitetag = origin.split(":", 1)[1]
|
|
t0 = time.time()
|
|
try:
|
|
links = _resolve_via_pornapp(sitetag=sitetag, page_url=page_url)
|
|
except Exception as e:
|
|
elapsed = time.time() - t0
|
|
print(f"{origin:28s} {'RESOLVE':10s} {elapsed:.1f}s {type(e).__name__}: {str(e)[:60]}")
|
|
continue
|
|
elapsed = time.time() - t0
|
|
if not links:
|
|
print(f"{origin:28s} {'EMPTY':10s} {elapsed:.1f}s no links returned")
|
|
continue
|
|
directs = [l for l in links if l.stream_url]
|
|
embeds = [l for l in links if not l.stream_url and l.embed_url]
|
|
if not directs:
|
|
print(f"{origin:28s} {'EMBED_ONLY':10s} {elapsed:.1f}s {len(embeds)} embed(s)")
|
|
continue
|
|
# Test first direct URL (with proxy-equivalent headers: UA + Referer)
|
|
url = directs[0].stream_url
|
|
ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/140.0.0.0 Safari/537.36"
|
|
try:
|
|
r = httpx.get(
|
|
url,
|
|
headers={"User-Agent": ua, "Referer": page_url, "Range": "bytes=0-1024"},
|
|
timeout=15,
|
|
follow_redirects=True,
|
|
)
|
|
if r.status_code in (200, 206):
|
|
print(f"{origin:28s} {'OK':10s} {elapsed:.1f}s {len(directs)}d/{len(embeds)}e {r.status_code} {(directs[0].quality or '?')}")
|
|
else:
|
|
print(f"{origin:28s} {'FETCH_FAIL':10s} {elapsed:.1f}s status={r.status_code}")
|
|
except Exception as e:
|
|
print(f"{origin:28s} {'FETCH_ERR':10s} {elapsed:.1f}s {type(e).__name__}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|