diff --git a/utils/system_jobs.py b/utils/system_jobs.py index a6141cd..7ef4176 100644 --- a/utils/system_jobs.py +++ b/utils/system_jobs.py @@ -43,6 +43,14 @@ def get_system_job_definitions(robot) -> List[Dict[str, Any]]: "trigger_config": {"seconds": 300}, "handler": _build_process_pending_images_handler(robot), }, + { + "job_key": "contact_avatar_cache_sync", + "name": "联系人头像缓存同步", + "description": "每小时扫描一次联系人头像差异并增量下载，避免启动阶段批量拉头像", + "trigger_type": "every_seconds", + "trigger_config": {"seconds": 3600}, + "handler": _build_contact_avatar_cache_sync_handler(robot), + }, ] def _build_process_pending_images_handler(robot) -> Callable[[], Awaitable[None]]: @@ -53,6 +61,20 @@ def _build_process_pending_images_handler(robot) -> Callable[[], Awaitable[None] return _handler +def _build_contact_avatar_cache_sync_handler(robot) -> Callable[[], Awaitable[None]]: + async def _handler(): + contact_manager = getattr(robot, "contact_manager", None) + if not contact_manager: + return + # 头像缓存同步内部包含 requests 下载和本地文件写入： + # 1. 这些都是阻塞式 IO，不适合直接卡在调度器事件循环里执行； + # 2. 这里统一切到线程池，保证系统其他异步任务的调度不受影响； + # 3. ContactManager 内部还会做“运行中跳过”保护，避免重复重入。 + await asyncio.to_thread(contact_manager.run_scheduled_avatar_cache_sync, "system_job_hourly") + + return _handler + + class SystemJobLoader: """系统任务加载器：从数据库读取调度配置并注册到 async_job。""" diff --git a/utils/wechat/contact_manager.py b/utils/wechat/contact_manager.py index d0b0225..1e0e443 100644 --- a/utils/wechat/contact_manager.py +++ b/utils/wechat/contact_manager.py @@ -35,7 +35,6 @@ class ContactManager: _avatar_manifest: Dict[str, Dict[str, str]] = {} _avatar_sync_state_lock = threading.Lock() _avatar_sync_running = False - _avatar_sync_pending = False # 定义公共好友列表 _PUBLIC_FRIENDS = { 'fmessage': '朋友推荐消息', @@ -88,60 +87,61 @@ class ContactManager: "gender": gender} chatroom_members: 所有的群成员昵称信息 """ - # 这些内存态会被后台头像同步线程读取，因此更新时统一加锁， - # 避免“刷新通讯录”和“后台下载头像”并发时出现字典遍历被修改的问题。 + # 这些内存态会被定时任务与按需补下载逻辑读取，因此更新时统一加锁， + # 避免“刷新通讯录”和“后台头像处理”并发时出现字典遍历被修改的问题。 with self._avatar_cache_lock: self._contacts = contacts self._friends = friends self._head_images = head_imgs self._group_members = chatroom_members - # 头像缓存同步改成后台异步触发： - # 1. set_contacts 常出现在“刷新通讯录”主链路里，若同步下载头像会明显阻塞系统； - # 2. 这里改成只提交后台任务，主线程先把联系人数据更新完立即返回； - # 3. 后台线程会基于当前最新 head_images 做增量同步，不会丢掉头像更新。 - self._schedule_avatar_cache_sync(reason="set_contacts") + # 联系人刷新链路不再主动触发批量头像同步： + # 1. 用户明确希望降低启动与刷新通讯录时的瞬时压力； + # 2. 头像全量/增量检查统一交给系统定时任务每小时处理； + # 3. 若后台页面提前访问某个头像，接口仍会走单头像按需补下载，不影响使用。 self._logger.info(f"联系人信息已更新，共 {len(contacts)} 个联系人") # 分类联系人 self._classify_contacts() - def _schedule_avatar_cache_sync(self, reason: str = "unknown") -> None: - """提交头像缓存后台同步任务，避免在主链路里同步下载头像。""" + def run_scheduled_avatar_cache_sync(self, reason: str = "system_job") -> Dict[str, int | str | bool]: + """供系统定时任务调用的头像缓存同步入口。 + + 设计说明： + 1. 启动、登录、刷新通讯录时都不再主动批量下载头像； + 2. 统一由系统任务定时做“manifest 与当前头像 URL”差异检查； + 3. 若某一小时任务尚未跑完，下一次重入会直接跳过，避免并发重复下载。 + """ with self._avatar_sync_state_lock: - self._avatar_sync_pending = True if self._avatar_sync_running: - self._logger.debug(f"头像缓存后台同步已在执行，合并本次请求 reason={reason}") - return + self._logger.debug(f"头像缓存定时同步跳过 reason={reason} busy=1") + return { + "success": False, + "skipped": True, + "reason": "busy", + } self._avatar_sync_running = True - worker = threading.Thread( - target=self._avatar_sync_worker, - args=(reason,), - daemon=True, - name="contact-avatar-sync", + + try: + self._logger.info(f"头像缓存定时同步开始 reason={reason}") + result = self._sync_avatar_cache() + self._logger.info( + "头像缓存定时同步完成 " + f"reason={reason} " + f"total_head_images={result.get('total_head_images', 0)} " + f"downloaded={result.get('downloaded', 0)} " + f"replaced={result.get('replaced', 0)} " + f"failed={result.get('failed', 0)} " + f"removed_contacts={result.get('removed_contacts', 0)} " + f"removed_files={result.get('removed_files', 0)}" ) - worker.start() - self._logger.debug(f"头像缓存后台同步已提交 reason={reason}") - - def _avatar_sync_worker(self, initial_reason: str) -> None: - """串行消费头像缓存同步请求，确保高频更新时只跑最新一轮。""" - trigger_reason = initial_reason - while True: + return { + "success": True, + "skipped": False, + "reason": reason, + **result, + } + finally: with self._avatar_sync_state_lock: - self._avatar_sync_pending = False - try: - self._logger.debug(f"头像缓存后台同步开始 reason={trigger_reason}") - self._sync_avatar_cache() - except Exception as exc: - self._logger.error(f"头像缓存后台同步失败 reason={trigger_reason}: {exc}") - - with self._avatar_sync_state_lock: - # 如果在本轮执行期间又收到了新的同步请求，就立刻再跑一轮， - # 这样可以把多次 set_contacts/update_head_image 合并成少量批处理。 - if self._avatar_sync_pending: - trigger_reason = "pending_update" - continue self._avatar_sync_running = False - self._logger.debug("头像缓存后台同步结束") - break def _init_avatar_cache(self) -> None: """初始化头像缓存目录和 manifest。""" @@ -288,22 +288,35 @@ class ContactManager: response.close() return str(target_path) - def _sync_avatar_cache(self) -> None: + def _sync_avatar_cache(self) -> Dict[str, int]: """按当前头像 URL 增量同步本地缓存。""" with self._avatar_cache_lock: head_images_snapshot = dict(self._head_images) if not head_images_snapshot: - return + return { + "total_head_images": 0, + "manifest_entries": len(self._avatar_manifest), + "reuse": 0, + "downloaded": 0, + "replaced": 0, + "failed": 0, + "removed_contacts": 0, + "removed_files": 0, + "orphan_deleted": 0, + "tmp_deleted": 0, + } manifest_changed = False # 统计字段用于打 debug 汇总日志，方便观察“初始化第一批头像”时到底发生了什么： # 1. reuse_count 表示命中本地缓存、无需下载； # 2. download_count 表示本轮真正新增或重拉的头像数量； # 3. replaced_count 表示头像 URL 变更导致发生了“新旧替换”； - # 4. removed_contact_count / removed_file_count 表示联系人消失后同步移除了多少记录和文件。 + # 4. failed_count 表示本轮有多少头像因为网络或写盘失败未能更新； + # 5. removed_contact_count / removed_file_count 表示联系人消失后同步移除了多少记录和文件。 reuse_count = 0 download_count = 0 replaced_count = 0 + failed_count = 0 removed_contact_count = 0 removed_file_count = 0 for wxid, avatar_url in head_images_snapshot.items(): @@ -320,6 +333,7 @@ class ContactManager: continue downloaded_path = self._download_avatar_to_cache(wxid, remote_url) if not downloaded_path: + failed_count += 1 self._logger.debug(f"头像缓存同步跳过 wxid={wxid} reason=download_failed") continue with self._avatar_cache_lock: @@ -364,11 +378,24 @@ class ContactManager: f"reuse={reuse_count} " f"downloaded={download_count} " f"replaced={replaced_count} " + f"failed={failed_count} " f"removed_contacts={removed_contact_count} " f"removed_files={removed_file_count} " f"orphan_deleted={cleanup_stats.get('orphan_deleted', 0)} " f"tmp_deleted={cleanup_stats.get('tmp_deleted', 0)}" ) + return { + "total_head_images": len(active_wxids), + "manifest_entries": len(self._avatar_manifest), + "reuse": reuse_count, + "downloaded": download_count, + "replaced": replaced_count, + "failed": failed_count, + "removed_contacts": removed_contact_count, + "removed_files": removed_file_count, + "orphan_deleted": cleanup_stats.get("orphan_deleted", 0), + "tmp_deleted": cleanup_stats.get("tmp_deleted", 0), + } def get_cached_head_image_path(self, wxid: str) -> str: """返回头像缓存本地路径，若缓存不存在则返回空字符串。""" @@ -553,11 +580,10 @@ class ContactManager: """ with self._avatar_cache_lock: self._head_images.update({wxid: head_image}) - # 单头像更新也改成后台异步： - # 1. 入群欢迎、成员变更等实时链路不能被头像下载阻塞； - # 2. 这里只更新最新远端 URL，并触发后台增量同步； - # 3. 若页面提前访问该头像，/api/avatar 仍会按需补下载，不影响可用性。 - self._schedule_avatar_cache_sync(reason=f"update_head_image:{wxid}") + # 单头像更新链路同样不再主动拉取文件： + # 1. 成员变更、欢迎语等实时场景优先保证主流程轻量； + # 2. 当前这里只维护最新头像 URL，批量下载统一交给定时任务； + # 3. 若后台刚好访问这个联系人头像，仍可通过按需补下载拿到本地缓存。 return True def get_group_name(self, roomid: str, wxid: str) -> str: