Upload app.py

app.py

@@ -2,28 +2,28 @@
 # 此文件专为 Hugging Face Spaces 部署优化
 
 import os
-import uuid
 import time
-from typing import Optional, List
+import uuid
 from contextlib import asynccontextmanager
+from typing import List, Optional
 
+import pymysql
+from dbutils.pooled_db import PooledDB
 from fastapi import FastAPI, HTTPException, Query
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import HTMLResponse, JSONResponse
 from pydantic import BaseModel
-import pymysql
 from pymysql.cursors import DictCursor
-from dbutils.pooled_db import PooledDB
 
 # 数据库配置 - 从环境变量读取
 DB_CONFIG = {
-    'host': os.getenv('DB_HOST', '114.116.200.230'),
-    'port': int(os.getenv('DB_PORT', 3306)),
-    'user': os.getenv('DB_USER', 'aistock_admin'),
-    'password': os.getenv('DB_PASSWORD', ''),  # 必须通过环境变量设置
-    'database': os.getenv('DB_NAME', 'aistock'),
-    'charset': 'utf8mb4',
-    'cursorclass': DictCursor
+    "host": os.getenv("DB_HOST", "114.116.200.230"),
+    "port": int(os.getenv("DB_PORT", 3306)),
+    "user": os.getenv("DB_USER", "aistock_admin"),
+    "password": os.getenv("DB_PASSWORD", ""),  # 必须通过环境变量设置
+    "database": os.getenv("DB_NAME", "aistock"),
+    "charset": "utf8mb4",
+    "cursorclass": DictCursor,
 }
 
 # 连接池
@@ -33,22 +33,24 @@ pool: Optional[PooledDB] = None
 def init_pool():
     """初始化数据库连接池"""
     global pool
-    if not DB_CONFIG['password']:
+    if not DB_CONFIG["password"]:
         print("⚠️ 警告: DB_PASSWORD 环境变量未设置!")
         return
-    
+
     pool = PooledDB(
         creator=pymysql,
-        maxconnections=5,   # Hugging Face 免费版资源有限
+        maxconnections=5,  # Hugging Face 免费版资源有限
         mincached=1,
         maxcached=3,
         blocking=True,
         maxusage=None,
         setsession=[],
         ping=1,
-        **DB_CONFIG
+        **DB_CONFIG,
+    )
+    print(
+        f"✅ MySQL 连接池已初始化: {DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}"
     )
-    print(f"✅ MySQL 连接池已初始化: {DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}")
 
 
 def get_connection():
@@ -72,7 +74,7 @@ app = FastAPI(
     title="Stock Analysis MySQL API",
     description="MySQL API Server for Stock Analysis Agent (Hugging Face Spaces)",
     version="1.0.0",
-    lifespan=lifespan
+    lifespan=lifespan,
 )
 
 # CORS 中间件 - 允许 Cloudflare Worker 访问
@@ -87,6 +89,7 @@ app.add_middleware(
 
 # ============ Pydantic 模型 ============
 
+
 class SubmitRequest(BaseModel):
     stock_code: str
     stock_name: str
@@ -116,6 +119,7 @@ class RemoveDuplicatesRequest(BaseModel):
 
 # ============ API 端点 ============
 
+
 @app.get("/")
 async def root():
     """健康检查"""
@@ -123,7 +127,7 @@ async def root():
         "status": "ok",
         "service": "MySQL API Server",
         "platform": "Hugging Face Spaces",
-        "db_configured": bool(DB_CONFIG['password'])
+        "db_configured": bool(DB_CONFIG["password"]),
     }
 
 
@@ -133,24 +137,28 @@ async def submit_task(req: SubmitRequest):
     conn = get_connection()
     # 获取 username（优先使用 username，其次 submitted_by）
     username = req.username or req.submitted_by or None
-    
+
     try:
         with conn.cursor() as cursor:
             cursor.execute(
                 "SELECT id, submit_count FROM requests WHERE stock_code = %s AND stock_name = %s AND status IN ('pending', 'processing') LIMIT 1",
-                (req.stock_code, req.stock_name)
+                (req.stock_code, req.stock_name),
             )
             existing = cursor.fetchone()
