1025 lines
44 KiB
Python
1025 lines
44 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from decimal import Decimal, ROUND_HALF_UP
|
||
from datetime import datetime, timezone
|
||
from typing import Any
|
||
|
||
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.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
|
||
from app.services.upstream_client import UpstreamClient
|
||
from app.services import webhook_service
|
||
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}"""
|
||
grp = group_name if group_name else group_id
|
||
return f"{upstream_name}-{grp}-{key_id}"
|
||
|
||
|
||
PRIORITY_BASE = 1
|
||
PRIORITY_STEP = 10
|
||
|
||
|
||
def _persist_upstream_auth_config(upstream: Upstream, updated_config: dict[str, Any]) -> None:
|
||
config_json = json.dumps(
|
||
normalize_auth_config(upstream.auth_type, updated_config),
|
||
ensure_ascii=False,
|
||
)
|
||
updated_at = datetime.now(timezone.utc)
|
||
db = SessionLocal()
|
||
try:
|
||
row = db.query(Upstream).filter(Upstream.id == upstream.id).first()
|
||
if row:
|
||
row.auth_config_json = config_json
|
||
row.updated_at = updated_at
|
||
db.commit()
|
||
upstream.auth_config_json = row.auth_config_json
|
||
upstream.updated_at = updated_at
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def priority_for_rate_rank(rank: int) -> int:
|
||
"""Convert a zero-based sorted rate rank to an account priority."""
|
||
return PRIORITY_BASE + rank * PRIORITY_STEP
|
||
|
||
|
||
def binding_sources(binding: WebsiteGroupBinding) -> list[dict[str, Any]]:
|
||
try:
|
||
data = json.loads(binding.source_groups_json or "[]")
|
||
except Exception:
|
||
return []
|
||
return data if isinstance(data, list) else []
|
||
|
||
|
||
def latest_rate_map(db: Session, upstream_id: int) -> dict[str, Any]:
|
||
row = (
|
||
db.query(UpstreamRateSnapshot)
|
||
.filter(UpstreamRateSnapshot.upstream_id == upstream_id)
|
||
.order_by(UpstreamRateSnapshot.captured_at.desc())
|
||
.first()
|
||
)
|
||
if not row:
|
||
return {}
|
||
snapshot = json.loads(row.snapshot_json or "{}")
|
||
groups = snapshot.get("groups") or {}
|
||
return groups if isinstance(groups, dict) else {}
|
||
|
||
|
||
def get_affected_bindings(db: Session, changes: list[dict[str, Any]], upstream_id: int) -> list[WebsiteGroupBinding]:
|
||
changed_ids = {str(change.get("group_id")) for change in changes if change.get("group_id") is not None}
|
||
if not changed_ids:
|
||
return []
|
||
result: list[WebsiteGroupBinding] = []
|
||
bindings = db.query(WebsiteGroupBinding).filter(WebsiteGroupBinding.enabled == True).all()
|
||
for binding in bindings:
|
||
for source in binding_sources(binding):
|
||
if int(source.get("upstream_id") or 0) == upstream_id and str(source.get("group_id")) in changed_ids:
|
||
result.append(binding)
|
||
break
|
||
return result
|
||
|
||
|
||
def _client_for(website: Website) -> Sub2ApiWebsiteClient:
|
||
return 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),
|
||
target_id=website.id,
|
||
target_name=website.name,
|
||
)
|
||
|
||
|
||
def _log(
|
||
db: Session,
|
||
binding: WebsiteGroupBinding,
|
||
website: Website,
|
||
source_rates: list[dict[str, Any]],
|
||
status: str,
|
||
message: str,
|
||
old_rate: Any = None,
|
||
new_rate: Any = None,
|
||
effective_percent: str = None,
|
||
) -> WebsiteSyncLog:
|
||
row = WebsiteSyncLog(
|
||
website_id=website.id,
|
||
binding_id=binding.id,
|
||
target_group_id=binding.target_group_id,
|
||
target_group_name=binding.target_group_name,
|
||
algorithm=binding.algorithm,
|
||
percent=effective_percent if effective_percent is not None else binding.percent,
|
||
source_rates_json=json.dumps(source_rates, ensure_ascii=False),
|
||
old_rate=fixed_decimal_string(old_rate, 2) if old_rate not in (None, "") else None,
|
||
new_rate=fixed_decimal_string(new_rate, 2) if new_rate not in (None, "") else None,
|
||
status=status,
|
||
message=message,
|
||
)
|
||
db.add(row)
|
||
db.commit()
|
||
db.refresh(row)
|
||
return row
|
||
|
||
|
||
def sync_binding(db: Session, binding: WebsiteGroupBinding, write: bool = True) -> WebsiteSyncLog:
|
||
website = db.query(Website).filter(Website.id == binding.website_id).first()
|
||
if not website:
|
||
raise WebsiteError("网站不存在")
|
||
sources = binding_sources(binding)
|
||
# ── 批量预查：收集所有上游 ID，一次查询上游名称 ──
|
||
upstream_ids = {int(s.get("upstream_id") or 0) for s in sources if s.get("upstream_id")}
|
||
upstreams = {}
|
||
if upstream_ids:
|
||
rows = db.query(Upstream).filter(Upstream.id.in_(upstream_ids)).all()
|
||
upstreams = {u.id: u for u in rows}
|
||
# ── 同一轮 sync 内的快照缓存（调用级，函数返回即释放）──
|
||
_snap_cache: dict[int, dict[str, Any]] = {}
|
||
|
||
def _get_snap(upstream_id: int) -> dict[str, Any]:
|
||
if upstream_id not in _snap_cache:
|
||
_snap_cache[upstream_id] = latest_rate_map(db, upstream_id)
|
||
return _snap_cache[upstream_id]
|
||
|
||
source_rates: list[dict[str, Any]] = []
|
||
for source in sources:
|
||
upstream_id = int(source.get("upstream_id") or 0)
|
||
group_id = str(source.get("group_id") or "")
|
||
groups = _get_snap(upstream_id)
|
||
group = groups.get(group_id) if group_id else None
|
||
upstream = upstreams.get(upstream_id)
|
||
source_rates.append({
|
||
"upstream_id": upstream_id,
|
||
"upstream_name": source.get("upstream_name") or (upstream.name if upstream else ""),
|
||
"group_id": group_id,
|
||
"group_name": source.get("group_name") or (group.get("group_name", "") if isinstance(group, dict) else ""),
|
||
"rate": group.get("rate") if isinstance(group, dict) else None,
|
||
})
|
||
try:
|
||
target_rate = calculate_target_rate([item.get("rate") for item in source_rates], binding.percent, binding.algorithm)
|
||
auto_percent_str = None
|
||
if binding.algorithm == "priority_weighted_plus_percent":
|
||
rates = [rate for rate in (parse_positive_decimal(item.get("rate")) for item in source_rates) if rate is not None]
|
||
if rates:
|
||
rates_sorted = sorted(rates)
|
||
N = len(rates_sorted)
|
||
if N == 1:
|
||
base = rates_sorted[0]
|
||
else:
|
||
w1 = Decimal("0.85")
|
||
w_others = Decimal("0.15") / Decimal(N - 1)
|
||
base = rates_sorted[0] * w1 + sum(rates_sorted[1:], Decimal("0")) * w_others
|
||
|
||
if base > 0:
|
||
effective_percent = (target_rate / base - Decimal("1")) * Decimal("100")
|
||
auto_percent_str = fixed_decimal_string(effective_percent.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), 2)
|
||
except Exception as exc:
|
||
return _log(db, binding, website, source_rates, "failed", str(exc))
|
||
|
||
old_rate = None
|
||
if write and website.enabled and website.auto_sync_enabled and binding.enabled:
|
||
try:
|
||
with _client_for(website) as client:
|
||
groups = client.get_groups(website.groups_endpoint)
|
||
target = next((item for item in groups if item.get("id") == binding.target_group_id), None)
|
||
old_rate = target.get("rate_multiplier") if target else None
|
||
client.update_group_rate(
|
||
website.group_update_endpoint,
|
||
binding.target_group_id,
|
||
target_rate,
|
||
)
|
||
if auto_percent_str is not None:
|
||
binding.percent = auto_percent_str
|
||
website.last_status = "healthy"
|
||
website.last_error = None
|
||
except Exception as exc:
|
||
website.last_status = "unhealthy"
|
||
website.last_error = str(exc)
|
||
db.commit()
|
||
return _log(db, binding, website, source_rates, "failed", f"写回失败：{exc}", old_rate, target_rate, auto_percent_str)
|
||
db.commit()
|
||
log = _log(db, binding, website, source_rates, "success", "同步成功", old_rate, target_rate, auto_percent_str)
|
||
old_rate_str = fixed_decimal_string(old_rate, 2) if old_rate not in (None, "") else None
|
||
new_rate_str = fixed_decimal_string(target_rate, 2)
|
||
if old_rate_str != new_rate_str:
|
||
webhook_service.send_website_rate_changed(
|
||
db,
|
||
website.id,
|
||
website.name,
|
||
website.base_url,
|
||
binding.id,
|
||
binding.target_group_id,
|
||
binding.target_group_name,
|
||
old_rate_str,
|
||
new_rate_str,
|
||
source_rates,
|
||
)
|
||
return log
|
||
|
||
message = "已计算建议倍率，未写回"
|
||
if not website.enabled or not website.auto_sync_enabled:
|
||
message = "网站未启用自动同步，未写回"
|
||
elif not binding.enabled:
|
||
message = "绑定未启用，未写回"
|
||
return _log(db, binding, website, source_rates, "success", message, old_rate, target_rate)
|
||
|
||
|
||
def _snapshot_group_rate(group: dict) -> float:
|
||
"""从快照分组数据中提取倍率（兼容多个字段名）。"""
|
||
raw = group.get("rate") or group.get("default_rate") or group.get("rate_multiplier") or 1
|
||
try:
|
||
return float(raw)
|
||
except (TypeError, ValueError):
|
||
return 1.0
|
||
|
||
|
||
def _strict_snapshot_group_rate(group: dict) -> float | None:
|
||
"""从快照分组数据中严格提取倍率，若缺失或非数值则返回 None。"""
|
||
raw = group.get("rate") or group.get("default_rate") or group.get("rate_multiplier")
|
||
if raw is None or raw == "":
|
||
return None
|
||
try:
|
||
return float(raw)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def build_rate_priority_map(db: Session, upstream_ids: set[int]) -> dict[str, int]:
|
||
"""根据上游分组倍率构建 f"{upstream_id}:{group_id}" → priority 映射。
|
||
|
||
使用 (upstream_id, group_id) 复合键避免不同上游的同名分组互相覆盖。
|
||
遍历所有涉及的上游的最新快照，收集分组的倍率，按倍率升序排列后赋值 priority。
|
||
倍率最低的 priority=1，次低的 priority=11，以此类推。相同倍率的分组共享同一 priority。
|
||
"""
|
||
group_rates: dict[str, float] = {}
|
||
for uid in upstream_ids:
|
||
groups = latest_rate_map(db, uid)
|
||
for gid, g in groups.items():
|
||
if not isinstance(g, dict):
|
||
continue
|
||
rate = _snapshot_group_rate(g)
|
||
key = f"{uid}:{gid}"
|
||
group_rates[key] = rate
|
||
unique_rates = sorted(set(group_rates.values()))
|
||
rate_to_priority = {rate: priority_for_rate_rank(idx) for idx, rate in enumerate(unique_rates)}
|
||
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,
|
||
account_name: str | None = None,
|
||
old_priority: Any = None,
|
||
source_rate: float | None = None,
|
||
target_group_name: str | None = None,
|
||
) -> dict:
|
||
"""构建统一的优先级同步结果 dict，支持丰富的展示和后向兼容性。"""
|
||
tg_id = row.imported_target_group_id or row.group_id
|
||
tg_name = target_group_name or 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": 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,
|
||
}
|
||
|
||
|
||
def _remote_account_single_group(account: dict[str, Any]) -> tuple[str | None, str | None]:
|
||
"""Return the only remote group id/name if the account belongs to exactly one group."""
|
||
group_ids: list[str] = []
|
||
group_names: dict[str, str] = {}
|
||
|
||
raw_group_ids = account.get("group_ids")
|
||
if isinstance(raw_group_ids, list):
|
||
for item in raw_group_ids:
|
||
group_id = item.get("id") if isinstance(item, dict) else item
|
||
if group_id is not None:
|
||
group_ids.append(str(group_id))
|
||
elif raw_group_ids is not None:
|
||
group_ids.append(str(raw_group_ids))
|
||
|
||
raw_groups = account.get("groups")
|
||
if isinstance(raw_groups, list):
|
||
for group in raw_groups:
|
||
if not isinstance(group, dict):
|
||
continue
|
||
group_id = group.get("id") or group.get("group_id")
|
||
if group_id is None:
|
||
continue
|
||
gid = str(group_id)
|
||
group_ids.append(gid)
|
||
if group.get("name") or group.get("group_name"):
|
||
group_names[gid] = str(group.get("name") or group.get("group_name"))
|
||
|
||
unique_ids = list(dict.fromkeys(gid for gid in group_ids if gid))
|
||
if len(unique_ids) != 1:
|
||
return None, None
|
||
gid = unique_ids[0]
|
||
return gid, group_names.get(gid)
|
||
|
||
|
||
def _backfill_missing_target_groups_from_remote_accounts(
|
||
db: Session,
|
||
website: Website,
|
||
rows: list[UpstreamGeneratedKey],
|
||
) -> int:
|
||
"""Backfill old imported rows whose target group is missing from the remote account list."""
|
||
missing_rows = [
|
||
row for row in rows
|
||
if row.imported_account_id and not row.imported_target_group_id
|
||
]
|
||
if not missing_rows:
|
||
return 0
|
||
|
||
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),
|
||
target_id=website.id,
|
||
target_name=website.name,
|
||
) as client:
|
||
accounts = client.list_accounts() or []
|
||
except Exception as exc:
|
||
logger.info("skip imported target group backfill for website %s: %s", website.id, exc)
|
||
return 0
|
||
|
||
account_map: dict[str, dict[str, Any]] = {}
|
||
for account in accounts:
|
||
account_id = _extract_id(account)
|
||
if account_id:
|
||
account_map[str(account_id)] = account
|
||
|
||
changed = 0
|
||
for row in missing_rows:
|
||
account = account_map.get(str(row.imported_account_id))
|
||
if not account:
|
||
continue
|
||
group_id, group_name = _remote_account_single_group(account)
|
||
if not group_id:
|
||
logger.info(
|
||
"skip imported target group backfill for account %s: remote account has zero or multiple groups",
|
||
row.imported_account_id,
|
||
)
|
||
continue
|
||
row.imported_target_group_id = group_id
|
||
row.imported_target_group_name = group_name
|
||
changed += 1
|
||
|
||
if changed:
|
||
db.commit()
|
||
logger.info("backfilled %d imported target group(s) for website %s", changed, website.id)
|
||
return changed
|
||
|
||
|
||
def _write_priority_sync_log_with_map(
|
||
db: Session, wid: int, upstream_name: str,
|
||
results: list[dict], priority_map: dict[str, int],
|
||
) -> None:
|
||
"""写入 priority_sync 日志，同时保存账号明细和 priority_map 快照。
|
||
|
||
source_rates_json 格式：[{"_meta": "priority_map", "data": {...}}, {"account_id": ..., ...}, ...]
|
||
兼容 WebsiteSyncLogResponse.source_rates: list[dict] 类型约束。
|
||
"""
|
||
log_results: list[dict] = [
|
||
{"_meta": "priority_map", "data": dict(priority_map)},
|
||
]
|
||
log_results.extend(results)
|
||
success = sum(1 for r in results if r["status"] == "success")
|
||
failed = sum(1 for r in results if r["status"] == "failed")
|
||
skipped = sum(1 for r in results if r["status"] == "skipped")
|
||
parts = []
|
||
if success:
|
||
parts.append(f"{success} 个更新成功")
|
||
if failed:
|
||
parts.append(f"{failed} 个失败")
|
||
if skipped:
|
||
parts.append(f"{skipped} 个跳过")
|
||
log = WebsiteSyncLog(
|
||
website_id=wid,
|
||
binding_id=None,
|
||
target_group_id="",
|
||
target_group_name="",
|
||
algorithm="priority_sync",
|
||
percent=0,
|
||
source_rates_json=json.dumps(log_results, ensure_ascii=False, default=str),
|
||
old_rate=None,
|
||
new_rate=None,
|
||
status="failed" if failed else "success",
|
||
message=f"优先级同步（上游={upstream_name}）：{'、'.join(parts)} / 共 {len(results)} 个",
|
||
)
|
||
db.add(log)
|
||
db.commit()
|
||
|
||
|
||
def _try_send_priority_webhook(
|
||
db: Session, wid: int, website_name: str,
|
||
upstream_id: int, upstream_name: str,
|
||
updates: list[dict],
|
||
) -> None:
|
||
"""发送 account_priority_changed webhook，失败不抛异常。"""
|
||
if not updates:
|
||
return
|
||
# 如果没传入名称，尝试从 DB 查
|
||
resolved_name = website_name
|
||
if not resolved_name:
|
||
row = db.query(Website.name).filter(Website.id == wid).first()
|
||
if row:
|
||
resolved_name = row[0]
|
||
else:
|
||
resolved_name = f"网站#{wid}"
|
||
try:
|
||
webhook_service.send_account_priority_changed(
|
||
db,
|
||
website_id=wid,
|
||
website_name=resolved_name,
|
||
upstream_id=upstream_id,
|
||
upstream_name=upstream_name,
|
||
updates=updates,
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("account_priority_changed webhook failed for website %s: %s", wid, exc)
|
||
|
||
|
||
def build_target_group_priority_map(
|
||
db: Session,
|
||
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 = _strict_snapshot_group_rate(g)
|
||
if rate is not None:
|
||
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]:
|
||
"""计算并同步某个网站上所有账号的优先级。
|
||
|
||
只处理同一目标分组内有多个账号（存在竞争）的情况：
|
||
- 竞争分组键：imported_target_group_id（老数据 fallback 到 group_id）
|
||
- 同一竞争分组内按倍率升序排序，priority 从 1 开始，每档间隔 10（相同倍率共享）
|
||
- 单账号分组：完全跳过，不调用 update_account，不发通知
|
||
- 无竞争分组：直接返回，不写日志，不发通知
|
||
"""
|
||
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),
|
||
UpstreamGeneratedKey.imported_account_id.isnot(None),
|
||
UpstreamGeneratedKey.status != "orphaned",
|
||
)
|
||
if website_id is not None:
|
||
key_query = key_query.filter(UpstreamGeneratedKey.imported_website_id == website_id)
|
||
key_rows = key_query.all()
|
||
if not key_rows:
|
||
return []
|
||
|
||
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
|
||
|
||
|
||
|
||
def _fetch_remote_managed_prefixes(db: Session, upstream_id: int) -> list[str]:
|
||
"""查询本地 distinct managed_prefix。
|
||
|
||
返回该上游所有已使用的 prefix 列表。空时回退 ["SmartUp"] 兼容旧数据。
|
||
"""
|
||
prefixes = [
|
||
row[0] for row in
|
||
db.query(UpstreamGeneratedKey.managed_prefix)
|
||
.filter(
|
||
UpstreamGeneratedKey.upstream_id == upstream_id,
|
||
UpstreamGeneratedKey.managed_prefix.isnot(None),
|
||
)
|
||
.distinct()
|
||
.all()
|
||
]
|
||
return prefixes if prefixes else ["SmartUp"]
|
||
|
||
|
||
def _fetch_remote_managed_key_ids(db: Session, client, upstream_id: int) -> set[str]:
|
||
"""查询本地 distinct managed_prefix，分别拉远端活跃 Key ID 集合。
|
||
|
||
返回全部找到的远端 Key ID（合并多个 prefix 的结果）。
|
||
"""
|
||
all_ids: set[str] = set()
|
||
for prefix in _fetch_remote_managed_prefixes(db, upstream_id):
|
||
remote_keys = client.list_api_keys(search=prefix, status="active")
|
||
all_ids.update(str(k["id"]) for k in remote_keys if k.get("id"))
|
||
return all_ids
|
||
|
||
|
||
def reconcile_upstream_keys(
|
||
db: Session,
|
||
upstream_id: int,
|
||
active_group_ids: set[str] | None,
|
||
remote_key_ids: set[str] | None,
|
||
captured_at: datetime,
|
||
prev_active_group_ids: set[str] | None = None,
|
||
) -> None:
|
||
"""对账上游 Key 的本地缓存与远端状态。
|
||
|
||
active_group_ids=None → 跳过分组级清理（避免登录失败时误删）。
|
||
remote_key_ids=None → 跳过远端 key_id 级清理（查询失败时安全）。
|
||
两者同时为 None 则完全跳过对账。
|
||
"""
|
||
if active_group_ids is None and remote_key_ids is None:
|
||
return
|
||
|
||
key_rows = (
|
||
db.query(UpstreamGeneratedKey)
|
||
.filter(
|
||
UpstreamGeneratedKey.upstream_id == upstream_id,
|
||
)
|
||
.all()
|
||
)
|
||
for row in key_rows:
|
||
# 1. 增加自恢复逻辑：已导入的 orphaned key 如果后续快照中来源分组恢复，且远端 key id 仍存在，则自动恢复为 imported 并清空错误。
|
||
if (
|
||
row.status == "orphaned"
|
||
and row.error == "来源分组已不存在"
|
||
and row.imported_website_id is not None
|
||
and row.imported_account_id is not None
|
||
and active_group_ids is not None
|
||
and row.group_id in active_group_ids
|
||
and remote_key_ids is not None
|
||
and row.key_id in remote_key_ids
|
||
):
|
||
row.status = "imported"
|
||
row.error = None
|
||
row.updated_at = captured_at
|
||
logger.info("recovered orphaned key %s to imported (group %s recovered and key_id %s exists remotely)", row.id, row.group_id, row.key_id)
|
||
|
||
# 2. 修改上游 key 对账逻辑：来源分组缺失时，不因单次快照缺失立即把已导入 key 标为 orphaned。
|
||
# 分组级 orphan 判断改为“两次连续快照都缺失该分组”才生效，避免上游接口临时返回不完整分组导致误标记。
|
||
if active_group_ids is not None and row.group_id not in active_group_ids:
|
||
if prev_active_group_ids is not None and row.group_id not in prev_active_group_ids:
|
||
if row.imported_website_id and row.imported_account_id:
|
||
row.status = "orphaned"
|
||
row.error = "来源分组已不存在"
|
||
row.updated_at = captured_at
|
||
logger.info("marked key %s orphaned (group %s no longer in snapshot consecutively)", row.id, row.group_id)
|
||
else:
|
||
db.delete(row)
|
||
logger.info("removed key %s (group %s no longer in snapshot consecutively)", row.id, row.group_id)
|
||
continue
|
||
else:
|
||
# 只有单次快照缺失，跳过标记/清理，避免误标
|
||
logger.info("skipped marking key %s orphaned (group %s missing in single snapshot, waiting for next consecutive check)", row.id, row.group_id)
|
||
|
||
# 3. 远端 key id 级清理的现有保守策略：只有远端 key 列表成功拉取后，才允许判断 key 是否不存在。
|
||
if row.key_id and remote_key_ids is not None and row.key_id not in remote_key_ids:
|
||
if row.imported_website_id and row.imported_account_id:
|
||
row.status = "orphaned"
|
||
row.error = "远端 Key 已不存在"
|
||
row.updated_at = captured_at
|
||
logger.info("marked key %s orphaned (key_id %s gone from remote)", row.id, row.key_id)
|
||
else:
|
||
db.delete(row)
|
||
logger.info("removed key %s (key_id %s gone from remote)", row.id, row.key_id)
|
||
continue
|
||
|
||
if remote_key_ids is not None and row.key_id in remote_key_ids:
|
||
row.updated_at = captured_at
|
||
|
||
|
||
def reconcile_upstream_keys_full(db: Session, upstream_id: int) -> bool:
|
||
"""完整的 Key 对账：拉取最新快照的分组 + 登录上游查远端 Key 列表 → 调用 reconcile_upstream_keys。
|
||
|
||
活跃分组 ID 从最新快照获取（与调度器一致），而非调用 live API 避免格式不一致。
|
||
安全规则：
|
||
- 快照存在 → 才允许分组级清理。
|
||
- 远端 Key 列表拉取成功 → 才允许 key_id 级清理。
|
||
- 两者均失败 → 不做任何删除/标记。
|
||
|
||
支持自定义 managed_prefix：查询本地 distinct prefix，分别查远端。
|
||
"""
|
||
from datetime import datetime, timezone
|
||
|
||
upstream = db.query(Upstream).filter(Upstream.id == upstream_id).first()
|
||
if not upstream:
|
||
return False
|
||
|
||
auth_config = json.loads(upstream.auth_config_json or "{}")
|
||
groups_fetched = False
|
||
keys_fetched = False
|
||
active_group_ids: set[str] | None = None
|
||
remote_key_ids: set[str] | None = None
|
||
now = datetime.now(timezone.utc)
|
||
|
||
# 从最新快照获取活跃分组 ID（与调度器 _sync_upstream_keys 一致）
|
||
prev_active_group_ids: set[str] | None = None
|
||
groups = latest_rate_map(db, upstream_id)
|
||
if groups:
|
||
active_group_ids = set(groups.keys())
|
||
groups_fetched = True
|
||
|
||
# 获取倒数第二个快照作为 previous snapshot
|
||
from app.models.snapshot import UpstreamRateSnapshot
|
||
snapshots = (
|
||
db.query(UpstreamRateSnapshot)
|
||
.filter(UpstreamRateSnapshot.upstream_id == upstream_id)
|
||
.order_by(UpstreamRateSnapshot.captured_at.desc())
|
||
.limit(2)
|
||
.all()
|
||
)
|
||
if len(snapshots) >= 2:
|
||
prev_snapshot = snapshots[1]
|
||
try:
|
||
prev_data = json.loads(prev_snapshot.snapshot_json or "{}")
|
||
prev_groups = prev_data.get("groups") or {}
|
||
if isinstance(prev_groups, dict):
|
||
prev_active_group_ids = set(prev_groups.keys())
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
with UpstreamClient(
|
||
base_url=upstream.base_url,
|
||
api_prefix=upstream.api_prefix,
|
||
auth_type=upstream.auth_type,
|
||
auth_config=auth_config,
|
||
timeout=float(upstream.timeout_seconds),
|
||
on_auth_config_update=lambda updated: _persist_upstream_auth_config(upstream, updated),
|
||
target_id=upstream.id,
|
||
target_name=upstream.name,
|
||
) as client:
|
||
client.ensure_authenticated()
|
||
# 获取远端 Key 列表（支持自定义 managed_prefix）
|
||
remote_key_ids = _fetch_remote_managed_key_ids(db, client, upstream_id)
|
||
keys_fetched = True
|
||
except Exception as exc:
|
||
logger.warning("reconcile_upstream_keys_full: upstream %s failed: %s", upstream_id, exc)
|
||
|
||
# 只传递成功获取的数据；失败的传 None 跳过对应检查
|
||
reconcile_upstream_keys(
|
||
db,
|
||
upstream_id,
|
||
active_group_ids if groups_fetched else None,
|
||
remote_key_ids if keys_fetched else None,
|
||
now,
|
||
prev_active_group_ids=prev_active_group_ids,
|
||
)
|
||
db.commit()
|
||
return keys_fetched
|
||
|
||
|
||
def sync_affected_bindings(db: Session, upstream_id: int, changes: list[dict[str, Any]]) -> None:
|
||
for binding in get_affected_bindings(db, changes, upstream_id):
|
||
try:
|
||
sync_binding(db, binding, write=True)
|
||
except Exception as exc:
|
||
logger.exception("website sync failed for binding %s: %s", binding.id, exc)
|