From fe40fcd4c99ec3c7ebcc5d0853c30cec744a6c3e Mon Sep 17 00:00:00 2001 From: liumangmang Date: Wed, 1 Jul 2026 09:00:29 +0800 Subject: [PATCH] feat(priority): group priority changed notifications by target group and fetch remote account details --- backend/app/services/website_sync.py | 73 ++++++++++++-- backend/app/utils/dingtalk.py | 38 ++++++- backend/test_priority_sync.py | 145 +++++++++++++++++++++++++++ 3 files changed, 246 insertions(+), 10 deletions(-) diff --git a/backend/app/services/website_sync.py b/backend/app/services/website_sync.py index 2aaac33..41cba74 100644 --- a/backend/app/services/website_sync.py +++ b/backend/app/services/website_sync.py @@ -239,16 +239,34 @@ def build_rate_priority_map(db: Session, upstream_ids: set[int]) -> dict[str, in return {key: rate_to_priority[rate] for key, rate in group_rates.items()} -def _priority_result(row, new_priority: int | None, status: str, message: str) -> dict: - """构建统一的优先级同步结果 dict。""" +def _priority_result( + row, + new_priority: int | None, + status: str, + message: str, + account_name: str | None = None, + old_priority: Any = None, + source_rate: float | None = None, +) -> dict: + """构建统一的优先级同步结果 dict，支持丰富的展示和后向兼容性。""" + tg_id = row.imported_target_group_id or row.group_id + tg_name = row.imported_target_group_name or tg_id return { + # 兼容旧字段 "account_id": row.imported_account_id, "group_id": row.group_id, "upstream_id": row.upstream_id, - "old_priority": None, + "old_priority": old_priority, "new_priority": new_priority, "status": status, "message": message, + # 新增增强型字段 + "target_group_id": tg_id, + "target_group_name": tg_name, + "account_name": account_name or row.imported_account_id, + "source_group_id": row.group_id, + "source_group_name": row.group_name or row.group_id, + "source_rate": source_rate, } @@ -529,12 +547,32 @@ def sync_account_priorities_for_upstream( auth_config=json.loads(website.auth_config_json or "{}"), timeout=float(website.timeout_seconds), ) as client: + # 尽量拉取目标网站账号列表以获取账号名称和旧 priority，失败不阻断 + account_info_map: dict[str, dict[str, Any]] = {} + try: + remote_accounts = client.list_accounts() or [] + for acc in remote_accounts: + acc_id = _extract_id(acc) + if acc_id: + account_info_map[str(acc_id)] = { + "name": acc.get("name") or acc.get("account_name") or str(acc_id), + "priority": acc.get("priority"), + } + except Exception as e: + logger.warning("failed to fetch remote accounts for website %s: %s", wid, e) + for comp_rows in competitive_buckets.values(): for row in comp_rows: account_id = row.imported_account_id new_priority = priority_assignment.get(account_id) if new_priority is None: continue + + acc_info = account_info_map.get(str(account_id)) or {} + acc_name = acc_info.get("name") or str(account_id) + old_prio = acc_info.get("priority") + src_rate = raw_rate_map.get(f"{row.upstream_id}:{row.group_id}") + try: client.update_account(account_id, {"priority": new_priority}) logger.info( @@ -544,7 +582,15 @@ def sync_account_priorities_for_upstream( row.imported_target_group_id or row.group_id, new_priority, ) site_results.append( - _priority_result(row, new_priority, "success", f"优先级已更新为 {new_priority}") + _priority_result( + row, + new_priority, + "success", + f"优先级已更新为 {new_priority}", + account_name=acc_name, + old_priority=old_prio, + source_rate=src_rate, + ) ) except Exception as exc: logger.warning( @@ -552,14 +598,29 @@ def sync_account_priorities_for_upstream( account_id, wid, exc, ) site_results.append( - _priority_result(row, new_priority, "failed", str(exc)) + _priority_result( + row, + new_priority, + "failed", + str(exc), + account_name=acc_name, + old_priority=old_prio, + source_rate=src_rate, + ) ) except Exception as exc: logger.warning("failed to connect website %s for account priority sync: %s", wid, exc) for comp_rows in competitive_buckets.values(): for row in comp_rows: + src_rate = raw_rate_map.get(f"{row.upstream_id}:{row.group_id}") site_results.append( - _priority_result(row, None, "failed", f"连接网站失败: {exc}") + _priority_result( + row, + None, + "failed", + f"连接网站失败: {exc}", + source_rate=src_rate, + ) ) # 只发送有实际成功/失败的通知（不包含单账号跳过项） diff --git a/backend/app/utils/dingtalk.py b/backend/app/utils/dingtalk.py index 81a0a8b..9af6416 100644 --- a/backend/app/utils/dingtalk.py +++ b/backend/app/utils/dingtalk.py @@ -79,11 +79,41 @@ def format_dingtalk_priority_changed( f"- **摘要**：{success} 更新 / {failed} 失败 / {skipped} 跳过", "", ] + + # 按目标分组聚合 + from collections import defaultdict + groups = defaultdict(list) for u in updates: - emoji = {"success": "✅", "failed": "❌", "skipped": "⏭️"}.get(u.get("status", ""), "➖") - gid = u.get("group_id", "?") - priority = u.get("new_priority", "—") - lines.append(f"{emoji} `{gid}` → priority={priority}") + tg_name = u.get("target_group_name") or u.get("target_group_id") or u.get("group_id") or "未知目标分组" + groups[tg_name].append(u) + + for tg_name, items in groups.items(): + lines.append(f"目标分组：{tg_name}") + for u in items: + emoji = {"success": "✅", "failed": "❌", "skipped": "⏭️"}.get(u.get("status", ""), "➖") + acc_name = u.get("account_name") or u.get("account_id") or "未知账号" + acc_id = u.get("account_id") or "未知ID" + + old_p = u.get("old_priority") + old_p_str = "未知" if old_p in (None, "", "未知") else str(old_p) + + new_p = u.get("new_priority") + new_p_str = "未知" if new_p in (None, "", "未知") else str(new_p) + + src_group = u.get("source_group_name") or u.get("source_group_id") or "未知来源" + + rate = u.get("source_rate") + rate_str = str(rate) if rate is not None else "—" + + line = f"- {emoji} {acc_name}（账号ID {acc_id}）：old={old_p_str} → new={new_p_str}，来源分组={src_group}，倍率={rate_str}" + if u.get("status") == "failed" and u.get("message"): + line += f"，原因={u.get('message')}" + lines.append(line) + lines.append("") + + while lines and not lines[-1]: + lines.pop() + return { "msgtype": "markdown", "markdown": { diff --git a/backend/test_priority_sync.py b/backend/test_priority_sync.py index 490f0d2..e82e5d7 100644 --- a/backend/test_priority_sync.py +++ b/backend/test_priority_sync.py @@ -533,3 +533,148 @@ def test_reorder_priority_endpoint_scopes_to_current_website(db_session, monkeyp updated = {account_id: payload["priority"] for account_id, payload in update_calls} assert response.success is True assert updated == {"W1A1": 1, "W1A2": 11} + + +def test_priority_sync_notification_grouped_by_target_group(): + """测试 DingTalk Markdown 通知按目标分组正确聚合。""" + from app.utils.dingtalk import format_dingtalk_priority_changed + + updates = [ + { + "target_group_id": "TG1", + "target_group_name": "PRO智能分组", + "account_id": "A123", + "account_name": "SmartUp-PRO-xxx", + "source_group_id": "sg1", + "source_group_name": "28", + "source_rate": 0.8, + "old_priority": 11, + "new_priority": 1, + "status": "success", + "message": "success", + }, + { + "target_group_id": "TG1", + "target_group_name": "PRO智能分组", + "account_id": "A456", + "account_name": "SmartUp-Claude-xxx", + "source_group_id": "sg2", + "source_group_name": "claude max渠道", + "source_rate": 1.2, + "old_priority": 21, + "new_priority": 11, + "status": "failed", + "message": "xxx", + }, + { + "target_group_id": "TG2", + "target_group_name": "Basic智能分组", + "account_id": "A789", + "account_name": "SmartUp-Basic-xxx", + "source_group_id": "sg3", + "source_group_name": "basic渠道", + "source_rate": 2.0, + "old_priority": "未知", + "new_priority": 31, + "status": "success", + "message": "success", + } + ] + + msg = format_dingtalk_priority_changed("SmartSite", "Upstream1", "2026-07-01 09:00:00", updates) + text = msg["markdown"]["text"] + + # 验证分组标题 + assert "目标分组：PRO智能分组" in text + assert "目标分组：Basic智能分组" in text + + # 验证账号明细 + assert "- ✅ SmartUp-PRO-xxx（账号ID A123）：old=11 → new=1，来源分组=28，倍率=0.8" in text + assert "- ❌ SmartUp-Claude-xxx（账号ID A456）：old=21 → new=11，来源分组=claude max渠道，倍率=1.2，原因=xxx" in text + assert "- ✅ SmartUp-Basic-xxx（账号ID A789）：old=未知 → new=31，来源分组=basic渠道，倍率=2.0" in text + + +def test_priority_sync_fetches_remote_accounts_and_backfills_info(db_session, monkeypatch): + """验证优先级同步能获取并回填远端账号的名称和旧优先级。""" + w = Website(name="W1", base_url="http://w1", enabled=True, + auth_config_json="{}", timeout_seconds=30) + u = Upstream(name="U1", base_url="http://u1") + db_session.add_all([w, u]) + db_session.commit() + db_session.refresh(w); db_session.refresh(u) + + _make_snapshot(db_session, u.id, {"G1": 1.0, "G2": 2.0}) + + _make_key(db_session, u.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1") + _make_key(db_session, u.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1") + + update_calls = [] + + class CustomMockClient: + def __init__(self, **kwargs): pass + def __enter__(self): return self + def __exit__(self, *a): pass + def list_accounts(self): + return [ + {"id": "A1", "name": "SmartUp-A1-Name", "priority": 15}, + {"id": "A2", "name": "SmartUp-A2-Name", "priority": 25}, + ] + def update_account(self, account_id, data): + update_calls.append((account_id, data)) + + monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", CustomMockClient) + + results = sync_account_priorities_for_upstream(db_session, u.id, website_id=w.id) + assert len(results) == 2 + + # 校验 site_results 中的丰富信息是否填充 + res_a1 = next(r for r in results if r["account_id"] == "A1") + assert res_a1["account_name"] == "SmartUp-A1-Name" + assert res_a1["old_priority"] == 15 + assert res_a1["new_priority"] == 1 + assert res_a1["source_rate"] == 1.0 + assert res_a1["target_group_id"] == "TG1" + assert res_a1["target_group_name"] == "TG1" # fallback to tg_id since target_group_name is not set + + res_a2 = next(r for r in results if r["account_id"] == "A2") + assert res_a2["account_name"] == "SmartUp-A2-Name" + assert res_a2["old_priority"] == 25 + assert res_a2["new_priority"] == 11 + assert res_a2["source_rate"] == 2.0 + + +def test_priority_sync_graceful_on_remote_list_failure(db_session, monkeypatch): + """验证 list_accounts 抛异常时，优先级同步依然成功更新，且各项降级为账号 ID 和未知。""" + w = Website(name="W1", base_url="http://w1", enabled=True, + auth_config_json="{}", timeout_seconds=30) + u = Upstream(name="U1", base_url="http://u1") + db_session.add_all([w, u]) + db_session.commit() + db_session.refresh(w); db_session.refresh(u) + + _make_snapshot(db_session, u.id, {"G1": 1.0, "G2": 2.0}) + + _make_key(db_session, u.id, "G1", "K1", "V1", w.id, "A1", imported_target_group_id="TG1") + _make_key(db_session, u.id, "G2", "K2", "V2", w.id, "A2", imported_target_group_id="TG1") + + class CustomMockClient: + def __init__(self, **kwargs): pass + def __enter__(self): return self + def __exit__(self, *a): pass + def list_accounts(self): + raise Exception("Remote server error 500") + def update_account(self, account_id, data): + pass + + monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", CustomMockClient) + + results = sync_account_priorities_for_upstream(db_session, u.id, website_id=w.id) + assert len(results) == 2 + + # 校验降级逻辑 + res_a1 = next(r for r in results if r["account_id"] == "A1") + assert res_a1["account_name"] == "A1" + assert res_a1["old_priority"] is None + assert res_a1["new_priority"] == 1 + assert res_a1["status"] == "success" +