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

feat: track imported accounts per upstream key across platforms

Browse Source 
Add upstream_key_account_links table mapping generated keys to remote
accounts per website/platform, surface imported_accounts on key
responses, and update website sync/routers to manage the links.
This commit is contained in:
liujing
2026-07-23 17:52:58 +08:00
parent ee89cd6119
commit df7e400e7b
17 changed files with 1081 additions and 537 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/website_client.py
+6 -2
View File
@@ -134,13 +134,17 @@ def normalize_groups(value: Any) -> list[dict[str, Any]]:
name = item.get("name") or item.get("group_name") or str(gid)
rate = item.get("rate_multiplier") or item.get("rateMultiplier") or item.get("ratio")
desc = item.get("description") or item.get("desc") or item.get("remark")
groups.append({
platform = item.get("platform")
group = {
"id": str(gid),
"name": str(name),
"rate_multiplier": fixed_decimal_string(rate, 2) if rate is not None else None,
"description": str(desc) if desc is not None else None,
"raw": item,
})
}
if platform is not None:
group["platform"] = str(platform)
groups.append(group)
return groups
backend/app/services/website_sync.py
+279 -59
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import json
import logging
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from datetime import datetime, timezone
from typing import Any
@@ -11,7 +12,7 @@ from sqlalchemy.orm import Session
from app.database import SessionLocal
from app.models.snapshot import UpstreamRateSnapshot
from app.models.upstream import Upstream
from app.models.upstream_key import UpstreamGeneratedKey
from app.models.upstream_key import UpstreamGeneratedKey, UpstreamKeyAccountLink
from app.models.website import Website, WebsiteGroupBinding, WebsiteSyncLog
from app.services.auth_config import normalize_auth_config
from app.services.website_client import Sub2ApiWebsiteClient, WebsiteError, _extract_id, calculate_target_rate, parse_positive_decimal
@@ -22,10 +23,219 @@ from app.utils.number import fixed_decimal_string
logger = logging.getLogger(__name__)
def build_imported_account_name(upstream_name: str, group_name: str | None, group_id: str, key_id: int) -> str:
"""构建导入的账号名称：{上游网站名}-{上游分组名或分组ID}-{KeyID}"""
@dataclass
class LegacyAccountLink:
"""Read-only compatibility view used only when a key has no mapping rows."""
upstream_key_id: int
website_id: int
target_group: str
remote_account_id: str
platform: str = ""
status: str = "active"
def get_account_mapping_pairs(
db: Session,
website_id: int,
*,
include_orphaned: bool = True,
) -> list[tuple[UpstreamGeneratedKey, UpstreamKeyAccountLink | LegacyAccountLink]]:
query = db.query(UpstreamGeneratedKey, UpstreamKeyAccountLink).join(
UpstreamKeyAccountLink, UpstreamKeyAccountLink.upstream_key_id == UpstreamGeneratedKey.id,
).filter(
UpstreamKeyAccountLink.website_id == website_id,
UpstreamKeyAccountLink.status == "active",
)
if not include_orphaned:
query = query.filter(UpstreamGeneratedKey.status != "orphaned")
pairs = list(query.all())
mapped_key_ids = {row.id for row, _ in pairs}
legacy_filters = [
UpstreamGeneratedKey.imported_website_id == website_id,
UpstreamGeneratedKey.imported_account_id.isnot(None),
]
if mapped_key_ids:
legacy_filters.append(~UpstreamGeneratedKey.id.in_(mapped_key_ids))
legacy_query = db.query(UpstreamGeneratedKey).filter(*legacy_filters)
if not include_orphaned:
legacy_query = legacy_query.filter(UpstreamGeneratedKey.status != "orphaned")
for row in legacy_query.all():
pairs.append((row, LegacyAccountLink(
upstream_key_id=row.id,
website_id=website_id,
target_group=str(row.imported_target_group_id or row.group_id),
remote_account_id=str(row.imported_account_id),
)))
return pairs
def deactivate_account_mapping(
db: Session,
key: UpstreamGeneratedKey,
link: UpstreamKeyAccountLink | LegacyAccountLink,
) -> None:
if isinstance(link, UpstreamKeyAccountLink):
link.status = "stale"
db.commit()
sync_key_legacy_fields(db, key.id)
return
key.imported_website_id = None
key.imported_account_id = None
key.imported_at = None
key.imported_target_group_id = None
key.imported_target_group_name = None
if key.status == "imported":
key.status = "created"
db.commit()
def _normalize_platform(platform: str | None) -> str:
if not platform:
return ""
p = str(platform).lower().strip()
if p in ("codex", "gpt", "openai"):
return "openai"
if p in ("claude", "anthropic"):
return "anthropic"
if p in ("xai", "grok"):
return "grok"
return p
def build_imported_account_name(
upstream_name: str,
group_name: str | None,
group_id: str,
key_id: int,
platform: str | None = None,
) -> str:
"""构建导入的账号名称：{上游网站名}-{上游分组名或分组ID}-{KeyID}（含平台后缀）。"""
grp = group_name if group_name else group_id
return f"{upstream_name}-{grp}-{key_id}"
base = f"{upstream_name}-{grp}-{key_id}"
norm_p = _normalize_platform(platform) if platform else None
if norm_p:
return f"{base}-{norm_p}"
return base
def sync_key_legacy_fields(db: Session, key_id: int) -> None:
"""将 UpstreamKeyAccountLink 中的首条活跃链接同步到 UpstreamGeneratedKey 的 legacy 字段。"""
from app.models.upstream_key import UpstreamKeyAccountLink
db.flush()
key = db.query(UpstreamGeneratedKey).filter(UpstreamGeneratedKey.id == key_id).first()
if not key:
return
active_links = (
db.query(UpstreamKeyAccountLink)
.filter(
UpstreamKeyAccountLink.upstream_key_id == key_id,
UpstreamKeyAccountLink.status == "active",
)
.order_by(
UpstreamKeyAccountLink.website_id.asc(),
UpstreamKeyAccountLink.platform.asc(),
UpstreamKeyAccountLink.id.asc(),
)
.all()
)
if active_links:
first = active_links[0]
key.imported_website_id = first.website_id
key.imported_account_id = first.remote_account_id
key.imported_target_group_id = first.target_group
# 查找绑定以获取目标分组名称
binding = (
db.query(WebsiteGroupBinding)
.filter(
WebsiteGroupBinding.website_id == first.website_id,
WebsiteGroupBinding.target_group_id == first.target_group,
)
.first()
)
if binding and binding.target_group_name:
key.imported_target_group_name = binding.target_group_name
elif not key.imported_target_group_name:
key.imported_target_group_name = first.target_group
if not key.imported_at:
key.imported_at = first.created_at or datetime.now(timezone.utc)
key.status = "imported"
else:
key.imported_website_id = None
key.imported_account_id = None
key.imported_target_group_id = None
key.imported_target_group_name = None
key.imported_at = None
if key.status in ("imported", "import_failed"):
key.status = "created"
db.commit()
def get_or_backfill_account_links(
db: Session,
website_id: int,
remote_account_map: dict[str, dict[str, Any]] | None = None,
target_group_map: dict[str, dict[str, Any]] | None = None,
) -> list[UpstreamKeyAccountLink]:
"""Backfill legacy markers only when account and target-group platforms agree."""
existing_links = (
db.query(UpstreamKeyAccountLink)
.filter(UpstreamKeyAccountLink.website_id == website_id)
.all()
)
linked_account_ids = {l.remote_account_id for l in existing_links}
# 查询只有 legacy imported_account_id 的 Key
legacy_keys = (
db.query(UpstreamGeneratedKey)
.filter(
UpstreamGeneratedKey.imported_website_id == website_id,
UpstreamGeneratedKey.imported_account_id.isnot(None),
)
.all()
)
created_any = False
for k in legacy_keys:
if k.imported_account_id in linked_account_ids:
continue
remote_acc = remote_account_map.get(str(k.imported_account_id)) if remote_account_map else None
target_group_id = str(k.imported_target_group_id or "")
target_group = target_group_map.get(target_group_id) if target_group_map and target_group_id else None
if not remote_acc or not target_group:
continue
remote_p = _normalize_platform(remote_acc.get("platform"))
target_p = _normalize_platform(target_group.get("platform"))
supported = {"openai", "anthropic", "gemini", "grok", "antigravity"}
if remote_p not in supported or target_p not in supported or remote_p != target_p:
continue
link = UpstreamKeyAccountLink(
upstream_key_id=k.id,
website_id=website_id,
target_group=target_group_id,
platform=remote_p,
remote_account_id=str(k.imported_account_id),
status="active",
)
db.add(link)
linked_account_ids.add(str(k.imported_account_id))
created_any = True
if created_any:
db.commit()
existing_links = (
db.query(UpstreamKeyAccountLink)
.filter(UpstreamKeyAccountLink.website_id == website_id)
.all()
)
return existing_links
PRIORITY_BASE = 1
@@ -286,13 +496,15 @@ def _priority_result(
old_priority: Any = None,
source_rate: float | None = None,
target_group_name: str | None = None,
link: UpstreamKeyAccountLink | LegacyAccountLink | None = None,
) -> dict:
"""构建统一的优先级同步结果 dict，支持丰富的展示和后向兼容性。"""
tg_id = row.imported_target_group_id or row.group_id
tg_id = link.target_group if link else (row.imported_target_group_id or row.group_id)
tg_name = target_group_name or row.imported_target_group_name or tg_id
account_id = link.remote_account_id if link else row.imported_account_id
return {
# 兼容旧字段
"account_id": row.imported_account_id,
"account_id": account_id,
"group_id": row.group_id,
"upstream_id": row.upstream_id,
"old_priority": old_priority,
@@ -302,7 +514,7 @@ def _priority_result(
# 新增增强型字段
"target_group_id": tg_id,
"target_group_name": tg_name,
"account_name": account_name or row.imported_account_id,
"account_name": account_name or account_id,
"source_group_id": row.group_id,
"source_group_name": row.group_name or row.group_id,
"source_rate": source_rate,
@@ -546,6 +758,8 @@ def sync_account_priorities_for_website(
# ── 1. 建立远端 client 连接并加载账号与分组列表 ────────────────────────
account_info_map: dict[str, dict[str, Any]] = {}
target_group_names: dict[str, str] = {}
remote_account_map: dict[str, dict[str, Any]] = {}
target_group_map: dict[str, dict[str, Any]] = {}
remote_account_ids: set[str] | None = None
site_results: list[dict] = []
competitive_buckets: dict[str, list] = {}
@@ -572,6 +786,7 @@ def sync_account_priorities_for_website(
"name": acc.get("name") or acc.get("account_name") or acc_id_str,
"priority": acc.get("priority"),
}
remote_account_map[acc_id_str] = acc
except Exception as e:
logger.warning("failed to fetch remote accounts for website %s: %s", website_id, e)
@@ -580,28 +795,40 @@ def sync_account_priorities_for_website(
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)
group_id = str(g_id)
target_group_names[group_id] = g.get("name") or group_id
target_group_map[group_id] = g
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",
legacy_missing_groups = db.query(UpstreamGeneratedKey).filter(
UpstreamGeneratedKey.imported_website_id == website_id,
UpstreamGeneratedKey.imported_account_id.isnot(None),
UpstreamGeneratedKey.imported_target_group_id.is_(None),
).all()
for row in legacy_missing_groups:
group_id, group_name = _remote_account_single_group(
remote_account_map.get(str(row.imported_account_id)) or {}
)
.all()
)
_backfill_missing_target_groups_from_remote_accounts(db, website, all_website_keys)
if group_id:
row.imported_target_group_id = group_id
row.imported_target_group_name = group_name or target_group_names.get(group_id)
if legacy_missing_groups:
db.commit()
get_or_backfill_account_links(db, website_id, remote_account_map, target_group_map)
# ── 2. 查询该网站所有有效账号映射（跨上游），用于倍率查询 ─────────
all_account_pairs = get_account_mapping_pairs(db, website_id, include_orphaned=False)
# ── 3. 自愈校验：清除远端已不存在的账号，并从后续计算中排除 ──────────
active_website_keys: list[UpstreamGeneratedKey] = []
for row in all_website_keys:
account_id = row.imported_account_id
active_account_pairs: list[
tuple[UpstreamGeneratedKey, UpstreamKeyAccountLink | LegacyAccountLink]
] = []
for row, link in all_account_pairs:
account_id = link.remote_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_id_str = str(link.target_group)
tg_name = target_group_names.get(tg_id_str) or row.imported_target_group_name
site_results.append(
_priority_result(
@@ -613,21 +840,17 @@ def sync_account_priorities_for_website(
old_priority=None,
source_rate=None,
target_group_name=tg_name,
link=link,
)
)
# 清理本地导入标记并恢复为 created
row.imported_website_id = None
row.imported_account_id = None
row.imported_at = None
row.status = "created"
db.commit()
deactivate_account_mapping(db, row, link)
else:
active_website_keys.append(row)
active_account_pairs.append((row, link))
# ── 4. 预取快照倍率 ────────────────────────────────────────────────
raw_rate_map: dict[str, float] = {}
if active_website_keys:
all_upstream_ids = {k.upstream_id for k in active_website_keys}
if active_account_pairs:
all_upstream_ids = {row.upstream_id for row, _ in active_account_pairs}
try:
# 构建 "{upstream_id}:{group_id}" → rate 查询表
for uid in all_upstream_ids:
@@ -644,10 +867,12 @@ def sync_account_priorities_for_website(
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)
buckets: dict[
str,
list[tuple[UpstreamGeneratedKey, UpstreamKeyAccountLink | LegacyAccountLink]],
] = defaultdict(list)
for row, link in active_account_pairs:
buckets[link.target_group].append((row, link))
# 只保留账号数 > 1 的分组（有竞争才需要排序）
competitive_buckets = {k: v for k, v in buckets.items() if len(v) > 1}
@@ -663,8 +888,8 @@ def sync_account_priorities_for_website(
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
((row, link), raw_rate_map[f"{row.upstream_id}:{row.group_id}"])
for row, link in comp_rows
if f"{row.upstream_id}:{row.group_id}" in raw_rate_map
]
# 过滤后有效账号不足 2 个 → 此分组无竞争意义，整组跳过
@@ -677,13 +902,13 @@ def sync_account_priorities_for_website(
# 组内按倍率升序排序（倍率低 → 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]
for (row, link), rate in rated:
priority_assignment[link.remote_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
for row, link in comp_rows:
account_id = link.remote_account_id
new_priority = priority_assignment.get(account_id)
if new_priority is None:
continue
@@ -693,7 +918,7 @@ def sync_account_priorities_for_website(
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_id_str = str(link.target_group)
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
@@ -705,7 +930,7 @@ def sync_account_priorities_for_website(
"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,
link.target_group, new_priority,
)
site_results.append(
_priority_result(
@@ -717,6 +942,7 @@ def sync_account_priorities_for_website(
old_priority=old_prio,
source_rate=src_rate,
target_group_name=tg_name or row.imported_target_group_name,
link=link,
)
)
except Exception as exc:
@@ -739,13 +965,10 @@ def sync_account_priorities_for_website(
old_priority=old_prio,
source_rate=src_rate,
target_group_name=tg_name or row.imported_target_group_name,
link=link,
)
)
row.imported_website_id = None
row.imported_account_id = None
row.imported_at = None
row.status = "created"
db.commit()
deactivate_account_mapping(db, row, link)
continue
logger.warning(
@@ -762,25 +985,22 @@ def sync_account_priorities_for_website(
old_priority=old_prio,
source_rate=src_rate,
target_group_name=tg_name or row.imported_target_group_name,
link=link,
)
)
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",
)
rows_to_fail = get_account_mapping_pairs(db, website_id, include_orphaned=False)
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:
rows_to_fail = [pair for pair in rows_to_fail if pair[0].upstream_id == trigger_upstream_id]
for row, link in rows_to_fail:
site_results.append(
_priority_result(
row,
None,
"failed",
f"连接网站失败: {exc}",
link=link,
)
)
@@ -788,10 +1008,10 @@ def sync_account_priorities_for_website(
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)
f"{row.upstream_id}:{row.group_id}": priority_assignment.get(link.remote_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
for row, link in comp_rows
if priority_assignment.get(link.remote_account_id) is not None
}
_write_priority_sync_log_with_map(db, website_id, upstream_name, site_results, priority_map_snapshot)