-            
+
             if existing:
-                new_count = (existing.get('submit_count') or 1) + 1
+                new_count = (existing.get("submit_count") or 1) + 1
                 cursor.execute(
                     "UPDATE requests SET submit_count = %s WHERE id = %s",
-                    (new_count, existing['id'])
+                    (new_count, existing["id"]),
                 )
                 conn.commit()
-                return {"success": True, "request_id": existing['id'], "duplicate": True}
-            
+                return {
+                    "success": True,
+                    "request_id": existing["id"],
+                    "duplicate": True,
+                }
+
             request_id = str(uuid.uuid4())
             now = int(time.time() * 1000)
             # 处理 is_public 字段：确保值为 0 或 1
@@ -161,13 +169,24 @@ async def submit_task(req: SubmitRequest):
                 is_public = 0
             else:
                 is_public = 1  # 默认公开
-            
+
             # 处理 is_vip 字段
             is_vip = 1 if req.is_vip == 1 or req.is_vip == True else 0
-            
+
             cursor.execute(
                 "INSERT INTO requests (id, stock_code, stock_name, market, status, created_at, submit_count, username, is_public, is_vip) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
-                (request_id, req.stock_code, req.stock_name, req.market, 'pending', now, 1, username, is_public, is_vip)
+                (
+                    request_id,
+                    req.stock_code,
+                    req.stock_name,
+                    req.market,
+                    "pending",
+                    now,
+                    1,
+                    username,
+                    is_public,
+                    is_vip,
+                ),
             )
             conn.commit()
             return {"success": True, "request_id": request_id, "username": username}
@@ -218,20 +237,20 @@ async def complete_task(req: CompleteRequest):
     try:
         with conn.cursor() as cursor:
             now = int(time.time() * 1000)
-            
+
             cursor.execute(
                 "UPDATE requests SET status = %s, completed_at = %s WHERE id = %s",
-                (req.status, now, req.id)
+                (req.status, now, req.id),
             )
-            
-            if req.status == 'completed' and req.html_content and req.stock_code:
+
+            if req.status == "completed" and req.html_content and req.stock_code:
                 # 生成报告 ID
                 report_id = str(uuid.uuid4())
                 cursor.execute(
                     "INSERT INTO reports (id, stock_code, stock_name, html_content, created_at) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE stock_name = VALUES(stock_name), html_content = VALUES(html_content), created_at = VALUES(created_at)",
-                    (report_id, req.stock_code, req.stock_name, req.html_content, now)
+                    (report_id, req.stock_code, req.stock_name, req.html_content, now),
                 )
-            
+
             conn.commit()
             return {"success": True}
     except Exception as e:
@@ -246,14 +265,18 @@ async def get_report(code: str = Query(...)):
     conn = get_connection()
     try:
         with conn.cursor() as cursor:
-            cursor.execute("SELECT html_content FROM reports WHERE stock_code = %s", (code,))
+            # reports 表可能存在同一 stock_code 的多条历史记录（按 created_at 递增）
+            # 这里必须取最新的一条，否则会出现“列表时间是最新、打开报告却是旧内容”的错配
+            cursor.execute(
+                "SELECT html_content FROM reports WHERE stock_code = %s ORDER BY created_at DESC LIMIT 1",
+                (code,),
+            )
             result = cursor.fetchone()
-            if not result or not result.get('html_content'):
+            if not result or not result.get("html_content"):
                 return HTMLResponse(
-                    content=f"<h1>Report not found for {code}</h1>",
-                    status_code=404
+                    content=f"<h1>Report not found for {code}</h1>", status_code=404
                 )
-            return HTMLResponse(content=result['html_content'])
+            return HTMLResponse(content=result["html_content"])
     except Exception as e:
         return JSONResponse(status_code=500, content={"error": str(e)})
     finally:
