fix(sync): 优先级同步前置过滤远端失效账号防排序污染，移除 locals 隐患，并将自愈结果纳入本地日志与摘要统计
This commit is contained in:
liumangmang
@@ -509,12 +509,15 @@ def reorder_account_priorities(
|
||||
failed_count = sum(1 for item in results if item.get("status") == "failed")
|
||||
success_count = sum(1 for item in results if item.get("status") == "success")
|
||||
skipped_count = sum(1 for item in results if item.get("status") == "skipped")
|
||||
stale_count = sum(1 for item in results if item.get("status") == "stale")
|
||||
|
||||
parts = []
|
||||
if success_count:
|
||||
parts.append(f"更新 {success_count}")
|
||||
if failed_count:
|
||||
parts.append(f"失败 {failed_count}")
|
||||
if stale_count:
|
||||
parts.append(f"清理失效账号 {stale_count}")
|
||||
if skipped_count:
|
||||
parts.append(f"跳过 {skipped_count}")
|
||||
message = "、".join(parts) + f" / 共 {len(results)} 个" if parts else "没有需要重排的账号"
|
||||
|
||||
@@ -225,7 +225,12 @@ class Sub2ApiWebsiteClient:
|
||||
|
||||
def update_account(self, account_id: str, body: dict[str, Any], endpoint: str = "/accounts") -> dict[str, Any]:
|
||||
"""更新远端账号（仅传入需要变更的字段）。"""
|
||||
resp = self._request("PUT", f"{endpoint}/{account_id}", body)
|
||||
try:
|
||||
resp = self._request("PUT", f"{endpoint}/{account_id}", body)
|
||||
except WebsiteError as exc:
|
||||
if isinstance(exc.__cause__, httpx.HTTPStatusError) and exc.__cause__.response.status_code == 404:
|
||||
raise WebsiteError(f"目标账号 {account_id} 不存在或已被删除") from exc.__cause__
|
||||
raise
|
||||
data = _unwrap_data(resp)
|
||||
return data if isinstance(data, dict) else {"value": data}
|
||||
|
||||
|
||||
@@ -472,74 +472,13 @@ def sync_account_priorities_for_upstream(
|
||||
# 不写日志、不发通知：网站不可用时账号无法更新，沉默跳过
|
||||
continue
|
||||
|
||||
# 查询该网站所有已导入 Key（跨上游），用于倍率查询
|
||||
all_website_keys = (
|
||||
db.query(UpstreamGeneratedKey)
|
||||
.filter(
|
||||
UpstreamGeneratedKey.imported_website_id == wid,
|
||||
UpstreamGeneratedKey.imported_account_id.isnot(None),
|
||||
UpstreamGeneratedKey.status != "orphaned",
|
||||
)
|
||||
.all()
|
||||
)
|
||||
_backfill_missing_target_groups_from_remote_accounts(db, website, all_website_keys)
|
||||
|
||||
# ── 按竞争分组分桶 ──────────────────────────────────────────────────
|
||||
# 竞争分组键：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:
|
||||
# 构建 "{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 map failed for website %s: %s", wid, exc)
|
||||
raw_rate_map = {}
|
||||
|
||||
# ── 每个竞争分组内独立计算 priority ────────────────────────────────
|
||||
# priority_assignment: account_id → new_priority
|
||||
priority_assignment: dict[str, int] = {}
|
||||
for comp_key, comp_rows in competitive_buckets.items():
|
||||
# 只保留快照中能查到倍率的账号；无数据的账号不参与排序
|
||||
rated = [
|
||||
(row, raw_rate_map[f"{row.upstream_id}:{row.group_id}"])
|
||||
for row in comp_rows
|
||||
if f"{row.upstream_id}:{row.group_id}" in raw_rate_map
|
||||
]
|
||||
# 过滤后有效账号不足 2 个 → 此分组无竞争意义，整组跳过
|
||||
if len(rated) < 2:
|
||||
logger.info(
|
||||
"skip competitive bucket %s for website %s: only %d account(s) have rate data",
|
||||
comp_key, wid, len(rated),
|
||||
)
|
||||
continue
|
||||
# 组内按倍率升序排序（倍率低 → priority 小 → 优先）
|
||||
unique_rates = sorted(set(r for _, r in rated))
|
||||
rate_to_prio = {rate: priority_for_rate_rank(idx) for idx, rate in enumerate(unique_rates)}
|
||||
for row, rate in rated:
|
||||
priority_assignment[row.imported_account_id] = rate_to_prio[rate]
|
||||
|
||||
# ── 调用 update_account（仅竞争分组的账号）───────────────────────
|
||||
# ── 1. 建立远端 client 连接并加载账号与分组列表 ────────────────────────
|
||||
account_info_map: dict[str, dict[str, Any]] = {}
|
||||
target_group_names: dict[str, str] = {}
|
||||
remote_account_ids: set[str] | None = None
|
||||
site_results: list[dict] = []
|
||||
competitive_buckets: dict[str, list] = {}
|
||||
priority_assignment: dict[str, int] = {}
|
||||
try:
|
||||
with Sub2ApiWebsiteClient(
|
||||
base_url=website.base_url,
|
||||
@@ -548,18 +487,19 @@ 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]] = {}
|
||||
target_group_names: dict[str, str] = {}
|
||||
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"),
|
||||
}
|
||||
remote_accounts = client.list_accounts()
|
||||
if remote_accounts is not None:
|
||||
remote_account_ids = set()
|
||||
for acc in remote_accounts:
|
||||
acc_id = _extract_id(acc)
|
||||
if acc_id:
|
||||
acc_id_str = str(acc_id)
|
||||
remote_account_ids.add(acc_id_str)
|
||||
account_info_map[acc_id_str] = {
|
||||
"name": acc.get("name") or acc.get("account_name") or acc_id_str,
|
||||
"priority": acc.get("priority"),
|
||||
}
|
||||
except Exception as e:
|
||||
logger.warning("failed to fetch remote accounts for website %s: %s", wid, e)
|
||||
|
||||
@@ -572,79 +512,201 @@ def sync_account_priorities_for_upstream(
|
||||
except Exception as e:
|
||||
logger.warning("failed to fetch groups for website %s at sync: %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}")
|
||||
# ── 2. 查询该网站所有已导入 Key（跨上游），用于倍率查询 ────────────
|
||||
all_website_keys = (
|
||||
db.query(UpstreamGeneratedKey)
|
||||
.filter(
|
||||
UpstreamGeneratedKey.imported_website_id == wid,
|
||||
UpstreamGeneratedKey.imported_account_id.isnot(None),
|
||||
UpstreamGeneratedKey.status != "orphaned",
|
||||
)
|
||||
.all()
|
||||
)
|
||||
_backfill_missing_target_groups_from_remote_accounts(db, website, all_website_keys)
|
||||
|
||||
# ── 3. 自愈校验：清除远端已不存在的账号，并从后续计算中排除 ──────────
|
||||
active_website_keys: list[UpstreamGeneratedKey] = []
|
||||
for row in all_website_keys:
|
||||
account_id = row.imported_account_id
|
||||
if remote_account_ids is not None and str(account_id) not in remote_account_ids:
|
||||
tg_id_str = str(row.imported_target_group_id or row.group_id)
|
||||
tg_name = target_group_names.get(tg_id_str)
|
||||
if tg_name and row.imported_target_group_name != tg_name:
|
||||
row.imported_target_group_name = tg_name
|
||||
db.commit()
|
||||
tg_name = target_group_names.get(tg_id_str) or row.imported_target_group_name
|
||||
site_results.append(
|
||||
_priority_result(
|
||||
row,
|
||||
None,
|
||||
"stale",
|
||||
"目标账号已不存在，已清除导入标记；请重新导入账号",
|
||||
account_name=account_id,
|
||||
old_priority=None,
|
||||
source_rate=None,
|
||||
target_group_name=tg_name,
|
||||
)
|
||||
)
|
||||
# 清理本地导入标记并恢复为 created
|
||||
row.imported_website_id = None
|
||||
row.imported_account_id = None
|
||||
row.imported_at = None
|
||||
row.status = "created"
|
||||
db.commit()
|
||||
else:
|
||||
active_website_keys.append(row)
|
||||
|
||||
try:
|
||||
client.update_account(account_id, {"priority": new_priority})
|
||||
# ── 4. 预取快照倍率 ────────────────────────────────────────────────
|
||||
raw_rate_map: dict[str, float] = {}
|
||||
if active_website_keys:
|
||||
all_upstream_ids = {k.upstream_id for k in active_website_keys}
|
||||
try:
|
||||
# 构建 "{upstream_id}:{group_id}" → rate 查询表
|
||||
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 map failed for website %s: %s", wid, exc)
|
||||
|
||||
# 补填 stale 账号结果的倍率
|
||||
for res in site_results:
|
||||
if res["status"] == "stale":
|
||||
res["source_rate"] = raw_rate_map.get(f"{res['upstream_id']}:{res['group_id']}")
|
||||
|
||||
# ── 5. 基于仍存在的账号分桶并校验竞争 ──────────────────────────
|
||||
buckets: dict[str, list[UpstreamGeneratedKey]] = defaultdict(list)
|
||||
for row in active_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,
|
||||
)
|
||||
else:
|
||||
# ── 6. 每个竞争分组内独立计算 priority ──────────────────────
|
||||
priority_assignment: dict[str, int] = {}
|
||||
for comp_key, comp_rows in competitive_buckets.items():
|
||||
# 只保留快照中能查到倍率的账号；无数据的账号不参与排序
|
||||
rated = [
|
||||
(row, raw_rate_map[f"{row.upstream_id}:{row.group_id}"])
|
||||
for row in comp_rows
|
||||
if f"{row.upstream_id}:{row.group_id}" in raw_rate_map
|
||||
]
|
||||
# 过滤后有效账号不足 2 个 → 此分组无竞争意义，整组跳过
|
||||
if len(rated) < 2:
|
||||
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,
|
||||
"skip competitive bucket %s for website %s: only %d account(s) have rate data",
|
||||
comp_key, wid, len(rated),
|
||||
)
|
||||
site_results.append(
|
||||
_priority_result(
|
||||
row,
|
||||
new_priority,
|
||||
"success",
|
||||
f"优先级已更新为 {new_priority}",
|
||||
account_name=acc_name,
|
||||
old_priority=old_prio,
|
||||
source_rate=src_rate,
|
||||
target_group_name=tg_name or row.imported_target_group_name,
|
||||
continue
|
||||
# 组内按倍率升序排序（倍率低 → priority 小 → 优先）
|
||||
unique_rates = sorted(set(r for _, r in rated))
|
||||
rate_to_prio = {rate: priority_for_rate_rank(idx) for idx, rate in enumerate(unique_rates)}
|
||||
for row, rate in rated:
|
||||
priority_assignment[row.imported_account_id] = rate_to_prio[rate]
|
||||
|
||||
# ── 7. 更新竞争账号的 priority ──────────────────────────────
|
||||
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}")
|
||||
|
||||
tg_id_str = str(row.imported_target_group_id or row.group_id)
|
||||
tg_name = target_group_names.get(tg_id_str)
|
||||
if tg_name and row.imported_target_group_name != tg_name:
|
||||
row.imported_target_group_name = tg_name
|
||||
db.commit()
|
||||
|
||||
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,
|
||||
)
|
||||
)
|
||||
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),
|
||||
account_name=acc_name,
|
||||
old_priority=old_prio,
|
||||
source_rate=src_rate,
|
||||
target_group_name=tg_name or row.imported_target_group_name,
|
||||
site_results.append(
|
||||
_priority_result(
|
||||
row,
|
||||
new_priority,
|
||||
"success",
|
||||
f"优先级已更新为 {new_priority}",
|
||||
account_name=acc_name,
|
||||
old_priority=old_prio,
|
||||
source_rate=src_rate,
|
||||
target_group_name=tg_name or row.imported_target_group_name,
|
||||
)
|
||||
)
|
||||
except Exception as exc:
|
||||
# 兜底清理：如果 PUT 404
|
||||
import httpx
|
||||
is_404 = False
|
||||
if isinstance(exc, WebsiteError) and "不存在或已被删除" in str(exc):
|
||||
is_404 = True
|
||||
elif isinstance(exc, WebsiteError) and isinstance(exc.__cause__, httpx.HTTPStatusError) and exc.__cause__.response.status_code == 404:
|
||||
is_404 = True
|
||||
|
||||
if is_404:
|
||||
site_results.append(
|
||||
_priority_result(
|
||||
row,
|
||||
new_priority,
|
||||
"stale",
|
||||
"目标账号已不存在，已清除导入标记；请重新导入账号",
|
||||
account_name=acc_name,
|
||||
old_priority=old_prio,
|
||||
source_rate=src_rate,
|
||||
target_group_name=tg_name or row.imported_target_group_name,
|
||||
)
|
||||
)
|
||||
row.imported_website_id = None
|
||||
row.imported_account_id = None
|
||||
row.imported_at = None
|
||||
row.status = "created"
|
||||
db.commit()
|
||||
continue
|
||||
|
||||
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),
|
||||
account_name=acc_name,
|
||||
old_priority=old_prio,
|
||||
source_rate=src_rate,
|
||||
target_group_name=tg_name or row.imported_target_group_name,
|
||||
)
|
||||
)
|
||||
)
|
||||
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}",
|
||||
source_rate=src_rate,
|
||||
)
|
||||
for row in rows:
|
||||
site_results.append(
|
||||
_priority_result(
|
||||
row,
|
||||
None,
|
||||
"failed",
|
||||
f"连接网站失败: {exc}",
|
||||
)
|
||||
)
|
||||
|
||||
# 只发送有实际成功/失败的通知（不包含单账号跳过项）
|
||||
# ── 8. 写入日志与钉钉通知 ──────────────────────────────────────────
|
||||
notify_updates = [r for r in site_results if r["status"] in ("success", "failed")]
|
||||
if notify_updates:
|
||||
if notify_updates or any(r["status"] == "stale" for r in site_results):
|
||||
# 构建简化的 priority_map 供日志参考（只包含竞争分组的账号）
|
||||
priority_map_snapshot = {
|
||||
f"{row.upstream_id}:{row.group_id}": priority_assignment.get(row.imported_account_id)
|
||||
@@ -652,7 +714,8 @@ def sync_account_priorities_for_upstream(
|
||||
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)
|
||||
if notify_updates:
|
||||
_try_send_priority_webhook(db, wid, website.name, upstream_id, upstream_name, notify_updates)
|
||||
|
||||
all_results.extend(site_results)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user