move server code into server/, update docs
- app.py + upstream.py -> server/ (relative import, __init__.py) - static/ resolved relative to repo root, works from any CWD - watch history migrated to server/data/history.json - README.md: quick start, features, layout, dev notes - SPEC.md: marked implemented, updated endpoints/run/deviations - ignore *.tar.gz (local curl-impersonate download at repo root)
This commit is contained in:
jozamudi committed
1 parent
a27ff5c931
commit
2c203990bc
7 files changed
+101
-37
No files matched your search
Whitespace-only changes.
+129
@@ -0,0 +1,129 @@
|
||||
"""app.py — ani-web: FastAPI over upstream.py + HLS proxy + static UI.
|
||||
|
||||
Browser playback: hls.anidb.app sends no CORS headers, so the player is
|
||||
pointed at /stream/{token}; the proxy fetches upstream (with the same
|
||||
Cloudflare workaround) and rewrites playlist URIs to go through itself.
|
||||
"""
|
||||
import base64
|
||||
import json
|
||||
import time
|
||||
from pathlib import Path
|
||||
from urllib.parse import urljoin
|
||||
|
||||
from fastapi import Body, FastAPI, HTTPException, Query
|
||||
from fastapi.responses import Response
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from . import upstream
|
||||
|
||||
ROOT = Path(__file__).parent.parent
|
||||
app = FastAPI(title="ani-web")
|
||||
|
||||
HIST_FILE = Path(__file__).parent / "data" / "history.json"
|
||||
|
||||
|
||||
def _load_hist():
|
||||
try:
|
||||
return json.loads(HIST_FILE.read_text())
|
||||
except (FileNotFoundError, json.JSONDecodeError):
|
||||
return []
|
||||
|
||||
|
||||
def _save_hist(entries):
|
||||
HIST_FILE.parent.mkdir(exist_ok=True)
|
||||
# ponytail: unguarded read-modify-write, fine for a single-user tool
|
||||
HIST_FILE.write_text(json.dumps(entries, indent=1))
|
||||
|
||||
|
||||
@app.get("/api/history")
|
||||
def api_history():
|
||||
return sorted(_load_hist(), key=lambda e: e.get("t", 0), reverse=True)[:5]
|
||||
|
||||
|
||||
@app.post("/api/history")
|
||||
def api_history_post(entry: dict = Body(...)):
|
||||
if "ep_id" not in entry:
|
||||
raise HTTPException(422, "ep_id required")
|
||||
entry["t"] = time.time()
|
||||
entries = _load_hist()
|
||||
for i, e in enumerate(entries):
|
||||
if e.get("slug_id") == entry.get("slug_id") and e.get("ep_id") == entry.get("ep_id"):
|
||||
entries[i] = {**e, **entry}
|
||||
break
|
||||
else:
|
||||
entries.append(entry)
|
||||
_save_hist(sorted(entries, key=lambda e: e.get("t", 0), reverse=True)[:50])
|
||||
return entry
|
||||
|
||||
|
||||
def _b64(url: str) -> str:
|
||||
return base64.urlsafe_b64encode(url.encode()).decode()
|
||||
|
||||
|
||||
def _url(token: str) -> str:
|
||||
return base64.urlsafe_b64decode(token.encode()).decode()
|
||||
|
||||
|
||||
def _upstream(fn, *a, **kw):
|
||||
try:
|
||||
return fn(*a, **kw)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(502, f"upstream error: {e}")
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def no_cache(request, call_next):
|
||||
# ponytail: no-store on everything except /stream (segments should cache)
|
||||
r = await call_next(request)
|
||||
if not request.url.path.startswith("/stream"):
|
||||
r.headers["Cache-Control"] = "no-store"
|
||||
return r
|
||||
|
||||
|
||||
@app.get("/api/search")
|
||||
def api_search(q: str = Query(min_length=1)):
|
||||
return _upstream(upstream.search, q)
|
||||
|
||||
|
||||
@app.get("/api/anime/{slug_id}")
|
||||
def api_anime(slug_id: str):
|
||||
return _upstream(upstream.details, slug_id)
|
||||
|
||||
|
||||
@app.get("/api/anime/{anime_id}/episodes")
|
||||
def api_episodes(anime_id: int):
|
||||
return _upstream(upstream.episodes, anime_id)
|
||||
|
||||
|
||||
@app.get("/api/episode/{ep_id}/stream")
|
||||
def api_stream(ep_id: int, lang: str = "eng"):
|
||||
master = _upstream(upstream.master_m3u8, ep_id, lang)
|
||||
if not master:
|
||||
raise HTTPException(404, f"no {lang} stream for episode {ep_id}")
|
||||
return {"token": _b64(master),
|
||||
"qualities": _upstream(upstream.master_variants, master)}
|
||||
|
||||
|
||||
def _rewrite_m3u8(text: str, base: str) -> str:
|
||||
# ponytail: rewrites plain URIs only; URI-list attributes in
|
||||
# #EXT-X-* tags (e.g. SUBTITLES) are ignored — fine for video-only playlists
|
||||
return "\n".join(
|
||||
line if line.startswith("#") or not line
|
||||
else "/stream/" + _b64(urljoin(base, line))
|
||||
for line in text.splitlines()) + "\n"
|
||||
|
||||
|
||||
@app.get("/stream/{token}")
|
||||
def stream(token: str):
|
||||
# ponytail: buffers whole response in memory; .ts segments are small
|
||||
url = _url(token)
|
||||
r = _upstream(upstream._get, url)
|
||||
body = r.content
|
||||
if url.endswith(".m3u8") or "mpegurl" in r.headers.get("content-type", ""):
|
||||
body = _rewrite_m3u8(body.decode(), url).encode()
|
||||
return Response(body, media_type=r.headers.get("content-type", "application/octet-stream"))
|
||||
|
||||
|
||||
app.mount("/", StaticFiles(directory=ROOT / "static", html=True), name="static")
|
||||
@@ -0,0 +1,112 @@
|
||||
"""upstream.py — the only module that talks to anidb.app.
|
||||
|
||||
anidb.app sits behind Cloudflare: plain curl gets 403. All requests use
|
||||
curl_cffi Chrome TLS impersonation (verified to pass the challenge).
|
||||
Parsers are regex against live markup — keep them here so the layout
|
||||
breaks in one place, not across the app.
|
||||
"""
|
||||
import re
|
||||
import sys
|
||||
|
||||
from curl_cffi import requests
|
||||
|
||||
BASE = "https://anidb.app"
|
||||
H = dict(impersonate="chrome", timeout=20)
|
||||
|
||||
_ANIME_LINK = re.compile(r'anime/([a-z0-9-]+)-(\d+)"[^>]*title="([^"]*)"')
|
||||
|
||||
|
||||
def _get(url, **kw):
|
||||
r = requests.get(url, **H, **kw)
|
||||
if r.status_code in (403, 503) or "Just a moment" in r.text[:500]:
|
||||
raise RuntimeError(f"blocked by cloudflare: {url}")
|
||||
r.raise_for_status()
|
||||
return r
|
||||
|
||||
|
||||
def search(q):
|
||||
"""Step 1: browse page -> [{id, slug_id, name, poster}]"""
|
||||
html = _get(f"{BASE}/browse", params={"q": q}).text
|
||||
out, seen = [], set()
|
||||
for m in re.finditer(
|
||||
r'<a href="https://anidb\.app/anime/([a-z0-9-]+-\d+)"[^>]*title="([^"]*)"[^>]*>(.*?)</a>',
|
||||
html, re.S):
|
||||
slug_id, name, block = m.group(1), m.group(2), m.group(3)
|
||||
if slug_id in seen:
|
||||
continue
|
||||
seen.add(slug_id)
|
||||
img = re.search(r'<img src="([^"]+)"', block)
|
||||
out.append({"id": int(slug_id.rsplit("-", 1)[1]), "slug_id": slug_id,
|
||||
"name": name, "poster": img.group(1) if img else None})
|
||||
return out
|
||||
|
||||
|
||||
def details(slug_id):
|
||||
"""Step 2: details page -> {name, synopsis, mal_id, seasons}"""
|
||||
p = _get(f"{BASE}/anime/{slug_id}").text
|
||||
mal = re.search(r'myanimelist\.net/anime/(\d+)', p)
|
||||
syn = re.search(r'>Synopsis</h2>.*?<p[^>]*>(.*?)</p>', p, re.S)
|
||||
h1 = re.search(r'<h1[^>]*>(.*?)</h1>', p, re.S)
|
||||
seasons, seen = [], set()
|
||||
i = p.find(">Seasons<")
|
||||
if i > 0:
|
||||
cnt = re.search(r">(\d+) entries in this series<", p[i:i + 400])
|
||||
n = int(cnt.group(1)) if cnt else 10 # fall back to cap if count markup changes
|
||||
for m in _ANIME_LINK.finditer(p[i:i + 20000]):
|
||||
sid = f"{m.group(1)}-{m.group(2)}"
|
||||
if sid in seen:
|
||||
continue
|
||||
seen.add(sid)
|
||||
seasons.append({"slug_id": sid, "name": m.group(3), "id": int(m.group(2))})
|
||||
if len(seasons) == n:
|
||||
break
|
||||
return {"id": int(slug_id.rsplit("-", 1)[1]),
|
||||
"name": h1.group(1).strip() if h1 else None,
|
||||
"synopsis": re.sub(r"<[^>]+>", "", syn.group(1)).strip() if syn else None,
|
||||
"mal_id": int(mal.group(1)) if mal else None,
|
||||
"seasons": seasons}
|
||||
|
||||
|
||||
def episodes(anime_id):
|
||||
"""Step 3: JSON API -> [{id, number, filler}]"""
|
||||
d = _get(f"{BASE}/api/frontend/anime/{anime_id}/episodes").json()
|
||||
return [{"id": e["id"], "number": e["number"], "filler": bool(e.get("filler"))}
|
||||
for e in d["episodes"]]
|
||||
|
||||
|
||||
def languages(ep_id):
|
||||
"""Step 4: JSON API -> [{code, name, embed_url}] (jpn=sub, eng=dub)"""
|
||||
return _get(f"{BASE}/api/frontend/episode/{ep_id}/languages").json()["languages"]
|
||||
|
||||
|
||||
def master_m3u8(ep_id, lang="eng"):
|
||||
"""Steps 5+6: embed page -> 'file: <master m3u8>' URL (None if unavailable)"""
|
||||
for l in languages(ep_id):
|
||||
if l["code"] == lang:
|
||||
m = re.search(r"file: '([^']*)'", _get(l["embed_url"]).text)
|
||||
if m:
|
||||
return m.group(1)
|
||||
return None
|
||||
|
||||
|
||||
def master_variants(master_url):
|
||||
"""Step 7: parse #EXT-X-STREAM-INF lines -> ['1920x1080', ...] (for display)"""
|
||||
text = _get(master_url).text
|
||||
return re.findall(r"RESOLUTION=(\d+x\d+)", text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# ponytail: assert self-check instead of a test framework
|
||||
results = search("chainsaw man")
|
||||
assert results and any(x["name"] == "Chainsaw Man" for x in results), results
|
||||
top = next(x for x in results if x["name"] == "Chainsaw Man")
|
||||
d = details(top["slug_id"])
|
||||
assert d["mal_id"] == 44511 and d["synopsis"], d
|
||||
assert [s["id"] for s in d["seasons"]] == [922, 923], d["seasons"]
|
||||
eps = episodes(top["id"])
|
||||
assert eps and eps[0]["number"] == 1, eps
|
||||
m = master_m3u8(eps[0]["id"], "eng")
|
||||
assert m and "EXT-X-STREAM" in _get(m).text, m
|
||||
print(f"OK: {top['name']} (MAL {d['mal_id']}, {len(d['seasons'])} seasons, "
|
||||
f"{len(eps)} eps) -> ep1 dub: {m[:70]}...")
|
||||
sys.exit(0)
|
||||
Reference in new issue
Block a user