refactor: split some logic into util func

backend/services/auth_service.py

@@ -381,6 +381,14 @@ class AuthService:
         Returns:
             包含打卡 token 验证结果的字典
         """
+        from backend.utils.time_helpers import (
+            parse_jwt_exp,
+            is_timestamp_expired,
+            days_until_expiry,
+            minutes_until_expiry,
+            seconds_until_expiry
+        )
+
         # 检查是否有 authorization token
         if not user.authorization or user.authorization == "":
             return {
@@ -389,52 +397,42 @@ class AuthService:
                 "reason": "no_token"
             }
 
-        # 检查 Token 是否过期
-        if not user.jwt_exp or user.jwt_exp == "0":
+        # 解析 jwt_exp
+        exp_timestamp = parse_jwt_exp(user.jwt_exp)
+        if not exp_timestamp:
             return {
                 "is_valid": False,
                 "message": "打卡凭证无效",
                 "reason": "invalid_expiry"
             }
 
-        try:
-            exp_timestamp = int(user.jwt_exp)
-            current_timestamp = int(datetime.now().timestamp())
-
-            if current_timestamp > exp_timestamp:
-                days_expired = (current_timestamp - exp_timestamp) // 86400
-                return {
-                    "is_valid": False,
-                    "message": f"打卡凭证已过期 {days_expired} 天",
-                    "reason": "expired",
-                    "days_expired": days_expired
-                }
-
-            # 计算剩余时间
-            seconds_remaining = exp_timestamp - current_timestamp
-            days_remaining = seconds_remaining // 86400
-            minutes_remaining = seconds_remaining // 60
-
-            # 判断是否即将过期（30分钟内）
-            expiring_soon = minutes_remaining <= 30
-
-            return {
-                "is_valid": True,
-                "message": "打卡凭证有效",
-                "days_remaining": days_remaining,
-                "minutes_remaining": minutes_remaining,
-                "expiring_soon": expiring_soon,
-                "expires_at": exp_timestamp
-            }
-
-        except ValueError:
-            logger.error(f"用户 {user.id} 的 jwt_exp 格式不正确: {user.jwt_exp}")
+        # 检查是否过期
+        if is_timestamp_expired(exp_timestamp):
+            days_expired = abs(days_until_expiry(exp_timestamp))
             return {
                 "is_valid": False,
-                "message": "打卡凭证格式错误",
-                "reason": "invalid_format"
+                "message": f"打卡凭证已过期 {days_expired} 天",
+                "reason": "expired",
+                "days_expired": days_expired
             }
 
+        # Token 有效，计算剩余时间
+        seconds_remaining = seconds_until_expiry(exp_timestamp)
+        days_remaining = days_until_expiry(exp_timestamp)
+        minutes_remaining = minutes_until_expiry(exp_timestamp)
+
+        # 判断是否即将过期（30分钟内）
+        expiring_soon = minutes_remaining <= 30
+
+        return {
+            "is_valid": True,
+            "message": "打卡凭证有效",
+            "days_remaining": days_remaining,
+            "minutes_remaining": minutes_remaining,
+            "expiring_soon": expiring_soon,
+            "expires_at": exp_timestamp
+        }
+
     @staticmethod
     def alias_login(alias: str, password: str, db: Session) -> Dict[str, Any]:
         """
@@ -532,15 +530,12 @@ class AuthService:
             token_warning = "token_invalid"
         else:
             # 检查 Token 是否过期
-            try:
-                exp_timestamp = int(user.jwt_exp)
-                current_timestamp = int(datetime.now().timestamp())
+            from backend.utils.time_helpers import parse_jwt_exp, is_timestamp_expired
 
