Upload music_api.py

music_api.py

@@ -0,0 +1,592 @@
+#!/usr/bin/env python3
+"""
+修复版音乐API - 适配 Hugging Face Spaces (含自动扫描功能)
+"""
+
+import os
+import sqlite3
+from pathlib import Path
+from typing import Optional, List, Dict
+import urllib.parse
+
+from fastapi import FastAPI, Query, HTTPException
+from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+
+# ==================== 第三方库导入 ====================
+# 用于读取音频元数据
+from mutagen import File as MutagenFile
+
+# ==================== 配置部分 ====================
+
+class Config:
+    """配置类"""
+    APP_NAME = "音乐API"
+    VERSION = "4.1.0" # 版本升级
+    
+    # 路径配置
+    BASE_DIR = Path(__file__).resolve().parent
+    MUSIC_DIR = BASE_DIR / "static" / "music"
+    COVERS_DIR = BASE_DIR / "static" / "covers"
+    DATABASE_PATH = BASE_DIR / "music.db"
+    
+    # 服务器配置
+    HOST = "0.0.0.0"
+    PORT = int(os.environ.get("PORT", 7860))
+    DEBUG = True
+    
+    def __init__(self):
+        """确保目录存在"""
+        self.MUSIC_DIR.mkdir(parents=True, exist_ok=True)
+        self.COVERS_DIR.mkdir(parents=True, exist_ok=True)
+
+config = Config()
+
+# ==================== 辅助函数 ====================
+
+def extract_audio_metadata(file_path: Path) -> Dict:
+    """
+    使用 mutagen 提取音频文件的元数据
+    """
+    try:
+        # 尝试加载文件
+        audio = MutagenFile(file_path)
+        if audio is None:
+            return None
+
+        # 初始化默认值
+        title = file_path.stem  # 默认使用文件名
+        artist = "Unknown Artist"
+        album = ""
+        genre = ""
+        duration = 0
+
+        # 尝试获取时长
+        if hasattr(audio, 'info') and audio.info.length:
+            duration = int(audio.info.length)
+
+        # 尝试获取标签 (兼容 MP3 ID3, MP4, OGG, FLAC 等)
+        # 常见的键名映射
+        common_keys = {
+            'title': ['TIT2', '\xa9nam', 'TITLE', 'Title', 'title'],
+            'artist': ['TPE1', '\xa9ART', 'ARTIST', 'Artist', 'artist'],
+            'album': ['TALB', '\xa9alb', 'ALBUM', 'Album', 'album'],
+            'genre': ['TCON', '\xa9gen', 'GENRE', 'Genre', 'genre']
+        }
+
+        # 辅助函数：获取第一个匹配的标签值
+        def get_tag(target_type):
+            keys = common_keys.get(target_type, [])
+            for k in keys:
+                if k in audio:
+                    val = audio[k]
+                    if isinstance(val, list) and len(val) > 0:
+                        return str(val[0])
+                    return str(val)
+            return None
+
+        # 覆盖默认值
+        found_title = get_tag('title')
+        if found_title:
+            title = found_title
+        
+        found_artist = get_tag('artist')
+        if found_artist:
+            artist = found_artist
+
+        found_album = get_tag('album')
+        if found_album:
+            album = found_album
+            
+        found_genre = get_tag('genre')
+        if found_genre:
+            genre = found_genre
+
+        return {
+            "title": title,
+            "artist": artist,
+            "album": album,
+            "genre": genre,
+            "duration": duration,
+            "file_size": file_path.stat().st_size
+        }
+
+    except Exception as e:
+        print(f"⚠️ 无法读取 {file_path.name} 的元数据: {e}")
+        return None
+
+# ==================== 数据库部分 ====================
+
+class Database:
+    """数据库管理类"""
+    
+    @staticmethod
+    def get_connection():
+        """获取数据库连接"""
+        conn = sqlite3.connect(str(config.DATABASE_PATH))
+        conn.row_factory = sqlite3.Row
+        return conn
+    
+    @staticmethod
+    def init_database():
+        """初始化数据库表结构"""
+        conn = Database.get_connection()
+        cursor = conn.cursor()
+        
+        cursor.execute('''
+        CREATE TABLE IF NOT EXISTS music (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            title TEXT NOT NULL,
+            artist TEXT NOT NULL,
+            album TEXT,
+            genre TEXT,
+            duration INTEGER,
+            file_name TEXT NOT NULL UNIQUE,
+            file_size INTEGER,
+            file_hash TEXT,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+        ''')
+        
+        cursor.execute('CREATE INDEX IF NOT EXISTS idx_title ON music(title)')
+        cursor.execute('CREATE INDEX IF NOT EXISTS idx_artist ON music(artist)')
+        cursor.execute('CREATE INDEX IF NOT EXISTS idx_filename ON music(file_name)')
+        
+        conn.commit()
+        conn.close()
+    
+    @staticmethod
+    def scan_music_files() -> Dict:
+        """
+        扫描 static/music 目录并更新数据库
+        """
+        added_count = 0
+        skipped_count = 0
+        error_files = []
+        
+        # 支持的音频后缀
+        audio_extensions = ['.mp3', '.flac', '.wav', '.ogg', '.m4a', '.wma']
+        
+        # 获取所有文件
+        if not config.MUSIC_DIR.exists():
+            return {"status": "error", "message": "音乐目录不存在"}
+            
+        files = list(config.MUSIC_DIR.rglob('*'))
+        
+        conn = Database.get_connection()
+        cursor = conn.cursor()
+        
+        for file_path in files:
+            if not file_path.is_file():
+                continue
+            if file_path.suffix.lower() not in audio_extensions:
+                continue
+            
+            file_name = file_path.name
+            
+            # 提取元数据
+            meta = extract_audio_metadata(file_path)
+            if not meta:
+                # 如果读取失败，创建一个基本记录
+                meta = {
+                    "title": file_path.stem,
+                    "artist": "Unknown",
+                    "album": "",
+                    "genre": "",
+                    "duration": 0,
+                    "file_size": file_path.stat().st_size
+                }
+                error_files.append(file_name)
+
+            # 尝试插入数据库 (如果 file_name 已存在则忽略)
+            try:
+                cursor.execute('''
+                    INSERT OR IGNORE INTO music 
+                    (title, artist, album, genre, duration, file_name, file_size)
+                    VALUES (?, ?, ?, ?, ?, ?, ?)
+                ''', (
+                    meta['title'],
+                    meta['artist'],
+                    meta['album'],
+                    meta['genre'],
+                    meta['duration'],
+                    file_name,
+                    meta['file_size']
+                ))
+                
+                if cursor.rowcount > 0:
+                    added_count += 1
+                    print(f"✅ 新增: {meta['title']} - {meta['artist']}")
+                else:
+                    skipped_count += 1
+                    print(f"⏭️ 跳过: {file_name} (已存在)")
+                    
+            except Exception as e:
+                print(f"❌ 数据库错误 ({file_name}): {e}")
+                error_files.append(file_name)
+        
+        conn.commit()
+        conn.close()
+        
+        return {
+            "status": "success",
+            "scanned": len(files),
+            "added": added_count,
+            "skipped": skipped_count,
+            "errors": len(error_files),
+            "message": f"扫描完成。新增 {added_count} 首歌曲。"
+        }
+    
+    @staticmethod
+    def get_song_cover_url(song_id: int) -> str:
+        """获取歌曲封面URL"""
+        cover_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.webp']
+        
+        for ext in cover_extensions:
+            cover_path = config.COVERS_DIR / f"{song_id}{ext}"
+            if cover_path.exists():
+                return f"/static/covers/{song_id}{ext}"
+        
+        default_path = config.COVERS_DIR / f"{song_id}_default.jpg"
+        if default_path.exists():
+            return f"/static/covers/{song_id}_default.jpg"
+        
+        return f"/api/music/{song_id}/default-cover"
+    
+    @staticmethod
+    def get_all_songs(limit: int = 1000) -> List[Dict]:
+        """获取所有歌曲"""
+        conn = Database.get_connection()
+        cursor = conn.cursor()
+        cursor.execute(f"SELECT * FROM music ORDER BY title LIMIT ?", (limit,))
+        rows = cursor.fetchall()
+        conn.close()
+        
+        songs = []
+        for row in rows:
+            song_id = row["id"]
+            songs.append({
+                "id": song_id,
+                "title": row["title"],
+                "artist": row["artist"],
+                "album": row["album"] or "",
+                "genre": row["genre"] or "",
+                "duration": row["duration"] or 0,
+                "file_name": row["file_name"],
+                "file_size": row["file_size"] or 0,
+                "cover_url": Database.get_song_cover_url(song_id),
+                "play_url": f"/api/music/{song_id}/play",
+                "download_url": f"/api/music/{song_id}/download"
+            })
+        return songs
+    
+    @staticmethod
+    def search_songs(query: str = None, artist: str = None, limit: int = 100) -> List[Dict]:
+        """搜索歌曲"""
+        conn = Database.get_connection()
+        cursor = conn.cursor()
+        
+        sql = "SELECT * FROM music WHERE 1=1"
+        params = []
+        
+        if query:
+            sql += " AND (title LIKE ? OR artist LIKE ? OR album LIKE ?)"
+            search_term = f"%{query}%"
+            params.extend([search_term, search_term, search_term])
+        
+        if artist:
+            sql += " AND artist LIKE ?"
+            params.append(f"%{artist}%")
+        
+        sql += " ORDER BY title LIMIT ?"
+        params.append(limit)
+        
+        cursor.execute(sql, params)
+        rows = cursor.fetchall()
+        conn.close()
+        
+        songs = []
+        for row in rows:
+            song_id = row["id"]
+            songs.append({
+                "id": song_id,
+                "title": row["title"],
+                "artist": row["artist"],
+                "album": row["album"] or "",
+                "genre": row["genre"] or "",
+                "duration": row["duration"] or 0,
+                "file_name": row["file_name"],
+                "file_size": row["file_size"] or 0,
+                "cover_url": Database.get_song_cover_url(song_id),
+                "play_url": f"/api/music/{song_id}/play",
+                "download_url": f"/api/music/{song_id}/download"
+            })
+        return songs
+    
+    @staticmethod
+    def get_song_by_id(song_id: int) -> Optional[Dict]:
+        """根据ID获取歌曲"""
+        conn = Database.get_connection()
+        cursor = conn.cursor()
+        cursor.execute("SELECT * FROM music WHERE id = ?", (song_id,))
+        row = cursor.fetchone()
+        conn.close()
+        
+        if not row:
+            return None
+        
+        return {
+            "id": song_id,
+            "title": row["title"],
+            "artist": row["artist"],
+            "album": row["album"] or "",
+            "genre": row["genre"] or "",
+            "duration": row["duration"] or 0,
+            "file_name": row["file_name"],
+            "file_size": row["file_size"] or 0,
+            "cover_url": Database.get_song_cover_url(song_id),
+            "play_url": f"/api/music/{song_id}/play",
+            "download_url": f"/api/music/{song_id}/download"
+        }
+
+# ==================== FastAPI应用部分 ====================
+
+app = FastAPI(
+    title=config.APP_NAME,
+    description="音乐API系统 - 封面URL直接绑定",
+    version=config.VERSION,
+    docs_url="/docs"
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.mount("/static", StaticFiles(directory="static"), name="static")
+
+# ==================== API 文档页面 HTML (保持不变，为节省空间此处省略完整HTML，请使用上一次的HTML内容) ====================
+# 为了方便您直接复制，这里仍然包含完整的 HTML
+API_DOCS_HTML = """
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>音乐API文档 - v4.1.0</title>
+    <style>
+        :root {
+            --bg-color: #1a1a2e; --card-bg: #16213e; --text-primary: #e0e0e0; --text-secondary: #a0a0a0;
+            --accent-color: #0f3460; --highlight: #e94560; --method-get: #61affe; --method-post: #49cc90;
+            --code-bg: #0f0f1a;
+        }
+        * { margin: 0; padding: 0; box-sizing: border-box; }
+        body { font-family: 'Segoe UI', sans-serif; background: var(--bg-color); color: var(--text-primary); line-height: 1.6; padding: 20px; }
+        .container { max-width: 1000px; margin: 0 auto; }
+        header { text-align: center; margin-bottom: 40px; padding-bottom: 20px; border-bottom: 1px solid var(--accent-color); }
+        h1 { font-size: 2.5rem; margin-bottom: 10px; color: #fff; }
+        .version { background: var(--highlight); padding: 2px 8px; border-radius: 4px; font-size: 0.9rem; }
+        .endpoint-card { background: var(--card-bg); border-radius: 8px; padding: 20px; margin-bottom: 20px; border: 1px solid var(--accent-color); }
+        .endpoint-header { display: flex; align-items: center; gap: 15px; margin-bottom: 15px; flex-wrap: wrap; }
+        .method { color: #fff; padding: 4px 10px; border-radius: 4px; font-weight: bold; font-size: 0.9rem; min-width: 60px; text-align: center; }
+        .method.get { background: var(--method-get); }
+        .method.post { background: var(--method-post); }
+        .path { font-family: 'Courier New', monospace; color: #fff; background: rgba(0,0,0,0.3); padding: 4px 10px; border-radius: 4px; flex: 1; word-break: break-all; }
+        table { width: 100%; border-collapse: collapse; margin-top: 10px; }
+        th, td { text-align: left; padding: 10px; border-bottom: 1px solid var(--accent-color); font-size: 0.9rem; }
+        .code-block { background: var(--code-bg); padding: 15px; border-radius: 6px; overflow-x: auto; margin-top: 10px; font-family: monospace; font-size: 0.9rem; color: #a5d6ff; }
+        .desc { color: var(--text-secondary); margin-bottom: 15px; }
+    </style>
+</head>
+<body>
+    <div class="container">
+        <header>
+            <h1>音乐API <span class="version">v4.1.0</span></h1>
+            <p style="color:#aaa">基于 FastAPI 的轻量级音乐流媒体服务 | <a href="/docs" style="color:#61affe">Swagger文档</a></p>
+        </header>
+        
+        <div style="background: var(--card-bg); padding: 20px; border-radius: 8px; margin-bottom: 30px; border-left: 4px solid #49cc90;">
+            <h2 style="margin-bottom: 10px;">🚀 快速开始</h2>
+            <p style="color:#ccc">1. 上传音乐文件到 <code>static/music/</code> 目录</p>
+            <p style="color:#ccc">2. 访问此接口扫描文件: <a href="/api/scan" target="_blank" style="color:#fff; background:#49cc90; padding:2px 8px; text-decoration:none; border-radius:4px;">GET /api/scan</a></p>
+            <p style="color:#ccc">3. 访问首页查看列表或使用 API 播放</p>
+        </div>
+
+        <h3 style="margin-bottom:20px;">接口列表</h3>
+
+        <div class="endpoint-card">
+            <div class="endpoint-header">
+                <span class="method get">GET</span>
+                <span class="path">/api/scan</span>
+            </div>
+            <p class="desc">扫描 static/music 目录并更新数据库（新增歌曲）</p>
+        </div>
+
+        <div class="endpoint-card">
+            <div class="endpoint-header">
+                <span class="method get">GET</span>
+                <span class="path">/api/music</span>
+            </div>
+            <p class="desc">获取歌曲列表或搜索</p>
+        </div>
+
+        <div class="endpoint-card">
+            <div class="endpoint-header">
+                <span class="method get">GET</span>
+                <span class="path">/api/music/{id}/play</span>
+            </div>
+            <p class="desc">播放指定ID的音乐</p>
+        </div>
+        
+        <div style="text-align: center; margin-top: 50px; color: #666; font-size: 0.8rem;">
+            Powered by FastAPI & HuggingFace Spaces
+        </div>
+    </div>
+</body>
+</html>
+"""
+
+# ==================== 启动事件 ====================
+
+@app.on_event("startup")
+def startup_event():
+    Database.init_database()
+
+# ==================== API路由部分 ====================
+
+@app.get("/")
+async def get_api_docs():
+    return HTMLResponse(API_DOCS_HTML)
+
+@app.get("/api/scan")
+async def scan_library():
+    """扫描音乐文件并导入数据库"""
+    result = Database.scan_music_files()
+    return JSONResponse(result)
+
+@app.get("/api/music")
+async def get_music(
+    q: Optional[str] = Query(None, description="搜索关键词"),
+    artist: Optional[str] = Query(None, description="艺术家"),
+    limit: int = Query(100, ge=1, le=1000, description="返回数量")
+):
+    return Database.search_songs(q, artist, limit)
+
+@app.get("/api/music/{music_id}")
+async def get_music_by_id(music_id: int):
+    song = Database.get_song_by_id(music_id)
+    if not song:
+        raise HTTPException(status_code=404, detail="歌曲不存在")
+    return song
+
+@app.get("/api/music/{music_id}/play")
+async def play_music(music_id: int):
+    song = Database.get_song_by_id(music_id)
+    if not song:
+        raise HTTPException(status_code=404, detail="歌曲不存在")
+    
+    file_path = config.MUSIC_DIR / song['file_name']
+    if not file_path.exists():
+        raise HTTPException(status_code=404, detail="音乐文件不存在")
+    
+    encoded_filename = urllib.parse.quote(song['file_name'])
+    
+    return FileResponse(
+        file_path,
+        filename=encoded_filename,
+        media_type="audio/mpeg",
+        headers={
+            "Content-Disposition": f"inline; filename*=UTF-8''{encoded_filename}",
+            "Accept-Ranges": "bytes"
+        }
+    )
+
+@app.get("/api/music/{music_id}/download")
+async def download_music(music_id: int):
+    song = Database.get_song_by_id(music_id)
+    if not song:
+        raise HTTPException(status_code=404, detail="歌曲不存在")
+    
+    file_path = config.MUSIC_DIR / song['file_name']
+    if not file_path.exists():
+        raise HTTPException(status_code=404, detail="音乐文件不存在")
+    
+    encoded_filename = urllib.parse.quote(song['file_name'])
+    
+    return FileResponse(
+        file_path,
+        filename=encoded_filename,
+        media_type="application/octet-stream",
+        headers={
+            "Content-Disposition": f"attachment; filename*=UTF-8''{encoded_filename}"
+        }
+    )
+
+@app.get("/api/music/{music_id}/default-cover")
+async def get_default_cover(music_id: int):
+    from io import BytesIO
+    from PIL import Image, ImageDraw, ImageFont
+    import base64
+    
+    img = Image.new('RGB', (300, 300), color=(106, 17, 203))
+    draw = ImageDraw.Draw(img)
+    try:
+        font = ImageFont.truetype("arial.ttf", 100)
+    except:
+        font = ImageFont.load_default()
+    draw.text((150, 150), "♪", fill=(255, 255, 255), font=font, anchor="mm")
+    
+    buffer = BytesIO()
+    img.save(buffer, format="PNG")
+    img_base64 = base64.b64encode(buffer.getvalue()).decode()
+    
+    return {
+        "image": f"data:image/png;base64,{img_base64}",
+        "song_id": music_id,
+        "is_default": True
+    }
+
+@app.get("/api/stats")
+async def get_stats():
+    conn = Database.get_connection()
+    cursor = conn.cursor()
+    
+    cursor.execute("SELECT COUNT(*) as total FROM music")
+    total = cursor.fetchone()[0]
+    
+    cursor.execute("SELECT SUM(file_size) as total_size FROM music")
+    total_size = cursor.fetchone()[0] or 0
+    
+    cover_files = list(config.COVERS_DIR.glob("*.jpg")) + list(config.COVERS_DIR.glob("*.png"))
+    cover_count = len(cover_files)
+    
+    conn.close()
+    
+    return {
+        "total_songs": total,
+        "total_size_gb": round(total_size / (1024**3), 2),
+        "cover_count": cover_count,
+        "cover_percentage": round(cover_count / total * 100, 1) if total > 0 else 0
+    }
+
+@app.get("/api/health")
+async def health_check():
+    return {
+        "status": "ok",
+        "service": config.APP_NAME,
+        "version": config.VERSION,
+        "endpoints": {
+            "API文档": "/",
+            "扫描歌曲": "/api/scan",
+            "Swagger交互文档": "/docs",
+            "搜索音乐": "/api/music?q=关键词",
+            "播放歌曲": "/api/music/{id}/play",
+            "下载歌曲": "/api/music/{id}/download",
+            "系统统计": "/api/stats"
+        }
+    }