diff --git a/auth_system/geo.py b/auth_system/geo.py index 2511271..da52f54 100644 --- a/auth_system/geo.py +++ b/auth_system/geo.py @@ -4,10 +4,13 @@ （自 auth_system.py 拆出，行为与原实现一致。） """ +import asyncio import json import sqlite3 import time +from core.http_client import client as _http + from .db import ( _x, ) @@ -31,9 +34,13 @@ def _is_private_ip(ip: str) -> bool: return False -def query_ip_province(ip: str) -> str: +async def query_ip_province(ip: str) -> str: """查询 IP 归属省份（内网/失败返回空串），带进程内缓存 - 改用官方路径 /json/{query} 与 zh-CN 语言参数""" + 改用官方路径 /json/{query} 与 zh-CN 语言参数。 + + 走 core.http_client 的原生异步通道（原实现经 core.sync_http 同步门面 + 在后台事件循环桥接——调用方本就是 async 的登录链路，无需再桥一次）。 + """ ip = (ip or '').split(',')[0].strip() if not ip or ip in ('-', '127.0.0.1', '0.0.0.0') or _is_private_ip(ip): return '' @@ -43,10 +50,10 @@ def query_ip_province(ip: str) -> str: try: import urllib.parse as _ul - from core import sync_http as _rq - r = _rq.get('http://ip-api.com/json/' + _ul.quote(ip), - params={'lang': 'zh-CN', 'fields': 'status,regionName'}, - timeout=5) + r = await _http.request( + 'GET', 'http://ip-api.com/json/' + _ul.quote(ip), + params={'lang': 'zh-CN', 'fields': 'status,regionName'}, + timeout=5) d = r.json() if d.get('status') == 'success': province = str(d.get('regionName') or '') @@ -67,12 +74,16 @@ PROVINCE_WINDOW = 24 * 3600 # 省份变化检测窗口：24 小时 PROVINCE_BAN_LIMIT = 3 # 窗口内出现 >=3 个不同省份即封禁 -def record_login_location(user: sqlite3.Row, ip: str) -> tuple: +async def record_login_location(user: sqlite3.Row, ip: str) -> tuple: """记录本次登录省份并检测异常：24h 内出现 >=3 个不同省份自动封禁。 - 返回 (是否被封禁, 说明)""" + 返回 (是否被封禁, 说明) + + 落库仍走同步 sqlite（auth_system 是同步 DB 层），但调用方在事件循环里， + 故经 asyncio.to_thread 卸载——避免 sqlite 的 busy 等待（最长 15s）冻结整个服务。 + """ if user['role'] == 'admin': return False, '' - province = query_ip_province(ip) + province = await query_ip_province(ip) if not province: return False, '' now = int(time.time()) @@ -86,11 +97,12 @@ def record_login_location(user: sqlite3.Row, ip: str) -> tuple: for h in history: if h.get('p') and h['p'] not in provinces: provinces.append(h['p']) - _x('UPDATE users SET recent_provinces = ? WHERE id = ?', - (json.dumps(history, ensure_ascii=False), user['id'])) + await asyncio.to_thread( + _x, 'UPDATE users SET recent_provinces = ? WHERE id = ?', + (json.dumps(history, ensure_ascii=False), user['id'])) if len(provinces) >= PROVINCE_BAN_LIMIT: reason = f'24小时内检测到 {len(provinces)} 个不同省份登录（{"/".join(provinces[-3:])}），疑似账号倒卖，已自动封禁' - ban_user(user['id'], reason) - add_log('fail', user['username'], ip, '自动封禁', reason) + await asyncio.to_thread(ban_user, user['id'], reason) + await asyncio.to_thread(add_log, 'fail', user['username'], ip, '自动封禁', reason) return True, reason return False, province diff --git a/core/utils.py b/core/utils.py index 816ad80..f0679b1 100644 --- a/core/utils.py +++ b/core/utils.py @@ -1,11 +1,10 @@ # -*- coding: utf-8 -*- """纯函数工具（框架无关，无 Web 框架依赖）。 -SRT 时间轴、LRC→SRT 转换、字段参数解析、163cn.tv 短链展开（含私有缓存）。 +SRT 时间轴、LRC→SRT 转换、字段参数解析。 """ import re -import threading def srt_time(t: float) -> str: @@ -69,32 +68,3 @@ def parse_meta_fields(raw) -> list[str] | None: 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 diff --git a/services/api_service.py b/services/api_service.py index 5c50181..4320240 100644 --- a/services/api_service.py +++ b/services/api_service.py @@ -3,14 +3,11 @@ """ import logging -import re from pathlib import Path -from urllib.parse import parse_qs, urlsplit from core.config import ( Settings, ) -from core.utils import expand_shortlink as _expand_shortlink from music_api import ( NeteaseAPI, load_cookies, @@ -80,37 +77,6 @@ class MusicAPIService: self.logger.error(f"读取 cookie.txt 异常: {e}") return {} - def _extract_music_id(self, id_or_url: str) -> str: - """提取音乐ID（正规 URL 解析替代子串/删除非数字启发式） - - 短链仅当整串就是一个 163cn.tv 短链 URL 时才请求展开（防 SSRF） - - music.163.com 链接用 parse_qs 取 id 参数（参数顺序无关） - """ - raw = str(id_or_url or '').strip() - try: - # 短链展开：仅接受"整个输入就是 163cn.tv 链接"（子串包含不算） - if re.match(r'^https?://(www\.)?163cn\.tv/\S+$', raw, re.IGNORECASE): - raw = _expand_shortlink(raw) - - # 网易云链接：正规解析 query 取 id（不再 find('id=') 误吃 userid= 尾部） - if 'music.163.com' in raw: - try: - q = parse_qs(urlsplit(raw).query) - for key in ('id', 'songid'): - if q.get(key): - return str(q[key][0]).strip() - except Exception: - pass - # 兜底：老格式 #/song?id=xxx 的片段部分 - m = re.search(r'[?&#]id=(\d+)', raw) - if m: - return m.group(1) - - # 纯数字/普通文本直接返回 - return raw - except Exception as e: - self.logger.error(f"提取音乐ID失败: {e}") - return str(id_or_url or '').strip() - def _format_file_size(self, size_bytes) -> str: """格式化文件大小（上游 size 可能为 null/垃圾值：容错返回"大小未知"而非 500）""" try: diff --git a/tests/test_login_flow.py b/tests/test_login_flow.py index a3cbe2d..45dbe96 100644 --- a/tests/test_login_flow.py +++ b/tests/test_login_flow.py @@ -133,7 +133,7 @@ def test_admin_without_email_blocked_fail_closed(env): def test_geo_ban_bumps_seq_again(env, monkeypatch): """异地多省封禁：login_seq 再递增一次（已签发中的会话全部顶下线）。""" - def _banned(user, ip): + async def _banned(user, ip): return True, {'province': 'X'} monkeypatch.setattr(auth, 'record_login_location', _banned) diff --git a/web/routers/auth.py b/web/routers/auth.py index f5c3397..99a7ec1 100644 --- a/web/routers/auth.py +++ b/web/routers/auth.py @@ -173,7 +173,7 @@ async def _user_session_response(user, ip: str) -> JSONResponse: seq_row = await repo.get_user_by_id(uid) seq = int((seq_row or {}).get('login_seq') or 1) - banned, _info = await asyncio.to_thread(auth.record_login_location, user, ip) + banned, _info = await auth.record_login_location(user, ip) if banned: await db.execute( 'UPDATE users SET login_seq = COALESCE(login_seq, 0) + 1 WHERE id = ?', (uid,)) @@ -320,7 +320,7 @@ async def admin_login_verify(request: Request): seq_row = await repo.get_user_by_id(uid) seq = int((seq_row or {}).get('login_seq') or 1) - banned, _info = await asyncio.to_thread(auth.record_login_location, user, ip) + banned, _info = await auth.record_login_location(user, ip) if banned: await db.execute('UPDATE users SET login_seq = COALESCE(login_seq, 0) + 1 WHERE id = ?', (uid,)) return error('检测到账号在多个省份异常登录，已自动封禁，请联系管理员')