@@ -261,7 +284,9 @@ async def get_report(code: str = Query(...)):
 
 
 @app.get("/api/history")
-async def get_history(date: Optional[str] = Query(None), username: Optional[str] = Query(None)):
+async def get_history(
+    date: Optional[str] = Query(None), username: Optional[str] = Query(None)
+):
     """获取历史报告列表，支持按日期和用户名筛选"""
     conn = get_connection()
     try:
@@ -270,29 +295,38 @@ async def get_history(date: Optional[str] = Query(None), username: Optional[str]
             if username:
                 if date:
                     from datetime import datetime
-                    start_dt = datetime.strptime(date, '%Y-%m-%d')
+
+                    start_dt = datetime.strptime(date, "%Y-%m-%d")
                     start_ts = int(start_dt.timestamp() * 1000)
                     end_ts = start_ts + 86400000
-                    cursor.execute("""
+                    cursor.execute(
+                        """
                         SELECT * FROM requests
                         WHERE username = %s AND status = 'completed' AND completed_at >= %s AND completed_at < %s
                         ORDER BY completed_at DESC
-                    """, (username, start_ts, end_ts))
+                    """,
+                        (username, start_ts, end_ts),
+                    )
                 else:
-                    cursor.execute("""
+                    cursor.execute(
+                        """
                         SELECT * FROM requests
                         WHERE username = %s AND status IN ('completed', 'error', 'pending', 'processing')
                         ORDER BY created_at DESC
                         LIMIT 100
-                    """, (username,))
+                    """,
+                        (username,),
+                    )
             elif date:
                 from datetime import datetime
-                start_dt = datetime.strptime(date, '%Y-%m-%d')
+
+                start_dt = datetime.strptime(date, "%Y-%m-%d")
                 start_ts = int(start_dt.timestamp() * 1000)
                 end_ts = start_ts + 86400000
-                
+
                 # 首页调用（无username）：只返回公开报告 (is_public = 1 或 is_public IS NULL)
-                cursor.execute("""
+                cursor.execute(
+                    """
                     SELECT r1.* FROM requests r1
                     INNER JOIN (
                         SELECT stock_code, MAX(completed_at) as max_completed_at
@@ -303,7 +337,9 @@ async def get_history(date: Optional[str] = Query(None), username: Optional[str]
                     ) r2 ON r1.stock_code = r2.stock_code AND r1.completed_at = r2.max_completed_at
                     WHERE r1.status = 'completed' AND (r1.is_public = 1 OR r1.is_public IS NULL)
                     ORDER BY r1.completed_at DESC
-                """, (start_ts, end_ts))
+                """,
+                    (start_ts, end_ts),
+                )
             else:
                 # 首页调用（无username，无date）：只返回公开报告 (is_public = 1 或 is_public IS NULL)
                 cursor.execute("""
@@ -318,7 +354,7 @@ async def get_history(date: Optional[str] = Query(None), username: Optional[str]
                     WHERE r1.status = 'completed' AND (r1.is_public = 1 OR r1.is_public IS NULL)
                     ORDER BY r1.completed_at DESC
                 """)
-            
+
             tasks = cursor.fetchall()
             return {"tasks": tasks}
     except Exception as e:
@@ -334,16 +370,16 @@ async def check_report(code: str = Query(...)):
     try:
         with conn.cursor() as cursor:
             cursor.execute(
-                "SELECT stock_code, stock_name, created_at FROM reports WHERE stock_code = %s",
-                (code,)
+                "SELECT stock_code, stock_name, created_at FROM reports WHERE stock_code = %s ORDER BY created_at DESC LIMIT 1",
+                (code,),
             )
             result = cursor.fetchone()
             if result:
                 return {
                     "exists": True,
-                    "stock_code": result['stock_code'],
-                    "stock_name": result['stock_name'],
-                    "created_at": result['created_at']
+                    "stock_code": result["stock_code"],
+                    "stock_name": result["stock_name"],
+                    "created_at": result["created_at"],
                 }
             return {"exists": False}
     except Exception as e:
@@ -362,12 +398,16 @@ async def reorder_task(req: ReorderRequest):
                 "SELECT MIN(created_at) as min_time FROM requests WHERE status = 'pending'"
             )
             result = cursor.fetchone()
-            min_time = result['min_time'] if result and result['min_time'] else int(time.time() * 1000)
+            min_time = (
+                result["min_time"]
+                if result and result["min_time"]
+                else int(time.time() * 1000)
+            )
             new_time = min_time - 1
-            
+
             cursor.execute(
                 "UPDATE requests SET created_at = %s WHERE id = %s AND status = 'pending'",
-                (new_time, req.id)
+                (new_time, req.id),
             )
             conn.commit()
             return {"success": True, "new_time": new_time}
@@ -381,15 +421,17 @@ async def reorder_task(req: ReorderRequest):
 async def remove_duplicates(req: RemoveDuplicatesRequest):
     """批量删除重复任务"""
     if not req.ids:
-        return JSONResponse(status_code=400, content={"error": "Missing or invalid task IDs"})
-    
+        return JSONResponse(
+            status_code=400, content={"error": "Missing or invalid task IDs"}
+        )
+
     conn = get_connection()
     try:
         with conn.cursor() as cursor:
-            placeholders = ', '.join(['%s'] * len(req.ids))
+            placeholders = ", ".join(["%s"] * len(req.ids))
             cursor.execute(
                 f"DELETE FROM requests WHERE id IN ({placeholders}) AND status = 'pending'",
-                tuple(req.ids)
+                tuple(req.ids),
             )
             deleted = cursor.rowcount
             conn.commit()
@@ -419,29 +461,40 @@ async def get_queue():
 
 # ============ 板块轮动报告 API ============
 
+
 @app.get("/api/sector-rotation-reports")
 async def get_sector_rotation_reports(limit: int = Query(default=10)):
     """获取最近的板块轮动报告列表"""
     conn = get_connection()
     try:
         with conn.cursor() as cursor:
-            cursor.execute("""
+            cursor.execute(
+                """
                 SELECT trade_date, updated_at
                 FROM sector_rotation_reports
                 ORDER BY trade_date DESC
                 LIMIT %s
-            """, (limit,))
+            """,
+                (limit,),
+            )
             reports = cursor.fetchall()
-            
+
             # 格式化返回数据
             result = []
             for report in reports:
-                result.append({
-                    "trade_date": report['trade_date'],
-                    "update_time": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(report['updated_at'] / 1000)) if report['updated_at'] else None,
-                    "updated_at": report['updated_at']
-                })
-            
+                result.append(
+                    {
+                        "trade_date": report["trade_date"],
+                        "update_time": time.strftime(
+                            "%Y-%m-%d %H:%M:%S",
+                            time.localtime(report["updated_at"] / 1000),
+                        )
+                        if report["updated_at"]
+                        else None,
+                        "updated_at": report["updated_at"],
+                    }
+                )
+
             return {"reports": result}
     except Exception as e:
         return JSONResponse(status_code=500, content={"error": str(e)})
