From 133beab492e0bef0ecef0c87329860ee25a5a9cd Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Mon, 28 Sep 2026 20:18:08 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=88=A0=E9=99=A4=20HTTP=20=E7=BA=BF?= =?UTF-8?q?=E4=B8=89=E4=B8=AA=E5=9E=AB=E7=89=87=EF=BC=8C=E5=90=8C=E6=AD=A5?= =?UTF-8?q?/=E5=BC=82=E6=AD=A5=E6=A1=A5=E6=8E=A5=E5=B1=82=E5=BD=BB?= =?UTF-8?q?=E5=BA=95=E6=B8=85=E9=9B=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 删 core/sync_http.py（requests 同形同步门面）、core/async_bridge.py（run_sync 同步↔异步桥）、core/http_errors.py（两垫片共用的 requests 同形异常层次）， 以及 tests/test_async_bridge.py。 为什么必须一次删完：三者是**自闭环三角**——sync_http 是 async_bridge 与 http_errors 的唯一消费者，删掉任何一个其余两个立刻悬空。前序 B1 把 worker 变成 协程、B2 把 auth_system 迁到 aiosqlite、更早一轮把 music_api 改走 core.http_client 之后，这条链上的生产消费者已归零；http_errors 更是在 B2 之后 只剩 sync_http 一个使用者（此前 sync_http 的 docstring 还提到 models._RetryFromScratch 继承它的 RequestException——B1 已改为继承 Exception）。 删除前的核查（避免误删活代码）： - run_sync / get_loop / SyncResponse / Session( 全仓零残留引用（只剩 aiohttp.ClientSession，那是真正的客户端，不受影响）。 - 所有 HTTPError 命中都是 core.http_client.UpstreamHTTPError——与 core.http_errors 的 HTTPError **名字像但不是一个东西**，删除时别被 grep 骗。 - requests 早已不是声明依赖、全仓零导入；删掉 http_errors 后连 requests 同形 异常的最后残影也没了。 - core/__init__.py 只有 docstring，无需改动；web/app.py 的 async_bridge.shutdown() 已在 B1 移除。 顺带修正两处过时措辞：tests/test_search_music.py 的错误映射注释 RequestException → UpstreamError（异常层次已换）；pyproject.toml 中 pyright include 的理由注释 requests→sync_http → requests→aiohttp（去掉已删模块名、 保留原意）。auth_system/geo.py 的历史注释本就是过去时，保持不动。 门禁：ruff 全过 / pyright 0-0-0 / pytest 382 项全绿（0 失败 0 error 0 skipped）。 项数 384→382 是 test_async_bridge.py 两条用例**随被测对象一起删除**的预期减项， 不是跳过或漏跑。data/site.db 读写前后 sha256 一致（a19f5db…6487）。 至此「同步调异步 / 异步桥同步」两类垫片全部拆净：DB 线（B1 worker + B2 auth_system）、HTTP 线（本批）。生产代码里 to_thread 仅剩正当的阻塞卸载 （PBKDF2 / smtplib / 备份文件 IO / ZIP 组装 / rmtree），threading.Event 仅剩 取消与暂停的跨线程置位载体。 --- core/async_bridge.py | 61 ------------ core/http_errors.py | 40 -------- core/sync_http.py | 196 ------------------------------------- pyproject.toml | 2 +- tests/test_async_bridge.py | 31 ------ tests/test_search_music.py | 2 +- 6 files changed, 2 insertions(+), 330 deletions(-) delete mode 100644 core/async_bridge.py delete mode 100644 core/http_errors.py delete mode 100644 core/sync_http.py delete mode 100644 tests/test_async_bridge.py diff --git a/core/async_bridge.py b/core/async_bridge.py deleted file mode 100644 index fb7c0a3..0000000 --- a/core/async_bridge.py +++ /dev/null @@ -1,61 +0,0 @@ -# -*- coding: utf-8 -*- -"""同步 ↔ 异步桥。 - -场景：部分调用方仍是同步代码（线程池下载器、后台任务线程），而服务层已异步。 -`run_sync(coro)` 把协程提交到**专用后台事件循环线程**执行并阻塞等待结果， -使同步代码得以调用异步实现，无需把调用链一次性全部改异步。 - -- 后台 loop 常驻单例（daemon 线程），避免每个请求新建 loop 的开销与 - aiohttp 连接池失效； -- 在已有事件循环中调用 `run_sync` 会死锁，故检测到运行中的 loop 时直接报错， - 提示改用 `await`。 -""" -import asyncio -import threading -from collections.abc import Coroutine -from typing import Any - -_loop: asyncio.AbstractEventLoop | None = None -_loop_lock = threading.Lock() - - -def _get_loop() -> asyncio.AbstractEventLoop: - global _loop - if _loop is not None and not _loop.is_closed(): - return _loop - with _loop_lock: - if _loop is None or _loop.is_closed(): - loop = asyncio.new_event_loop() - t = threading.Thread(target=loop.run_forever, name='async-bridge', daemon=True) - t.start() - _loop = loop - return _loop - - -def get_loop() -> asyncio.AbstractEventLoop: - """后台事件循环（供需要直接提交协程的调用方使用）。""" - return _get_loop() - - -def run_sync[T](coro: Coroutine[Any, Any, T], timeout: float | None = None) -> T: - """在同步代码中运行协程并等待结果。 - - Raises: - RuntimeError: 在已有事件循环内调用（应直接 await 而非桥接）。 - """ - try: - asyncio.get_running_loop() - except RuntimeError: - pass - else: - raise RuntimeError('run_sync 不能在事件循环内调用，请直接 await 协程') - fut = asyncio.run_coroutine_threadsafe(coro, _get_loop()) - return fut.result(timeout) - - -def shutdown() -> None: - """停止后台 loop（应用退出时调用）。""" - global _loop - if _loop is not None and not _loop.is_closed(): - _loop.call_soon_threadsafe(_loop.stop) - _loop = None diff --git a/core/http_errors.py b/core/http_errors.py deleted file mode 100644 index 67710ee..0000000 --- a/core/http_errors.py +++ /dev/null @@ -1,40 +0,0 @@ -# -*- coding: utf-8 -*- -"""requests 同形异常层次（同步 HTTP 门面 core.sync_http 使用）。 - -历史：`core.aiohttp_compat`（异步垫片）与 `core.sync_http`（同步门面）曾各自 -定义一份 `RequestException` / `HTTPError`——两份定义互不相干，于是同一条调用链上 -「同步段抛的异常捕获不到异步段的同类失败」。异步垫片已删除、其调用方（music_api） -全面改走 core.http_client（自带 UpstreamError 层次）；此处保留唯一来源， -供仍在线程里同步取数据的 core.sync_http 使用。 - -命名沿用 requests（`Timeout` / `ConnectionError` 与内置异常同名，这是 requests -自身的约定——调用方应带模块前缀引用，例如 `requests.ConnectionError`， -裸 `ConnectionError` 会命中内置的那个）。 - -层次： - RequestException 网络 / 协议层失败（可捕获的总入口） - ├── HTTPError raise_for_status() 抛出的状态异常（携带 response） - ├── Timeout 请求超时 - └── ConnectionError 连接错误 -""" -from typing import Any - - -class RequestException(Exception): - """请求异常基类（网络/协议层）。""" - - -class HTTPError(RequestException): - """raise_for_status() 抛出的 HTTP 状态异常。""" - - def __init__(self, message: str, response: Any = None): - super().__init__(message) - self.response = response - - -class Timeout(RequestException): - """请求超时。""" - - -class ConnectionError(RequestException): - """连接错误。""" diff --git a/core/sync_http.py b/core/sync_http.py deleted file mode 100644 index 652b0ba..0000000 --- a/core/sync_http.py +++ /dev/null @@ -1,196 +0,0 @@ -# -*- coding: utf-8 -*- -"""同步 HTTP 门面（底层 aiohttp）——彻底移除 requests 依赖。 - -为什么需要它：部分代码天然是同步的（worker 线程里的流式下载、纯函数里的短链展开/ -IP 归属地查询）。本模块提供 requests 风格的同步 API，但底层复用 aiohttp 连接池， -不再引入 requests。 - -流式读取的实现：在后台事件循环起一个读取协程，把块推入有界队列（背压）， -同步侧从队列消费——每次搬运只过一次队列，不做「每块一次跨线程调度」。 -""" -import asyncio -import queue -import threading -from collections.abc import Iterator, Mapping -from typing import Any - -from core import async_bridge -from core.http_client import client as _client -from core.http_errors import HTTPError, RequestException - -# 异常层次见 core.http_errors（与调用方共用的 requests 同形定义）。RequestException -# 在此再导出：music_downloader.models 的 _RetryFromScratch 继承它，既有 except -# 与测试也直接导入它。 -__all__ = ['HTTPError', 'RequestException', 'Session', 'SyncResponse', 'get', 'post'] - -_DEFAULT_TIMEOUT = 30 - - -class SyncResponse: - """requests.Response 的同步子集（底层 aiohttp）。""" - - def __init__(self, status: int, headers: Mapping[str, str], content: bytes | None, - url: str, stream_state: dict | None = None): - self.status_code = status - self.headers = dict(headers) - self.url = url - self._content = content - self._stream = stream_state - - @property - def content(self) -> bytes: - if self._content is None: - self._content = b''.join(self.iter_content(64 * 1024)) - return self._content - - @property - def text(self) -> str: - return self.content.decode('utf-8', errors='replace') - - def json(self, **kw) -> Any: - import json as _json - return _json.loads(self.text, **kw) - - def raise_for_status(self) -> None: - if self.status_code >= 400: - raise HTTPError(f'HTTP {self.status_code}', self) - - def iter_content(self, chunk_size: int = 64 * 1024) -> Iterator[bytes]: - """流式读取响应体（大小写与 requests 一致：chunk_size 为字节数）。""" - if self._content is not None: - for i in range(0, len(self._content), chunk_size): - yield self._content[i:i + chunk_size] - return - st = self._stream - assert st is not None - q: queue.Queue = st['queue'] - while True: - item = q.get() - if item is None: # 正常结束 - break - if isinstance(item, BaseException): # 读取出错 - raise item - yield item - - def close(self) -> None: - st = self._stream - if st and st.get('cancel') is not None: - try: - async_bridge.run_sync(_cancel(st)) - except Exception: - pass - - def __enter__(self) -> 'SyncResponse': - return self - - def __exit__(self, *exc) -> bool: - self.close() - return False - - -async def _cancel(st: dict) -> None: - st['cancelled'] = True - resp = st.get('resp') - if resp is not None: - resp.close() - - -def _stream_reader(st: dict, chunk_size: int) -> None: - """在后台事件循环读块 → 有界队列（背压），同步侧消费。""" - async def _run(): - try: - async for chunk in st['resp'].content.iter_chunked(chunk_size): - if st.get('cancelled'): - break - while True: # 背压：队列满则稍等，避免内存无界 - try: - st['queue'].put(chunk, timeout=0.5) - break - except queue.Full: - if st.get('cancelled'): - return - except Exception as e: - st['queue'].put(e) - finally: - try: - st['resp'].close() - except Exception: - pass - st['queue'].put(None) - - asyncio.run_coroutine_threadsafe(_run(), async_bridge.get_loop()) - - -class Session: - """requests.Session 的同步子集（连接池由底层 aiohttp 统一维护）。""" - - trust_env = False - - def __init__(self, **kw) -> None: - # headers：会话级默认头（与 requests.Session().headers.update 等价） - self.headers: dict[str, str] = dict(kw.pop('headers', {})) - - def _merge(self, kw: dict) -> dict: - h = dict(self.headers) - h.update(kw.pop('headers', None) or {}) - if h: - kw['headers'] = h - return kw - - def get(self, url: str, **kw) -> SyncResponse: - return _request('GET', url, **self._merge(kw)) - - def post(self, url: str, **kw) -> SyncResponse: - return _request('POST', url, **self._merge(kw)) - - def close(self) -> None: - pass - - -def _norm_timeout(timeout): - """requests 允许 (connect, read) 元组；aiohttp 只接受单一总时长——取较大者。""" - if isinstance(timeout, (tuple, list)): - vals = [float(v) for v in timeout if isinstance(v, (int, float))] - return max(vals) if vals else _DEFAULT_TIMEOUT - return timeout - - -def _request(method: str, url: str, *, headers: Mapping[str, str] | None = None, - cookies: Mapping[str, str] | None = None, params: Mapping[str, Any] | None = None, - data: Any = None, json: Any = None, timeout: float | None = _DEFAULT_TIMEOUT, - allow_redirects: bool = True, stream: bool = False) -> SyncResponse: - timeout = _norm_timeout(timeout) - if stream: - return _open_stream(method, url, headers=headers, cookies=cookies, params=params, - data=data, json=json, timeout=timeout, - allow_redirects=allow_redirects) - resp = async_bridge.run_sync(_client.request( - method, url, headers=headers, cookies=cookies, params=params, data=data, - json_body=json, timeout=timeout, allow_redirects=allow_redirects)) - return SyncResponse(resp.status, resp.headers, resp.text.encode('utf-8'), - resp.url) - - -def _open_stream(method: str, url: str, **kw) -> SyncResponse: - timeout = _norm_timeout(kw.pop('timeout', _DEFAULT_TIMEOUT)) - # AsyncHTTPClient.open 只接受 headers/cookies/timeout/allow_redirects； - # 流式场景不支持 params/data/json（调用方直接把查询串拼进 URL） - for unsupported in ('params', 'data', 'json'): - kw.pop(unsupported, None) - resp = async_bridge.run_sync(_client.open(method, url, timeout=timeout, **kw)) - st = {'resp': resp, 'queue': queue.Queue(maxsize=8), 'cancel': object(), 'cancelled': False} - th = threading.Thread(target=_stream_reader, args=(st, 64 * 1024), daemon=True) - th.start() - return SyncResponse(resp.status, resp.headers, None, str(resp.url), stream_state=st) - - -# 模块级便捷函数（等价于共享 Session 的 get/post） -_default_session = Session() - - -def get(url: str, **kw) -> SyncResponse: - return _default_session.get(url, **kw) - - -def post(url: str, **kw) -> SyncResponse: - return _default_session.post(url, **kw) diff --git a/pyproject.toml b/pyproject.toml index cc71899..e85b405 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -90,7 +90,7 @@ quote-style = "preserve" [tool.pyright] # 全量纳入（Flask 巨石删除后不再有例外清单） # 顶层模块与 tests 此前不在名单里——「pyright 0 错误」只覆盖了 core/web/services， -# 而 requests→sync_http 迁移的遗留恰好全部躲在漏检的顶层模块中，故补齐。 +# 而早期 requests→aiohttp 迁移的遗留恰好全部躲在漏检的顶层模块中，故补齐。 include = [ "core", "web", diff --git a/tests/test_async_bridge.py b/tests/test_async_bridge.py deleted file mode 100644 index 72bdbb0..0000000 --- a/tests/test_async_bridge.py +++ /dev/null @@ -1,31 +0,0 @@ -# -*- coding: utf-8 -*- -"""同步 ↔ 异步桥测试。 - -（原 test_aiohttp_compat.py：异步 HTTP 垫片 core.aiohttp_compat 已随其调用方 -music_api 全面改走 core.http_client 而删除，仅保留桥本身的行为契约。） -""" -import asyncio - -import pytest - -from core import async_bridge - - -def test_bridge_runs_coro_from_sync(): - async def coro(): - await asyncio.sleep(0.01) - return 42 - - assert async_bridge.run_sync(coro()) == 42 - - -async def test_bridge_rejects_inside_loop(): - async def coro(): - return 1 - - c = coro() - try: - with pytest.raises(RuntimeError): - async_bridge.run_sync(c) - finally: - c.close() # run_sync 拒绝执行，需自行关闭以免「未 await」告警 diff --git a/tests/test_search_music.py b/tests/test_search_music.py index ee07e92..c5d315f 100644 --- a/tests/test_search_music.py +++ b/tests/test_search_music.py @@ -3,7 +3,7 @@ 为什么：``music_api/api.py::search_music`` 是站内搜索的唯一上游通道 （101 行 / 四路条目映射 / 嵌套 7），此前**零测试覆盖**。它的错误映射 -（RequestException/JSONDecodeError→APIException）与四路字段的容错口径 +（UpstreamError/JSONDecodeError→APIException）与四路字段的容错口径 （残缺单条跳过、空 artists 不兜底、alias 缺省 '' 等）都是对外可观察行为， 拆解时改坏任意一路都不会有测试报警。