refactor: resolve_episode_url 阶梯构造拆解(58 行 / 圈 19 / 嵌套 3 → 36 行 / 圈 12 / 嵌套 2)

为什么:播客单集直链解析的唯一入口,前 20 行是降级阶梯的嵌套构造
(if 套 for 套 if),与尝试循环混在同一个函数里,改音质口径要通读全函数。

拆出:
- _falldown_order(模块级纯函数):降级尝试序列——auto→通用阶梯原序 /
  已知音质→向下后缀 / 未知音质→该音质优先+通用阶梯,去重保序收敛为
  dict.fromkeys;_FALLDOWN_LADDER 从函数内上移模块级
- resolve_episode_url 只剩尝试循环;song_id 的 int 防御上提循环外
  (循环不变量:原实现首轮即判、失败直接 None,url_v1 从未收到过坏 sid)

先写测试后动代码:新增 tests/test_resolve_episode_url.py 11 例,在未改动
实现上一次跑绿,钉住核心口径——「只向下不向上」(standard 绝不回退到
exhigh,历史坑)、未知音质前置、垃圾 song_id 零请求、APIException 降级、
data[0] 无 url 降级、结果映射缺省值。
门禁:ruff 全过 · pyright 0/0 · pytest 370 passed(359→370)。
This commit is contained in:
snowy-yang committed 2026-09-27 11:00:50 +08:00
1 parent d2e6856899
commit 4d3204213e
2 files changed
+148 -28

No files matched your search

+27 -28
View File
@@ -434,6 +434,27 @@ def get_hot_radios(cookies: dict[str, str] | None = None,
return []
_FALLDOWN_LADDER = ['hires', 'lossless', 'exhigh', 'standard']
def _falldown_order(level: str) -> list[str]:
"""音质降级尝试序列:只向下不向上(原实现指定 standard 会回退到更高的 exhigh)。
- auto/空 → 通用阶梯原序
- 已知音质 → 该音质 + 阶梯中低于它的部分
- 未知音质 → 该音质优先 + 通用阶梯(去重保序)
"""
level = str(level or 'auto').strip()
if not level or level == 'auto':
order = list(PodcastConstants.QUALITY_LADDER)
elif level in _FALLDOWN_LADDER:
idx = _FALLDOWN_LADDER.index(level)
order = [level, *_FALLDOWN_LADDER[idx + 1:]]
else:
order = [level, *PodcastConstants.QUALITY_LADDER]
return list(dict.fromkeys(order)) # 去重保序
def resolve_episode_url(song_id: Any, cookies: dict[str, str] | None = None,
level: str = 'auto') -> dict[str, Any] | None:
"""解析单集音频直链(带音质降级)
@@ -446,35 +467,13 @@ def resolve_episode_url(song_id: Any, cookies: dict[str, str] | None = None,
Returns:
{'url','level','type','size','br','md5'} 或 None(全部失败)
"""
# 降级阶梯只向下不向上——原实现指定 standard 会回退到更高的 exhigh
_FALLDOWN_LADDER = ['hires', 'lossless', 'exhigh', 'standard']
order: list[str] = []
level = str(level or 'auto').strip()
if level and level != 'auto':
order.append(level)
# 追加"低于"请求音质的等级(hires > lossless > exhigh > standard)
if level in _FALLDOWN_LADDER:
idx = _FALLDOWN_LADDER.index(level)
for lv in _FALLDOWN_LADDER[idx + 1:]:
order.append(lv)
else:
# 未知等级:退到通用阶梯
order.extend(PodcastConstants.QUALITY_LADDER)
else:
order.extend(PodcastConstants.QUALITY_LADDER)
# 去重保序
dedup: list[str] = []
order = _falldown_order(level)
# song_id 可能是 None/垃圾值(防御:int 失败按无音频处理,不再 TypeError 500)
try:
sid = int(str(song_id).strip())
except (TypeError, ValueError):
return None
for lv in order:
if lv not in dedup:
dedup.append(lv)
order = dedup
for lv in order:
try:
# song_id 可能是 None/垃圾值(防御:int 失败按无音频处理,不再 TypeError 500)
sid = int(str(song_id).strip())
except (TypeError, ValueError):
return None
try:
res = url_v1(sid, lv, cookies or {})
except (APIException, ValueError):