Upload api_playlist.py

api_playlist.py

@@ -1,40 +1,108 @@
-from fastapi import APIRouter
-from core import get_db, save_db, PlaylistCreateReq, PlaylistAddReq
+import os
+import uuid
+from fastapi import APIRouter, UploadFile, File, Form
+from pydantic import BaseModel
+from huggingface_hub import upload_file
+from core import get_db, save_db, REPO_ID, TOKEN
 
 router = APIRouter()
 
+class PlaylistAddReq(BaseModel):
+    username: str
+    type: str
+    playlist_name: str
+    filename: str
+
+class PlaylistDeleteReq(BaseModel):
+    username: str
+    type: str
+    name: str
+
 @router.get("/api/cloud_pl/get")
 def get_cloud_playlists(): 
     return {"status": "success", "data": get_db("cloud_playlists.json")}
 
 @router.post("/api/cloud_pl/create")
-def create_cloud_playlist(req: PlaylistCreateReq):
+async def create_cloud_playlist(username: str = Form(...), type: str = Form(...), name: str = Form(...), description: str = Form(""), image: UploadFile = File(None)):
     try:
         db = get_db("cloud_playlists.json")
         if "admin" not in db: db["admin"] = {}
         if "users" not in db: db["users"] = {}
-        if req.type == "admin":
-            if req.username != "猪的飞行梦": return {"status": "error", "message": "权限不足"}
-            if req.name in db["admin"]: return {"status": "error", "message": "已存在"}
-            db["admin"][req.name] = []
-        else:
-            if req.username not in db["users"]: db["users"][req.username] = {}
-            if len(db["users"][req.username]) >= 3: return {"status": "error", "message": "上限3个"}
-            if req.name in db["users"][req.username]: return {"status": "error", "message": "已存在"}
-            db["users"][req.username][req.name] = []
+        
+        target_dict = db["admin"] if type == "admin" else db["users"].setdefault(username, {})
+        if type == "admin" and username != "猪的飞行梦": return {"status": "error", "message": "权限不足"}
+        if type == "user" and len(target_dict) >= 3: return {"status": "error", "message": "上限3个"}
+        if name in target_dict: return {"status": "error", "message": "歌单已存在"}
+
+        cover_name = ""
+        if image:
+            ext = os.path.splitext(image.filename)[1]
+            cover_name = f"pl_{uuid.uuid4().hex[:8]}{ext}"
+            temp_path = f"temp_{cover_name}"
+            with open(temp_path, "wb") as f: f.write(await image.read())
+            upload_file(path_or_fileobj=temp_path, path_in_repo=f"community_images/{cover_name}", repo_id=REPO_ID, repo_type="dataset", token=TOKEN)
+            os.remove(temp_path)
+
+        target_dict[name] = {"description": description[:200], "cover": cover_name, "songs": []}
         save_db(db, "cloud_playlists.json")
         return {"status": "success"}
     except Exception as e: return {"status": "error", "message": str(e)}
 
+@router.post("/api/cloud_pl/edit")
+async def edit_cloud_playlist(username: str = Form(...), type: str = Form(...), old_name: str = Form(...), name: str = Form(...), description: str = Form(""), image: UploadFile = File(None)):
+    try:
+        db = get_db("cloud_playlists.json")
+        target_dict = db.get("admin", {}) if type == "admin" else db.get("users", {}).get(username, {})
+        if old_name not in target_dict: return {"status": "error", "message": "歌单不存在"}
+        if type == "admin" and username != "猪的飞行梦": return {"status": "error", "message": "权限不足"}
+        if old_name != name and name in target_dict: return {"status": "error", "message": "新名称已存在"}
+
+        pl_data = target_dict.pop(old_name)
+        # 兼容旧版本纯数组数据
+        if isinstance(pl_data, list): pl_data = {"description": "", "cover": "", "songs": pl_data}
+        
+        pl_data["description"] = description[:200]
+        if image:
+            ext = os.path.splitext(image.filename)[1]
+            cover_name = f"pl_{uuid.uuid4().hex[:8]}{ext}"
+            temp_path = f"temp_{cover_name}"
+            with open(temp_path, "wb") as f: f.write(await image.read())
+            upload_file(path_or_fileobj=temp_path, path_in_repo=f"community_images/{cover_name}", repo_id=REPO_ID, repo_type="dataset", token=TOKEN)
+            os.remove(temp_path)
+            pl_data["cover"] = cover_name
+
+        target_dict[name] = pl_data
+        save_db(db, "cloud_playlists.json")
+        return {"status": "success"}
+    except Exception as e: return {"status": "error", "message": str(e)}
+
+@router.post("/api/cloud_pl/delete")
+def delete_cloud_playlist(req: PlaylistDeleteReq):
+    try:
+        db = get_db("cloud_playlists.json")
+        target_dict = db.get("admin", {}) if req.type == "admin" else db.get("users", {}).get(req.username, {})
+        if req.type == "admin" and req.username != "猪的飞行梦": return {"status": "error", "message": "权限不足"}
+        if req.name in target_dict:
+            del target_dict[req.name]
+            save_db(db, "cloud_playlists.json")
+            return {"status": "success"}
+        return {"status": "error", "message": "歌单不存在"}
+    except Exception as e: return {"status": "error", "message": str(e)}
+
 @router.post("/api/cloud_pl/add")
 def add_cloud_playlist(req: PlaylistAddReq):
     try:
         db = get_db("cloud_playlists.json")
-        if req.type == "admin":
-            if req.username != "猪的飞行梦": return {"status": "error", "message": "权限不足"}
-            if req.filename not in db.get("admin", {}).get(req.playlist_name, []): db["admin"][req.playlist_name].append(req.filename)
+        target_dict = db.get("admin", {}) if req.type == "admin" else db.get("users", {}).get(req.username, {})
+        pl = target_dict.get(req.playlist_name)
+        if pl is None: return {"status": "error", "message": "歌单不存在"}
+        
+        # 兼容旧版本列表
+        if isinstance(pl, list):
+            if req.filename not in pl: pl.append(req.filename)
         else:
-            if req.filename not in db.get("users", {}).get(req.username, {}).get(req.playlist_name, []): db["users"][req.username][req.playlist_name].append(req.filename)
+            if req.filename not in pl["songs"]: pl["songs"].append(req.filename)
+            
         save_db(db, "cloud_playlists.json")
         return {"status": "success"}
     except Exception as e: return {"status": "error", "message": str(e)}