Upload 2 files

app.py

@@ -10,7 +10,9 @@ import uuid
 import threading
 import time
 import random
+from fastapi import FastAPI
 from fastapi.responses import FileResponse, JSONResponse
+import uvicorn
 
 # 环境变量
 ACCESS_PASSWORD = os.environ.get("ACCESS_PASSWORD", "changeme")
@@ -395,6 +397,89 @@ init_success = init_db()
 keep_alive_thread = threading.Thread(target=keep_alive, daemon=True)
 keep_alive_thread.start()
 
+# 创建FastAPI应用
+app = FastAPI(title="私人图床服务")
+
+# 添加图片访问API路由（在Gradio之前）
+@app.get("/api/img/{image_hash}")
+async def get_image(image_hash: str):
+    """通过hash获取图片"""
+    try:
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        cursor.execute("SELECT * FROM images WHERE hash = ?", (image_hash,))
+        row = cursor.fetchone()
+        conn.close()
+        
+        if not row:
+            return JSONResponse(
+                status_code=404,
+                content={"error": "Image not found"}
+            )
+        
+        file_path = Path(row['file_path'])
+        
+        if not file_path.exists():
+            return JSONResponse(
+                status_code=404,
+                content={"error": "Image file not found"}
+            )
+        
+        # 返回图片文件
+        return FileResponse(
+            file_path,
+            media_type=row['mime_type'],
+            headers={
+                "Cache-Control": "public, max-age=31536000",
+                "Content-Disposition": f'inline; filename="{row["original_filename"]}"'
+            }
+        )
+    
+    except Exception as e:
+        return JSONResponse(
+            status_code=500,
+            content={"error": str(e)}
+        )
+
+@app.get("/api/img/{image_hash}/info")
+async def get_image_info(image_hash: str, password: str = None):
+    """获取图片信息（需要密码）"""
+    if not password or not check_password(password):
+        return JSONResponse(
+            status_code=401,
+            content={"error": "Unauthorized"}
+        )
+    
+    try:
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        cursor.execute("SELECT * FROM images WHERE hash = ?", (image_hash,))
+        row = cursor.fetchone()
+        conn.close()
+        
+        if not row:
+            return JSONResponse(
+                status_code=404,
+                content={"error": "Image not found"}
+            )
+        
+        return JSONResponse(content={
+            "hash": row['hash'],
+            "filename": row['filename'],
+            "original_filename": row['original_filename'],
+            "file_size": row['file_size'],
+            "mime_type": row['mime_type'],
+            "upload_time": row['upload_time'],
+            "description": row['description'],
+            "url": generate_full_url(row['hash'])
+        })
+    
+    except Exception as e:
+        return JSONResponse(
+            status_code=500,
+            content={"error": str(e)}
+        )
+
 # 自定义CSS
 custom_css = """
 .compact-gallery .gallery {
@@ -540,85 +625,8 @@ with gr.Blocks(title="My图床", theme=gr.themes.Soft(), css=custom_css) as grad
                     outputs=[export_output, export_file_output]
                 )
 
-# 添加FastAPI路由到Gradio的底层FastAPI应用
-@gradio_app.app.get("/api/img/{image_hash}")
-async def get_image(image_hash: str):
-    """通过hash获取图片"""
-    try:
-        conn = get_db_connection()
-        cursor = conn.cursor()
-        cursor.execute("SELECT * FROM images WHERE hash = ?", (image_hash,))
-        row = cursor.fetchone()
-        conn.close()
-        
-        if not row:
-            return JSONResponse(
-                status_code=404,
-                content={"error": "Image not found"}
-            )
-        
-        file_path = Path(row['file_path'])
-        
-        if not file_path.exists():
-            return JSONResponse(
-                status_code=404,
-                content={"error": "Image file not found"}
-            )
-        
-        # 返回图片文件
-        return FileResponse(
-            file_path,
-            media_type=row['mime_type'],
-            headers={
-                "Cache-Control": "public, max-age=31536000",
-                "Content-Disposition": f'inline; filename="{row["original_filename"]}"'
-            }
-        )
-    
-    except Exception as e:
-        return JSONResponse(
-            status_code=500,
-            content={"error": str(e)}
-        )
-
-@gradio_app.app.get("/api/img/{image_hash}/info")
-async def get_image_info(image_hash: str, password: str = None):
-    """获取图片信息（需要密码）"""
-    if not password or not check_password(password):
-        return JSONResponse(
-            status_code=401,
-            content={"error": "Unauthorized"}
-        )
-    
-    try:
-        conn = get_db_connection()
-        cursor = conn.cursor()
-        cursor.execute("SELECT * FROM images WHERE hash = ?", (image_hash,))
-        row = cursor.fetchone()
-        conn.close()
-        
-        if not row:
-            return JSONResponse(
-                status_code=404,
-                content={"error": "Image not found"}
-            )
-        
-        return JSONResponse(content={
-            "hash": row['hash'],
-            "filename": row['filename'],
-            "original_filename": row['original_filename'],
-            "file_size": row['file_size'],
-            "mime_type": row['mime_type'],
-            "upload_time": row['upload_time'],
-            "description": row['description'],
-            "url": generate_full_url(row['hash'])
-        })
-    
-    except Exception as e:
-        return JSONResponse(
-            status_code=500,
-            content={"error": str(e)}
-        )
+# 将Gradio应用挂载到FastAPI
+app = gr.mount_gradio_app(app, gradio_app, path="/")
 
 if __name__ == "__main__":
-    gradio_app.launch(server_name="0.0.0.0", server_port=7860)
+    uvicorn.run(app, host="0.0.0.0", port=7860)

requirements.txt

@@ -1,3 +1,5 @@
 gradio==4.44.0
 Pillow==10.4.0
+fastapi>=0.104.0
+uvicorn[standard]>=0.24.0