@@ -455,23 +508,26 @@ async def get_sector_rotation_report(trade_date: str):
     conn = get_connection()
     try:
         with conn.cursor() as cursor:
-            cursor.execute("""
+            cursor.execute(
+                """
                 SELECT trade_date, html_content, updated_at
                 FROM sector_rotation_reports
                 WHERE trade_date = %s
-            """, (trade_date,))
+            """,
+                (trade_date,),
+            )
             report = cursor.fetchone()
-            
+
             if not report:
                 return JSONResponse(
                     status_code=404,
-                    content={"error": f"未找到 {trade_date} 的板块轮动报告"}
+                    content={"error": f"未找到 {trade_date} 的板块轮动报告"},
                 )
-            
+
             return {
-                "trade_date": report['trade_date'],
-                "html_content": report['html_content'],
-                "updated_at": report['updated_at']
+                "trade_date": report["trade_date"],
+                "html_content": report["html_content"],
+                "updated_at": report["updated_at"],
             }
     except Exception as e:
         return JSONResponse(status_code=500, content={"error": str(e)})
@@ -479,128 +535,149 @@ async def get_sector_rotation_report(trade_date: str):
         conn.close()
 
 
-@app.get("/api/sector-rotation-reports/{trade_date}/html")
-async def get_sector_rotation_report_html(trade_date: str):
-    """获取指定交易日的板块轮动报告（直接返回HTML页面）"""
-    conn = get_connection()
-    try:
+@app.get("/api/sector-rotation-reports/{trade_date}/html")
+async def get_sector_rotation_report_html(trade_date: str):
+    """获取指定交易日的板块轮动报告（直接返回HTML页面）"""
+    conn = get_connection()
+    try:
         with conn.cursor() as cursor:
