v3.1 + 批次0: 智能回复重构基线 - ApprovalMatcher + 关键词降级 + 文档速修 + v4.0任务书面化

backend/app/services/token_service.py

@@ -244,6 +244,80 @@ class TokenService:
 
         logger.info(f"Token 已失效: {token[:10]}...")
 
+    async def record_token_ip(self, token: str, ip_address: str) -> None:
+        """记录Token使用的IP地址。
+
+        用于后续的异常检测（如同一Token多IP使用）。
+
+        Args:
+            token: Token 字符串
+            ip_address: 客户端IP地址
+        """
+        import hashlib
+        import uuid
+
+        # 计算token hash，避免存储明文token
+        token_hash = hashlib.sha256(token.encode()).hexdigest()[:16]
+
+        # Redis Key: token_ip:token_hash -> 逗号分隔的IP列表
+        key = f"token_ip:{token_hash}"
+        timestamp = datetime.now().isoformat()
+
+        # 追加新IP和访问时间
+        # 格式: "ip1@2026-07-14T10:00:00,ip2@2026-07-14T10:05:00"
+        existing = await self.redis.get(key)
+        if existing:
+            existing_str = existing.decode("utf-8") if isinstance(existing, bytes) else existing
+            # 只保留最近1小时的记录
+            entries = existing_str.split(",")
+            from datetime import timedelta
+            one_hour_ago = datetime.now() - timedelta(hours=1)
+            filtered = []
+            for entry in entries:
+                if "@" in entry:
+                    ip, ts_str = entry.rsplit("@", 1)
+                    try:
+                        ts = datetime.fromisoformat(ts_str)
+                        if ts > one_hour_ago:
+                            filtered.append(entry)
+                    except ValueError:
+                        pass
+            filtered.append(f"{ip_address}@{timestamp}")
+            new_value = ",".join(filtered[-50:])  # 最多保留50条
+        else:
+            new_value = f"{ip_address}@{timestamp}"
+
+        # 保留1小时
+        await self.redis.setex(key, 3600, new_value)
+        logger.debug(f"记录Token IP: token_hash={token_hash}, ip={ip_address}")
+
+    async def get_token_ips(self, token: str) -> List[Dict]:
+        """获取Token使用的IP列表。
+
+        Args:
+            token: Token 字符串
+
+        Returns:
+            List[Dict]: [{"ip": "x.x.x.x", "timestamp": "..."}]
+        """
+        import hashlib
+
+        token_hash = hashlib.sha256(token.encode()).hexdigest()[:16]
+        key = f"token_ip:{token_hash}"
+
+        data = await self.redis.get(key)
+        if not data:
+            return []
+
+        data_str = data.decode("utf-8") if isinstance(data, bytes) else data
+        result = []
+        for entry in data_str.split(","):
+            if "@" in entry:
+                ip, ts_str = entry.rsplit("@", 1)
+                result.append({"ip": ip, "timestamp": ts_str})
+
+        return result
+
     def _get_default_role(self, roles: List[str]) -> str:
         """获取默认角色。