""" core/counter.py ────────────── 极简持久化计数器（核心业务指标：总搜索、总访问、复制次数、成功交互、词频聚合、bad_case）。 数据持久化到阿里云 OSS，通过 platform_utils 屏蔽细节。 无任何本地落盘操作。 count.json 结构： { "total": int, "visits": int, "copies": int, "successes": int, "hot_keywords": {word: count, ...}, "bad_cases": [{q, t}, ...], "history": [{"date": "YYYY-MM-DD", "total": int}, ...] # 每日快照，最近 180 天 } """ from __future__ import annotations import asyncio import json import time from collections import Counter from datetime import datetime, timezone, timedelta, date from typing import Optional from platform_utils import get_counter_cfg, read_bytes, upload_bytes, CounterConfig # ── 核心状态变量 ──────────────────────────────── _memory_count: int = 0 _dirty_count: int = 0 _memory_visits: int = 0 _dirty_visits: int = 0 BASE_VISITS: int = 0 _memory_copies: int = 0 _dirty_copies: int = 0 BASE_COPIES: int = 0 _memory_mcp: int = 0 _dirty_mcp: int = 0 _memory_successes: int = 0 _dirty_successes: int = 0 _memory_keywords: Counter = Counter() _dirty_keywords: Counter = Counter() MAX_KEYWORDS_LIMIT = 200 _memory_bad_cases: list[dict] = [] _dirty_bad_cases: list[dict] = [] MAX_BAD_CASES = 50 _memory_history: list[dict] = [] # [{"date": "YYYY-MM-DD", "total": int}, ...] MAX_HISTORY_DAYS = 180 _last_sync: float = 0.0 _sync_lock: Optional[asyncio.Lock] = None SYNC_INTERVAL = 1800 # 每 30 分钟同步一次 SYNC_THRESHOLD = 500 # 或各项增量之和达到 200 次时同步 def _get_sync_lock() -> asyncio.Lock: global _sync_lock if _sync_lock is None: _sync_lock = asyncio.Lock() return _sync_lock # ── 历史快照工具 ────────────────────────────────────────────────────────────── def _today_str() -> str: return datetime.now(timezone(timedelta(hours=8))).date().isoformat() def _merge_history( remote: list[dict], current_total: int, memory: list[dict] | None = None, ) -> list[dict]: """ 将当天的 total 写入（或更新）历史列表，保留最近 MAX_HISTORY_DAYS 天。 同一天只保留最新一条（total 取较大值）。 remote : 从 OSS 读回的历史（可能因缓存导致不完整） memory : 内存中已加载的历史（init() 时从 OSS 完整读取，用于兜底） """ today = _today_str() by_date: dict[str, int] = {} # 先用内存数据打底，再用远端数据覆盖（远端是最终写入态，优先级更高） for r in (memory or []): by_date[r['date']] = r['total'] for r in remote: by_date[r['date']] = max(by_date.get(r['date'], 0), r['total']) # 当天记录取较大值，避免同步时序问题导致数据回退 by_date[today] = max(by_date.get(today, 0), current_total) sorted_records = sorted(by_date.items(), reverse=True)[:MAX_HISTORY_DAYS] # 返回时按时间正序，方便绘图 return [{'date': d, 'total': t} for d, t in sorted(sorted_records)] # ── 远端 IO ─────────────────────────────────────────────────────────────────── _COUNTER_FILE = 'count.json' def _parse_remote_data(raw: bytes | None) -> tuple[int, int, int, int, int, dict, list, list]: """解析远端 JSON bytes，返回 (total, visits, copies, mcp, successes, keywords, bad_cases, history)。""" if raw is None: return 0, BASE_VISITS, BASE_COPIES, 0, 0, {}, [], [] try: data = json.loads(raw.decode('utf-8')) r_total = int(data.get('total', 0)) r_visits = int(data.get('visits', BASE_VISITS)) r_copies = int(data.get('copies', BASE_COPIES)) r_mcp = int(data.get('mcp', 0)) r_successes = int(data.get('successes', int(r_total * 0.75))) r_keywords = data.get('hot_keywords', {}) r_bad_cases = data.get('bad_cases', []) r_history = data.get('history', []) return r_total, r_visits, r_copies, r_mcp, r_successes, r_keywords, r_bad_cases, r_history except Exception: return 0, BASE_VISITS, BASE_COPIES, 0, 0, {}, [], [] def _read_remote() -> tuple[int, int, int, int, int, dict, list, list]: cfg = get_counter_cfg() if not cfg.available: return 0, BASE_VISITS, BASE_COPIES, 0, 0, {}, [], [] try: raw = read_bytes(_COUNTER_FILE, cfg) except Exception as e: print(f'[Counter] 启动读取远端数据异常: {e}') return 0, BASE_VISITS, BASE_COPIES, 0, 0, {}, [], [] if raw is None: return 0, BASE_VISITS, BASE_COPIES, 0, 0, {}, [], [] return _parse_remote_data(raw) def _sync_remote_task( adds_count: int, adds_visits: int, adds_copies: int, adds_mcp: int, adds_successes: int, adds_keywords: dict, adds_bad_cases: list, memory_history_snapshot: list, ) -> tuple[bool, int, int, int, int, int, dict, list, list]: cfg: CounterConfig = get_counter_cfg() if not cfg.available: return False, 0, 0, 0, 0, 0, {}, [], [] try: raw = read_bytes(_COUNTER_FILE, cfg) except Exception as e: print(f'[Counter] 远端读取异常，中止本次同步以保护数据: {e}') return False, 0, 0, 0, 0, 0, {}, [], [] r_total, r_visits, r_copies, r_mcp, r_successes, r_keywords, r_bad_cases, r_history = ( _parse_remote_data(raw) ) n_total = r_total + adds_count n_visits = r_visits + adds_visits n_copies = r_copies + adds_copies n_mcp = r_mcp + adds_mcp n_successes = r_successes + adds_successes merged_keywords = Counter(r_keywords) for word, count in adds_keywords.items(): merged_keywords[word] += count top_keywords = dict(merged_keywords.most_common(MAX_KEYWORDS_LIMIT)) merged_bad_cases = (adds_bad_cases + r_bad_cases)[:MAX_BAD_CASES] n_history = _merge_history(r_history, n_total, memory=memory_history_snapshot) content = json.dumps({ 'total': n_total, 'visits': n_visits, 'copies': n_copies, 'mcp': n_mcp, 'successes': n_successes, 'hot_keywords': top_keywords, 'bad_cases': merged_bad_cases, 'history': n_history, }, ensure_ascii=False, indent=2).encode('utf-8') commit_msg = ( f'Sync: 搜索:{n_total} | MCP:{n_mcp} | 成功:{n_successes} | ' f'复制:{n_copies} | 访问:{n_visits} | bad_cases:{len(merged_bad_cases)}' ) ok = upload_bytes(content, _COUNTER_FILE, cfg, commit_msg, retries=3, retry_delay=1.0) if ok: print( f'[Counter] 同步成功！搜索:{n_total}, MCP:{n_mcp}, 成功交互:{n_successes}, ' f'复制:{n_copies}, bad_cases:{len(merged_bad_cases)}' ) return True, n_total, n_visits, n_copies, n_mcp, n_successes, top_keywords, merged_bad_cases, n_history return False, 0, 0, 0, 0, 0, {}, [], [] # ── 后台同步协程 ────────────────────────────────────────────────────────────── async def _perform_sync(): global _dirty_count, _dirty_visits, _dirty_copies, _dirty_mcp, _dirty_successes global _dirty_keywords, _dirty_bad_cases, _last_sync global _memory_count, _memory_visits, _memory_copies, _memory_mcp, _memory_successes global _memory_keywords, _memory_bad_cases, _memory_history lock = _get_sync_lock() if lock.locked(): return async with lock: has_dirty = ( _dirty_count + _dirty_visits + _dirty_copies + _dirty_mcp + _dirty_successes + len(_dirty_keywords) + len(_dirty_bad_cases) ) > 0 if not has_dirty: return c_adds, v_adds, cp_adds, m_adds, s_adds = ( _dirty_count, _dirty_visits, _dirty_copies, _dirty_mcp, _dirty_successes ) k_adds = dict(_dirty_keywords) bc_adds = list(_dirty_bad_cases) history_snapshot = list(_memory_history) _dirty_count = 0 _dirty_visits = 0 _dirty_copies = 0 _dirty_mcp = 0 _dirty_successes = 0 _dirty_keywords.clear() _dirty_bad_cases.clear() loop = asyncio.get_running_loop() result = await loop.run_in_executor( None, _sync_remote_task, c_adds, v_adds, cp_adds, m_adds, s_adds, k_adds, bc_adds, history_snapshot, ) success, l_total, l_visits, l_copies, l_mcp, l_successes, l_keywords, l_bad_cases, l_history = result if success: _last_sync = time.time() _memory_count = max(_memory_count, l_total) _memory_visits = max(_memory_visits, l_visits) _memory_copies = max(_memory_copies, l_copies) _memory_mcp = max(_memory_mcp, l_mcp) _memory_successes = max(_memory_successes, l_successes) _memory_keywords.clear() _memory_keywords.update(l_keywords) _memory_bad_cases.clear() _memory_bad_cases.extend(l_bad_cases) _memory_history.clear() _memory_history.extend(l_history) else: _dirty_count += c_adds _dirty_visits += v_adds _dirty_copies += cp_adds _dirty_mcp += m_adds _dirty_successes += s_adds _dirty_keywords.update(k_adds) _dirty_bad_cases.extend(bc_adds) # ── 公共 API ────────────────────────────────────────────────────────────────── async def init(): """启动时从 OSS 拉取最新计数，初始化内存状态。""" global _memory_count, _memory_visits, _memory_copies, _memory_mcp, _memory_successes global _memory_keywords, _memory_bad_cases, _memory_history, _last_sync cfg = get_counter_cfg() if not cfg.available: print(f'[Counter] 计数器未配置（platform={cfg.platform}），仅使用内存计数。') return loop = asyncio.get_running_loop() ( _memory_count, _memory_visits, _memory_copies, _memory_mcp, _memory_successes, r_keys, r_bad_cases, r_history, ) = await loop.run_in_executor(None, _read_remote) _memory_keywords.update(r_keys) _memory_bad_cases.clear() _memory_bad_cases.extend(r_bad_cases) _memory_history.clear() _memory_history.extend(r_history) _last_sync = time.time() print( f'[Counter] 初始化完成（{cfg.platform}）：搜索={_memory_count}, MCP={_memory_mcp}, ' f'访问={_memory_visits}, 复制={_memory_copies}, 历史={len(_memory_history)}天' ) def _check_sync(): cfg = get_counter_cfg() if not cfg.available: return if (time.time() - _last_sync > SYNC_INTERVAL) or ( _dirty_count + _dirty_visits + _dirty_copies + _dirty_mcp + _dirty_successes + len(_dirty_keywords) + len(_dirty_bad_cases) ) >= SYNC_THRESHOLD: asyncio.create_task(_perform_sync()) async def increment() -> int: global _memory_count, _dirty_count _memory_count += 1; _dirty_count += 1 _check_sync() return _memory_count async def increment_visit() -> int: global _memory_visits, _dirty_visits _memory_visits += 1; _dirty_visits += 1 _check_sync() return _memory_visits async def increment_copy() -> int: global _memory_copies, _dirty_copies _memory_copies += 1; _dirty_copies += 1 _check_sync() return _memory_copies async def increment_mcp() -> int: global _memory_mcp, _dirty_mcp _memory_mcp += 1; _dirty_mcp += 1 _check_sync() return _memory_mcp async def increment_success() -> int: global _memory_successes, _dirty_successes _memory_successes += 1; _dirty_successes += 1 _check_sync() return _memory_successes async def add_keywords(words: list[str]): global _dirty_keywords, _memory_keywords valid_words = [w for w in words if len(w) > 1 and w not in {'一个', '的', '在', '了', '是', '有'}] for word in valid_words: _memory_keywords[word] += 1 _dirty_keywords[word] += 1 _check_sync() async def add_bad_case(query: str, platform: str = '', settings: dict | None = None) -> None: global _dirty_bad_cases, _memory_bad_cases entry = { 'q': query, 't': datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), } if platform: entry['platform'] = platform if settings: entry['settings'] = settings print(f'[Counter] bad_case 上报: {json.dumps(entry, ensure_ascii=False)}') _memory_bad_cases.insert(0, entry) if len(_memory_bad_cases) > MAX_BAD_CASES: _memory_bad_cases[:] = _memory_bad_cases[:MAX_BAD_CASES] _dirty_bad_cases.insert(0, entry) _check_sync() def get() -> int: return _memory_count def get_visits() -> int: return _memory_visits def get_copies() -> int: return _memory_copies def get_mcp() -> int: return _memory_mcp def get_successes() -> int: return _memory_successes def get_hot_keywords(top_n: int = 10) -> dict: return dict(_memory_keywords.most_common(top_n)) def get_bad_cases() -> list[dict]: return list(_memory_bad_cases) def get_history() -> list[dict]: """返回每日搜索量快照，格式 [{"date": "YYYY-MM-DD", "total": int}, ...]，按日期正序。""" return list(_memory_history) async def force_sync(): await _perform_sync()