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

feat: 修正优先级自动计算算法为分目标分组独立排序，并优化一键分组整理完成后仅触发一次网站级优先级重排

Browse Source
This commit is contained in:
liumangmang
2026-07-01 20:19:54 +08:00
parent fd46492012
commit dd3649a6ad
4 changed files with 536 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/website_sync.py
+336 -269
View File
@@ -432,12 +432,61 @@ def _try_send_priority_webhook(
logger.warning("account_priority_changed webhook failed for website %s: %s", wid, exc)
def sync_account_priorities_for_upstream(
def build_target_group_priority_map(
db: Session,
upstream_id: int,
website_id: int | None = None,
target_group_sources: dict[str, list[tuple[int, str]]],
) -> dict[tuple[str, int, str], int]:
"""根据目标分组内的上游分组倍率构建 (target_group_id, upstream_id, group_id) → priority 映射。
每个目标分组独立排序，倍率低到高对应 1, 11, 21, 31...
相同倍率共享同一 priority。
找不到倍率快照的来源不参与排序。
"""
priority_map: dict[tuple[str, int, str], int] = {}
# 提取所有涉及的 upstream_id
all_upstream_ids = set()
for sources in target_group_sources.values():
for upstream_id, _ in sources:
all_upstream_ids.add(upstream_id)
# 预载最新倍率快照
rate_maps = {}
for uid in all_upstream_ids:
try:
rate_maps[uid] = latest_rate_map(db, uid)
except Exception:
rate_maps[uid] = {}
for tg_id, sources in target_group_sources.items():
# 收集有有效倍率的来源
rated_sources: list[tuple[tuple[int, str], float]] = []
for upstream_id, group_id in sources:
groups = rate_maps.get(upstream_id) or {}
g = groups.get(group_id)
if isinstance(g, dict):
rate = _snapshot_group_rate(g)
rated_sources.append(((upstream_id, group_id), rate))
if not rated_sources:
continue
# 独立排序并赋予 priority (间隔 10)
unique_rates = sorted(set(r for _, r in rated_sources))
rate_to_priority = {rate: priority_for_rate_rank(idx) for idx, rate in enumerate(unique_rates)}
for (upstream_id, group_id), rate in rated_sources:
priority_map[(str(tg_id), upstream_id, str(group_id))] = rate_to_priority[rate]
return priority_map
def sync_account_priorities_for_website(
db: Session,
website_id: int,
trigger_upstream_id: int | None = None,
) -> list[dict]:
"""上游倍率变化后，自动更新已导入下游账号的 priority。
"""计算并同步某个网站上所有账号的优先级。
只处理同一目标分组内有多个账号（存在竞争）的情况：
- 竞争分组键：imported_target_group_id（老数据 fallback 到 group_id）
@@ -447,6 +496,284 @@ def sync_account_priorities_for_upstream(
"""
from collections import defaultdict
website = db.query(Website).filter(Website.id == website_id).first()
if not website or not website.enabled:
logger.info("skip account priority sync: website %s not found or disabled", website_id)
return []
if trigger_upstream_id is not None:
upstream_name = db.query(Upstream.name).filter(Upstream.id == trigger_upstream_id).scalar() or f"#{trigger_upstream_id}"
else:
upstream_name = "分组整理/一键整理"
# ── 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,
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:
try:
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", website_id, e)
try:
grps = client.get_groups(endpoint=website.groups_endpoint) or []
for g in grps:
g_id = g.get("id")
if g_id is not None:
target_group_names[str(g_id)] = g.get("name") or str(g_id)
except Exception as e:
logger.warning("failed to fetch groups for website %s at sync: %s", website_id, e)
# ── 2. 查询该网站所有已导入 Key（跨上游），用于倍率查询 ────────────
all_website_keys = (
db.query(UpstreamGeneratedKey)
.filter(
UpstreamGeneratedKey.imported_website_id == website_id,
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) 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)
# ── 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", website_id, 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)",
website_id,
)
else:
# ── 6. 每个竞争分组内独立计算 priority ──────────────────────
priority_assignment = {}
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, website_id, 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]
# ── 7. 更新竞争账号 the 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, website_id, 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}",
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, website_id, 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", website_id, exc)
key_query = db.query(UpstreamGeneratedKey).filter(
UpstreamGeneratedKey.imported_website_id == website_id,
UpstreamGeneratedKey.imported_account_id.isnot(None),
UpstreamGeneratedKey.status != "orphaned",
)
if trigger_upstream_id is not None:
key_query = key_query.filter(UpstreamGeneratedKey.upstream_id == trigger_upstream_id)
rows_to_fail = key_query.all()
for row in rows_to_fail:
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 or any(r["status"] == "stale" for r in site_results):
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
if priority_assignment.get(row.imported_account_id) is not None
}
_write_priority_sync_log_with_map(db, website_id, upstream_name, site_results, priority_map_snapshot)
_try_send_priority_webhook(
db, website_id, website.name,
trigger_upstream_id or 0, upstream_name,
notify_updates,
)
return site_results
def sync_account_priorities_for_upstream(
db: Session,
upstream_id: int,
website_id: int | None = None,
) -> list[dict]:
"""上游倍率变化后，自动更新已导入下游账号的 priority。
"""
key_query = db.query(UpstreamGeneratedKey).filter(
UpstreamGeneratedKey.upstream_id == upstream_id,
UpstreamGeneratedKey.imported_website_id.isnot(None),
@@ -459,271 +786,11 @@ def sync_account_priorities_for_upstream(
if not key_rows:
return []
upstream_name = db.query(Upstream.name).filter(Upstream.id == upstream_id).scalar() or f"#{upstream_id}"
# 按 imported_website_id 分组
website_groups: dict[int, list[UpstreamGeneratedKey]] = {}
for row in key_rows:
wid = row.imported_website_id
website_groups.setdefault(wid, []).append(row)
all_results: list[dict] = []
for wid, rows in website_groups.items():
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)
# 不写日志、不发通知：网站不可用时账号无法更新，沉默跳过
continue
# ── 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,
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:
try:
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)
try:
grps = client.get_groups(endpoint=website.groups_endpoint) or []
for g in grps:
g_id = g.get("id")
if g_id is not None:
target_group_names[str(g_id)] = g.get("name") or str(g_id)
except Exception as e:
logger.warning("failed to fetch groups for website %s at sync: %s", wid, e)
# ── 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) 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)
# ── 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(
"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]
# ── 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,
)
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 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 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)
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)
if notify_updates:
_try_send_priority_webhook(db, wid, website.name, upstream_id, upstream_name, notify_updates)
all_results.extend(site_results)
return all_results
wids = sorted(list({row.imported_website_id for row in key_rows}))
results = []
for wid in wids:
results.extend(sync_account_priorities_for_website(db, wid, trigger_upstream_id=upstream_id))
return results