[add] Standardize time zones; Reuse a single Redis client; Use "mget" for batch writing requests

api/app/repositories/implicit_emotions_storage_repository.py

@@ -142,17 +142,32 @@ class ImplicitEmotionsStorageRepository:
                 if not batch:
                     break
 
-                for end_user_id, updated_at in batch:
+                # 批量获取当前批次所有用户的 last_done 时间戳（一次网络往返）
-                    raw = redis_client.get(f"write_message:last_done:{end_user_id}")
+                keys = [f"write_message:last_done:{end_user_id}" for end_user_id, _ in batch]
+                raw_values = redis_client.mget(keys)
+
+                for (end_user_id, updated_at), raw in zip(batch, raw_values):
                     if raw is None:
-                        # 该用户从未有过 write_message 成功记录，跳过
                         continue
                     try:
+                        CST = timezone(timedelta(hours=8))
                         last_done = datetime.fromisoformat(raw)
-                        # 统一去掉时区信息做 naive 比较
+                        # 统一转为 CST naive 时间做比较
-                        if last_done.tzinfo is not None:
+                        if last_done.tzinfo is None:
-                            last_done = last_done.astimezone(timezone.utc).replace(tzinfo=None)
+                            last_done = last_done.replace(tzinfo=timezone.utc).astimezone(CST).replace(tzinfo=None)
-                        if updated_at is None or last_done > updated_at:
+                        else:
+                            last_done = last_done.astimezone(CST).replace(tzinfo=None)
+
+                        if updated_at is None:
+                            yield end_user_id
+                            continue
+                        # updated_at 同样转为 CST naive
+                        if updated_at.tzinfo is None:
+                            updated_at_cst = updated_at.replace(tzinfo=timezone.utc).astimezone(CST).replace(tzinfo=None)
+                        else:
+                            updated_at_cst = updated_at.astimezone(CST).replace(tzinfo=None)
+
+                        if last_done > updated_at_cst:
                             yield end_user_id
                     except Exception as e:
                         logger.warning(f"解析 last_done 时间戳失败: end_user_id={end_user_id}, raw={raw}, error={e}")

api/app/tasks.py

@@ -15,6 +15,29 @@ from uuid import UUID
 import redis
 import requests
 
+# 模块级同步 Redis 客户端单例，供 Celery 任务共享使用（避免每次任务新建连接）
+# 连接 CELERY_BACKEND DB，与 write_message:last_done 时间戳写入保持一致
+def _build_sync_redis_client():
+    try:
+        return redis.StrictRedis(
+            host=settings.REDIS_HOST,
+            port=settings.REDIS_PORT,
+            db=settings.REDIS_DB_CELERY_BACKEND,
+            password=settings.REDIS_PASSWORD,
+            decode_responses=True,
+        )
+    except Exception:
+        return None
+
+_sync_redis_client: redis.StrictRedis = None
+
+def get_sync_redis_client() -> redis.StrictRedis:
+    """获取模块级同步 Redis 客户端（懒初始化单例）"""
+    global _sync_redis_client
+    if _sync_redis_client is None:
+        _sync_redis_client = _build_sync_redis_client()
+    return _sync_redis_client
+
 # Import a unified Celery instance
 from app.celery_app import celery_app
 from app.core.config import settings
@@ -1090,22 +1113,18 @@ def write_message_task(self, end_user_id: str, message: list[dict], config_id: s
         logger.info(
             f"[CELERY WRITE] Task completed successfully - elapsed_time={elapsed_time:.2f}s, task_id={self.request.id}")
 
-        # 记录该用户最后一次 write_message 成功的时间，供 init_implicit_emotions_for_users 做时间轴筛选
+        # 记录该用户最后一次 write_message 成功的时间，供时间轴筛选使用
         try:
-            import redis as _redis
+            _r = get_sync_redis_client()
-            from urllib.parse import quote as _quote
+            if _r is not None:
-            _r = _redis.StrictRedis(
+                from datetime import timezone as _tz, timedelta as _td
-                host=settings.REDIS_HOST,
+                _CST = _tz(timedelta(hours=8))
-                port=settings.REDIS_PORT,
+                _now_cst = datetime.now(_CST).replace(tzinfo=None).isoformat()
-                db=settings.REDIS_DB_CELERY_BACKEND,
+                _r.set(
-                password=settings.REDIS_PASSWORD,
+                    f"write_message:last_done:{end_user_id}",
-                decode_responses=True,
+                    _now_cst,
-            )
+                    ex=86400 * 30,
-            _r.set(
+                )
-                f"write_message:last_done:{end_user_id}",
-                datetime.utcnow().isoformat(),
-                ex=86400 * 30,  # 30天过期
-            )
         except Exception as _e:
             logger.warning(f"[CELERY WRITE] 写入 last_done 时间戳失败（不影响主流程）: {_e}")
 
@@ -2196,14 +2215,7 @@ def update_implicit_emotions_storage(self) -> Dict[str, Any]:
                 logger.info(f"表中存量用户总数: {total_users}，开始时间轴筛选")
 
                 # 构建 Redis 同步客户端，用于时间轴筛选
-                import redis as _redis
+                _redis_client = get_sync_redis_client()
-                _redis_client = _redis.StrictRedis(
-                    host=settings.REDIS_HOST,
-                    port=settings.REDIS_PORT,
-                    db=settings.REDIS_DB_CELERY_BACKEND,
-                    password=settings.REDIS_PASSWORD,
-                    decode_responses=True,
-                )
 
                 # 只处理 last_done > updated_at 的用户（有新记忆写入的用户）
                 for end_user_id in repo.get_users_needing_refresh(_redis_client, batch_size=100):