NeteaseUrl/core/utils.py

# -*- coding: utf-8 -*-
"""纯函数工具（框架无关，无 Web 框架依赖）。

安全整数解析、控制台输出兜底、SRT 时间轴、
LRC→SRT 转换、字段参数解析、163cn.tv 短链展开（含私有缓存）。
"""

import re
import sys
import threading


def safe_int(value, default: int) -> int:
    """用户输入的安全整数解析：垃圾值回退默认值（而非 500）"""
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def safe_console_print(text: str) -> None:
    """控制台输出兜底：stdout 被重定向到文件且系统编码非 UTF-8 时
    （典型：NSSM 服务日志 + 中文 Windows GBK 控制台），emoji 等字符会触发
    UnicodeEncodeError 中断启动流程——实测崩溃循环。此处降级为可编码字符。"""
    try:
        print(text)
    except UnicodeEncodeError:
        enc = getattr(sys.stdout, 'encoding', None) or 'ascii'
        print(text.encode(enc, errors='replace').decode(enc, errors='replace'))


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