fix: update JWT expiration to use timezone-aware UTC datetime and enhance caching logic for search responses

Author: binaryYuki

_auth.py

@@ -19,7 +19,8 @@ async def generateJWT(payload: dict):
     生成 JWT Token
     :return: str
     """
-    payload['exp'] = datetime.datetime.utcnow() + datetime.timedelta(hours=12)
+    # Use timezone-aware UTC datetime per deprecation guidance
+    payload['exp'] = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=12)
     return jwt.encode(payload, os.getenv('SESSION_SECRET'), algorithm="HS256")
 
 

_search.py

@@ -128,10 +128,14 @@ async def search(request: Request, background_tasks: BackgroundTasks):
     try:
         id = f"search_{keyword}_{page}_{size}_{datetime.datetime.now().strftime('%Y-%m-%d')}"
         if await redis_key_exists(id):
-            data = json.loads(await redis_get_key(id))
-            data["msg"] = "cached"
-            return JSONResponse(data)
+            cached = await redis_get_key(id)
+            if cached is not None:
+                cached_data = json.loads(cached)
+                if isinstance(cached_data, dict):
+                    cached_data["msg"] = "cached"
+                return JSONResponse(cached_data, status_code=200, headers={"X-Cache": "HIT" if "msg" in cached_data and cached_data["msg"] == "cached" else "MISS"})
     except Exception as e:
+        logging.info(f"Invalid Request: {data}, {e}")
         pass
     try:
         result = await search_api(keyword, page, size)
@@ -167,10 +171,12 @@ async def keyword(request: Request):
              "msg": "ok"}, status_code=200, headers={"X-Info": "Success"})
     redis_key = f"keyword_{datetime.datetime.now().strftime('%Y-%m-%d')}_{_keyword}"
     try:
-        if await redis_get_key(redis_key):
-            data = await redis_get_key(redis_key)
-            data = json.loads(data)
-            data["msg"] = "cached"
+        cached = await redis_get_key(redis_key)
+        if cached is not None:
+            parsed = json.loads(cached)
+            if isinstance(parsed, dict):
+                parsed["msg"] = "cached"
+            data = parsed
         else:
             data = await link_keywords(_keyword)
             await redis_set_key(redis_key, json.dumps(data), ex=86400)  # 缓存一天

_utils.py

@@ -3,7 +3,9 @@
 import json
 import logging
 import urllib
+import urllib.parse
 import uuid
+from typing import Optional
 
 from fake_useragent import UserAgent
 from httpx import AsyncClient
@@ -83,17 +85,16 @@ def _getRandomUserAgent():
 
 
 async def pushNotification(baseURL: str, msg: str, icon: str = '', click_url: str = '', is_passive: bool = False,
-                           headers=None):
+                           headers: Optional[dict] = None):
     """
     推送通知
-    :param baseURL: str
-    :param msg: str
-    :param icon: str
-    :param click_url: str
-    :param is_passive: bool
-    :param headers: dict
-    :param data: dict
-    :param log_data: dict -> {'push_id': str, 'push_receiver': str}
+    :param baseURL: 推送服务基础 URL
+    :param msg: 推送消息内容
+    :param icon: 图标 URL，可选
+    :param click_url: 点击跳转 URL，可选
+    :param is_passive: 是否被动推送（静默）
+    :param headers: 额外请求头，可选
+    :return: bool 是否推送成功
     """
     # url = https://api.day.app/uKeSrwm3ainGgn5SAmRyg9/{msg}?icon={icon}&url={url}&passive={is_passive}
     if headers is None:
@@ -118,25 +119,29 @@ async def pushNotification(baseURL: str, msg: str, icon: str = '', click_url: st
 
 # url 编码关键词
 def url_encode(keyword):
-    return urllib.parse.quote(keyword.encode())
+    # ensure str input and use urllib.parse.quote for encoding
+    if isinstance(keyword, bytes):
+        keyword = keyword.decode('utf-8', errors='ignore')
+    return urllib.parse.quote(str(keyword))
 
 
 async def generatePushTask(baseURL: str, msg: str, user_id: str, receiver: str, icon=None, click_url=None,
-                           is_passive=None, headers: dict = None, taskID: str = uuid.uuid4().hex,
+                           is_passive=None, headers: Optional[dict] = None, taskID: str = uuid.uuid4().hex,
                            push_receiver: str = "yuki", push_by: str = "system"):
     """
-    :param push_by:  推送者 默认为system
-    :param push_channel:  推送渠道 默认为bark
-    :param push_receiver: 用户email
-    :param taskID:  任务ID 默认为uuid
-    :param headers:  请求头
-    :param icon:  图标
-    :param user_id:  用户ID
-    :param msg:  消息
-    :param baseURL:  基础URL
-    :param click_url: 点击通知后跳转的URL
-    :param receiver: 接收者
-    :param is_passive: 是否被动推送 就是不会有声音
+    生成推送任务并暂存到 Redis 队列
+    :param baseURL: 推送服务基础 URL
+    :param msg: 推送消息内容
+    :param user_id: 用户 ID
+    :param receiver: 接收者标识
+    :param icon: 图标 URL，可选
+    :param click_url: 点击跳转 URL，可选
+    :param is_passive: 是否被动推送（静默）
+    :param headers: 请求头，可选
+    :param taskID: 任务 ID（默认随机）
+    :param push_receiver: 推送接收者（日志）
+    :param push_by: 推送发起者（日志）
+    :return: bool
     示例： generatePushTask("https://api.day.app/uKeSrwm3ainGgn5SAmRyg9/", "You have a new notification!", str(12345),
                            "https://example.com", False, None, None, None, uuid.uuid4().hex, "system",
                             "bark")

app.py

@@ -281,5 +281,7 @@ async def add_process_time_header(request, call_next):
     import uvicorn
     import watchfiles
 
-    watchfiles.filters = ["*venv", "\\.env$"]
+    # Some versions of watchfiles don't expose a `filters` attribute; guard access.
+    if hasattr(watchfiles, 'filters'):
+        watchfiles.filters = ["*venv", "\\.env$"]
     uvicorn.run(app, host="0.0.0.0", port=8000)