Upload 2 files

app.py

@@ -1,100 +1,135 @@
 import gradio as gr
 import os
-import base64
-from datetime import datetime
-from libsql_client import create_client
+import sqlite3
 import hashlib
-import io
+import shutil
+from datetime import datetime
+from pathlib import Path
 from PIL import Image
+import json
 
 # 环境变量
-TURSO_DATABASE_URL = os.environ.get("TURSO_DATABASE_URL")
-TURSO_AUTH_TOKEN = os.environ.get("TURSO_AUTH_TOKEN")
 ACCESS_PASSWORD = os.environ.get("ACCESS_PASSWORD", "changeme")
 
-# 创建数据库客户端
-client = None
+# 文件存储路径
+IMAGE_DIR = Path("uploaded_images")
+DB_PATH = "image_database.db"
+
+# 创建图片存储目录
+IMAGE_DIR.mkdir(exist_ok=True)
 
 def init_db():
-    """初始化数据库连接和表"""
-    global client
+    """初始化数据库"""
     try:
-        client = create_client(
-            url=TURSO_DATABASE_URL,
-            auth_token=TURSO_AUTH_TOKEN
-        )
+        conn = sqlite3.connect(DB_PATH)
+        cursor = conn.cursor()
         
         # 创建图片表
-        client.execute("""
+        cursor.execute("""
             CREATE TABLE IF NOT EXISTS images (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 filename TEXT NOT NULL,
-                image_data TEXT NOT NULL,
+                original_filename TEXT NOT NULL,
+                file_path TEXT NOT NULL,
                 file_size INTEGER,
                 mime_type TEXT,
                 upload_time TEXT NOT NULL,
                 description TEXT
             )
         """)
-        print("数据库初始化成功")
+        
+        conn.commit()
+        conn.close()
+        print("✅ 数据库初始化成功")
         return True
     except Exception as e:
-        print(f"数据库初始化失败: {e}")
+        print(f"❌ 数据库初始化失败: {e}")
         return False
 
+def get_db_connection():
+    """获取数据库连接"""
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
 def check_password(password):
     """验证密码"""
-    if password == ACCESS_PASSWORD:
-        return True
-    return False
+    return password == ACCESS_PASSWORD
+
+def generate_filename(original_filename):
+    """生成唯一的文件名"""
+    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+    hash_suffix = hashlib.md5(str(datetime.now().timestamp()).encode()).hexdigest()[:8]
+    ext = Path(original_filename).suffix
+    return f"{timestamp}_{hash_suffix}{ext}"
 
 def upload_image(image, description, password):
-    """上传图片到数据库"""
+    """上传图片"""
     if not check_password(password):
-        return "❌ 密码错误！", None
+        return "❌ 密码错误！", None, None
     
     if image is None:
-        return "❌ 请选择要上传的图片！", None
+        return "❌ 请选择要上传的图片！", None, None
     
     try:
-        # 读取图片
+        # 获取原始文件名
         if isinstance(image, str):
             # 如果是文件路径
-            with open(image, "rb") as f:
-                image_bytes = f.read()
-            filename = os.path.basename(image)
+            original_filename = Path(image).name
+            img = Image.open(image)
         else:
             # 如果是 PIL Image
-            img_byte_arr = io.BytesIO()
-            image.save(img_byte_arr, format='PNG')
-            image_bytes = img_byte_arr.getvalue()
-            filename = f"image_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
+            original_filename = "uploaded_image.png"
+            img = image
         
-        # 转换为 base64
-        image_base64 = base64.b64encode(image_bytes).decode('utf-8')
+        # 生成新文件名
+        new_filename = generate_filename(original_filename)
+        file_path = IMAGE_DIR / new_filename
+        
+        # 保存图片
+        img.save(file_path, quality=95, optimize=True)
         
         # 获取文件信息
-        file_size = len(image_bytes)
-        mime_type = "image/png"
+        file_size = file_path.stat().st_size
+        mime_type = f"image/{file_path.suffix[1:]}"
         upload_time = datetime.now().isoformat()
         
         # 插入数据库
-        client.execute(
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        cursor.execute(
             """
-            INSERT INTO images (filename, image_data, file_size, mime_type, upload_time, description)
-            VALUES (?, ?, ?, ?, ?, ?)
+            INSERT INTO images (filename, original_filename, file_path, file_size, mime_type, upload_time, description)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
             """,
-            [filename, image_base64, file_size, mime_type, upload_time, description or ""]
+            (new_filename, original_filename, str(file_path), file_size, mime_type, upload_time, description or "")
         )
+        image_id = cursor.lastrowid
+        conn.commit()
+        conn.close()
+        
+        # 生成访问 URL
+        url = f"/file={file_path}"
         
-        # 获取插入的 ID
-        result = client.execute("SELECT last_insert_rowid() as id")
-        image_id = result.rows[0]["id"]
+        result_text = f"""✅ 上传成功！
+
+**图片ID**: {image_id}
+**文件名**: {original_filename}
+**大小**: {file_size / 1024:.2f} KB
+**上传时间**: {upload_time}
+
+**访问URL**: 
+```
+{url}
+```
+
+💡 提示：可以在"查看图片"标签页中输入 ID: {image_id} 来查看图片
+"""
         
-        return f"✅ 上传成功！图片ID: {image_id}\n文件名: {filename}\n大小: {file_size / 1024:.2f} KB", get_image_list(password)
+        return result_text, get_image_list(password), url
     
     except Exception as e:
-        return f"❌ 上传失败: {str(e)}", None
+        return f"❌ 上传失败: {str(e)}", None, None
 
 def get_image_list(password):
     """获取图片列表"""
@@ -102,21 +137,26 @@ def get_image_list(password):
         return "❌ 密码错误！"
     
     try:
-        result = client.execute(
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        cursor.execute(
             """
-            SELECT id, filename, file_size, upload_time, description
+            SELECT id, filename, original_filename, file_size, upload_time, description
             FROM images
             ORDER BY upload_time DESC
             """
         )
+        rows = cursor.fetchall()
+        conn.close()
         
-        if not result.rows:
+        if not rows:
             return "暂无图片"
         
         # 格式化输出
         output = "## 📸 图片列表\n\n"
-        for row in result.rows:
-            output += f"**ID: {row['id']}** | {row['filename']}\n"
+        for row in rows:
+            output += f"**ID: {row['id']}** | {row['original_filename']}\n"
+            output += f"- 存储文件名: {row['filename']}\n"
             output += f"- 大小: {row['file_size'] / 1024:.2f} KB\n"
             output += f"- 上传时间: {row['upload_time']}\n"
             if row['description']:
@@ -131,30 +171,51 @@ def get_image_list(password):
 def view_image(image_id, password):
     """查看图片"""
     if not check_password(password):
-        return "❌ 密码错误！", None
+        return "❌ 密码错误！", None, None
     
     if not image_id:
-        return "❌ 请输入图片ID！", None
+        return "❌ 请输入图片ID！", None, None
     
     try:
-        result = client.execute(
-            "SELECT filename, image_data, mime_type FROM images WHERE id = ?",
-            [int(image_id)]
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT * FROM images WHERE id = ?",
+            (int(image_id),)
         )
+        row = cursor.fetchone()
+        conn.close()
         
-        if not result.rows:
-            return "❌ 图片不存在！", None
+        if not row:
+            return "❌ 图片不存在！", None, None
         
-        row = result.rows[0]
-        image_data = base64.b64decode(row['image_data'])
+        file_path = Path(row['file_path'])
         
-        # 转换为 PIL Image
-        image = Image.open(io.BytesIO(image_data))
+        if not file_path.exists():
+            return "❌ 图片文件已损坏或丢失！", None, None
         
-        return f"✅ 图片: {row['filename']}", image
+        # 生成访问 URL
+        url = f"/file={file_path}"
+        
+        info_text = f"""✅ 图片信息
+
+**ID**: {row['id']}
+**原始文件名**: {row['original_filename']}
+**存储文件名**: {row['filename']}
+**大小**: {row['file_size'] / 1024:.2f} KB
+**上传时间**: {row['upload_time']}
+**描述**: {row['description'] or '无'}
+
+**访问URL**:
+```
+{url}
+```
+"""
+        
+        return info_text, str(file_path), url
     
     except Exception as e:
-        return f"❌ 查看失败: {str(e)}", None
+        return f"❌ 查看失败: {str(e)}", None, None
 
 def delete_image(image_id, password):
     """删除图片"""
@@ -165,49 +226,104 @@ def delete_image(image_id, password):
         return "❌ 请输入图片ID！", None
     
     try:
-        # 检查图片是否存在
-        result = client.execute(
-            "SELECT filename FROM images WHERE id = ?",
-            [int(image_id)]
-        )
+        conn = get_db_connection()
+        cursor = conn.cursor()
         
-        if not result.rows:
+        # 获取图片信息
+        cursor.execute("SELECT * FROM images WHERE id = ?", (int(image_id),))
+        row = cursor.fetchone()
+        
+        if not row:
+            conn.close()
             return "❌ 图片不存在！", None
         
-        filename = result.rows[0]['filename']
+        file_path = Path(row['file_path'])
+        original_filename = row['original_filename']
+        
+        # 删除数据库记录
+        cursor.execute("DELETE FROM images WHERE id = ?", (int(image_id),))
+        conn.commit()
+        conn.close()
         
-        # 删除图片
-        client.execute("DELETE FROM images WHERE id = ?", [int(image_id)])
+        # 删除文件
+        if file_path.exists():
+            file_path.unlink()
         
-        return f"✅ 已删除图片: {filename} (ID: {image_id})", get_image_list(password)
+        return f"✅ 已删除图片: {original_filename} (ID: {image_id})", get_image_list(password)
     
     except Exception as e:
         return f"❌ 删除失败: {str(e)}", None
 
-def get_image_url(image_id, password):
-    """获取图片的 base64 URL"""
+def get_stats(password):
+    """获取统计信息"""
     if not check_password(password):
         return "❌ 密码错误！"
     
-    if not image_id:
-        return "❌ 请输入图片ID！"
+    try:
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        
+        # 总图片数
+        cursor.execute("SELECT COUNT(*) as count FROM images")
+        total_count = cursor.fetchone()['count']
+        
+        # 总大小
+        cursor.execute("SELECT SUM(file_size) as total_size FROM images")
+        total_size = cursor.fetchone()['total_size'] or 0
+        
+        # 最近上传
+        cursor.execute("SELECT upload_time FROM images ORDER BY upload_time DESC LIMIT 1")
+        last_upload = cursor.fetchone()
+        
+        conn.close()
+        
+        stats = f"""## 📊 统计信息
+
+**总图片数**: {total_count} 张
+**总存储大小**: {total_size / 1024 / 1024:.2f} MB
+**最近上传**: {last_upload['upload_time'] if last_upload else '暂无'}
+**存储路径**: {IMAGE_DIR.absolute()}
+**数据库**: {Path(DB_PATH).absolute()}
+"""
+        
+        return stats
+    
+    except Exception as e:
+        return f"❌ 获取统计失败: {str(e)}"
+
+def export_data(password):
+    """导出数据（仅元数据）"""
+    if not check_password(password):
+        return "❌ 密码错误！", None
     
     try:
-        result = client.execute(
-            "SELECT image_data, mime_type FROM images WHERE id = ?",
-            [int(image_id)]
-        )
+        conn = get_db_connection()
+        cursor = conn.cursor()
+        cursor.execute("SELECT * FROM images ORDER BY upload_time DESC")
+        rows = cursor.fetchall()
+        conn.close()
         
-        if not result.rows:
-            return "❌ 图片不存在！"
+        # 转换为 JSON
+        data = []
+        for row in rows:
+            data.append({
+                'id': row['id'],
+                'filename': row['filename'],
+                'original_filename': row['original_filename'],
+                'file_size': row['file_size'],
+                'upload_time': row['upload_time'],
+                'description': row['description']
+            })
         
-        row = result.rows[0]
-        data_url = f"data:{row['mime_type']};base64,{row['image_data']}"
+        # 保存为文件
+        export_path = "image_metadata_export.json"
+        with open(export_path, 'w', encoding='utf-8') as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
         
-        return f"✅ 图片URL（可用于 markdown/html）:\n\n```\n{data_url[:200]}...\n```\n\n⚠️ 注意：完整URL很长，建议使用图片ID查看"
+        return f"✅ 元数据已导出到: {export_path}", export_path
     
     except Exception as e:
-        return f"❌ 获取URL失败: {str(e)}"
+        return f"❌ 导出失败: {str(e)}", None
 
 # 初始化数据库
 init_success = init_db()
@@ -215,10 +331,10 @@ init_success = init_db()
 # 创建 Gradio 界面
 with gr.Blocks(title="私人图床服务", theme=gr.themes.Soft()) as app:
     gr.Markdown("# 🖼️ 私人图床服务")
-    gr.Markdown("⚠️ 请输入密码以使用此服务")
+    gr.Markdown("⚠️ 请输入密码以使用此服务 | 图片存储在服务器本地文件系统")
     
     if not init_success:
-        gr.Markdown("## ❌ 数据库连接失败！请检查环境变量配置。")
+        gr.Markdown("## ❌ 数据库初始化失败！")
     else:
         with gr.Tabs():
             # 上传图片标签页
@@ -231,13 +347,14 @@ with gr.Blocks(title="私人图床服务", theme=gr.themes.Soft()) as app:
                         upload_btn = gr.Button("上传", variant="primary")
                     
                     with gr.Column():
-                        upload_output = gr.Textbox(label="上传结果", lines=5)
+                        upload_output = gr.Textbox(label="上传结果", lines=10)
+                        upload_url_output = gr.Textbox(label="图片URL", lines=2)
                         upload_list_output = gr.Markdown(label="图片列表")
                 
                 upload_btn.click(
                     upload_image,
                     inputs=[upload_image_input, upload_desc_input, upload_password_input],
-                    outputs=[upload_output, upload_list_output]
+                    outputs=[upload_output, upload_list_output, upload_url_output]
                 )
             
             # 查看图片标签页
@@ -247,15 +364,16 @@ with gr.Blocks(title="私人图床服务", theme=gr.themes.Soft()) as app:
                         view_id_input = gr.Textbox(label="图片ID", placeholder="输入图片ID")
                         view_password_input = gr.Textbox(label="密码", type="password", placeholder="输入访问密码")
                         view_btn = gr.Button("查看", variant="primary")
-                        view_output = gr.Textbox(label="结果")
+                        view_output = gr.Textbox(label="图片信息", lines=10)
+                        view_url_output = gr.Textbox(label="图片URL", lines=2)
                     
                     with gr.Column():
-                        view_image_output = gr.Image(label="图片")
+                        view_image_output = gr.Image(label="图片预览")
                 
                 view_btn.click(
                     view_image,
                     inputs=[view_id_input, view_password_input],
-                    outputs=[view_output, view_image_output]
+                    outputs=[view_output, view_image_output, view_url_output]
                 )
             
             # 图片列表标签页
@@ -288,23 +406,33 @@ with gr.Blocks(title="私人图床服务", theme=gr.themes.Soft()) as app:
                     outputs=[delete_output, delete_list_output]
                 )
             
-            # 获取URL标签页
-            with gr.Tab("🔗 获取URL"):
-                gr.Markdown("⚠️ 注意：由于图片以 base64 编码存储，URL会非常长，建议直接使用图片ID查看")
-                url_id_input = gr.Textbox(label="图片ID", placeholder="输入图片ID")
-                url_password_input = gr.Textbox(label="密码", type="password", placeholder="输入访问密码")
-                url_btn = gr.Button("获取URL", variant="primary")
-                url_output = gr.Textbox(label="URL", lines=10)
+            # 统计信息标签页
+            with gr.Tab("📊 统计信息"):
+                stats_password_input = gr.Textbox(label="密码", type="password", placeholder="输入访问密码")
+                stats_btn = gr.Button("查看统计", variant="primary")
+                stats_output = gr.Markdown(label="统计信息")
+                
+                gr.Markdown("---")
+                gr.Markdown("### 导出数据")
+                export_btn = gr.Button("导出元数据（JSON）", variant="secondary")
+                export_output = gr.Textbox(label="导出结果", lines=2)
+                export_file_output = gr.File(label="下载文件")
                 
-                url_btn.click(
-                    get_image_url,
-                    inputs=[url_id_input, url_password_input],
-                    outputs=[url_output]
+                stats_btn.click(
+                    get_stats,
+                    inputs=[stats_password_input],
+                    outputs=[stats_output]
+                )
+                
+                export_btn.click(
+                    export_data,
+                    inputs=[stats_password_input],
+                    outputs=[export_output, export_file_output]
                 )
     
     gr.Markdown("---")
-    gr.Markdown("💡 提示：所有图片以 base64 编码存储在 Turso 数据库中")
+    gr.Markdown("💡 提示：图片存储在服务器本地，URL可以直接在外部访问（需要 Space 为 Public）")
+    gr.Markdown("🔒 安全提示：建议将 Space 设为 Private 以保护隐私")
 
 if __name__ == "__main__":
     app.launch()
-

requirements.txt

@@ -1,4 +1,3 @@
 gradio==4.44.0
-libsql-client==0.3.1
 Pillow==10.4.0