"""FAQ sec_ids 共享缓存(基于 Redis) + Leader 定时刷新。 设计要点: - 数据存 Redis 的 KEY_CACHE_DATA(JSON 格式),所有 worker 共享 - 多 worker 启动时通过 KEY_LOCK_LEADER 选举出唯一 leader - 仅 leader 执行刷新 + 跑 scheduler_loop;follower 只读 Redis - leader 心跳续期;leader 进程挂掉后锁自然过期,其他 worker 抢锁接管 """ from __future__ import annotations import asyncio import json import logging import os import uuid from datetime import datetime, time, timedelta from typing import Any from app.core.config import settings from app.services.redis_client import ( KEY_CACHE_DATA, KEY_LOCK_LEADER, KEY_LOCK_REFRESH, get_redis, release_lock, renew_lock, try_acquire_lock, ) from app.services.zendesk_client import list_all_faq_sections logger = logging.getLogger(__name__) # 当前进程的 leader 标识(用于锁续期校验) WORKER_ID = f"{os.getpid()}-{uuid.uuid4().hex[:8]}" # 空快照(Redis 不可用或缓存未就绪时使用,保证业务不崩) _EMPTY_SNAPSHOT: dict[str, Any] = { "sec_ids": [], "faq_sections": [], "last_updated_at": None, "next_refresh_at": None, } def _next_refresh_time(now: datetime) -> datetime: """计算下一次定时刷新时间(每天 hh:mm,本地时区)。""" target_time = time( hour=settings.cache_refresh_hour, minute=settings.cache_refresh_minute, ) today_target = datetime.combine(now.date(), target_time) if now < today_target: return today_target return today_target + timedelta(days=1) async def get_snapshot() -> dict[str, Any]: """从 Redis 读快照;Redis 不可用或无数据时返回空快照。""" try: r = await get_redis() raw = await r.get(KEY_CACHE_DATA) if not raw: return dict(_EMPTY_SNAPSHOT) return json.loads(raw) except Exception as exc: logger.warning("读取 Redis 缓存失败,使用空快照: %s", exc) return dict(_EMPTY_SNAPSHOT) async def refresh_cache(force: bool = False) -> bool | None: """从 Zendesk 拉取最新数据并写入 Redis。 使用 KEY_LOCK_REFRESH 防并发刷新(避免重复打 Zendesk)。 Args: force: True 时跳过去重锁直接刷新(管理员手动触发) Returns: True 实际执行了刷新;False 被去重锁挡住 """ if not force: got = await try_acquire_lock( KEY_LOCK_REFRESH, WORKER_ID, ttl_seconds=30 ) if not got: logger.info("另一进程正在刷新缓存,本次跳过") return False try: logger.info("[%s] 开始刷新 FAQ 缓存…", WORKER_ID) try: sec_ids, faq_sections = await list_all_faq_sections() except Exception as exc: # noqa: BLE001 logger.exception("FAQ 缓存刷新失败:%s", exc) return False if not sec_ids: logger.warning("Zendesk 没有 FAQ sec_ids,无法刷新缓存") return False now = datetime.now() payload = { "sec_ids": list(sec_ids), "faq_sections": list(faq_sections), "last_updated_at": now.isoformat(), "next_refresh_at": _next_refresh_time(now).isoformat(), } try: r = await get_redis() await r.set(KEY_CACHE_DATA, json.dumps(payload, ensure_ascii=False)) except Exception as exc: # noqa: BLE001 logger.exception("写入 Redis 缓存失败: %s", exc) return False logger.info( "[%s] FAQ 缓存刷新完成:sec_ids=%s 条,下一次刷新 %s", WORKER_ID, len(payload["sec_ids"]), payload["next_refresh_at"], ) return True finally: if not force: await release_lock(KEY_LOCK_REFRESH, WORKER_ID) async def try_become_leader() -> bool: """尝试成为 leader(仅一个 worker 会成功)。""" try: return await try_acquire_lock( KEY_LOCK_LEADER, WORKER_ID, ttl_seconds=settings.leader_lock_ttl ) except Exception as exc: # noqa: BLE001 logger.warning("Leader 选举异常: %s", exc) return False async def leader_heartbeat() -> None: """Leader 心跳:周期性续期 leader 锁。 锁过期则尝试重新抢锁;抢到继续当 leader,抢不到说明已被他人接管,退出。 """ interval = max(1, settings.leader_lock_ttl // 3) while True: try: await asyncio.sleep(interval) except asyncio.CancelledError: raise try: ok = await renew_lock( KEY_LOCK_LEADER, WORKER_ID, settings.leader_lock_ttl ) if not ok: # 锁丢失,尝试重新抢 logger.warning("[%s] Leader 锁丢失,尝试重新选举", WORKER_ID) regained = await try_become_leader() if not regained: logger.info("[%s] 已不再是 leader,停止心跳", WORKER_ID) return logger.info("[%s] 重新成为 leader", WORKER_ID) except Exception as exc: # noqa: BLE001 logger.warning("Leader 心跳异常: %s", exc) async def scheduler_loop() -> None: """Leader 专属:每天 hh:mm 刷新一次 FAQ 缓存。""" while True: now = datetime.now() next_run = _next_refresh_time(now) wait_seconds = (next_run - now).total_seconds() logger.info( "[%s] 下一次 FAQ 缓存刷新 %s(%.0f 秒后)", WORKER_ID, next_run.isoformat(), wait_seconds, ) try: await asyncio.sleep(wait_seconds) except asyncio.CancelledError: logger.info("scheduler_loop 收到取消信号,退出") raise await refresh_cache() async def wait_cache_ready(timeout_seconds: int | None = None) -> bool: """Follower 启动时等待 leader 把缓存写入 Redis。 Returns: True 缓存已就绪;False 超时仍未就绪(应用照样启动,搜索降级) """ timeout = timeout_seconds or settings.cache_ready_wait deadline = asyncio.get_event_loop().time() + timeout while asyncio.get_event_loop().time() < deadline: try: r = await get_redis() exists = await r.exists(KEY_CACHE_DATA) if exists: return True except Exception as exc: # noqa: BLE001 logger.warning("等待缓存就绪时 Redis 异常: %s", exc) return False await asyncio.sleep(0.5) return False