diff --git a/backend/app/services/scheduler.py b/backend/app/services/scheduler.py index 4395d5f..c1f8365 100644 --- a/backend/app/services/scheduler.py +++ b/backend/app/services/scheduler.py @@ -248,6 +248,27 @@ def _sync_upstream_keys(upstream_id: int, snapshot: dict[str, Any], captured_at: db = SessionLocal() try: active_group_ids = set(snapshot.get("groups", {}).keys()) + + # 获取倒数第二个快照作为 previous snapshot + prev_active_group_ids: set[str] | None = None + from app.models.snapshot import UpstreamRateSnapshot + snapshots = ( + db.query(UpstreamRateSnapshot) + .filter(UpstreamRateSnapshot.upstream_id == upstream_id) + .order_by(UpstreamRateSnapshot.captured_at.desc()) + .limit(2) + .all() + ) + if len(snapshots) >= 2: + prev_snapshot = snapshots[1] + try: + prev_data = json.loads(prev_snapshot.snapshot_json or "{}") + prev_groups = prev_data.get("groups") or {} + if isinstance(prev_groups, dict): + prev_active_group_ids = set(prev_groups.keys()) + except Exception: + pass + # 用 UpstreamClient 查询远端活跃 Key ID 集合 remote_key_ids: set[str] | None = None try: @@ -269,7 +290,7 @@ def _sync_upstream_keys(upstream_id: int, snapshot: dict[str, Any], captured_at: except Exception as exc: logger.warning("sync upstream keys list failed for %s: %s", upstream_id, exc) - website_sync.reconcile_upstream_keys(db, upstream_id, active_group_ids, remote_key_ids, captured_at) + website_sync.reconcile_upstream_keys(db, upstream_id, active_group_ids, remote_key_ids, captured_at, prev_active_group_ids=prev_active_group_ids) db.commit() except Exception: logger.exception("key sync failed for upstream %s", upstream_id) diff --git a/backend/app/services/website_sync.py b/backend/app/services/website_sync.py index d268acd..f7cf48a 100644 --- a/backend/app/services/website_sync.py +++ b/backend/app/services/website_sync.py @@ -848,6 +848,7 @@ def reconcile_upstream_keys( active_group_ids: set[str] | None, remote_key_ids: set[str] | None, captured_at: datetime, + prev_active_group_ids: set[str] | None = None, ) -> None: """对账上游 Key 的本地缓存与远端状态。 @@ -857,6 +858,7 @@ def reconcile_upstream_keys( """ if active_group_ids is None and remote_key_ids is None: return + key_rows = ( db.query(UpstreamGeneratedKey) .filter( @@ -865,16 +867,40 @@ def reconcile_upstream_keys( .all() ) for row in key_rows: + # 1. 增加自恢复逻辑：已导入的 orphaned key 如果后续快照中来源分组恢复，且远端 key id 仍存在，则自动恢复为 imported 并清空错误。 + if ( + row.status == "orphaned" + and row.error == "来源分组已不存在" + and row.imported_website_id is not None + and row.imported_account_id is not None + and active_group_ids is not None + and row.group_id in active_group_ids + and remote_key_ids is not None + and row.key_id in remote_key_ids + ): + row.status = "imported" + row.error = None + row.updated_at = captured_at + logger.info("recovered orphaned key %s to imported (group %s recovered and key_id %s exists remotely)", row.id, row.group_id, row.key_id) + + # 2. 修改上游 key 对账逻辑：来源分组缺失时，不因单次快照缺失立即把已导入 key 标为 orphaned。 + # 分组级 orphan 判断改为“两次连续快照都缺失该分组”才生效，避免上游接口临时返回不完整分组导致误标记。 if active_group_ids is not None and row.group_id not in active_group_ids: - if row.imported_website_id and row.imported_account_id: - row.status = "orphaned" - row.error = "来源分组已不存在" - row.updated_at = captured_at - logger.info("marked key %s orphaned (group %s no longer in snapshot)", row.id, row.group_id) + if prev_active_group_ids is not None and row.group_id not in prev_active_group_ids: + if row.imported_website_id and row.imported_account_id: + row.status = "orphaned" + row.error = "来源分组已不存在" + row.updated_at = captured_at + logger.info("marked key %s orphaned (group %s no longer in snapshot consecutively)", row.id, row.group_id) + else: + db.delete(row) + logger.info("removed key %s (group %s no longer in snapshot consecutively)", row.id, row.group_id) + continue else: - db.delete(row) - logger.info("removed key %s (group %s no longer in snapshot)", row.id, row.group_id) - continue + # 只有单次快照缺失，跳过标记/清理，避免误标 + logger.info("skipped marking key %s orphaned (group %s missing in single snapshot, waiting for next consecutive check)", row.id, row.group_id) + + # 3. 远端 key id 级清理的现有保守策略：只有远端 key 列表成功拉取后，才允许判断 key 是否不存在。 if row.key_id and remote_key_ids is not None and row.key_id not in remote_key_ids: if row.imported_website_id and row.imported_account_id: row.status = "orphaned" @@ -885,6 +911,7 @@ def reconcile_upstream_keys( db.delete(row) logger.info("removed key %s (key_id %s gone from remote)", row.id, row.key_id) continue + if remote_key_ids is not None and row.key_id in remote_key_ids: row.updated_at = captured_at @@ -914,11 +941,31 @@ def reconcile_upstream_keys_full(db: Session, upstream_id: int) -> bool: now = datetime.now(timezone.utc) # 从最新快照获取活跃分组 ID（与调度器 _sync_upstream_keys 一致） + prev_active_group_ids: set[str] | None = None groups = latest_rate_map(db, upstream_id) if groups: active_group_ids = set(groups.keys()) groups_fetched = True + # 获取倒数第二个快照作为 previous snapshot + from app.models.snapshot import UpstreamRateSnapshot + snapshots = ( + db.query(UpstreamRateSnapshot) + .filter(UpstreamRateSnapshot.upstream_id == upstream_id) + .order_by(UpstreamRateSnapshot.captured_at.desc()) + .limit(2) + .all() + ) + if len(snapshots) >= 2: + prev_snapshot = snapshots[1] + try: + prev_data = json.loads(prev_snapshot.snapshot_json or "{}") + prev_groups = prev_data.get("groups") or {} + if isinstance(prev_groups, dict): + prev_active_group_ids = set(prev_groups.keys()) + except Exception: + pass + try: with UpstreamClient( base_url=upstream.base_url, @@ -944,6 +991,7 @@ def reconcile_upstream_keys_full(db: Session, upstream_id: int) -> bool: active_group_ids if groups_fetched else None, remote_key_ids if keys_fetched else None, now, + prev_active_group_ids=prev_active_group_ids, ) db.commit() return keys_fetched diff --git a/backend/test_upstream_key_sync.py b/backend/test_upstream_key_sync.py index fe88887..bf7e7e6 100644 --- a/backend/test_upstream_key_sync.py +++ b/backend/test_upstream_key_sync.py @@ -10,6 +10,7 @@ from sqlalchemy.pool import StaticPool from app.database import Base from app.models.upstream import Upstream from app.models.website import Website # noqa: F401 — registers table for FK refs +from app.models.snapshot import UpstreamRateSnapshot # noqa: F401 — registers table for FK refs from app.models.upstream_key import UpstreamGeneratedKey @@ -1834,3 +1835,200 @@ def test_ensure_group_key_rollback_prevents_leak(db_session): assert vip_db.key_value == "sk-vip-local" +def test_upstream_key_reconciliation_robustness(db_session, monkeypatch): + """测试对账逻辑的健壮性：单次快照缺失不标记、两次连续快照缺失标记、自恢复、拉取失败时不误清理/误恢复。""" + from app.services import scheduler as sched_mod + from app.models.upstream_key import UpstreamGeneratedKey + from app.models.snapshot import UpstreamRateSnapshot + from app.services.upstream_client import UpstreamClient + from app.services import website_sync + import json + from datetime import datetime, timezone + + # 1. 准备基础数据 + website = Website( + name="Target", + site_type="sub2api", + base_url="http://target.local", + api_prefix="/api/v1/admin", + auth_type="api_key", + auth_config_json="{}", + ) + upstream = Upstream( + name="TestUpstream", + base_url="http://upstream.local", + api_prefix="/api/v1", + auth_type="bearer", + auth_config_json="{}", + ) + db_session.add_all([website, upstream]) + db_session.commit() + db_session.refresh(website) + db_session.refresh(upstream) + + # 验证 UpstreamRateSnapshot 表是否成功创建并可正常读写 + snap = UpstreamRateSnapshot( + upstream_id=upstream.id, + snapshot_json="{}", + captured_at=datetime.now(timezone.utc) + ) + db_session.add(snap) + db_session.commit() + assert db_session.query(UpstreamRateSnapshot).count() == 1 + + # 插入一个初始的已导入 key，目前状态是 imported + key_row = UpstreamGeneratedKey( + upstream_id=upstream.id, + group_id="plus_group", + group_name="Plus智能分组", + key_name="SmartUp-TestUpstream-plus_group", + key_value="sk-plus-value", + managed_prefix="SmartUp", + key_id="key-12345", + status="imported", + imported_website_id=website.id, + imported_account_id="acc-999", + imported_target_group_id="target-3", + ) + db_session.add(key_row) + db_session.commit() + db_session.refresh(key_row) + + # 准备 Mock UpstreamClient + mock_keys = [{"id": "key-12345", "name": "SmartUp-TestUpstream-plus_group", "group_id": "plus_group"}] + monkeypatch.setattr(UpstreamClient, "list_api_keys", lambda self, **kw: mock_keys) + monkeypatch.setattr(UpstreamClient, "login", lambda self: None) + monkeypatch.setattr(UpstreamClient, "close", lambda self: None) + monkeypatch.setattr(UpstreamClient, "__enter__", lambda self: self) + monkeypatch.setattr(UpstreamClient, "__exit__", lambda self, *a: None) + monkeypatch.setattr(sched_mod, "SessionLocal", lambda: db_session) + original_close = db_session.close + monkeypatch.setattr(db_session, "close", lambda: None) + + try: + # === 场景 A: 单次异常快照缺少来源分组 === + # 运行对账，传入前一次快照存在该分组，当前快照不含该分组 + active_group_ids_1 = {"other_group"} + prev_active_group_ids_1 = {"plus_group", "other_group"} + remote_key_ids_1 = {"key-12345"} + captured_at_1 = datetime.now(timezone.utc) + + website_sync.reconcile_upstream_keys( + db_session, + upstream.id, + active_group_ids_1, + remote_key_ids_1, + captured_at_1, + prev_active_group_ids=prev_active_group_ids_1 + ) + db_session.commit() + + # 验证：单次异常快照缺少该分组，key 不应被标记为 orphaned + db_session.refresh(key_row) + assert key_row.status == "imported", "单次缺失分组不应该将 key 标为 orphaned" + assert key_row.error is None + + # === 场景 B: 两次连续快照都缺少来源分组 === + # 运行对账，传入前一个和当前快照都不含该分组 + active_group_ids_2 = {"other_group"} + prev_active_group_ids_2 = {"other_group"} + remote_key_ids_2 = {"key-12345"} + captured_at_2 = datetime.now(timezone.utc) + + website_sync.reconcile_upstream_keys( + db_session, + upstream.id, + active_group_ids_2, + remote_key_ids_2, + captured_at_2, + prev_active_group_ids=prev_active_group_ids_2 + ) + db_session.commit() + + # 验证：两次连续快照都缺少该分组，应标记为 orphaned + db_session.refresh(key_row) + assert key_row.status == "orphaned", "连续两次缺失分组应该将 key 标为 orphaned" + assert key_row.error == "来源分组已不存在" + + # === 场景 C: 自恢复逻辑 === + # 运行对账，传入当前快照分组已恢复，且远端 key 存在 + active_group_ids_3 = {"plus_group", "other_group"} + remote_key_ids_3 = {"key-12345"} + captured_at_3 = datetime.now(timezone.utc) + + website_sync.reconcile_upstream_keys( + db_session, + upstream.id, + active_group_ids_3, + remote_key_ids_3, + captured_at_3 + ) + db_session.commit() + + # 验证：分组恢复且远端 key id 存在时，应恢复为 imported 并清空错误 + db_session.refresh(key_row) + assert key_row.status == "imported", "恢复分组且远端 key 存在时应该恢复为 imported" + assert key_row.error is None + + # === 场景 D: 远端 key 列表拉取失败时，不做 key id 级清理或恢复 === + key_row.status = "orphaned" + key_row.error = "临时标记" + db_session.commit() + + active_group_ids_4 = {"plus_group", "other_group"} + remote_key_ids_4 = None + captured_at_4 = datetime.now(timezone.utc) + + website_sync.reconcile_upstream_keys( + db_session, + upstream.id, + active_group_ids_4, + remote_key_ids_4, + captured_at_4 + ) + db_session.commit() + + db_session.refresh(key_row) + assert key_row.status == "orphaned" + assert key_row.error == "临时标记" + + # === 场景 E: 重复旧 key 级防误恢复 === + # 创建一个重复旧 key (orphaned，error="重复旧 Key...", imported_account_id=None) + dup_key_row = UpstreamGeneratedKey( + upstream_id=upstream.id, + group_id="plus_group", + group_name="Plus智能分组", + key_name="SmartUp-TestUpstream-plus_group-dup", + key_value="sk-dup-value", + managed_prefix="SmartUp", + key_id="key-dup-999", + status="orphaned", + error="重复旧 Key；已由 key #73 对应当前分组导入账号", + imported_website_id=None, + imported_account_id=None, + imported_target_group_id=None, + ) + db_session.add(dup_key_row) + db_session.commit() + + # 运行对账，传入分组恢复且远端存在 key-dup-999 + active_group_ids_5 = {"plus_group", "other_group"} + remote_key_ids_5 = {"key-12345", "key-dup-999"} + captured_at_5 = datetime.now(timezone.utc) + + website_sync.reconcile_upstream_keys( + db_session, + upstream.id, + active_group_ids_5, + remote_key_ids_5, + captured_at_5 + ) + db_session.commit() + + # 验证：即使分组和 key_id 存在，因为它是重复旧 key 且 imported_account_id=None，也必须保持 orphaned 并且 error 不变 + db_session.refresh(dup_key_row) + assert dup_key_row.status == "orphaned", "重复旧 key 即使恢复也不应被误标记为 imported" + assert dup_key_row.error == "重复旧 Key；已由 key #73 对应当前分组导入账号" + + finally: + monkeypatch.setattr(db_session, "close", original_close) diff --git a/backend/test_website_client.py b/backend/test_website_client.py index 4ef7fae..ad39628 100644 --- a/backend/test_website_client.py +++ b/backend/test_website_client.py @@ -326,3 +326,76 @@ def test_list_accounts_pagination(): assert "page=2" in str(requests_called[1].url) finally: client.close() + + +def test_list_accounts_pagination_single_large_page_does_not_refetch(): + from app.services.website_client import Sub2ApiWebsiteClient + + requests_called = [] + + def handler(req: httpx.Request) -> httpx.Response: + requests_called.append(req) + return httpx.Response(200, json={ + "total": 2, + "page": 1, + "page_size": 100, + "pages": 1, + "data": [ + {"id": "a1", "name": "acc1"}, + {"id": "a2", "name": "acc2"}, + ], + }, request=req) + + client = Sub2ApiWebsiteClient( + base_url="https://target.example", + api_prefix="/api/v1/admin", + auth_type="api_key", + auth_config={"key": "admin-key"}, + ) + client._client = httpx.Client(transport=httpx.MockTransport(handler)) + + try: + accounts = client.list_accounts("/accounts") + assert accounts == [ + {"id": "a1", "name": "acc1"}, + {"id": "a2", "name": "acc2"}, + ] + assert len(requests_called) == 1 + assert "page_size=100" in str(requests_called[0].url) + finally: + client.close() + + +def test_list_accounts_pagination_returns_none_when_any_page_fails(): + from app.services.website_client import Sub2ApiWebsiteClient + + requests_called = [] + + def handler(req: httpx.Request) -> httpx.Response: + requests_called.append(req) + if "page=1" in str(req.url): + return httpx.Response(200, json={ + "total": 3, + "page": 1, + "page_size": 2, + "pages": 2, + "data": [ + {"id": "a1", "name": "acc1"}, + {"id": "a2", "name": "acc2"}, + ], + }, request=req) + return httpx.Response(500, json={"error": "failed"}, request=req) + + client = Sub2ApiWebsiteClient( + base_url="https://target.example", + api_prefix="/api/v1/admin", + auth_type="api_key", + auth_config={"key": "admin-key"}, + ) + client._client = httpx.Client(transport=httpx.MockTransport(handler)) + + try: + assert client.list_accounts("/accounts") is None + assert len(requests_called) == 2 + finally: + client.close()