fix(torrent): refresh TTG path-signed download links

- 将 TTG 的 /dl/{id}/{auth} 下载链接识别为短时签名链接,403 后复用既有的
  同资源刷新重试;仅匹配 totheglory.im 及其子域名,其他站点相同形态的路径不受影响。
- 从 TTG 详情页 /t/{id}/ 与下载路径 /dl/{id}/{auth} 提取稳定种子 ID,
  末段 auth 不参与资源身份,刷新时可匹配回同一种子。
- RSS 条目使用 rssdd.php?par= 链接,不受本次路径规则影响。

Refs #7191

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
LIN student
2026-10-26 19:13:52 +08:00
parent a11bc444f4
commit de9fefe90f
2 changed files with 112 additions and 8 deletions

View File

@@ -7,11 +7,16 @@ import binascii
import hashlib
import re
from typing import TYPE_CHECKING, Optional, Union
from urllib.parse import parse_qs, urlsplit
from urllib.parse import SplitResult, parse_qs, urlsplit
if TYPE_CHECKING:
from app.domain.context import TorrentInfo
# TTG 详情页为 /t/{id}/,下载链接为 /dl/{id}/{auth};auth 由每次搜索生成,几分钟后或多次请求后失效
_TTG_HOST = "totheglory.im"
_TTG_DETAIL_PATH = re.compile(r"/t/(\d+)/?")
_TTG_DOWNLOAD_PATH = re.compile(r"/dl/(\d+)/[^/]+/?")
def is_magnet_link(content: Union[str, bytes]) -> bool:
"""
@@ -141,24 +146,52 @@ def resource_identity(torrent: "TorrentInfo") -> Optional[tuple[str, str]]:
if values and len(values) == 1:
prefix = "hash" if name == "hash" else "id"
return site or parsed.netloc, f"{prefix}={values[0]}"
# 馒头详情页把稳定种子 ID 放在路径中,换票请求参数不参与资源身份。
match = re.fullmatch(r"/detail/(\d+)/?", parsed.path)
if match:
return site or parsed.netloc, f"id={match[1]}"
resource_id = _path_resource_id(parsed)
if resource_id:
return site or parsed.netloc, f"id={resource_id}"
return None
def _is_ttg_url(parsed: SplitResult) -> bool:
"""
判断地址是否属于 TTG 站点(含子域名)。
"""
host = parsed.hostname or ""
return host == _TTG_HOST or host.endswith(f".{_TTG_HOST}")
def _path_resource_id(parsed: SplitResult) -> Optional[str]:
"""
从已知站点的详情页或下载链接路径中提取稳定种子 ID。
馒头详情页为 /detail/{id},换票请求参数不参与资源身份;
TTG 详情页为 /t/{id}/,下载链接为 /dl/{id}/{auth},末段 auth 是临时签名,不参与资源身份。
:param parsed: 已拆分的 http(s) 地址
:return: 站内种子 ID;不是已知路径形态时返回 None
"""
match = re.fullmatch(r"/detail/(\d+)/?", parsed.path)
if match is None and _is_ttg_url(parsed):
match = _TTG_DETAIL_PATH.fullmatch(parsed.path) or _TTG_DOWNLOAD_PATH.fullmatch(parsed.path)
return match[1] if match else None
def is_expiring_download_url(url: Optional[str]) -> bool:
"""
识别已知的时间戳加签名下载链接;不推断签名的具体有效期。
识别已知的短时签名下载链接;不推断签名的具体有效期。
支持查询参数中的时间戳加签名(t、sign),以及 TTG 放在下载路径末段的 auth 签名。
"""
if not url or not url.startswith(("http://", "https://")):
return False
try:
params = parse_qs(urlsplit(url).query)
parsed = urlsplit(url)
params = parse_qs(parsed.query)
except ValueError:
return False
return bool(params.get("t") and params.get("sign"))
if params.get("t") and params.get("sign"):
return True
return _is_ttg_url(parsed) and _TTG_DOWNLOAD_PATH.fullmatch(parsed.path) is not None
def rate_limit_cooldown(message: Optional[str]) -> Optional[int]:

View File

@@ -499,3 +499,74 @@ def test_only_known_timestamp_signature_pattern_is_expiring(url, expected):
签名模式判断不把空参数、普通链接或换票协议误认为时效链接。
"""
assert torrent_rules.is_expiring_download_url(url) is expected
TTG_PAGE_URL = "https://totheglory.im/t/395931/"
TTG_OLD_URL = "https://totheglory.im/dl/395931/3576"
TTG_NEW_URL = "https://totheglory.im/dl/395931/8067"
# TTG RSS 条目的下载链接是 rssdd.php?par=,par 为 base64 编码的 "vvv{id}...",不使用 /dl/ 路径签名
TTG_RSS_URL = "https://totheglory.im/rssdd.php?par=dnZ2Mzk1OTMx&ssl=yes"
@pytest.mark.parametrize("page_url,enclosure,expected", [
(TTG_PAGE_URL, TTG_OLD_URL, ("1", "id=395931")),
(None, TTG_OLD_URL, ("1", "id=395931")),
("https://totheglory.im/t/395931", None, ("1", "id=395931")),
("https://www.totheglory.im/t/395931/", None, ("1", "id=395931")),
(TTG_PAGE_URL, TTG_RSS_URL, ("1", "id=395931")),
(None, TTG_RSS_URL, None),
("https://tracker.example/t/395931/", "https://tracker.example/dl/395931/3576", None),
("https://totheglory.im/browse.php?c=M", "https://totheglory.im/dl/395931/", None),
("https://nottheglory.im/t/395931/", None, None),
])
def test_ttg_path_identity_ignores_signed_auth(page_url, enclosure, expected):
"""
TTG 详情页和下载路径中的种子 ID 是稳定身份,下载路径末段的 auth 不参与身份;其他站点不套用该路径规则。
"""
torrent = _torrent(page_url=page_url, enclosure=enclosure)
assert torrent_rules.resource_identity(torrent) == expected
if expected is not None:
renewed = replace(torrent, enclosure=TTG_NEW_URL)
assert torrent_rules.resource_identity(renewed) == expected
@pytest.mark.parametrize("url,expected", [
(TTG_OLD_URL, True),
("https://totheglory.im/dl/395931/3576/", True),
("https://www.totheglory.im/dl/395931/3576", True),
("https://totheglory.im/dl/395931/", False),
("https://totheglory.im/dl/395931/3576/extra", False),
("https://totheglory.im/t/395931/", False),
(TTG_RSS_URL, False),
("https://tracker.example/dl/395931/3576", False),
("https://nottheglory.im/dl/395931/3576", False),
])
def test_ttg_path_signature_is_expiring(url, expected):
"""
只有 TTG 的 /dl/{id}/{auth} 下载路径按短时签名处理,其他站点相同形态的路径保持普通链接。
"""
assert torrent_rules.is_expiring_download_url(url) is expected
def test_ttg_expired_path_signature_refreshes_same_resource(download_chain):
"""
TTG 路径签名下载返回 403 后,重新搜索同一种子 ID 的新 auth 并重试一次。
"""
chain, helper = download_chain
torrent = _torrent(page_url=TTG_PAGE_URL, enclosure=TTG_OLD_URL)
refreshed = replace(torrent, enclosure=TTG_NEW_URL, site_cookie="session=new")
other = replace(torrent, page_url="https://totheglory.im/t/395932/",
enclosure="https://totheglory.im/dl/395932/9429")
chain.search_site_torrents.return_value = [other, refreshed]
helper.download_torrent.side_effect = [(*FAILED[:4], "下载种子出错,状态码:403"), DOWNLOADED]
assert chain.download_torrent(torrent) == DOWNLOADED[1:4]
chain.search_site_torrents.assert_called_once_with(site=SITE, keyword=torrent.title)
assert [call.kwargs["url"] for call in helper.download_torrent.call_args_list] == [
TTG_OLD_URL, TTG_NEW_URL,
]
assert all(call.kwargs["cache_invalid"] is False for call in helper.download_torrent.call_args_list)
assert torrent.enclosure == TTG_NEW_URL
assert torrent.site_cookie == "session=new"
chain.post_message.assert_not_called()