# -*- coding: utf-8 -*- """路由公共工具:请求取参、ID 归一化、Cookie、安全转换。 这些函数原先在 11 个 router 模块里各复制一份(约 200 行重复), 现集中于此,行为与各副本完全一致(原逻辑逐字保留)。 """ import re from typing import Any from fastapi import Request # music.163.com 链接 / 纯数字 → ID(与 Flask MusicAPIService._extract_music_id 的 # 查询参数口径一致;短链展开由调用方按需先做) _ID_QUERY_RE = re.compile(r'[?&#]id=(\d+)') async def request_data(request: Request) -> dict[str, Any]: """三路兼容取参:GET query 优先,其次 POST JSON,最后表单。""" data: dict[str, Any] = {} if request.method == 'GET': data.update(dict(request.query_params)) return data try: body = await request.json() if isinstance(body, dict): data.update(body) except Exception: pass if not data: try: form = await request.form() data.update(dict(form.items())) except Exception: pass return data def safe_int(value: Any, default: int) -> int: """用户输入的安全整数解析:垃圾值回退默认值(而非 500)。""" try: return int(value) except (TypeError, ValueError): return default def extract_id(raw: Any) -> str: """URL / 分享链接 / 纯数字 → ID 字符串。""" s = str(raw or '').strip() if 'music.163.com' in s: from urllib.parse import parse_qs, urlsplit try: q = parse_qs(urlsplit(s).query) for key in ('id', 'songid'): if q.get(key): return str(q[key][0]).strip() except Exception: pass m = _ID_QUERY_RE.search(s) if m: return m.group(1) return s def cookies() -> dict[str, str]: """全局账号 Cookie(管理员 cookie.txt)。""" import music_api return music_api.load_cookies()