-                if current_timestamp > exp_timestamp:
-                    logger.info(f"用户 {alias} Token 已过期，允许密码登录但需提示用户更新")
-                    token_warning = "token_expired"
-            except ValueError:
-                logger.error(f"用户 {user.id} 的 jwt_exp 格式不正确: {user.jwt_exp}")
+            exp_timestamp = parse_jwt_exp(user.jwt_exp)
+            if exp_timestamp and is_timestamp_expired(exp_timestamp):
+                logger.info(f"用户 {alias} Token 已过期，允许密码登录但需提示用户更新")
+                token_warning = "token_expired"
 
         # 登录成功
         logger.info(f"✅ 用户 {alias} (ID: {user.id}) 别名登录成功")

backend/services/check_in_service.py

@@ -2,7 +2,6 @@ import logging
 from typing import List, Dict, Any, Optional
 from datetime import datetime
 from sqlalchemy.orm import Session
-import json
 import threading
 
 from backend.models import User, CheckInTask, CheckInRecord
@@ -34,20 +33,10 @@ class CheckInService:
 
         try:
             from backend.services.email_service import EmailService
+            from backend.utils.json_helpers import build_task_info
 
-            # 构建 task_info
-            task_info = {
-                'thread_id': '未知',
-                'name': task.name or f'Task-{task.id}'
-            }
-
-            # 尝试从 payload_config 中获取 ThreadId
-            if task.payload_config:
-                try:
-                    payload = json.loads(task.payload_config)
-                    task_info['thread_id'] = payload.get('ThreadId', '未知')
-                except (json.JSONDecodeError, KeyError):
-                    pass
+            # 使用辅助函数构建 task_info
+            task_info = build_task_info(task)
 
             # 发送打卡失败通知（内容包含 Token 失效说明和刷新指引）
             EmailService.notify_check_in_result(user, task_info, False, "Token 已失效，需要重新授权")
@@ -553,12 +542,8 @@ class CheckInService:
             task_name = task.name
 
             # 从 payload_config 提取 ThreadId
