goon/scripts/test_resolve_endpoint.py
goon-foss ad0284585b Initial commit
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.
2026-05-20 10:10:22 +02:00

59 lines
2.2 KiB
Python

"""Test /resolve endpoint for sample playback URLs across multiple tubes — sprawdza
czy stream_url jest niepuste (= direct video, nie embed)."""
from __future__ import annotations
import sys
import httpx
from app.db import session_scope
from app.models.playback_source import PlaybackSource
from sqlalchemy import select
def main() -> None:
base = "http://localhost:8000"
api_key = sys.argv[1] if len(sys.argv) > 1 else None
if not api_key:
print("usage: test_resolve_endpoint.py <api_key>")
sys.exit(1)
targets = [
("pornapp:porntrexcom", "porntrex"),
("pornapp:porn4dayspw", "porn4days"),
("pornapp:hqpornercom", "hqporner"),
]
with session_scope() as s:
for origin, label in targets:
row = s.execute(
select(PlaybackSource.id, PlaybackSource.scene_id, PlaybackSource.page_url)
.where(PlaybackSource.origin == origin)
.where(PlaybackSource.dead_at.is_(None))
.limit(1)
).first()
if row is None:
print(f"[{label}] NO PLAYBACK SOURCE")
continue
pb_id, scene_id, page_url = row
url = f"{base}/scenes/{scene_id}/playback/{pb_id}/resolve"
try:
r = httpx.post(url, headers={"X-API-Key": api_key}, timeout=60.0)
except Exception as e:
print(f"[{label}] ERROR: {e}")
continue
print(f"[{label}] status={r.status_code} page={page_url[:60]}")
if r.status_code == 200:
data = r.json()
best = data.get("best") or {}
print(f" best.stream_url: {(best.get('stream_url') or '')[:90]}")
print(f" best.embed_url: {(best.get('embed_url') or '')[:90]}")
print(f" best.type: {best.get('type')}")
print(f" best.quality: {best.get('quality')}")
links = data.get("links", [])
print(f" total links: {len(links)} ({sum(1 for l in links if l.get('stream_url'))} direct)")
else:
print(f" body: {r.text[:200]}")
if __name__ == "__main__":
main()