fix(types): type indexer spider response boundaries

This commit is contained in:
jxxghp
2026-10-17 18:01:20 +08:00
parent 653180c8f0
commit 967780db88

View File

@@ -1,7 +1,7 @@
import datetime
import re
import traceback
from typing import Any, List, Optional
from typing import Any, List, Optional, cast
from urllib.parse import parse_qs, quote, urlparse
from jinja2 import Template
@@ -24,6 +24,9 @@ _FAILURE_TITLE_SCAN_CHARS = 4096
_HTML_TITLE_PATTERN = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
_HTML_TAG_PATTERN = re.compile(r"<[^>]+>")
JsonDict = dict[str, Any]
JsonList = list[JsonDict]
def _describe_unusable_response(response: Any) -> str:
"""
@@ -48,7 +51,7 @@ def _describe_unusable_response(response: Any) -> str:
return f"HTTP {status_code},{label}:{summary}"
def select_media_categories(category: Optional[dict], mtype: Optional[MediaType]) -> list[dict]:
def select_media_categories(category: Optional[JsonDict], mtype: Optional[MediaType]) -> JsonList:
"""根据媒体类型选择站点索引配置中的分类列表。"""
if not category:
return []
@@ -65,7 +68,7 @@ def select_media_categories(category: Optional[dict], mtype: Optional[MediaType]
)
def resolve_category_media_type(category_value: Any, category: Optional[dict]) -> MediaType:
def resolve_category_media_type(category_value: Any, category: Optional[JsonDict]) -> MediaType:
"""将站点分类 ID 映射为统一的电影、电视剧或音乐类型。"""
if category_value is None or not category:
return MediaType.UNKNOWN
@@ -92,26 +95,34 @@ class SiteSpider:
"""
_default_result_num = 100
search: Optional[JsonDict]
batch: Optional[JsonDict]
browse: Optional[JsonDict]
category: Optional[JsonDict]
list: JsonDict
fields: JsonDict
domain: Optional[str]
requested_result_media_type: Optional[MediaType]
@property
def __class__(self):
@property # type: ignore[misc]
def __class__(self) -> type[object]: # type: ignore[override]
"""隐藏 Spider 的真实类型,避免模板或脚本读取内部实现。"""
return object
@property
def __dict__(self):
def __dict__(self) -> JsonDict: # type: ignore[override]
"""隐藏 Spider 实例属性,避免外部枚举请求上下文。"""
return {}
@property
def __dir__(self):
def __dir__(self) -> Any:
"""拒绝外部列举 Spider 的受保护属性。"""
raise AttributeError("Cannot read protected attribute!")
def __init__(self,
indexer: dict,
indexer: JsonDict,
keyword: Optional[str] = None,
mtype: MediaType = None,
mtype: Optional[MediaType] = None,
cat: Optional[str] = None,
page: Optional[int] = 0,
referer: Optional[str] = None,
@@ -133,7 +144,8 @@ class SiteSpider:
self.search_type = search_type or "torrents"
self.indexerid = indexer.get('id')
self.indexername = indexer.get('name')
self.site_media_type = MediaType.from_agent(indexer.get('media_type'))
media_type = indexer.get('media_type')
self.site_media_type = MediaType.from_agent(str(media_type)) if media_type else None
if self.search_type == "subtitles":
subtitle_conf = indexer.get('subtitles') or {}
self.search = subtitle_conf.get('search')
@@ -159,7 +171,8 @@ class SiteSpider:
)
self.requested_result_media_type = None
self._field_templates = self.__build_field_templates()
self.domain = indexer.get('domain')
domain = indexer.get('domain')
self.domain = str(domain) if domain else None
self.result_num = int(result_num or self.default_result_num())
self._timeout = int(indexer.get('timeout') or 15)
self.page = page
@@ -175,10 +188,10 @@ class SiteSpider:
self.is_error = False
# is_error 为真时的失败原因,供搜索失败提示展示
self.error_detail: Optional[str] = None
self.torrents_info = {}
self.torrents_info_array = []
self.torrents_info: JsonDict = {}
self.torrents_info_array: JsonList = []
def __build_field_templates(self) -> dict:
def __build_field_templates(self) -> dict[str, Template]:
"""
预编译字段模板,避免按每条种子重复构造 Jinja Template。
"""
@@ -198,13 +211,16 @@ class SiteSpider:
"""
return cls._default_result_num
def __get_search_url(self):
def __get_search_url(self) -> str:
"""
获取搜索URL
"""
self._result_offset = 0
search = self.search or {}
domain = self.domain or ""
category = self.category or {}
# 种子搜索相对路径
paths = self.search.get('paths', [])
paths = search.get('paths', [])
torrentspath = ""
# 是否选中了媒体类型专用路径,浏览模式下专用路径优先于 browse 配置
typed_path_selected = False
@@ -218,7 +234,7 @@ class SiteSpider:
MediaType.MOVIE: "movie",
MediaType.TV: "tv",
MediaType.MUSIC: "music",
}.get(self.mtype)
}.get(self.mtype or MediaType.UNKNOWN)
for path in paths:
path_type = path.get("type")
if path_type == "all" and not fallback_path:
@@ -254,7 +270,7 @@ class SiteSpider:
search_word = self.__format_search_word(search_word)
# 搜索URL
indexer_params = self.search.get("params", {}).copy()
indexer_params = search.get("params", {}).copy()
if indexer_params:
search_area = indexer_params.get('search_area')
# search_area非0表示支持imdbid搜索
@@ -278,21 +294,22 @@ class SiteSpider:
"%s" % key: str(value).format(**inputs_dict)
})
# 分类条件
if self.category:
cats = select_media_categories(self.category, self.mtype)
if category:
cats = select_media_categories(category, self.mtype)
allowed_cats = set(self.cat.split(',')) if self.cat else None
for cat in cats:
if allowed_cats and str(cat.get('id')) not in allowed_cats:
continue
if self.category.get("field"):
if category.get("field"):
category_filter_selected = True
value = params.get(self.category.get("field"), "")
value = params.get(str(category.get("field")), "")
params.update({
"%s" % self.category.get("field"): value + self.category.get("delimiter",
' ') + cat.get("id")
"%s" % category.get("field"): str(value)
+ str(category.get("delimiter", ' '))
+ str(cat.get("id", ""))
})
else:
category_param = cat.get("param") or self.category.get("param")
category_param = cat.get("param") or category.get("param")
if category_param:
category_filter_selected = True
# 某些站点(例如憨憨)使用重复的 cat[] 参数,字典值列表可由
@@ -312,16 +329,16 @@ class SiteSpider:
})
# 分类项可以附带站点要求的额外开关,例如音乐专用展示模式。
if isinstance(cat, dict) and cat.get("params"):
params.update(cat.get("params"))
searchurl = UrlUtils.combine_url(self.domain, torrentspath, params)
params.update(cast(JsonDict, cat.get("params")))
searchurl = UrlUtils.combine_url(domain, torrentspath, params)
else:
# 变量字典
inputs_dict = {
"keyword": quote(search_word),
"page": self.page or 0
"page": str(self.page or 0)
}
# 无额外参数
searchurl = self.domain + str(torrentspath).format(**inputs_dict)
searchurl = domain + str(torrentspath).format(**inputs_dict)
# 没有页码入口的搜索一次返回完整列表,需在解析前按原始行分页。
self._result_offset = 0 if "{page}" in str(torrentspath) else int(self.page or 0) * self.result_num
@@ -329,23 +346,23 @@ class SiteSpider:
else:
# 变量字典
inputs_dict = {
"page": self.page or 0,
"page": str(self.page or 0),
"keyword": ""
}
# 有单独浏览路径;指定了媒体类型专用路径时不覆盖,确保音乐等专用入口可达
if self.browse and not typed_path_selected:
torrentspath = self.browse.get("path")
torrentspath = str(self.browse.get("path") or "")
if self.browse.get("start"):
start_page = int(self.browse.get("start")) + int(self.page or 0)
start_page = int(self.browse.get("start") or 0) + int(self.page or 0)
inputs_dict.update({
"page": start_page
"page": str(start_page)
})
elif self.page and "{page}" not in str(torrentspath):
# 按路径是否已带查询参数选择连接符,避免拼出两个问号的非法地址
separator = "&" if "?" in str(torrentspath) else "?"
torrentspath = torrentspath + f"{separator}page={self.page}"
# 搜索Url
searchurl = self.domain + str(torrentspath).format(**inputs_dict)
searchurl = domain + str(torrentspath).format(**inputs_dict)
if self.result_media_type_from_request \
and self.mtype \
@@ -353,7 +370,7 @@ class SiteSpider:
self.requested_result_media_type = self.mtype
else:
self.requested_result_media_type = None
return searchurl
return str(searchurl)
def __format_search_word(self, search_word: str) -> str:
"""
@@ -362,7 +379,7 @@ class SiteSpider:
if not search_word or not isinstance(search_word, str):
return search_word
if re.fullmatch(r"tt\d+", search_word):
imdbid_format = self.search.get("imdbid_format")
imdbid_format = (self.search or {}).get("imdbid_format")
if imdbid_format:
return str(imdbid_format).format(
keyword=search_word,
@@ -386,7 +403,7 @@ class SiteSpider:
confidence_threshold=get_runtime_setting('ENCODING_DETECTION_MIN_CONFIDENCE')
)
def get_torrents(self) -> List[dict]:
def get_torrents(self) -> JsonList:
"""
开始请求
"""
@@ -412,7 +429,7 @@ class SiteSpider:
# 解析返回
return self.parse(html_text)
async def async_get_torrents(self) -> List[dict]:
async def async_get_torrents(self) -> JsonList:
"""
异步请求
"""
@@ -436,9 +453,9 @@ class SiteSpider:
if not html_text:
self.error_detail = _describe_unusable_response(ret)
# 解析返回
return await run_in_threadpool(self.parse, html_text)
return cast(JsonList, await run_in_threadpool(self.parse, html_text))
def __get_title(self, torrent: Any):
def __get_title(self, torrent: Any) -> None:
"""按站点字段配置提取并清洗种子标题。"""
if 'title' not in self.fields:
return
@@ -460,7 +477,7 @@ class SiteSpider:
self.torrents_info['title'] = self.__filter_text(self.torrents_info.get('title'),
selector.get('filters'))
def __get_description(self, torrent: Any):
def __get_description(self, torrent: Any) -> None:
"""按选择器或模板生成种子副标题。"""
if 'description' not in self.fields:
return
@@ -494,7 +511,7 @@ class SiteSpider:
self.torrents_info['description'] = self.__filter_text(self.torrents_info.get('description'),
selector.get('filters'))
def __get_detail(self, torrent: Any):
def __get_detail(self, torrent: Any) -> None:
"""提取并规范化种子详情页地址。"""
if 'details' not in self.fields:
return
@@ -504,15 +521,15 @@ class SiteSpider:
if detail_link:
if not detail_link.startswith("http"):
if detail_link.startswith("//"):
self.torrents_info['page_url'] = self.domain.split(":")[0] + ":" + detail_link
self.torrents_info['page_url'] = (self.domain or "").split(":")[0] + ":" + detail_link
elif detail_link.startswith("/"):
self.torrents_info['page_url'] = self.domain + detail_link[1:]
self.torrents_info['page_url'] = (self.domain or "") + detail_link[1:]
else:
self.torrents_info['page_url'] = self.domain + detail_link
self.torrents_info['page_url'] = (self.domain or "") + detail_link
else:
self.torrents_info['page_url'] = detail_link
def __get_download(self, torrent: Any):
def __get_download(self, torrent: Any) -> None:
"""提取并规范化种子下载地址。"""
if 'download' not in self.fields:
return
@@ -522,7 +539,7 @@ class SiteSpider:
if download_link:
if not download_link.startswith("http") \
and not download_link.startswith("magnet"):
_scheme, _domain = url_tools.split_netloc(self.domain)
_scheme, _domain = url_tools.split_netloc(self.domain or "")
if _domain in download_link:
if download_link.startswith("/"):
self.torrents_info['enclosure'] = f"{_scheme}:{download_link}"
@@ -536,7 +553,7 @@ class SiteSpider:
else:
self.torrents_info['enclosure'] = download_link
def __get_report_url(self, torrent: Any):
def __get_report_url(self, torrent: Any) -> None:
"""
获取字幕举报页面链接。
"""
@@ -548,7 +565,7 @@ class SiteSpider:
if report_link:
self.torrents_info['report_url'] = self.__normalize_link(report_link)
def __get_language_icon(self, torrent: Any):
def __get_language_icon(self, torrent: Any) -> None:
"""
获取字幕语言图标链接。
"""
@@ -560,7 +577,7 @@ class SiteSpider:
if icon_link:
self.torrents_info['language_icon'] = self.__normalize_link(icon_link)
def __get_imdbid(self, torrent: Any):
def __get_imdbid(self, torrent: Any) -> None:
"""提取种子关联的 IMDb 标识。"""
if "imdbid" not in self.fields:
return
@@ -568,7 +585,7 @@ class SiteSpider:
item = self._safe_query(torrent, selector)
self.torrents_info['imdbid'] = self.__filter_text(item, selector.get('filters'))
def __get_size(self, torrent: Any):
def __get_size(self, torrent: Any) -> None:
"""提取种子大小并转换为字节数。"""
if 'size' not in self.fields:
return
@@ -577,12 +594,12 @@ class SiteSpider:
if item is not None and item != "":
size_val = item.replace("\n", "").strip()
size_val = self.__filter_text(size_val,
selector.get('filters'))
selector.get('filters')) or ""
self.torrents_info['size'] = size_tools.parse_size(size_val)
else:
self.torrents_info['size'] = 0
def __get_leechers(self, torrent: Any):
def __get_leechers(self, torrent: Any) -> None:
"""提取种子的下载人数。"""
if 'leechers' not in self.fields:
return
@@ -591,12 +608,12 @@ class SiteSpider:
if item:
peers_val = item.split("/")[0]
peers_val = peers_val.replace(",", "")
peers_val = self.__filter_text(peers_val, selector.get('filters'))
peers_val = self.__filter_text(peers_val, selector.get('filters')) or ""
self.torrents_info['peers'] = int(peers_val) if peers_val and peers_val.isdigit() else 0
else:
self.torrents_info['peers'] = 0
def __get_seeders(self, torrent: Any):
def __get_seeders(self, torrent: Any) -> None:
"""提取种子的做种人数。"""
if 'seeders' not in self.fields:
return
@@ -605,12 +622,12 @@ class SiteSpider:
if item:
seeders_val = item.split("/")[0]
seeders_val = seeders_val.replace(",", "")
seeders_val = self.__filter_text(seeders_val, selector.get('filters'))
seeders_val = self.__filter_text(seeders_val, selector.get('filters')) or ""
self.torrents_info['seeders'] = int(seeders_val) if seeders_val and seeders_val.isdigit() else 0
else:
self.torrents_info['seeders'] = 0
def __get_grabs(self, torrent: Any):
def __get_grabs(self, torrent: Any) -> None:
"""提取种子的完成人数。"""
if 'grabs' not in self.fields:
return
@@ -619,12 +636,12 @@ class SiteSpider:
if item is not None and item != "":
grabs_val = item.split("/")[0]
grabs_val = grabs_val.replace(",", "")
grabs_val = self.__filter_text(grabs_val, selector.get('filters'))
grabs_val = self.__filter_text(grabs_val, selector.get('filters')) or ""
self.torrents_info['grabs'] = int(grabs_val) if grabs_val and grabs_val.isdigit() else 0
else:
self.torrents_info['grabs'] = 0
def __get_pubdate(self, torrent: Any):
def __get_pubdate(self, torrent: Any) -> None:
"""按站点日期配置提取并规范化发布时间。"""
if 'date_added' not in self.fields and 'date' not in self.fields:
return
@@ -647,7 +664,7 @@ class SiteSpider:
if self.__is_invalid_pubdate_text(self.torrents_info.get('pubdate')):
self.torrents_info.pop('pubdate', None)
def __get_date(self, torrent: Any, selector: dict) -> Optional[str]:
def __get_date(self, torrent: Any, selector: JsonDict) -> Optional[str]:
"""
从 date 模板解析发布时间。
"""
@@ -673,9 +690,9 @@ class SiteSpider:
pubdate_str = template.render(fields=render_dict)
if pubdate_str == "now" or self.__is_relative_pubdate_text(pubdate_str):
return None
return pubdate_str
return str(pubdate_str)
def __get_date_from_cell(self, torrent: Any, selector: dict) -> Optional[str]:
def __get_date_from_cell(self, torrent: Any, selector: JsonDict) -> Optional[str]:
"""
兼容 NexusPHP 发生时间模式下不再渲染 span 的时间单元格。
"""
@@ -728,7 +745,7 @@ class SiteSpider:
except (ValueError, TypeError):
return True
def __get_date_elapsed(self, torrent: Any):
def __get_date_elapsed(self, torrent: Any) -> None:
"""提取相对发布时间对应的已过秒数。"""
if 'date_elapsed' not in self.fields:
return
@@ -736,7 +753,7 @@ class SiteSpider:
date_elapsed = self._safe_query(torrent, selector)
self.torrents_info['date_elapsed'] = self.__filter_text(date_elapsed, selector.get('filters'))
def __get_downloadvolumefactor(self, torrent: Any):
def __get_downloadvolumefactor(self, torrent: Any) -> None:
"""按促销标签解析种子下载系数。"""
selector = self.fields.get('downloadvolumefactor', {})
if not selector:
@@ -760,7 +777,7 @@ class SiteSpider:
if downloadvolumefactor:
self.torrents_info['downloadvolumefactor'] = int(downloadvolumefactor.group(1))
def __get_uploadvolumefactor(self, torrent: Any):
def __get_uploadvolumefactor(self, torrent: Any) -> None:
"""按促销标签解析种子上传系数。"""
selector = self.fields.get('uploadvolumefactor', {})
if not selector:
@@ -784,7 +801,7 @@ class SiteSpider:
if uploadvolumefactor:
self.torrents_info['uploadvolumefactor'] = int(uploadvolumefactor.group(1))
def __get_labels(self, torrent: Any):
def __get_labels(self, torrent: Any) -> None:
"""提取并清洗种子标签列表。"""
if 'labels' not in self.fields:
return
@@ -806,7 +823,7 @@ class SiteSpider:
labels.clear()
del labels
def __get_free_date(self, torrent: Any):
def __get_free_date(self, torrent: Any) -> None:
"""提取种子免费促销截止时间。"""
if 'freedate' not in self.fields:
return
@@ -814,7 +831,7 @@ class SiteSpider:
freedate = self._safe_query(torrent, selector)
self.torrents_info['freedate'] = self.__filter_text(freedate, selector.get('filters'))
def __get_hit_and_run(self, torrent: Any):
def __get_hit_and_run(self, torrent: Any) -> None:
"""判定种子是否带有 Hit and Run 要求。"""
if 'hr' not in self.fields:
return
@@ -829,7 +846,7 @@ class SiteSpider:
hit_and_run.clear()
del hit_and_run
def __get_category(self, torrent: Any):
def __get_category(self, torrent: Any) -> None:
"""将站点分类字段投影为统一媒体类型。"""
if self.requested_result_media_type:
self.torrents_info['category'] = self.requested_result_media_type.value
@@ -846,7 +863,7 @@ class SiteSpider:
resolved_type = self.site_media_type
self.torrents_info['category'] = resolved_type.value
def __apply_result_media_type(self, torrents: Optional[List[dict]]) -> List[dict]:
def __apply_result_media_type(self, torrents: Optional[JsonList]) -> JsonList:
"""
为解析结果补充统一媒体类型。
@@ -870,7 +887,7 @@ class SiteSpider:
torrent["category"] = self.site_media_type.value
return results
def __get_subtitle_field(self, torrent: Any, field_name: str):
def __get_subtitle_field(self, torrent: Any, field_name: str) -> None:
"""
按配置读取字幕字段。
"""
@@ -882,7 +899,7 @@ class SiteSpider:
if value is not None:
self.torrents_info[field_name] = value
def __fill_subtitle_ids(self):
def __fill_subtitle_ids(self) -> None:
"""
从字幕下载链接中补充站点种子ID和字幕ID。
"""
@@ -910,13 +927,13 @@ class SiteSpider:
return link
if not link.startswith("http"):
if link.startswith("//"):
return self.domain.split(":")[0] + ":" + link
return (self.domain or "").split(":")[0] + ":" + link
if link.startswith("/"):
return self.domain + link[1:]
return self.domain + link
return (self.domain or "") + link[1:]
return (self.domain or "") + link
return link
def _safe_query(self, torrent: Any, selector_config: Optional[dict]) -> Optional[str]:
def _safe_query(self, torrent: Any, selector_config: Optional[JsonDict]) -> Optional[str]:
"""
安全地执行PyQuery查询并自动清理资源
:param torrent: PyQuery对象
@@ -939,7 +956,7 @@ class SiteSpider:
query_obj.clear()
del query_obj
def get_info(self, torrent: Any) -> dict:
def get_info(self, torrent: Any) -> JsonDict:
"""
解析单条种子数据
"""
@@ -988,7 +1005,7 @@ class SiteSpider:
finally:
self.torrents_info.clear()
def get_subtitle_info(self, subtitle: Any) -> dict:
def get_subtitle_info(self, subtitle: Any) -> JsonDict:
"""
解析单条字幕数据。
"""
@@ -1019,46 +1036,47 @@ class SiteSpider:
self.torrents_info.clear()
@staticmethod
def __filter_text(text: Optional[str], filters: Optional[List[dict]]) -> str:
def __filter_text(text: Optional[str], filters: Optional[List[JsonDict]]) -> Optional[str]:
"""
对文件进行处理
"""
if not text or not filters or not isinstance(filters, list):
return text
filtered_text: Any = text
if not isinstance(text, str):
text = str(text)
filtered_text = str(text)
for filter_item in filters:
if not text:
if not filtered_text:
break
method_name = filter_item.get("name")
try:
args = filter_item.get("args")
if method_name == "re_search" and isinstance(args, list):
rematch = re.search(r"%s" % args[0], text)
rematch = re.search(r"%s" % args[0], filtered_text)
if rematch:
text = rematch.group(args[-1])
filtered_text = rematch.group(args[-1])
elif method_name == "split" and isinstance(args, list):
text = text.split(r"%s" % args[0])[args[-1]]
filtered_text = filtered_text.split(r"%s" % args[0])[args[-1]]
elif method_name == "replace" and isinstance(args, list):
text = text.replace(r"%s" % args[0], r"%s" % args[-1])
filtered_text = filtered_text.replace(r"%s" % args[0], r"%s" % args[-1])
elif method_name == "dateparse" and isinstance(args, str):
text = text.replace("\n", " ").strip()
text = datetime.datetime.strptime(text, r"%s" % args)
filtered_text = filtered_text.replace("\n", " ").strip()
filtered_text = datetime.datetime.strptime(filtered_text, r"%s" % args)
elif method_name == "strip":
text = text.strip()
filtered_text = filtered_text.strip()
elif method_name == "appendleft":
text = f"{args}{text}"
filtered_text = f"{args}{filtered_text}"
elif method_name == "querystring":
parsed_url = urlparse(str(text))
query_params = parse_qs(parsed_url.query)
param_value = query_params.get(args)
text = param_value[0] if param_value else ''
param_value = query_params.get(str(args))
filtered_text = param_value[0] if param_value else ''
except Exception as err:
logger.debug(f'过滤器 {method_name} 处理失败:{str(err)} - {traceback.format_exc()}')
return text.strip() if isinstance(text, str) else text
return filtered_text.strip() if isinstance(filtered_text, str) else str(filtered_text)
@staticmethod
def __remove(item: Any, selector: Optional[dict]):
def __remove(item: Any, selector: Optional[JsonDict]) -> None:
"""
移除元素
"""
@@ -1068,12 +1086,12 @@ class SiteSpider:
item.remove(v)
@staticmethod
def __attribute_or_text(item: Any, selector: Optional[dict]) -> list:
def __attribute_or_text(item: Any, selector: Optional[JsonDict]) -> List[Any]:
"""
获取查询结果的属性或文本列表。
"""
if not selector:
return item
return cast(List[Any], item)
if not item:
return []
if 'attribute' in selector:
@@ -1083,7 +1101,7 @@ class SiteSpider:
return items
@staticmethod
def __index(items: Optional[list], selector: Optional[dict]) -> Optional[str]:
def __index(items: Optional[List[str]], selector: Optional[JsonDict]) -> Optional[str]:
"""
按配置下标读取查询结果。
"""
@@ -1091,11 +1109,11 @@ class SiteSpider:
return None
if selector:
if "contents" in selector \
and len(items) > int(selector.get("contents")):
item = items[0].split("\n")[selector.get("contents")]
and len(items) > int(selector.get("contents") or 0):
item = items[0].split("\n")[int(selector.get("contents") or 0)]
elif "index" in selector \
and len(items) > int(selector.get("index")):
item = items[int(selector.get("index"))]
and len(items) > int(selector.get("index") or 0):
item = items[int(selector.get("index") or 0)]
else:
item = items[0]
else:
@@ -1130,7 +1148,7 @@ class SiteSpider:
PyQuery(rows[self._result_offset + self.result_num:]).remove()
return str(html_doc)
def parse(self, html_text: str) -> List[dict]:
def parse(self, html_text: str) -> JsonList:
"""
解析页面或 RSS 索引,统一使用 HTML 选择器语义。
"""
@@ -1161,7 +1179,7 @@ class SiteSpider:
if self.search_type == "subtitles":
rust_subtitles = rust_accel.parse_indexer_subtitles(
html_text=html_text,
domain=self.domain,
domain=self.domain or "",
list_config=self.list,
fields=self.fields,
result_num=self.result_num
@@ -1171,7 +1189,7 @@ class SiteSpider:
else:
rust_torrents = rust_accel.parse_indexer_torrents(
html_text=html_text,
domain=self.domain,
domain=self.domain or "",
list_config=self.list,
fields=self.fields,
category=self.category,