LiuMangMang/SmartUp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(priority-sync): narrow account priority update to competitive groups only

Browse Source 
Root cause: sync_account_priorities_for_upstream() was doing a global
priority re-rank across ALL imported accounts on a website whenever any
upstream rate changed, triggering spurious account_priority_changed
notifications for accounts in different target groups with no competition.

Fix:
- Add imported_target_group_id / imported_target_group_name to
  UpstreamGeneratedKey (nullable; old data falls back to group_id)
- Writ imported_target_group_id on account import in websites.py
- Rewrite sync_account_priorities_for_upstream():
  * bucket accounts by competition_group = imported_target_group_id or group_id
  * only process buckets with count > 1 (genuine competition)
  * each competitive bucket independently sorted by rate; priority starts at 1
  * single-account groups: completely skipped (no update_account, no notification)
  * no competitive groups at all: early return, no log, no notification
- Remove auto priority update in re-import idempotency path (was also
  incorrect; now fully delegated to sync_account_priorities_for_upstream)
- Fix Sub2ApiWebsiteClient local import in sync fn → use module-level name
  so monkeypatch works correctly in tests

Tests: rewrite test_priority_sync.py
- REMOVED: test_priority_sync_full_website_update (was asserting the buggy behavior)
- NEW: test_no_update_when_different_groups_single_account_each
- NEW: test_same_target_group_two_accounts_updated
- NEW: test_two_target_groups_independent_priority
- NEW: test_old_data_null_target_group_fallback
- NEW: test_single_account_in_mixed_website
- UPDATED: test_priority_sync_log_structure (now requires competitive group)
- KEPT: test_priority_sync_cross_upstream_group, test_import_auto_priority_by_rate