-            try:
-                payload = json.loads(str(task.payload_config))
-                thread_id = payload.get('ThreadId')
-            except (json.JSONDecodeError, KeyError, TypeError, AttributeError) as e:
-                logger.debug(f"解析任务 {task.id} 的 payload_config 失败: {e}")
-                pass
+            from backend.utils.json_helpers import extract_thread_id
+            thread_id = extract_thread_id(task.payload_config)  # type: ignore
 
         # 转换为字典并添加额外字段
         record_dict = {

backend/services/email_service.py

@@ -428,12 +428,10 @@ class EmailService:
             return False
 
         # 计算剩余时间
-        try:
-            exp_timestamp = int(jwt_exp)
-            current_timestamp = int(datetime.now().timestamp())
-            minutes_left = (exp_timestamp - current_timestamp) // 60
-        except ValueError:
-            minutes_left = 0
+        from backend.utils.time_helpers import parse_jwt_exp, minutes_until_expiry
+
+        exp_timestamp = parse_jwt_exp(jwt_exp)
+        minutes_left = minutes_until_expiry(exp_timestamp) if exp_timestamp else 0
 
         # 构建邮件内容
         subject = f"【接龙自动打卡系统】登录凭证即将过期 - {user.alias}"

backend/services/scheduler_service.py

@@ -1,7 +1,6 @@
 import logging
 import os
 import time
-from datetime import datetime
 from pathlib import Path
 from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.cron import CronTrigger
@@ -154,6 +153,8 @@ def check_token_expiration():
     检查所有用户的打卡 authorization token，如果在 30 分钟内过期，发送提醒邮件
     注意：检查的是打卡业务 token，不是网站登录 JWT token
     """
+    from backend.utils.time_helpers import seconds_until_expiry
+
     logger.info("Scheduler: 正在执行打卡 Token 过期检查...")
 
     try:
@@ -165,8 +166,6 @@ def check_token_expiration():
             from backend.services.auth_service import AuthService
 
             users = db.query(User).all()
-            current_timestamp = int(datetime.now().timestamp())
-
             notified_count = 0
 
             for user in users:
@@ -178,7 +177,7 @@ def check_token_expiration():
                 if not exp_timestamp:
                     continue
 
-                time_until_expiry = exp_timestamp - current_timestamp
+                time_until_expiry = seconds_until_expiry(exp_timestamp)
 
                 # 情况1：Token 即将过期（过期前 30 分钟内，且还未过期）
                 if 0 < time_until_expiry < 1800:  # 30分钟 = 1800秒

backend/services/task_service.py

@@ -2,7 +2,6 @@ import logging
 from typing import List, Optional, Dict, Any
 from sqlalchemy.orm import Session
 from sqlalchemy import desc
-import json
 
 from backend.models import User, CheckInTask, CheckInRecord
 from backend.schemas.task import TaskCreate, TaskUpdate
@@ -34,35 +33,26 @@ class TaskService:
             raise ValueError(f"用户 ID {user_id} 不存在")
 
         # 2. 从 payload_config 中提取 ThreadId 用于唯一性校验
-        try:
-            payload = json.loads(task_data.payload_config)
-            thread_id = payload.get('ThreadId')
-            if not thread_id:
-                raise ValueError("payload_config 中缺少 ThreadId")
-        except json.JSONDecodeError:
-            raise ValueError("payload_config 格式错误，必须是有效的 JSON")
+        from backend.utils.json_helpers import safe_parse_payload, extract_thread_id
+
+        payload = safe_parse_payload(task_data.payload_config)
+        thread_id = payload.get('ThreadId')
+        if not thread_id:
+            raise ValueError("payload_config 中缺少 ThreadId")
 
         # 3. 验证唯一性：同一用户在同一个接龙中不能有重复的任务
-        # 优化：只查询必要的字段（id 和 payload_config），避免加载完整对象
         existing_tasks = db.query(
-            CheckInTask.id,
             CheckInTask.payload_config
         ).filter(
             CheckInTask.user_id == user_id
         ).all()
 
-        for task_id, payload_config in existing_tasks:
-            try:
-                existing_payload = json.loads(payload_config)
-                if existing_payload.get('ThreadId') == thread_id:
-                    logger.warning(f"⚠️ 任务创建冲突 - User: {user.alias}({user_id}), ThreadId: {thread_id}")
-                    raise ValueError(
-                        f"该接龙中已存在任务。ThreadId: {thread_id}"
-                    )
-            except (json.JSONDecodeError, AttributeError, TypeError):
-                # 跳过无法解析的 payload_config
-                logger.debug(f"跳过无法解析的任务配置 - Task ID: {task_id}")
-                continue
+        for (payload_config,) in existing_tasks:
+            existing_thread_id = extract_thread_id(payload_config)
+            # extract_thread_id 已处理异常，失败时返回 None
+            if existing_thread_id and existing_thread_id == thread_id:
+                logger.warning(f"⚠️ 任务创建冲突 - User: {user.alias}({user_id}), ThreadId: {thread_id}")
+                raise ValueError(f"该接龙中已存在任务。ThreadId: {thread_id}")
 
         # 4. 记录日志
         task_name = task_data.name or f"接龙任务 {thread_id}"
@@ -118,19 +108,15 @@ class TaskService:
         Returns:
             包含额外信息的任务字典
         """
+        from backend.utils.json_helpers import extract_thread_id
+
         # 获取最后一次打卡记录
         last_record = db.query(CheckInRecord).filter(
             CheckInRecord.task_id == task.id
         ).order_by(desc(CheckInRecord.check_in_time)).first()
 
         # 从 payload_config 提取 ThreadId
-        thread_id = None
-        try:
-            payload = json.loads(str(task.payload_config))
-            thread_id = payload.get('ThreadId')
-        except (json.JSONDecodeError, AttributeError, TypeError):
-            logger.debug(f"无法从任务 {task.id} 的 payload_config 中提取 ThreadId")
-            pass
+        thread_id = extract_thread_id(task.payload_config)  # type: ignore
 
         # 转换为字典并添加额外字段
         task_dict = {