ad1e6fb8c4
* fix: calendar_yearview * external search bangumi * bgm.tv: fecth volumes of series * bgm.tv: mark ova as tv season * bangumi performance * add bangumi performance test * pagesize
354 lines
12 KiB
Python
354 lines
12 KiB
Python
import logging
|
||
from typing import Any
|
||
from collections import OrderedDict
|
||
from loguru import logger
|
||
import httpx
|
||
|
||
from django.conf import settings
|
||
|
||
from catalog.book.utils import detect_isbn_asin
|
||
from catalog.common import *
|
||
from catalog.game.models import GameReleaseType
|
||
from catalog.models import *
|
||
from common.models.lang import detect_language
|
||
|
||
_logger = logging.getLogger(__name__)
|
||
|
||
|
||
@SiteManager.register
|
||
class Bangumi(AbstractSite):
|
||
SITE_NAME = SiteName.Bangumi
|
||
ID_TYPE = IdType.Bangumi
|
||
URL_PATTERNS = [
|
||
r"https://bgm\.tv/subject/(\d+)",
|
||
r"https://bangumi\.tv/subject/(\d+)",
|
||
r"https://chii\.in/subject/(\d+)",
|
||
]
|
||
WIKI_PROPERTY_ID = ""
|
||
DEFAULT_MODEL = None
|
||
|
||
@classmethod
|
||
def get_category(
|
||
cls, o: dict[str, Any], fetch_resources: bool = False
|
||
) -> tuple[ItemCategory, dict[str, Any]]:
|
||
dt = o.get("date")
|
||
pub_year = None
|
||
pub_month = None
|
||
release_year = None
|
||
release_type = None
|
||
showtime = None
|
||
year = None
|
||
related_resources = []
|
||
match o["type"]:
|
||
case 1:
|
||
model = "Edition"
|
||
category = ItemCategory.Book
|
||
if o["series"] and fetch_resources:
|
||
# model = "Series" TODO
|
||
res = (
|
||
BasicDownloader(
|
||
f"https://api.bgm.tv/v0/subjects/{o['id']}/subjects",
|
||
headers={
|
||
"User-Agent": settings.NEODB_USER_AGENT,
|
||
},
|
||
)
|
||
.download()
|
||
.json()
|
||
)
|
||
|
||
for s in res:
|
||
if s["relation"] != "单行本":
|
||
continue
|
||
related_resources.append(
|
||
{
|
||
"url": cls.id_to_url(s["id"]),
|
||
}
|
||
)
|
||
if dt:
|
||
d = dt.split("-")
|
||
pub_year = d[0]
|
||
pub_month = d[1]
|
||
case 2 | 6:
|
||
is_season = o["platform"] in {
|
||
"TV",
|
||
"OVA", # may be movie in other sites
|
||
"WEB",
|
||
"电视剧",
|
||
"欧美剧",
|
||
"日剧",
|
||
"华语剧",
|
||
"综艺",
|
||
}
|
||
category = ItemCategory.TV if is_season else ItemCategory.Movie
|
||
model = "TVSeason" if is_season else "Movie"
|
||
if "舞台剧" in [
|
||
t["name"] for t in o["tags"]
|
||
]: # 只能这样判断舞台剧了，bangumi三次元分类太少
|
||
category = ItemCategory.Performance
|
||
model = "Performance"
|
||
if dt:
|
||
year = dt.split("-")[0]
|
||
showtime = [
|
||
{"time": dt, "region": "首播日期" if is_season else "发布日期"}
|
||
]
|
||
case 3:
|
||
model = "Album"
|
||
category = ItemCategory.Music
|
||
case 4:
|
||
model = "Game"
|
||
category = ItemCategory.Game
|
||
match o["platform"]:
|
||
case "游戏":
|
||
release_type = GameReleaseType.GAME
|
||
case "扩展包":
|
||
release_type = GameReleaseType.DLC
|
||
case _:
|
||
raise ValueError(
|
||
f"Unknown type {o['type']} for bangumi subject {o["id"]}"
|
||
)
|
||
return category, {
|
||
"preferred_model": model,
|
||
"related_resources": related_resources,
|
||
"pub_year": pub_year,
|
||
"pub_month": pub_month,
|
||
"release_year": release_year,
|
||
"release_type": release_type,
|
||
"showtime": showtime,
|
||
"year": year,
|
||
}
|
||
|
||
@classmethod
|
||
def id_to_url(cls, id_value):
|
||
return f"https://bgm.tv/subject/{id_value}"
|
||
|
||
@classmethod
|
||
async def search_task(
|
||
cls, query: str, page: int, category: str, page_size: int
|
||
) -> list[ExternalSearchResultItem]:
|
||
results = []
|
||
bgm_type = {
|
||
"all": None,
|
||
"movietv": [2, 6],
|
||
"movie": [2, 6],
|
||
"tv": [2, 6],
|
||
"book": [1],
|
||
"game": [4],
|
||
"performance": [6],
|
||
"music": [3],
|
||
}
|
||
if category not in bgm_type:
|
||
return results
|
||
search_url = f"https://api.bgm.tv/v0/search/subjects?limit={page_size}&offset={(page-1)*page_size}"
|
||
async with httpx.AsyncClient() as client:
|
||
try:
|
||
response = await client.post(
|
||
search_url,
|
||
headers={"User-Agent": settings.NEODB_USER_AGENT},
|
||
json={"keyword": query, "filter": {"type": bgm_type[category]}},
|
||
timeout=2,
|
||
)
|
||
r = response.json()
|
||
for s in r["data"]:
|
||
cat, _ = cls.get_category(s)
|
||
results.append(
|
||
ExternalSearchResultItem(
|
||
category=cat,
|
||
source_site=cls.SITE_NAME,
|
||
source_url=cls.id_to_url(s["id"]),
|
||
title=s["name"],
|
||
subtitle="",
|
||
brief=s.get("summary", ""),
|
||
cover_url=s["images"].get("common"),
|
||
)
|
||
)
|
||
except Exception as e:
|
||
logger.error(
|
||
"Bangumi search error", extra={"query": query, "exception": e}
|
||
)
|
||
return results
|
||
|
||
def scrape(self):
|
||
api_url = f"https://api.bgm.tv/v0/subjects/{self.id_value}"
|
||
o = (
|
||
BasicDownloader(
|
||
api_url,
|
||
headers={
|
||
"User-Agent": settings.NEODB_USER_AGENT,
|
||
},
|
||
)
|
||
.download()
|
||
.json()
|
||
)
|
||
category, data = self.get_category(o, True)
|
||
title = o.get("name_cn") or o.get("name")
|
||
orig_title = o.get("name") if o.get("name") != title else None
|
||
brief = o.get("summary")
|
||
episodes = o.get("total_episodes", 0)
|
||
genre = None
|
||
platform = None
|
||
other_title = []
|
||
imdb_code = None
|
||
isbn_type = None
|
||
isbn = None
|
||
language = None
|
||
pub_house = None
|
||
orig_creator = None
|
||
authors = []
|
||
site = None
|
||
director = None
|
||
playwright = None
|
||
actor = None
|
||
pages = None
|
||
price = None
|
||
opening_date = None
|
||
closing_date = None
|
||
location = None
|
||
for i in o.get("infobox", []):
|
||
k = i["key"].lower()
|
||
v = i["value"]
|
||
match k:
|
||
case "别名":
|
||
other_title = (
|
||
[d["v"] for d in v]
|
||
if isinstance(v, list)
|
||
else ([v] if isinstance(v, str) else [])
|
||
)
|
||
case "话数":
|
||
try:
|
||
episodes = int(v)
|
||
except ValueError:
|
||
pass
|
||
case "imdb_id":
|
||
imdb_code = v
|
||
case "isbn":
|
||
isbn_type, isbn = detect_isbn_asin(v)
|
||
case "语言":
|
||
language = v
|
||
case "出版社":
|
||
pub_house = v
|
||
case "导演":
|
||
director = v
|
||
case "编剧" | "脚本":
|
||
playwright = (
|
||
[d["v"] for d in v]
|
||
if isinstance(v, list)
|
||
else ([v] if isinstance(v, str) else [])
|
||
)
|
||
case "原作":
|
||
match category:
|
||
case ItemCategory.Book:
|
||
authors.append(v)
|
||
case ItemCategory.Performance:
|
||
orig_creator = (
|
||
[d["v"] for d in v]
|
||
if isinstance(v, list)
|
||
else ([v] if isinstance(v, str) else [])
|
||
)
|
||
case "作画":
|
||
authors.append(v)
|
||
case "作者":
|
||
authors.extend(
|
||
[d["v"] for d in v]
|
||
if isinstance(v, list)
|
||
else ([v] if isinstance(v, str) else [])
|
||
)
|
||
case "平台":
|
||
platform = (
|
||
[d["v"] for d in v]
|
||
if isinstance(v, list)
|
||
else ([v] if isinstance(v, str) else [])
|
||
)
|
||
case "游戏类型" | "类型":
|
||
genre = (
|
||
[d["v"] for d in v]
|
||
if isinstance(v, list)
|
||
else ([v] if isinstance(v, str) else [])
|
||
)
|
||
case "官方网站" | "website":
|
||
site = v[0] if isinstance(v, list) else v
|
||
case "页数":
|
||
pages = v
|
||
case "价格":
|
||
price = v
|
||
case "开始":
|
||
opening_date = v
|
||
case "结束":
|
||
closing_date = v
|
||
case "演出":
|
||
if category == ItemCategory.Performance:
|
||
director = v
|
||
case "主演":
|
||
actor = (
|
||
[{"name": d["v"], "role": None} for d in v]
|
||
if isinstance(v, list)
|
||
else (
|
||
[{"name": w, "role": None} for w in v.split("、")]
|
||
if isinstance(v, str)
|
||
else []
|
||
)
|
||
)
|
||
case "会场" | "演出地点":
|
||
location = v
|
||
|
||
img_url = o["images"].get("large") or o["images"].get("common")
|
||
raw_img = None
|
||
ext = None
|
||
if img_url:
|
||
raw_img, ext = BasicImageDownloader.download_image(
|
||
img_url, None, headers={}
|
||
)
|
||
titles = OrderedDict.fromkeys(
|
||
[title] + (other_title or []) + ([orig_title] if orig_title else [])
|
||
)
|
||
if o.get("name_cn"):
|
||
titles[o.get("name_cn")] = "zh-cn"
|
||
localized_title = [
|
||
{"lang": lang or detect_language(t), "text": t}
|
||
for t, lang in titles.items()
|
||
]
|
||
localized_desc = (
|
||
[{"lang": detect_language(brief), "text": brief}] if brief else []
|
||
)
|
||
data.update(
|
||
{
|
||
"localized_title": localized_title,
|
||
"localized_description": localized_desc,
|
||
"title": title,
|
||
"orig_title": orig_title,
|
||
"other_title": other_title or None,
|
||
"orig_creator": orig_creator,
|
||
"author": authors,
|
||
"genre": genre,
|
||
"translator": None,
|
||
"director": director,
|
||
"playwright": playwright,
|
||
"actor": actor,
|
||
"language": language,
|
||
"platform": platform,
|
||
"imdb_code": imdb_code,
|
||
"pub_house": pub_house,
|
||
"binding": None,
|
||
"episode_count": episodes or None,
|
||
"official_site": site,
|
||
"site": site,
|
||
"isbn": isbn,
|
||
"brief": brief,
|
||
"cover_image_url": img_url,
|
||
"pages": pages,
|
||
"price": price,
|
||
"opening_date": opening_date,
|
||
"closing_date": closing_date,
|
||
"location": location,
|
||
}
|
||
)
|
||
lookup_ids = {}
|
||
if isbn:
|
||
lookup_ids[isbn_type] = isbn
|
||
if imdb_code:
|
||
lookup_ids[IdType.IMDB] = imdb_code
|
||
return ResourceContent(
|
||
metadata={k: v for k, v in data.items() if v is not None},
|
||
cover_image=raw_img,
|
||
cover_image_extention=ext,
|
||
lookup_ids=lookup_ids,
|
||
)
|