All 25 tests pass (8 priority_sync + 17 existing upstream tests).
This commit is contained in:
liumangmang
2026-06-01 19:13:14 +08:00
parent 871557e4ae
commit e519d1804b
5 changed files with 415 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/database.py
+4
View File
@@ -150,6 +150,10 @@ def _migrate_upstream_generated_keys():
conn.execute(text("UPDATE upstream_generated_keys SET updated_at = created_at WHERE updated_at IS NULL"))
if "managed_prefix" not in columns:
conn.execute(text("ALTER TABLE upstream_generated_keys ADD COLUMN managed_prefix VARCHAR(64)"))
if "imported_target_group_id" not in columns:
conn.execute(text("ALTER TABLE upstream_generated_keys ADD COLUMN imported_target_group_id VARCHAR(255)"))
if "imported_target_group_name" not in columns:
conn.execute(text("ALTER TABLE upstream_generated_keys ADD COLUMN imported_target_group_name VARCHAR(255)"))
# ——— 历史数据迁移：回填 managed_prefix + 清理重复 ———
with engine.begin() as conn:
backend/app/models/upstream_key.py
+2
View File
@@ -25,6 +25,8 @@ class UpstreamGeneratedKey(Base):
imported_website_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey("websites.id", ondelete="SET NULL"), nullable=True, index=True)
imported_account_id: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
imported_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
imported_target_group_id: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
imported_target_group_name: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at: Mapped[datetime] = mapped_column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
backend/app/routers/websites.py
+3 -11
View File
@@ -558,16 +558,6 @@ def import_upstream_keys_as_accounts(
old_account_id = row.imported_account_id
exists = c.account_exists(row.imported_account_id)
if exists is True:
# 自动更新已有账号的 priority（分步导入时全局倍率排序可能已变）
new_priority = rate_priority_map.get(f"{row.upstream_id}:{row.group_id}") if body.auto_priority_by_rate else None
priority_msg = "已导入过，已跳过"
if new_priority is not None:
try:
c.update_account(old_account_id, {"priority": new_priority})
priority_msg = f"已导入过，优先级已更新为 {new_priority}"
except Exception as exc:
logger.warning("update priority failed account=%s: %s", old_account_id, exc)
priority_msg = f"已导入过，优先级更新失败: {exc}"
items.append(ImportAccountItem(
upstream_key_id=row.id,
source_group_id=row.group_id,
@@ -578,7 +568,7 @@ def import_upstream_keys_as_accounts(
platform=platform,
upstream_base_url=upstream_base_url,
status="exists",
message=priority_msg,
message="已导入过，已跳过",
))
continue
elif exists is False:
@@ -639,6 +629,8 @@ def import_upstream_keys_as_accounts(
row.imported_website_id = wid
row.imported_account_id = account_id or None
row.imported_at = datetime.now(timezone.utc)
row.imported_target_group_id = target_group_id or None
row.imported_target_group_name = None # target_group_map 只存 ID，name 展示用可留 NULL
row.status = "imported"
row.error = None
db.commit()
backend/app/services/website_sync.py
+96 -70
View File
@@ -288,12 +288,13 @@ def _try_send_priority_webhook(
def sync_account_priorities_for_upstream(db: Session, upstream_id: int) -> list[dict]:
"""上游倍率变化后，自动更新已导入下游账号的 priority。
查询该上游下所有已导入（非 orphaned）的 Key，按目标网站分组后重新计算全局优先级，
并通过 update_account API 推送到下游网站。返回详细结果列表。
同时写入 WebsiteSyncLog 持久化审计日志，并通过 webhook 发送通知。
只处理同一目标分组内有多个账号（存在竞争）的情况：
- 竞争分组键：imported_target_group_id（老数据 fallback 到 group_id）
- 同一竞争分组内按倍率升序排序，priority 从 1 开始（相同倍率共享）
- 单账号分组：完全跳过，不调用 update_account，不发通知
- 无竞争分组：直接返回，不写日志，不发通知
"""
from app.services.website_client import Sub2ApiWebsiteClient as Client
from collections import defaultdict
key_rows = (
db.query(UpstreamGeneratedKey)
@@ -314,9 +315,7 @@ def sync_account_priorities_for_upstream(db: Session, upstream_id: int) -> list[
website_groups: dict[int, list[UpstreamGeneratedKey]] = {}
for row in key_rows:
wid = row.imported_website_id
if wid not in website_groups:
website_groups[wid] = []
website_groups[wid].append(row)
website_groups.setdefault(wid, []).append(row)
all_results: list[dict] = []
@@ -324,16 +323,10 @@ def sync_account_priorities_for_upstream(db: Session, upstream_id: int) -> list[
website = db.query(Website).filter(Website.id == wid).first()
if not website or not website.enabled:
logger.info("skip account priority sync: website %s not found or disabled", wid)
site_results = []
for row in rows:
r = _priority_result(row, None, "failed", "网站不可用")
site_results.append(r)
all_results.append(r)
_write_priority_sync_log_with_map(db, wid, upstream_name, site_results, {})
_try_send_priority_webhook(db, wid, "", upstream_id, upstream_name, site_results)
# 不写日志、不发通知：网站不可用时账号无法更新，沉默跳过
continue
# 查询该网站所有已导入 Key（跨上游），实现全局优先级排序
# 查询该网站所有已导入 Key（跨上游），用于倍率查询
all_website_keys = (
db.query(UpstreamGeneratedKey)
.filter(
@@ -343,81 +336,114 @@ def sync_account_priorities_for_upstream(db: Session, upstream_id: int) -> list[
)
.all()
)
# ── 按竞争分组分桶 ──────────────────────────────────────────────────
# 竞争分组键：imported_target_group_id（老数据为 NULL 时 fallback 到 group_id）
buckets: dict[str, list[UpstreamGeneratedKey]] = defaultdict(list)
for row in all_website_keys:
comp_key = row.imported_target_group_id or row.group_id
buckets[comp_key].append(row)
# 只保留账号数 > 1 的分组（有竞争才需要排序）
competitive_buckets = {k: v for k, v in buckets.items() if len(v) > 1}
if not competitive_buckets:
logger.info(
"skip account priority sync for website %s: no competitive groups (all single-account)",
wid,
)
continue # 不写日志，不发通知
# ── 预取快照倍率 ────────────────────────────────────────────────────
all_upstream_ids = {k.upstream_id for k in all_website_keys}
try:
priority_map = build_rate_priority_map(db, all_upstream_ids)
# 构建 "{upstream_id}:{group_id}" → rate 查询表
raw_rate_map: dict[str, float] = {}
for uid in all_upstream_ids:
groups = latest_rate_map(db, uid)
for gid, g in groups.items():
if isinstance(g, dict):
raw_rate_map[f"{uid}:{gid}"] = _snapshot_group_rate(g)
except Exception as exc:
logger.warning("build_rate_priority_map failed for website %s: %s", wid, exc)
site_results = []
for row in all_website_keys:
r = _priority_result(row, None, "failed", f"构建优先级映射失败: {exc}")
site_results.append(r)
all_results.append(r)
_write_priority_sync_log_with_map(db, wid, upstream_name, site_results, {})
_try_send_priority_webhook(db, wid, "", upstream_id, upstream_name, site_results)
continue
logger.warning("build rate map failed for website %s: %s", wid, exc)
raw_rate_map = {}
if not priority_map:
logger.info("skip account priority sync for website %s: empty priority map", wid)
site_results = []
for row in all_website_keys:
r = _priority_result(row, None, "skipped", "无上游倍率数据")
site_results.append(r)
all_results.append(r)
_write_priority_sync_log_with_map(db, wid, upstream_name, site_results, {})
_try_send_priority_webhook(db, wid, "", upstream_id, upstream_name, site_results)
continue
# ── 每个竞争分组内独立计算 priority ────────────────────────────────
# priority_assignment: account_id → new_priority
priority_assignment: dict[str, int] = {}
for comp_key, comp_rows in competitive_buckets.items():
# 取每行的倍率（查不到则 fallback 1.0）
rated = [
(row, raw_rate_map.get(f"{row.upstream_id}:{row.group_id}", 1.0))
for row in comp_rows
]
# 组内按倍率升序排序（倍率低 → priority 小 → 优先）
unique_rates = sorted(set(r for _, r in rated))
rate_to_prio = {rate: idx + 1 for idx, rate in enumerate(unique_rates)}
for row, rate in rated:
priority_assignment[row.imported_account_id] = rate_to_prio[rate]
# ── 调用 update_account（仅竞争分组的账号）───────────────────────
site_results: list[dict] = []
try:
with Client(
with Sub2ApiWebsiteClient(
base_url=website.base_url,
api_prefix=website.api_prefix,
auth_type=website.auth_type,
auth_config=json.loads(website.auth_config_json or "{}"),
timeout=float(website.timeout_seconds),
) as client:
for row in all_website_keys:
account_id = row.imported_account_id
if not account_id:
continue
new_priority = priority_map.get(f"{row.upstream_id}:{row.group_id}")
if new_priority is None:
site_results.append(
_priority_result(row, None, "skipped", "无倍率数据，跳过")
)
continue
try:
client.update_account(account_id, {"priority": new_priority})
logger.info(
"updated priority for account %s (website=%s, upstream=%s, group=%s): %s",
account_id, wid, row.upstream_id, row.group_id, new_priority,
)
site_results.append(
_priority_result(row, new_priority, "success", f"优先级已更新为 {new_priority}")
)
except Exception as exc:
logger.warning(
"failed to update priority for account %s (website=%s): %s",
account_id, wid, exc,
)
site_results.append(
_priority_result(row, new_priority, "failed", str(exc))
)
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
try:
client.update_account(account_id, {"priority": new_priority})
logger.info(
"updated priority for account %s (website=%s, upstream=%s, group=%s"
", comp_group=%s): %s",
account_id, wid, row.upstream_id, row.group_id,
row.imported_target_group_id or row.group_id, new_priority,
)
site_results.append(
_priority_result(row, new_priority, "success", f"优先级已更新为 {new_priority}")
)
except Exception as exc:
logger.warning(
"failed to update priority for account %s (website=%s): %s",
account_id, wid, exc,
)
site_results.append(
_priority_result(row, new_priority, "failed", str(exc))
)
except Exception as exc:
logger.warning("failed to connect website %s for account priority sync: %s", wid, exc)
for row in all_website_keys:
site_results.append(
_priority_result(row, None, "failed", f"连接网站失败: {exc}")
)
for comp_rows in competitive_buckets.values():
for row in comp_rows:
site_results.append(
_priority_result(row, None, "failed", f"连接网站失败: {exc}")
)
# 只发送有实际成功/失败的通知（不包含单账号跳过项）
notify_updates = [r for r in site_results if r["status"] in ("success", "failed")]
if notify_updates:
# 构建简化的 priority_map 供日志参考（只包含竞争分组的账号）
priority_map_snapshot = {
f"{row.upstream_id}:{row.group_id}": priority_assignment.get(row.imported_account_id)
for comp_rows in competitive_buckets.values()
for row in comp_rows
}
_write_priority_sync_log_with_map(db, wid, upstream_name, site_results, priority_map_snapshot)
_try_send_priority_webhook(db, wid, website.name, upstream_id, upstream_name, notify_updates)
all_results.extend(site_results)
_write_priority_sync_log_with_map(db, wid, upstream_name, site_results, priority_map)
_try_send_priority_webhook(db, wid, website.name, upstream_id, upstream_name, site_results)
return all_results
def _fetch_remote_managed_prefixes(db: Session, upstream_id: int) -> list[str]:
"""查询本地 distinct managed_prefix。