shishantbiswas/aniscrap
1
0
Fork 0
mirror of https://huggingface.co/spaces/Catapang1989/aniscrap synced 2026-04-17 15:51:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update main.py

Browse Source
This commit is contained in:
Aira Catapang
2026-03-17 03:49:45 +00:00
committed by system
parent 47bf692c67
commit d408de57b6
1 changed files with 275 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

387
main.py
View File

@@ -11,21 +11,22 @@ from playwright.async_api import async_playwright, BrowserContext
BASE_URL = "https://animepahe.si"
ANILIST_API = "https://graphql.anilist.co"
JIKAN_API = "https://api.jikan.moe/v4"
KITSU_API = "https://kitsu.io/api/edge"
IS_HEADLESS = os.environ.get("HEADLESS", "true").lower() == "true"
# In-memory caches
_info_cache: dict = {} # keyed by anilist_id — full merged result
_mal_synopsis_cache: dict = {} # keyed by mal_id
_info_cache: dict = {}
_mal_synopsis_cache: dict = {}
_kitsu_relations_cache: dict = {}
# AniList relation types considered "direct"
DIRECT_RELATION_TYPES = {
"SEQUEL",
"PREQUEL",
"SIDE_STORY",
"PARENT",
"FULL_STORY",
KITSU_HEADERS = {
"Accept": "application/vnd.api+json",
"Content-Type": "application/vnd.api+json",
}
# Direct relation types (shown first)
DIRECT_RELATION_TYPES = {"sequel", "prequel", "parent", "full_story", "side_story"}
class AnimePahe:
def __init__(self):
@@ -91,9 +92,6 @@ class AnimePahe:
# ---------------- SCRAPE IDs ONLY ----------------
async def _scrape_ids(self, session: str) -> dict:
"""
Open AnimePahe anime page and collect only the external IDs.
"""
page = await self.context.new_page()
try:
await page.goto(
@@ -138,13 +136,8 @@ class AnimePahe:
# ---------------- MAL SYNOPSIS ----------------
async def _fetch_mal_synopsis(self, mal_id: str) -> Optional[str]:
"""
Fetch synopsis from MyAnimeList via Jikan API (no auth needed).
Falls back to None if unavailable.
"""
if mal_id in _mal_synopsis_cache:
return _mal_synopsis_cache[mal_id]
try:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(
@@ -152,23 +145,94 @@ class AnimePahe:
headers={"Accept": "application/json"},
)
resp.raise_for_status()
data = resp.json()
synopsis = data.get("data", {}).get("synopsis")
synopsis = resp.json().get("data", {}).get("synopsis")
_mal_synopsis_cache[mal_id] = synopsis
return synopsis
except Exception as e:
print(f"[mal_synopsis] fetch failed for mal_id={mal_id}: {e}")
print(f"[mal_synopsis] failed for mal_id={mal_id}: {e}")
_mal_synopsis_cache[mal_id] = None
return None
# ---------------- KITSU RELATIONS ----------------
async def _fetch_kitsu_relations(self, kitsu_id: str) -> list:
"""
Fetch ALL related anime from Kitsu — full chain including all seasons,
movies, OVAs, specials. Direct types listed first.
"""
if kitsu_id in _kitsu_relations_cache:
return _kitsu_relations_cache[kitsu_id]
try:
async with httpx.AsyncClient(timeout=15) as client:
url = (
f"{KITSU_API}/anime/{kitsu_id}/media-relationships"
f"?include=destination"
f"&fields[anime]=canonicalTitle,posterImage,episodeCount,status,subtype,startDate"
f"&page[limit]=20"
)
resp = await client.get(url, headers=KITSU_HEADERS)
resp.raise_for_status()
data = resp.json()
except Exception as e:
print(f"[kitsu_relations] failed for kitsu_id={kitsu_id}: {e}")
_kitsu_relations_cache[kitsu_id] = []
return []
# Build lookup of included resources by id
included = {}
for item in data.get("included", []):
included[item["id"]] = item
direct = []
indirect = []
for rel in data.get("data", []):
attrs = rel.get("attributes", {})
role = (attrs.get("role") or "").lower()
dest_data = (
rel.get("relationships", {}).get("destination", {}).get("data", {})
)
dest_type = dest_data.get("type", "")
dest_id = dest_data.get("id", "")
# Only include anime destinations
if dest_type != "anime":
continue
dest = included.get(dest_id, {})
dest_attrs = dest.get("attributes", {})
poster = dest_attrs.get("posterImage") or {}
entry = {
"kitsu_id": dest_id,
"title": dest_attrs.get("canonicalTitle"),
"format": dest_attrs.get("subtype"),
"status": dest_attrs.get("status"),
"episodes": dest_attrs.get("episodeCount"),
"start_date": dest_attrs.get("startDate"),
"image": (
poster.get("small")
or poster.get("medium")
or poster.get("original")
),
"url": f"https://kitsu.io/anime/{dest_id}",
"relation_type": role,
}
if role in DIRECT_RELATION_TYPES:
direct.append(entry)
else:
indirect.append(entry)
combined = direct + indirect
_kitsu_relations_cache[kitsu_id] = combined
return combined
# ---------------- ANILIST ----------------
async def _fetch_anilist(self, anilist_id: str) -> dict:
"""
Query AniList GraphQL API.
Relations: direct (Sequel/Prequel/etc.) + indirect combined into
a single "Related" list — direct entries first.
"""
query = """
query ($id: Int) {
Media(id: $id, type: ANIME) {
@@ -243,22 +307,6 @@ class AnimePahe:
}
}
}
relations {
edges {
relationType(version: 2)
node {
id
idMal
type
title { romaji english }
format
status
episodes
coverImage { medium }
siteUrl
}
}
}
recommendations(perPage: 20, sort: RATING_DESC) {
nodes {
rating
@@ -301,26 +349,26 @@ class AnimePahe:
resp.raise_for_status()
result = resp.json()
except Exception as e:
print(f"[anilist] fetch failed for id={anilist_id}: {e}")
print(f"[anilist] failed for id={anilist_id}: {e}")
return {"error": f"AniList fetch failed: {str(e)}"}
media = result.get("data", {}).get("Media")
if not media:
return {"error": "AniList returned no data"}
# ── MAL synopsis — cleaner than AniList's HTML-heavy description ──
# MAL synopsis
mal_id = str(media.get("idMal") or "")
mal_synopsis = await self._fetch_mal_synopsis(mal_id) if mal_id else None
synopsis = mal_synopsis or media.get("description")
# ── Format dates ──────────────────────────────────────────────
# Format dates
def fmt_date(d):
if not d or not d.get("year"):
return None
parts = [d.get("year"), d.get("month"), d.get("day")]
return "-".join(str(p).zfill(2) for p in parts if p)
# ── Trailer URL ───────────────────────────────────────────────
# Trailer
trailer = None
if media.get("trailer"):
t = media["trailer"]
@@ -329,39 +377,7 @@ class AnimePahe:
elif t.get("site") == "dailymotion":
trailer = f"https://www.dailymotion.com/video/{t['id']}"
# ── Relations — direct first, indirect after, all in "Related" ─
direct = []
indirect = []
for edge in media.get("relations", {}).get("edges", []):
rel_type = edge.get("relationType", "OTHER")
node = edge.get("node", {})
# Skip non-anime relations (manga, novel, one-shot, etc.)
if node.get("type") != "ANIME":
continue
entry = {
"id": node.get("id"),
"mal_id": node.get("idMal"),
"title": node["title"].get("english") or node["title"].get("romaji"),
"format": node.get("format"),
"status": node.get("status"),
"episodes": node.get("episodes"),
"image": node.get("coverImage", {}).get("medium"),
"url": node.get("siteUrl"),
"relation_type": rel_type,
}
if rel_type in DIRECT_RELATION_TYPES:
direct.append(entry)
else:
indirect.append(entry)
# Combined: direct first, indirect after — all under one "Related" key
combined = direct + indirect
relations = {"Related": combined} if combined else {}
# ── Recommendations ───────────────────────────────────────────
# Recommendations
recommendations = []
for node in media.get("recommendations", {}).get("nodes", []):
rec = node.get("mediaRecommendation")
@@ -382,7 +398,7 @@ class AnimePahe:
}
)
# ── Characters ────────────────────────────────────────────────
# Characters
characters = []
for edge in media.get("characters", {}).get("edges", []):
node = edge.get("node", {})
@@ -403,7 +419,7 @@ class AnimePahe:
}
)
# ── Staff ─────────────────────────────────────────────────────
# Staff
staff = []
for edge in media.get("staff", {}).get("edges", []):
node = edge.get("node", {})
@@ -463,7 +479,7 @@ class AnimePahe:
],
"characters": characters,
"staff": staff,
"relations": relations,
"relations": {}, # filled by get_info() from Kitsu
"recommendations": recommendations,
}
@@ -480,16 +496,39 @@ class AnimePahe:
# ---------------- EPISODES ----------------
async def get_episodes(self, anime_id: str, p: int = 1):
return await self._fetch_json(
async def get_episodes(self, anime_id: str, p: int = 1, resolve: bool = False):
"""
Fetch episode list. If resolve=True, also resolve the highest-res
stream URL and download link for each episode concurrently.
"""
data = await self._fetch_json(
f"{BASE_URL}/api?m=release&id={anime_id}&sort=episode_desc&page={p}"
)
if not data or not resolve:
return data
episodes = data.get("data", [])
async def enrich(ep):
ep_session = ep.get("session")
if not ep_session:
return ep
stream = await self._resolve_episode(anime_id, ep_session)
ep["url"] = stream.get("url")
ep["download"] = stream.get("download")
ep["resolution"] = stream.get("resolution")
ep["fansub"] = stream.get("fansub")
return ep
data["data"] = list(await asyncio.gather(*[enrich(ep) for ep in episodes]))
return data
# ---------------- INFO ----------------
async def get_info(self, session: str):
try:
# Step 1 — scrape IDs from AnimePahe page
# Step 1 — scrape IDs from AnimePahe
ids = await self._scrape_ids(session)
anilist_id = ids.get("anilist")
@@ -503,24 +542,37 @@ class AnimePahe:
if anilist_id in _info_cache:
return _info_cache[anilist_id]
# Step 3 — fetch everything from AniList (includes relations)
data = await self._fetch_anilist(anilist_id)
# Step 3 — fetch AniList data + Kitsu relations concurrently
kitsu_id = ids.get("kitsu")
async def empty_relations():
return []
anilist_task = self._fetch_anilist(anilist_id)
kitsu_task = (
self._fetch_kitsu_relations(kitsu_id) if kitsu_id else empty_relations()
)
data, kitsu_relations = await asyncio.gather(anilist_task, kitsu_task)
if "error" in data:
return {"error": data["error"], "ids": ids}
# Step 4 — inject all scraped IDs
# Step 4 — inject Kitsu relations under "Related"
data["relations"] = {"Related": kitsu_relations} if kitsu_relations else {}
# Step 5 — inject all IDs
data["ids"] = {
"animepahe": ids.get("animepahe"),
"anilist": anilist_id,
"mal": ids.get("mal"),
"anidb": ids.get("anidb"),
"kitsu": ids.get("kitsu"),
"kitsu": kitsu_id,
"ann": ids.get("ann"),
"animePlanet": ids.get("animePlanet"),
}
# Step 5 — cache and return
# Step 6 — cache fully merged result
_info_cache[anilist_id] = data
return data
@@ -528,65 +580,176 @@ class AnimePahe:
print(f"[get_info] ERROR: {e}")
return {"error": f"Failed: {str(e)}"}
# --- THE FIXED RESOLVER ---
async def resolve(self, anime_session: str, episode_session: str):
# ---------------- RESOLVE (single episode → highest res only) ----------------
async def _resolve_episode(self, anime_session: str, episode_session: str) -> dict:
"""
Open the play page, collect all resolution buttons, resolve only the
highest-resolution embed to its m3u8, and return url + download link.
"""
play_url = f"{BASE_URL}/play/{anime_session}/{episode_session}"
page = await self.context.new_page()
try:
await page.goto(play_url, wait_until="domcontentloaded")
await page.wait_for_selector("#resolutionMenu button", timeout=5000)
await page.wait_for_selector(
"#resolutionMenu button",
state="attached",
timeout=15000,
)
buttons = await page.locator("#resolutionMenu button").all()
res_data = []
for btn in buttons:
text = (await btn.inner_text()).strip()
res_match = re.search(r"(\d+)", text)
res_data.append(
{
"embed": await btn.get_attribute("data-src"),
"res": (re.search(r"(\d+)", text) or ["720"])[0],
"fanSub": text.split("·")[0].strip()
"res": int(res_match.group(1)) if res_match else 720,
"fansub": text.split("·")[0].strip()
if "·" in text
else "Unknown",
}
)
await page.close()
# Parallel resolution using the "Request Capture" method
async def get_single_mp4(item):
await page.close()
page = None
if not res_data:
return {
"url": None,
"download": None,
"resolution": None,
"fansub": None,
}
# Pick highest resolution
best = max(res_data, key=lambda x: x["res"])
# Resolve best embed to m3u8
p = await self.context.new_page()
m3u8 = None
def capture(req):
nonlocal m3u8
if ".m3u8" in req.url:
m3u8 = req.url
p.on("request", capture)
try:
await p.set_extra_http_headers({"Referer": BASE_URL})
await p.goto(best["embed"], wait_until="domcontentloaded")
for _ in range(10):
if m3u8:
break
await p.evaluate(
"document.querySelectorAll('button, video, [class*=play]')"
".forEach(el => el.click())"
)
await asyncio.sleep(0.5)
finally:
await p.close()
res_str = str(best["res"])
return {
"url": m3u8,
"download": self._generate_mp4(m3u8, anime_session, res_str),
"resolution": res_str,
"fansub": best["fansub"],
}
except Exception as e:
return {
"url": None,
"download": None,
"resolution": None,
"fansub": None,
"error": str(e),
}
finally:
if page:
await page.close()
async def resolve(self, anime_session: str, episode_session: str):
"""Resolve all sources for a single episode (all resolutions)."""
play_url = f"{BASE_URL}/play/{anime_session}/{episode_session}"
page = await self.context.new_page()
try:
await page.goto(play_url, wait_until="domcontentloaded")
await page.wait_for_selector(
"#resolutionMenu button",
state="attached",
timeout=15000,
)
buttons = await page.locator("#resolutionMenu button").all()
res_data = []
for btn in buttons:
text = (await btn.inner_text()).strip()
res_match = re.search(r"(\d+)", text)
res_data.append(
{
"embed": await btn.get_attribute("data-src"),
"res": res_match.group(1) if res_match else "720",
"fansub": text.split("·")[0].strip()
if "·" in text
else "Unknown",
}
)
await page.close()
page = None
async def get_single_source(item):
p = await self.context.new_page()
m3u8 = None
def log_req(req):
def capture(req):
nonlocal m3u8
if ".m3u8" in req.url:
m3u8 = req.url
p.on("request", log_req)
p.on("request", capture)
try:
await p.set_extra_http_headers({"Referer": BASE_URL})
await p.goto(item["embed"], wait_until="domcontentloaded")
# Force the player to trigger the m3u8 request
for _ in range(5):
for _ in range(10):
if m3u8:
break
await p.evaluate(
"document.querySelectorAll('button, video').forEach(el => el.click())"
"document.querySelectorAll('button, video, [class*=play]')"
".forEach(el => el.click())"
)
await asyncio.sleep(0.5)
item["url"] = m3u8
item["download"] = self._generate_mp4(
m3u8, anime_session, item["res"]
)
return item
return {
"resolution": item["res"],
"fansub": item["fansub"],
"url": m3u8,
"download": self._generate_mp4(
m3u8, anime_session, item["res"]
),
}
except Exception as e:
return {
"resolution": item["res"],
"fansub": item["fansub"],
"url": None,
"download": None,
"error": str(e),
}
finally:
await p.close()
sources = await asyncio.gather(*[get_single_mp4(i) for i in res_data])
return {"anime": anime_session, "sources": sources}
sources = await asyncio.gather(*[get_single_source(i) for i in res_data])
return {"anime": anime_session, "sources": list(sources)}
except Exception as e:
return {"error": str(e)}
finally:
if page:
await page.close()
pahe = AnimePahe()
@@ -618,8 +781,8 @@ async def api_info(session: str):
@app.get("/episodes/{session}")
async def api_episodes(session: str, p: int = 1):
return await pahe.get_episodes(session, p)
async def api_episodes(session: str, p: int = 1, resolve: bool = False):
return await pahe.get_episodes(session, p, resolve)
@app.get("/resolve/{anime}/{episode}")