Archived
snowy-yang
0569b50737
为什么：用 AST 级引用分析（而非文本 grep——同名符号会互相掩盖）确认这些 符号全项目零引用。文本 grep 查不出的原因是它们都有「活着的同名兄弟」： services/downloads.py（共 6 个，93 行） _parse_cookie_str / _global_music_u / _vip_name / _detect_vip / _admin_client / _user_client —— 活的是 core.media.parse_cookie_str、web/routers/music.py、account.py 的版本 随之失效的 import：USER_COOKIE_FILE / GLOBAL_MUSIC_U_CACHE / VIP_PROBE_LOCK / account_info / detect_vip_by_probe / QRLoginClient services/api_service.py APIResponse 类（29 行）—— web/response.py 才是活的响应封装 services/_shared.py _rate_ok —— 逐字复制 web.ratelimit.rate_ok core/utils.py safe_int / safe_console_print —— 活的是 web/request_utils.safe_int core/repo.py all_settings / get_user_by_name —— auth_system.get_user_by_name 才是活的 core/db.py / web/auth_state.py / web/request_utils.py execute_many / _safe_register_credits + secrets_rand / dedup_ids 结果：生产代码死代码候选 13 → 0（余下 118 个全是 pytest 测试函数，属正常）。
101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""纯函数工具（框架无关，无 Web 框架依赖）。
|
|
|
|
SRT 时间轴、LRC→SRT 转换、字段参数解析、163cn.tv 短链展开（含私有缓存）。
|
|
"""
|
|
|
|
import re
|
|
import threading
|
|
|
|
|
|
def srt_time(t: float) -> str:
|
|
"""秒 → SRT 时间格式 HH:MM:SS,mmm"""
|
|
h = int(t // 3600)
|
|
m = int((t % 3600) // 60)
|
|
s = int(t % 60)
|
|
ms = round((t - int(t)) * 1000)
|
|
if ms >= 1000:
|
|
ms = 999
|
|
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
|
|
|
|
|
|
def format_lyric(lrc: str, lyric_format: str) -> str:
|
|
"""将 LRC 歌词转换为指定格式（lrc / srt）
|
|
时间轴先规范化再转 SRT——支持一行多时间标签（同句多时段各自成条）、
|
|
分钟超两位（[100:00]）、乱序行（排序后区间不再倒挂）、纯文本行（无时间戳则原样保留）。"""
|
|
if lyric_format != 'srt':
|
|
return lrc
|
|
tag_re = re.compile(r'\[(\d{1,3}):(\d{1,2}(?:[.:]\d+)?)\]')
|
|
parsed = []
|
|
for line in lrc.split('\n'):
|
|
tags = tag_re.findall(line)
|
|
text = tag_re.sub('', line).strip()
|
|
if not tags:
|
|
# 无时间戳的纯文本：保留为 0 时长行不合适，直接附在输出尾部
|
|
if text:
|
|
parsed.append((None, text))
|
|
continue
|
|
if not text:
|
|
continue
|
|
for m, s in tags:
|
|
try:
|
|
secs = int(m) * 60 + float(s.replace(':', '.'))
|
|
except ValueError:
|
|
continue
|
|
parsed.append((secs, text))
|
|
timed = [(t, x) for t, x in parsed if t is not None]
|
|
untimed = [x for t, x in parsed if t is None]
|
|
if not timed:
|
|
return lrc
|
|
timed.sort(key=lambda p: p[0])
|
|
out = []
|
|
for i, (t, text) in enumerate(timed):
|
|
end = timed[i + 1][0] if i + 1 < len(timed) else t + 5
|
|
if end <= t:
|
|
end = t + 1 # 同时间多行：给 1 秒最小区间，不再生成倒挂区间
|
|
out.append(f"{i + 1}\n{srt_time(t)} --> {srt_time(end)}\n{text}\n")
|
|
if untimed:
|
|
out.append('\n'.join(untimed))
|
|
return '\n'.join(out)
|
|
|
|
|
|
def parse_meta_fields(raw) -> list[str] | None:
|
|
"""解析写入字段参数，空返回 None（表示全部写入）"""
|
|
if raw is None or raw == '':
|
|
return None
|
|
if isinstance(raw, str):
|
|
return [f.strip() for f in raw.split(',') if f.strip()] or None
|
|
try:
|
|
return list(raw) or None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# 短链展开结果缓存：用户反复解析同一分享链接不再重复请求 163cn.tv（有界防膨胀）
|
|
_SHORTLINK_CACHE: dict[str, str] = {}
|
|
_SHORTLINK_CACHE_LOCK = threading.Lock()
|
|
|
|
|
|
def expand_shortlink(url: str) -> str:
|
|
"""展开 163cn.tv 短链（带进程内缓存；失败原样返回）"""
|
|
with _SHORTLINK_CACHE_LOCK:
|
|
hit = _SHORTLINK_CACHE.get(url)
|
|
if hit is not None:
|
|
return hit
|
|
expanded = url
|
|
try:
|
|
from core import sync_http
|
|
|
|
response = sync_http.get(url, allow_redirects=False, timeout=10)
|
|
expanded = response.headers.get('Location', url)
|
|
except Exception:
|
|
expanded = url
|
|
# 展开失败（Location 缺失/网络异常，expanded 原样等于 url）不写
|
|
# 缓存——旧口径一次网络抖动会让该分享链接在整个缓存存续期内必然解析失败
|
|
if expanded != url:
|
|
with _SHORTLINK_CACHE_LOCK:
|
|
if len(_SHORTLINK_CACHE) > 500:
|
|
_SHORTLINK_CACHE.clear()
|
|
_SHORTLINK_CACHE[url] = expanded
|
|
return expanded
|