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)