adding start.sh
This commit is contained in:
Blitz08
@@ -1 +1 @@
|
||||
#!/usr/bin/env python3
|
||||
#!/usr/bin/env python3
|
||||
|
||||
@@ -1,141 +1,141 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import date, timedelta
|
||||
|
||||
ANILIST_URL = "https://graphql.anilist.co"
|
||||
|
||||
|
||||
def yyyymmdd(value: date) -> int:
|
||||
return value.year * 10000 + value.month * 100 + value.day
|
||||
|
||||
|
||||
def post_graphql(query: str, variables: dict) -> dict:
|
||||
payload = json.dumps({"query": query, "variables": variables}).encode("utf-8")
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
"User-Agent": "anime-movies-script/1.0",
|
||||
}
|
||||
req = urllib.request.Request(ANILIST_URL, data=payload, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
return json.loads(resp.read().decode("utf-8"))
|
||||
except urllib.error.HTTPError as exc:
|
||||
body = exc.read().decode("utf-8", "replace")
|
||||
raise RuntimeError(f"HTTP {exc.code}: {body}") from exc
|
||||
|
||||
|
||||
def map_anilist_movie(media: dict | None) -> dict:
|
||||
media = media or {}
|
||||
title = media.get("title") or {}
|
||||
start_date = media.get("startDate") or {}
|
||||
studios = ((media.get("studios") or {}).get("nodes") or [])
|
||||
|
||||
studio_names = []
|
||||
for node in studios:
|
||||
name = ((node or {}).get("name") or "").strip()
|
||||
if name:
|
||||
studio_names.append(name)
|
||||
|
||||
genres = []
|
||||
for entry in media.get("genres") or []:
|
||||
text = str(entry or "").strip()
|
||||
if text:
|
||||
genres.append(text)
|
||||
|
||||
tags = []
|
||||
for tag in media.get("tags") or []:
|
||||
if not isinstance(tag, dict):
|
||||
continue
|
||||
name = (tag.get("name") or "").strip()
|
||||
rank = int(tag.get("rank") or 0)
|
||||
if name and rank >= 70:
|
||||
tags.append(name)
|
||||
|
||||
return {
|
||||
"anilist_id": int(media.get("id") or 0),
|
||||
"title_english": (title.get("english") or "").strip(),
|
||||
"title_romaji": (title.get("romaji") or "").strip(),
|
||||
"title_native": (title.get("native") or "").strip(),
|
||||
"cover_image": (((media.get("coverImage") or {}).get("large")) or "").strip(),
|
||||
"anilist_url": (media.get("siteUrl") or "").strip(),
|
||||
"start_year": int(start_date.get("year") or 0),
|
||||
"format": (media.get("format") or "").strip() or "MOVIE",
|
||||
"description": (media.get("description") or "").strip(),
|
||||
"genres_text": ", ".join(genres),
|
||||
"tags_text": ", ".join(tags),
|
||||
"studio_text": ", ".join(studio_names),
|
||||
}
|
||||
|
||||
|
||||
def fetch_anilist_movie_candidates(today: date, years_window: int = 1) -> list[dict]:
|
||||
start_date = today - timedelta(days=365 * years_window)
|
||||
end_date = today + timedelta(days=365 * years_window)
|
||||
|
||||
query = """
|
||||
query ($page: Int, $perPage: Int, $start: FuzzyDateInt, $end: FuzzyDateInt) {
|
||||
Page(page: $page, perPage: $perPage) {
|
||||
pageInfo { hasNextPage }
|
||||
media(
|
||||
type: ANIME
|
||||
format: MOVIE
|
||||
countryOfOrigin: JP
|
||||
sort: [POPULARITY_DESC, START_DATE_DESC]
|
||||
startDate_greater: $start
|
||||
startDate_lesser: $end
|
||||
) {
|
||||
id
|
||||
title { english romaji native }
|
||||
startDate { year }
|
||||
format
|
||||
description(asHtml: false)
|
||||
genres
|
||||
tags { name rank }
|
||||
coverImage { large }
|
||||
studios { nodes { name } }
|
||||
siteUrl
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
page = 1
|
||||
results = []
|
||||
seen_ids = set()
|
||||
|
||||
while True:
|
||||
payload = post_graphql(
|
||||
query,
|
||||
{
|
||||
"page": page,
|
||||
"perPage": 50,
|
||||
"start": yyyymmdd(start_date) - 1,
|
||||
"end": yyyymmdd(end_date) + 1,
|
||||
},
|
||||
)
|
||||
|
||||
if "errors" in payload:
|
||||
raise RuntimeError(payload["errors"])
|
||||
|
||||
page_data = payload.get("data", {}).get("Page", {})
|
||||
for media in page_data.get("media", []):
|
||||
mapped = map_anilist_movie(media)
|
||||
anilist_id = mapped.get("anilist_id") or 0
|
||||
if anilist_id <= 0 or anilist_id in seen_ids:
|
||||
continue
|
||||
seen_ids.add(anilist_id)
|
||||
|
||||
if not mapped.get("title_english") and not mapped.get("title_romaji") and not mapped.get("title_native"):
|
||||
continue
|
||||
|
||||
results.append(mapped)
|
||||
|
||||
if not page_data.get("pageInfo", {}).get("hasNextPage"):
|
||||
break
|
||||
page += 1
|
||||
|
||||
return results
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import date, timedelta
|
||||
|
||||
ANILIST_URL = "https://graphql.anilist.co"
|
||||
|
||||
|
||||
def yyyymmdd(value: date) -> int:
|
||||
return value.year * 10000 + value.month * 100 + value.day
|
||||
|
||||
|
||||
def post_graphql(query: str, variables: dict) -> dict:
|
||||
payload = json.dumps({"query": query, "variables": variables}).encode("utf-8")
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
"User-Agent": "anime-movies-script/1.0",
|
||||
}
|
||||
req = urllib.request.Request(ANILIST_URL, data=payload, headers=headers)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
return json.loads(resp.read().decode("utf-8"))
|
||||
except urllib.error.HTTPError as exc:
|
||||
body = exc.read().decode("utf-8", "replace")
|
||||
raise RuntimeError(f"HTTP {exc.code}: {body}") from exc
|
||||
|
||||
|
||||
def map_anilist_movie(media: dict | None) -> dict:
|
||||
media = media or {}
|
||||
title = media.get("title") or {}
|
||||
start_date = media.get("startDate") or {}
|
||||
studios = ((media.get("studios") or {}).get("nodes") or [])
|
||||
|
||||
studio_names = []
|
||||
for node in studios:
|
||||
name = ((node or {}).get("name") or "").strip()
|
||||
if name:
|
||||
studio_names.append(name)
|
||||
|
||||
genres = []
|
||||
for entry in media.get("genres") or []:
|
||||
text = str(entry or "").strip()
|
||||
if text:
|
||||
genres.append(text)
|
||||
|
||||
tags = []
|
||||
for tag in media.get("tags") or []:
|
||||
if not isinstance(tag, dict):
|
||||
continue
|
||||
name = (tag.get("name") or "").strip()
|
||||
rank = int(tag.get("rank") or 0)
|
||||
if name and rank >= 70:
|
||||
tags.append(name)
|
||||
|
||||
return {
|
||||
"anilist_id": int(media.get("id") or 0),
|
||||
"title_english": (title.get("english") or "").strip(),
|
||||
"title_romaji": (title.get("romaji") or "").strip(),
|
||||
"title_native": (title.get("native") or "").strip(),
|
||||
"cover_image": (((media.get("coverImage") or {}).get("large")) or "").strip(),
|
||||
"anilist_url": (media.get("siteUrl") or "").strip(),
|
||||
"start_year": int(start_date.get("year") or 0),
|
||||
"format": (media.get("format") or "").strip() or "MOVIE",
|
||||
"description": (media.get("description") or "").strip(),
|
||||
"genres_text": ", ".join(genres),
|
||||
"tags_text": ", ".join(tags),
|
||||
"studio_text": ", ".join(studio_names),
|
||||
}
|
||||
|
||||
|
||||
def fetch_anilist_movie_candidates(today: date, years_window: int = 1) -> list[dict]:
|
||||
start_date = today - timedelta(days=365 * years_window)
|
||||
end_date = today + timedelta(days=365 * years_window)
|
||||
|
||||
query = """
|
||||
query ($page: Int, $perPage: Int, $start: FuzzyDateInt, $end: FuzzyDateInt) {
|
||||
Page(page: $page, perPage: $perPage) {
|
||||
pageInfo { hasNextPage }
|
||||
media(
|
||||
type: ANIME
|
||||
format: MOVIE
|
||||
countryOfOrigin: JP
|
||||
sort: [POPULARITY_DESC, START_DATE_DESC]
|
||||
startDate_greater: $start
|
||||
startDate_lesser: $end
|
||||
) {
|
||||
id
|
||||
title { english romaji native }
|
||||
startDate { year }
|
||||
format
|
||||
description(asHtml: false)
|
||||
genres
|
||||
tags { name rank }
|
||||
coverImage { large }
|
||||
studios { nodes { name } }
|
||||
siteUrl
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
page = 1
|
||||
results = []
|
||||
seen_ids = set()
|
||||
|
||||
while True:
|
||||
payload = post_graphql(
|
||||
query,
|
||||
{
|
||||
"page": page,
|
||||
"perPage": 50,
|
||||
"start": yyyymmdd(start_date) - 1,
|
||||
"end": yyyymmdd(end_date) + 1,
|
||||
},
|
||||
)
|
||||
|
||||
if "errors" in payload:
|
||||
raise RuntimeError(payload["errors"])
|
||||
|
||||
page_data = payload.get("data", {}).get("Page", {})
|
||||
for media in page_data.get("media", []):
|
||||
mapped = map_anilist_movie(media)
|
||||
anilist_id = mapped.get("anilist_id") or 0
|
||||
if anilist_id <= 0 or anilist_id in seen_ids:
|
||||
continue
|
||||
seen_ids.add(anilist_id)
|
||||
|
||||
if not mapped.get("title_english") and not mapped.get("title_romaji") and not mapped.get("title_native"):
|
||||
continue
|
||||
|
||||
results.append(mapped)
|
||||
|
||||
if not page_data.get("pageInfo", {}).get("hasNextPage"):
|
||||
break
|
||||
page += 1
|
||||
|
||||
return results
|
||||
|
||||
@@ -1,261 +1,261 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
ANIMESCHEDULE_API_BASE = "https://animeschedule.net/api/v3"
|
||||
|
||||
|
||||
def normalize_title(text: str) -> str:
|
||||
cleaned = "".join(ch.lower() if ch.isalnum() else " " for ch in (text or ""))
|
||||
return " ".join(cleaned.split())
|
||||
|
||||
|
||||
def flatten_items(payload: dict | list) -> list[dict]:
|
||||
if isinstance(payload, list):
|
||||
return [item for item in payload if isinstance(item, dict)]
|
||||
if not isinstance(payload, dict):
|
||||
return []
|
||||
|
||||
for key in ("anime", "items", "data", "results"):
|
||||
value = payload.get(key)
|
||||
if isinstance(value, list):
|
||||
return [item for item in value if isinstance(item, dict)]
|
||||
return []
|
||||
|
||||
|
||||
def extract_names(item: dict) -> list[str]:
|
||||
names = []
|
||||
direct = [item.get("title"), item.get("name"), item.get("romaji"), item.get("english"), item.get("native")]
|
||||
for candidate in direct:
|
||||
text = (candidate or "").strip()
|
||||
if text:
|
||||
names.append(text)
|
||||
|
||||
nested = item.get("names") or {}
|
||||
if isinstance(nested, dict):
|
||||
for candidate in nested.values():
|
||||
if isinstance(candidate, list):
|
||||
for value in candidate:
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
names.append(text)
|
||||
continue
|
||||
|
||||
text = (str(candidate) if candidate is not None else "").strip()
|
||||
if text:
|
||||
names.append(text)
|
||||
|
||||
unique = []
|
||||
seen = set()
|
||||
for name in names:
|
||||
lowered = name.lower()
|
||||
if lowered in seen:
|
||||
continue
|
||||
seen.add(lowered)
|
||||
unique.append(name)
|
||||
return unique
|
||||
|
||||
|
||||
def extract_url(item: dict) -> str:
|
||||
slug = (item.get("slug") or item.get("route") or item.get("id") or "").strip()
|
||||
if slug:
|
||||
return f"https://animeschedule.net/anime/{slug}"
|
||||
|
||||
websites = item.get("websites") or item.get("links") or []
|
||||
if isinstance(websites, list):
|
||||
for entry in websites:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
url = (entry.get("url") or "").strip()
|
||||
if url:
|
||||
return url
|
||||
return ""
|
||||
|
||||
|
||||
def extract_english_title(item: dict) -> str:
|
||||
direct = [item.get("english"), item.get("titleEnglish"), item.get("englishTitle")]
|
||||
for candidate in direct:
|
||||
text = (candidate or "").strip()
|
||||
if text:
|
||||
return text
|
||||
|
||||
names = item.get("names") or {}
|
||||
if isinstance(names, dict):
|
||||
for key, value in names.items():
|
||||
key_norm = str(key).strip().lower()
|
||||
if key_norm in {"en", "eng", "english", "titleenglish"}:
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
return text
|
||||
return ""
|
||||
|
||||
|
||||
def extract_list(item: dict, *keys: str) -> str:
|
||||
for key in keys:
|
||||
value = item.get(key)
|
||||
if not value:
|
||||
continue
|
||||
if isinstance(value, str):
|
||||
text = value.strip()
|
||||
if text:
|
||||
return text
|
||||
if isinstance(value, list):
|
||||
names = []
|
||||
for entry in value:
|
||||
if isinstance(entry, str):
|
||||
text = entry.strip()
|
||||
elif isinstance(entry, dict):
|
||||
text = (entry.get("name") or entry.get("title") or "").strip()
|
||||
else:
|
||||
text = ""
|
||||
if text:
|
||||
names.append(text)
|
||||
if names:
|
||||
return ", ".join(names)
|
||||
return ""
|
||||
|
||||
|
||||
def extract_format(item: dict) -> str:
|
||||
for key in ("format", "mediaType", "type"):
|
||||
value = item.get(key)
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
return text
|
||||
return ""
|
||||
|
||||
|
||||
def extract_description(item: dict) -> str:
|
||||
for key in ("description", "synopsis", "overview", "summary"):
|
||||
value = item.get(key)
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
return text
|
||||
|
||||
details = item.get("details") or {}
|
||||
if isinstance(details, dict):
|
||||
for key in ("description", "synopsis", "overview", "summary"):
|
||||
value = details.get(key)
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
return text
|
||||
return ""
|
||||
|
||||
|
||||
def empty_result() -> dict:
|
||||
return {
|
||||
"url": "",
|
||||
"title": "",
|
||||
"english_title": "",
|
||||
"format": "",
|
||||
"genres": "",
|
||||
"studios": "",
|
||||
"description": "",
|
||||
"names": [],
|
||||
"match_score": 0,
|
||||
"raw": {},
|
||||
}
|
||||
|
||||
|
||||
def title_match_score(wanted: str, names: list[str]) -> int:
|
||||
wanted_norm = normalize_title(wanted)
|
||||
if not wanted_norm:
|
||||
return 0
|
||||
|
||||
best = 0
|
||||
wanted_parts = wanted_norm.split(" ")
|
||||
|
||||
for candidate in names:
|
||||
normalized = normalize_title(candidate)
|
||||
if not normalized:
|
||||
continue
|
||||
|
||||
if normalized == wanted_norm:
|
||||
best = max(best, 4)
|
||||
continue
|
||||
|
||||
if wanted_norm in normalized or normalized in wanted_norm:
|
||||
best = max(best, 3)
|
||||
continue
|
||||
|
||||
normalized_parts = normalized.split(" ")
|
||||
if normalized_parts[:2] == wanted_parts[:2] and len(normalized_parts) >= 2 and len(wanted_parts) >= 2:
|
||||
best = max(best, 2)
|
||||
continue
|
||||
|
||||
if set(normalized_parts) & set(wanted_parts):
|
||||
best = max(best, 1)
|
||||
|
||||
return best
|
||||
|
||||
|
||||
def fetch_animeschedule_anime_by_title(title: str, token: str, cache: dict[str, dict]) -> dict:
|
||||
key = normalize_title(title)
|
||||
if key in cache:
|
||||
return cache[key]
|
||||
|
||||
if not token:
|
||||
cache[key] = empty_result()
|
||||
return cache[key]
|
||||
|
||||
params = {"search": title, "take": "10"}
|
||||
url = ANIMESCHEDULE_API_BASE + "/anime?" + urllib.parse.urlencode(params)
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
headers={
|
||||
"Accept": "application/json",
|
||||
"Authorization": f"Bearer {token}",
|
||||
"User-Agent": "anime-movies-script/1.0",
|
||||
},
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=20) as resp:
|
||||
payload = json.loads(resp.read().decode("utf-8"))
|
||||
except Exception:
|
||||
cache[key] = empty_result()
|
||||
return cache[key]
|
||||
|
||||
candidates = flatten_items(payload)
|
||||
if not candidates:
|
||||
cache[key] = empty_result()
|
||||
return cache[key]
|
||||
|
||||
best = None
|
||||
best_score = -1
|
||||
|
||||
for item in candidates:
|
||||
names = extract_names(item)
|
||||
if not names:
|
||||
continue
|
||||
|
||||
score = title_match_score(title, names)
|
||||
if score <= 0:
|
||||
continue
|
||||
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best = item
|
||||
if score == 4:
|
||||
break
|
||||
|
||||
if not best:
|
||||
cache[key] = empty_result()
|
||||
return cache[key]
|
||||
|
||||
names = extract_names(best)
|
||||
cache[key] = {
|
||||
"url": extract_url(best),
|
||||
"title": names[0] if names else "",
|
||||
"english_title": extract_english_title(best),
|
||||
"format": extract_format(best),
|
||||
"genres": extract_list(best, "genres", "genre", "categories"),
|
||||
"studios": extract_list(best, "studios", "studio"),
|
||||
"description": extract_description(best),
|
||||
"names": names,
|
||||
"match_score": best_score,
|
||||
"raw": best,
|
||||
}
|
||||
return cache[key]
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
ANIMESCHEDULE_API_BASE = "https://animeschedule.net/api/v3"
|
||||
|
||||
|
||||
def normalize_title(text: str) -> str:
|
||||
cleaned = "".join(ch.lower() if ch.isalnum() else " " for ch in (text or ""))
|
||||
return " ".join(cleaned.split())
|
||||
|
||||
|
||||
def flatten_items(payload: dict | list) -> list[dict]:
|
||||
if isinstance(payload, list):
|
||||
return [item for item in payload if isinstance(item, dict)]
|
||||
if not isinstance(payload, dict):
|
||||
return []
|
||||
|
||||
for key in ("anime", "items", "data", "results"):
|
||||
value = payload.get(key)
|
||||
if isinstance(value, list):
|
||||
return [item for item in value if isinstance(item, dict)]
|
||||
return []
|
||||
|
||||
|
||||
def extract_names(item: dict) -> list[str]:
|
||||
names = []
|
||||
direct = [item.get("title"), item.get("name"), item.get("romaji"), item.get("english"), item.get("native")]
|
||||
for candidate in direct:
|
||||
text = (candidate or "").strip()
|
||||
if text:
|
||||
names.append(text)
|
||||
|
||||
nested = item.get("names") or {}
|
||||
if isinstance(nested, dict):
|
||||
for candidate in nested.values():
|
||||
if isinstance(candidate, list):
|
||||
for value in candidate:
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
names.append(text)
|
||||
continue
|
||||
|
||||
text = (str(candidate) if candidate is not None else "").strip()
|
||||
if text:
|
||||
names.append(text)
|
||||
|
||||
unique = []
|
||||
seen = set()
|
||||
for name in names:
|
||||
lowered = name.lower()
|
||||
if lowered in seen:
|
||||
continue
|
||||
seen.add(lowered)
|
||||
unique.append(name)
|
||||
return unique
|
||||
|
||||
|
||||
def extract_url(item: dict) -> str:
|
||||
slug = (item.get("slug") or item.get("route") or item.get("id") or "").strip()
|
||||
if slug:
|
||||
return f"https://animeschedule.net/anime/{slug}"
|
||||
|
||||
websites = item.get("websites") or item.get("links") or []
|
||||
if isinstance(websites, list):
|
||||
for entry in websites:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
url = (entry.get("url") or "").strip()
|
||||
if url:
|
||||
return url
|
||||
return ""
|
||||
|
||||
|
||||
def extract_english_title(item: dict) -> str:
|
||||
direct = [item.get("english"), item.get("titleEnglish"), item.get("englishTitle")]
|
||||
for candidate in direct:
|
||||
text = (candidate or "").strip()
|
||||
if text:
|
||||
return text
|
||||
|
||||
names = item.get("names") or {}
|
||||
if isinstance(names, dict):
|
||||
for key, value in names.items():
|
||||
key_norm = str(key).strip().lower()
|
||||
if key_norm in {"en", "eng", "english", "titleenglish"}:
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
return text
|
||||
return ""
|
||||
|
||||
|
||||
def extract_list(item: dict, *keys: str) -> str:
|
||||
for key in keys:
|
||||
value = item.get(key)
|
||||
if not value:
|
||||
continue
|
||||
if isinstance(value, str):
|
||||
text = value.strip()
|
||||
if text:
|
||||
return text
|
||||
if isinstance(value, list):
|
||||
names = []
|
||||
for entry in value:
|
||||
if isinstance(entry, str):
|
||||
text = entry.strip()
|
||||
elif isinstance(entry, dict):
|
||||
text = (entry.get("name") or entry.get("title") or "").strip()
|
||||
else:
|
||||
text = ""
|
||||
if text:
|
||||
names.append(text)
|
||||
if names:
|
||||
return ", ".join(names)
|
||||
return ""
|
||||
|
||||
|
||||
def extract_format(item: dict) -> str:
|
||||
for key in ("format", "mediaType", "type"):
|
||||
value = item.get(key)
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
return text
|
||||
return ""
|
||||
|
||||
|
||||
def extract_description(item: dict) -> str:
|
||||
for key in ("description", "synopsis", "overview", "summary"):
|
||||
value = item.get(key)
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
return text
|
||||
|
||||
details = item.get("details") or {}
|
||||
if isinstance(details, dict):
|
||||
for key in ("description", "synopsis", "overview", "summary"):
|
||||
value = details.get(key)
|
||||
text = (str(value) if value is not None else "").strip()
|
||||
if text:
|
||||
return text
|
||||
return ""
|
||||
|
||||
|
||||
def empty_result() -> dict:
|
||||
return {
|
||||
"url": "",
|
||||
"title": "",
|
||||
"english_title": "",
|
||||
"format": "",
|
||||
"genres": "",
|
||||
"studios": "",
|
||||
"description": "",
|
||||
"names": [],
|
||||
"match_score": 0,
|
||||
"raw": {},
|
||||
}
|
||||
|
||||
|
||||
def title_match_score(wanted: str, names: list[str]) -> int:
|
||||
wanted_norm = normalize_title(wanted)
|
||||
if not wanted_norm:
|
||||
return 0
|
||||
|
||||
best = 0
|
||||
wanted_parts = wanted_norm.split(" ")
|
||||
|
||||
for candidate in names:
|
||||
normalized = normalize_title(candidate)
|
||||
if not normalized:
|
||||
continue
|
||||
|
||||
if normalized == wanted_norm:
|
||||
best = max(best, 4)
|
||||
continue
|
||||
|
||||
if wanted_norm in normalized or normalized in wanted_norm:
|
||||
best = max(best, 3)
|
||||
continue
|
||||
|
||||
normalized_parts = normalized.split(" ")
|
||||
if normalized_parts[:2] == wanted_parts[:2] and len(normalized_parts) >= 2 and len(wanted_parts) >= 2:
|
||||
best = max(best, 2)
|
||||
continue
|
||||
|
||||
if set(normalized_parts) & set(wanted_parts):
|
||||
best = max(best, 1)
|
||||
|
||||
return best
|
||||
|
||||
|
||||
def fetch_animeschedule_anime_by_title(title: str, token: str, cache: dict[str, dict]) -> dict:
|
||||
key = normalize_title(title)
|
||||
if key in cache:
|
||||
return cache[key]
|
||||
|
||||
if not token:
|
||||
cache[key] = empty_result()
|
||||
return cache[key]
|
||||
|
||||
params = {"search": title, "take": "10"}
|
||||
url = ANIMESCHEDULE_API_BASE + "/anime?" + urllib.parse.urlencode(params)
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
headers={
|
||||
"Accept": "application/json",
|
||||
"Authorization": f"Bearer {token}",
|
||||
"User-Agent": "anime-movies-script/1.0",
|
||||
},
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=20) as resp:
|
||||
payload = json.loads(resp.read().decode("utf-8"))
|
||||
except Exception:
|
||||
cache[key] = empty_result()
|
||||
return cache[key]
|
||||
|
||||
candidates = flatten_items(payload)
|
||||
if not candidates:
|
||||
cache[key] = empty_result()
|
||||
return cache[key]
|
||||
|
||||
best = None
|
||||
best_score = -1
|
||||
|
||||
for item in candidates:
|
||||
names = extract_names(item)
|
||||
if not names:
|
||||
continue
|
||||
|
||||
score = title_match_score(title, names)
|
||||
if score <= 0:
|
||||
continue
|
||||
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best = item
|
||||
if score == 4:
|
||||
break
|
||||
|
||||
if not best:
|
||||
cache[key] = empty_result()
|
||||
return cache[key]
|
||||
|
||||
names = extract_names(best)
|
||||
cache[key] = {
|
||||
"url": extract_url(best),
|
||||
"title": names[0] if names else "",
|
||||
"english_title": extract_english_title(best),
|
||||
"format": extract_format(best),
|
||||
"genres": extract_list(best, "genres", "genre", "categories"),
|
||||
"studios": extract_list(best, "studios", "studio"),
|
||||
"description": extract_description(best),
|
||||
"names": names,
|
||||
"match_score": best_score,
|
||||
"raw": best,
|
||||
}
|
||||
return cache[key]
|
||||
|
||||
@@ -1,105 +1,105 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from datetime import date
|
||||
|
||||
TMDB_API_BASE = "https://api.themoviedb.org/3"
|
||||
|
||||
|
||||
def parse_release_date(value: str) -> date | None:
|
||||
text = (value or "").strip()
|
||||
if not text:
|
||||
return None
|
||||
try:
|
||||
return date.fromisoformat(text[:10])
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def tmdb_get_json(path: str, params: dict, token: str) -> dict:
|
||||
url = TMDB_API_BASE + path
|
||||
if params:
|
||||
url += "?" + urllib.parse.urlencode(params)
|
||||
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
headers={
|
||||
"Accept": "application/json",
|
||||
"Authorization": f"Bearer {token}",
|
||||
"User-Agent": "anime-movies-script/1.0",
|
||||
},
|
||||
)
|
||||
|
||||
with urllib.request.urlopen(req, timeout=20) as resp:
|
||||
return json.loads(resp.read().decode("utf-8"))
|
||||
|
||||
|
||||
def search_tmdb_movies(title: str, token: str, language: str = "en-US") -> list[dict]:
|
||||
query = (title or "").strip()
|
||||
if not query:
|
||||
return []
|
||||
|
||||
payload = tmdb_get_json(
|
||||
"/search/movie",
|
||||
{
|
||||
"query": query,
|
||||
"include_adult": "false",
|
||||
"language": language,
|
||||
"page": "1",
|
||||
},
|
||||
token,
|
||||
)
|
||||
|
||||
results = []
|
||||
for item in payload.get("results", []):
|
||||
movie_id = item.get("id")
|
||||
if not movie_id:
|
||||
continue
|
||||
|
||||
results.append(
|
||||
{
|
||||
"tmdb_id": int(movie_id),
|
||||
"title": (item.get("title") or "").strip(),
|
||||
"original_title": (item.get("original_title") or "").strip(),
|
||||
"release_date": parse_release_date(str(item.get("release_date") or "")),
|
||||
}
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def fetch_tmdb_release_dates(movie_id: int, token: str) -> dict:
|
||||
return tmdb_get_json(f"/movie/{movie_id}/release_dates", {}, token)
|
||||
|
||||
|
||||
def extract_de_theatrical_dates(release_payload: dict) -> list[date]:
|
||||
german_block = None
|
||||
for result in release_payload.get("results", []):
|
||||
if str(result.get("iso_3166_1") or "").upper() == "DE":
|
||||
german_block = result
|
||||
break
|
||||
|
||||
if not german_block:
|
||||
return []
|
||||
|
||||
dates = []
|
||||
for entry in german_block.get("release_dates", []):
|
||||
release_type = int(entry.get("type") or 0)
|
||||
if release_type not in {2, 3}:
|
||||
continue
|
||||
|
||||
release_date = parse_release_date(str(entry.get("release_date") or ""))
|
||||
if release_date:
|
||||
dates.append(release_date)
|
||||
|
||||
return sorted(dates)
|
||||
|
||||
|
||||
def select_release_in_range(release_dates: list[date], start_date: date, end_date: date) -> date | None:
|
||||
for release_date in sorted(release_dates):
|
||||
if start_date <= release_date <= end_date:
|
||||
return release_date
|
||||
return None
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from datetime import date
|
||||
|
||||
TMDB_API_BASE = "https://api.themoviedb.org/3"
|
||||
|
||||
|
||||
def parse_release_date(value: str) -> date | None:
|
||||
text = (value or "").strip()
|
||||
if not text:
|
||||
return None
|
||||
try:
|
||||
return date.fromisoformat(text[:10])
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def tmdb_get_json(path: str, params: dict, token: str) -> dict:
|
||||
url = TMDB_API_BASE + path
|
||||
if params:
|
||||
url += "?" + urllib.parse.urlencode(params)
|
||||
|
||||
req = urllib.request.Request(
|
||||
url,
|
||||
headers={
|
||||
"Accept": "application/json",
|
||||
"Authorization": f"Bearer {token}",
|
||||
"User-Agent": "anime-movies-script/1.0",
|
||||
},
|
||||
)
|
||||
|
||||
with urllib.request.urlopen(req, timeout=20) as resp:
|
||||
return json.loads(resp.read().decode("utf-8"))
|
||||
|
||||
|
||||
def search_tmdb_movies(title: str, token: str, language: str = "en-US") -> list[dict]:
|
||||
query = (title or "").strip()
|
||||
if not query:
|
||||
return []
|
||||
|
||||
payload = tmdb_get_json(
|
||||
"/search/movie",
|
||||
{
|
||||
"query": query,
|
||||
"include_adult": "false",
|
||||
"language": language,
|
||||
"page": "1",
|
||||
},
|
||||
token,
|
||||
)
|
||||
|
||||
results = []
|
||||
for item in payload.get("results", []):
|
||||
movie_id = item.get("id")
|
||||
if not movie_id:
|
||||
continue
|
||||
|
||||
results.append(
|
||||
{
|
||||
"tmdb_id": int(movie_id),
|
||||
"title": (item.get("title") or "").strip(),
|
||||
"original_title": (item.get("original_title") or "").strip(),
|
||||
"release_date": parse_release_date(str(item.get("release_date") or "")),
|
||||
}
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def fetch_tmdb_release_dates(movie_id: int, token: str) -> dict:
|
||||
return tmdb_get_json(f"/movie/{movie_id}/release_dates", {}, token)
|
||||
|
||||
|
||||
def extract_de_theatrical_dates(release_payload: dict) -> list[date]:
|
||||
german_block = None
|
||||
for result in release_payload.get("results", []):
|
||||
if str(result.get("iso_3166_1") or "").upper() == "DE":
|
||||
german_block = result
|
||||
break
|
||||
|
||||
if not german_block:
|
||||
return []
|
||||
|
||||
dates = []
|
||||
for entry in german_block.get("release_dates", []):
|
||||
release_type = int(entry.get("type") or 0)
|
||||
if release_type not in {2, 3}:
|
||||
continue
|
||||
|
||||
release_date = parse_release_date(str(entry.get("release_date") or ""))
|
||||
if release_date:
|
||||
dates.append(release_date)
|
||||
|
||||
return sorted(dates)
|
||||
|
||||
|
||||
def select_release_in_range(release_dates: list[date], start_date: date, end_date: date) -> date | None:
|
||||
for release_date in sorted(release_dates):
|
||||
if start_date <= release_date <= end_date:
|
||||
return release_date
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user