[codex] harden today top news AI response parsing

app.py

@@ -60,6 +60,25 @@ class Config:
         title = "hot_topic_detail"
 
 
+def parse_hot_topics_response(text):
+    if not text or not text.strip():
+        raise ValueError("AI response is empty")
+
+    parsed = json.loads(repair_json(text))
+    if not isinstance(parsed, dict):
+        raise ValueError("AI response is invalid: expected object with hot_topics")
+
+    topics = parsed.get("hot_topics")
+    if not isinstance(topics, list):
+        raise ValueError("AI response is invalid: hot_topics must be a list")
+
+    for index, topic in enumerate(topics):
+        if not isinstance(topic, dict):
+            raise ValueError(f"AI response is invalid: hot_topics[{index}] must be an object")
+
+    return topics
+
+
 limiter = Limiter(key_func=get_remote_address, default_limits=["30 per minute"], storage_uri=f"redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}")
 backoff = ExponentialBackoff(cap=2, base=2)
 retry = Retry(backoff=backoff, retries=10)
@@ -379,8 +398,8 @@ async def getTodayTopNews():
                             "user": "请从下方数据中选出5条你认为最应该让我知道的内容,返回json格式数据,不要改变原有的数据内容,返回格式{'hot_topics': [{hot_label:'',hot_url:'',hot_value:''}]}\ndata:" + json.dumps(true_sites_data)
                         }
                     , HotTopics.model_json_schema())
-                today_top_news_data = json.loads(repair_json(text))
-                need_knows = await parse_detail(today_top_news_data.get("hot_topics", []))
+                hot_topics = parse_hot_topics_response(text)
+                need_knows = await parse_detail(hot_topics)
                 summarizes = []
                 for needKnow in need_knows:
                     err = 3

tests/test_today_top_news.py

@@ -0,0 +1,80 @@
+import unittest
+import sys
+import types
+from unittest.mock import patch
+
+config = types.ModuleType("config")
+config.REDIS_PASSWORD = ""
+config.REDIS_HOST = "localhost"
+config.REDIS_PORT = 6379
+config.REDIS_DB = 0
+config.PG_HOST = "localhost"
+config.PG_PORT = 5432
+config.PG_USER = "user"
+config.PG_PASSWORD = "password"
+config.PG_DB = "db"
+config.api_url = "https://example.com"
+config.api_headers = {}
+config.news_sites = ["澎湃新闻"]
+config.EMAIL = {
+    "user": "test@example.com",
+    "host": "smtp.example.com",
+    "port": 587,
+    "password": "password",
+}
+sys.modules["config"] = config
+
+import app
+
+
+class FakeRedis:
+    async def get(self, key):
+        return None
+
+    async def set(self, *args, **kwargs):
+        return True
+
+    async def delete(self, *args, **kwargs):
+        return 1
+
+    async def setex(self, *args, **kwargs):
+        return True
+
+
+class TodayTopNewsTest(unittest.IsolatedAsyncioTestCase):
+    async def test_empty_model_response_returns_clear_error(self):
+        async def fake_get_data(item_id):
+            return {
+                "code": 200,
+                "msg": "success",
+                "data": [
+                    {
+                        "name": "澎湃新闻",
+                        "data": [
+                            {
+                                "hot_label": "A",
+                                "hot_url": "https://example.com",
+                                "hot_value": "1",
+                            }
+                        ],
+                    }
+                ],
+            }
+
+        async def fake_chat_with_model(messages, response_format):
+            return ""
+
+        with patch.object(app, "redis_client", FakeRedis()), patch.object(
+            app, "get_data", fake_get_data
+        ), patch.object(app, "chatWithModel", fake_chat_with_model), patch.object(
+            app.logging, "error"
+        ):
+            result = await app.getTodayTopNews()
+
+        self.assertEqual(result["code"], 500)
+        self.assertIn("AI response is empty", result["msg"])
+        self.assertNotIn("object has no attribute", result["msg"])
+
+
+if __name__ == "__main__":
+    unittest.main()