-            cursor.execute("""
+            cursor.execute(
+                """
                 SELECT html_content
                 FROM sector_rotation_reports
                 WHERE trade_date = %s
-            """, (trade_date,))
+            """,
+                (trade_date,),
+            )
             report = cursor.fetchone()
-            
-            if not report or not report['html_content']:
+
+            if not report or not report["html_content"]:
                 return HTMLResponse(
                     content=f"<html><body><h1>未找到 {trade_date} 的板块轮动报告</h1></body></html>",
-                    status_code=404
+                    status_code=404,
                 )
-            
-            return HTMLResponse(content=report['html_content'])
+
+            return HTMLResponse(content=report["html_content"])
     except Exception as e:
         return HTMLResponse(
             content=f"<html><body><h1>加载失败</h1><p>{str(e)}</p></body></html>",
-            status_code=500
+            status_code=500,
         )
-    finally:
-        conn.close()
-
-
-# ============ 龙头股对比分析报告 API ============
-
-@app.get("/api/longtou-compare-reports")
-async def get_longtou_compare_reports(limit: int = Query(default=10)):
-    """获取最近的龙头股对比分析报告列表"""
-    conn = get_connection()
-    try:
-        with conn.cursor() as cursor:
-            cursor.execute("""
-                SELECT trade_date, updated_at
-                FROM longtou_compare_reports
-                ORDER BY trade_date DESC
-                LIMIT %s
-            """, (limit,))
-            reports = cursor.fetchall()
-            
-            result = []
-            for report in reports:
-                result.append({
-                    "trade_date": report['trade_date'],
-                    "update_time": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(report['updated_at'] / 1000)) if report['updated_at'] else None,
-                    "updated_at": report['updated_at']
-                })
-            
-            return {"reports": result}
-    except Exception as e:
-        return JSONResponse(status_code=500, content={"error": str(e)})
-    finally:
-        conn.close()
-
-
-@app.get("/api/longtou-compare-reports/{trade_date}")
-async def get_longtou_compare_report(trade_date: str):
-    """获取指定交易日的龙头股对比分析报告"""
-    conn = get_connection()
-    try:
-        with conn.cursor() as cursor:
-            cursor.execute("""
-                SELECT trade_date, html_content, updated_at
-                FROM longtou_compare_reports
-                WHERE trade_date = %s
-            """, (trade_date,))
-            report = cursor.fetchone()
-            
-            if not report:
-                return JSONResponse(
-                    status_code=404,
-                    content={"error": f"未找到 {trade_date} 的龙头股对比分析报告"}
-                )
-            
-            return {
-                "trade_date": report['trade_date'],
-                "html_content": report['html_content'],
-                "updated_at": report['updated_at']
-            }
-    except Exception as e:
-        return JSONResponse(status_code=500, content={"error": str(e)})
-    finally:
-        conn.close()
-
-
-@app.get("/api/longtou-compare-reports/{trade_date}/html")
-async def get_longtou_compare_report_html(trade_date: str):
-    """获取指定交易日的龙头股对比分析报告（直接返回HTML页面）"""
-    conn = get_connection()
-    try:
-        with conn.cursor() as cursor:
-            cursor.execute("""
-                SELECT html_content
-                FROM longtou_compare_reports
-                WHERE trade_date = %s
-            """, (trade_date,))
-            report = cursor.fetchone()
-            
-            if not report or not report['html_content']:
-                return HTMLResponse(
-                    content=f"<html><body><h1>未找到 {trade_date} 的龙头股对比分析报告</h1></body></html>",
-                    status_code=404
-                )
-            
-            return HTMLResponse(content=report['html_content'])
-    except Exception as e:
-        return HTMLResponse(
-            content=f"<html><body><h1>加载失败</h1><p>{str(e)}</p></body></html>",
-            status_code=500
-        )
-    finally:
-        conn.close()
-
-
-# Hugging Face Spaces 使用 7860 端口
-if __name__ == "__main__":
-    import uvicorn
-    port = int(os.getenv('PORT', 7860))
+    finally:
+        conn.close()
+
+
+# ============ 龙头股对比分析报告 API ============
+
+
+@app.get("/api/longtou-compare-reports")
+async def get_longtou_compare_reports(limit: int = Query(default=10)):
+    """获取最近的龙头股对比分析报告列表"""
+    conn = get_connection()
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT trade_date, updated_at
+                FROM longtou_compare_reports
+                ORDER BY trade_date DESC
+                LIMIT %s
+            """,
+                (limit,),
+            )
+            reports = cursor.fetchall()
+
+            result = []
+            for report in reports:
+                result.append(
+                    {
+                        "trade_date": report["trade_date"],
+                        "update_time": time.strftime(
+                            "%Y-%m-%d %H:%M:%S",
+                            time.localtime(report["updated_at"] / 1000),
+                        )
+                        if report["updated_at"]
+                        else None,
+                        "updated_at": report["updated_at"],
+                    }
+                )
+
+            return {"reports": result}
+    except Exception as e:
+        return JSONResponse(status_code=500, content={"error": str(e)})
+    finally:
+        conn.close()
+
+
+@app.get("/api/longtou-compare-reports/{trade_date}")
+async def get_longtou_compare_report(trade_date: str):
+    """获取指定交易日的龙头股对比分析报告"""
+    conn = get_connection()
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT trade_date, html_content, updated_at
+                FROM longtou_compare_reports
+                WHERE trade_date = %s
+            """,
+                (trade_date,),
+            )
+            report = cursor.fetchone()
+
+            if not report:
+                return JSONResponse(
+                    status_code=404,
+                    content={"error": f"未找到 {trade_date} 的龙头股对比分析报告"},
+                )
+
+            return {
+                "trade_date": report["trade_date"],
+                "html_content": report["html_content"],
+                "updated_at": report["updated_at"],
+            }
+    except Exception as e:
+        return JSONResponse(status_code=500, content={"error": str(e)})
+    finally:
+        conn.close()
+
+
+@app.get("/api/longtou-compare-reports/{trade_date}/html")
+async def get_longtou_compare_report_html(trade_date: str):
+    """获取指定交易日的龙头股对比分析报告（直接返回HTML页面）"""
+    conn = get_connection()
+    try:
+        with conn.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT html_content
+                FROM longtou_compare_reports
+                WHERE trade_date = %s
+            """,
+                (trade_date,),
+            )
+            report = cursor.fetchone()
+
+            if not report or not report["html_content"]:
+                return HTMLResponse(
+                    content=f"<html><body><h1>未找到 {trade_date} 的龙头股对比分析报告</h1></body></html>",
+                    status_code=404,
+                )
+
+            return HTMLResponse(content=report["html_content"])
+    except Exception as e:
+        return HTMLResponse(
+            content=f"<html><body><h1>加载失败</h1><p>{str(e)}</p></body></html>",
+            status_code=500,
+        )
+    finally:
+        conn.close()
+
+
+# Hugging Face Spaces 使用 7860 端口
+if __name__ == "__main__":
+    import uvicorn
+
+    port = int(os.getenv("PORT", 7860))
     print(f"🚀 Starting MySQL API Server on port {port}...")
     uvicorn.run(app, host="0.0.0.0", port=port)