improvement(sns): 精简朋友圈媒体链路并移除手动挑图入口

- 移除 Server-Timing/X-SNS 阶段透出与前端媒体来源标记
- 移除本地候选图与手动匹配接口及预览交互
- /api/sns/media 默认仅保留远程下载解密路径,未命中直接返回 404
- 将导出所需的本地缓存匹配逻辑下沉到 sns_export_service
This commit is contained in:
2977094657
2026-04-16 01:19:09 +08:00
Unverified
parent 94a8b124c3
commit 7870cd54fc
8 changed files with 343 additions and 1267 deletions
@@ -0,0 +1,39 @@
import asyncio
import sys
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest import mock
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from wechat_decrypt_tool.routers import sns # noqa: E402 pylint: disable=wrong-import-position
class TestSnsMediaRouteWeFlowDefault(unittest.TestCase):
def test_route_stops_after_remote_miss_by_default(self):
with TemporaryDirectory() as td:
account_dir = Path(td) / "acc"
account_dir.mkdir(parents=True, exist_ok=True)
with mock.patch("wechat_decrypt_tool.routers.sns._resolve_account_dir", return_value=account_dir):
with mock.patch("wechat_decrypt_tool.routers.sns._try_fetch_and_decrypt_sns_remote", return_value=None):
with self.assertRaises(sns.HTTPException) as ctx:
asyncio.run(
sns.get_sns_media(
account="acc",
url="https://mmsns.qpic.cn/sns/test/0",
key="123",
token="tkn",
use_cache=1,
)
)
self.assertEqual(ctx.exception.status_code, 404)
if __name__ == "__main__":
unittest.main()
-40
View File
@@ -1,40 +0,0 @@
import sys
import unittest
from pathlib import Path
from starlette.responses import Response
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from wechat_decrypt_tool.sns_stage_timing import add_sns_stage_timing_headers # noqa: E402 pylint: disable=wrong-import-position
class TestSnsStageServerTiming(unittest.TestCase):
def test_injects_server_timing_when_missing(self):
resp = Response(content=b"ok")
add_sns_stage_timing_headers(resp.headers, source="proxy")
st = str(resp.headers.get("Server-Timing") or "")
self.assertIn("sns_source_", st)
self.assertIn("proxy", st)
def test_appends_when_upstream_server_timing_exists(self):
resp = Response(content=b"ok")
resp.headers["Server-Timing"] = "edge;dur=1"
add_sns_stage_timing_headers(resp.headers, source="proxy")
st = str(resp.headers.get("Server-Timing") or "")
self.assertIn("edge;dur=1", st)
self.assertIn("sns_source_", st)
def test_does_not_duplicate_existing_sns_source_metric(self):
resp = Response(content=b"ok")
resp.headers["Server-Timing"] = 'sns_source_proxy;dur=0;desc="proxy"'
add_sns_stage_timing_headers(resp.headers, source="proxy")
st = str(resp.headers.get("Server-Timing") or "")
self.assertEqual(st.count("sns_source_"), 1)
if __name__ == "__main__":
unittest.main()