Spaces:
Sleeping
Sleeping
| import os | |
| import uuid | |
| import time | |
| import asyncio | |
| import tempfile | |
| import logging | |
| from fastapi import APIRouter, UploadFile, File, Form | |
| from fastapi.responses import FileResponse, JSONResponse | |
| from huggingface_hub import upload_file, hf_hub_download | |
| from core import get_db, save_db, check_write_auth, is_admin, api, REPO_ID, TOKEN, PlaylistAddReq, PlaylistBatchAddReq, PlaylistDeleteReq, PlaylistLikeReq, PlaylistCommentReq | |
| logger = logging.getLogger("飞行音乐云端") | |
| router = APIRouter() | |
| IMAGE_EXTS = ('.png', '.jpg', '.jpeg', '.gif', '.webp') | |
| MAX_IMAGE_SIZE = 10 * 1024 * 1024 | |
| READ_CHUNK = 1024 * 1024 | |
| async def _save_upload_cover(image, temp_dir): | |
| """分块保存封面图片到临时目录，返回 (本地路径, 扩展名)；超限抛 ValueError""" | |
| ext = os.path.splitext(image.filename)[1].lower() | |
| if ext not in IMAGE_EXTS: | |
| raise ValueError("不支持的图片格式") | |
| fd, temp_path = tempfile.mkstemp(suffix=ext, dir=temp_dir) | |
| written = 0 | |
| with os.fdopen(fd, "wb") as f: | |
| while True: | |
| chunk = await image.read(READ_CHUNK) | |
| if not chunk: | |
| break | |
| written += len(chunk) | |
| if written > MAX_IMAGE_SIZE: | |
| raise ValueError("图片过大（超过10MB）") | |
| f.write(chunk) | |
| return temp_path, ext | |
| async def get_cloud_playlists(): | |
| db = get_db("cloud_playlists.json") | |
| # 🌟 网络歌单统一收口 admin 区：存量 users 区数据一次性并入（同名自动改名避让），迁移后 users 区清空 | |
| users_area = db.get("users") or {} | |
| if users_area: | |
| admin_area = db.setdefault("admin", {}) | |
| for user, pls in users_area.items(): | |
| if not isinstance(pls, dict): continue | |
| for name, pl in pls.items(): | |
| if not isinstance(pl, dict): continue | |
| final_name = name | |
| suffix = 2 | |
| while final_name in admin_area: | |
| final_name = f"{name}（{user}{suffix if suffix > 2 else ''}）" | |
| suffix += 1 | |
| pl.setdefault("creator", user) | |
| admin_area[final_name] = pl | |
| db["users"] = {} | |
| save_db(db, "cloud_playlists.json") | |
| return {"status": "success", "data": db} | |
| async def create_cloud_playlist(username: str = Form(...), type: str = Form(...), name: str = Form(...), description: str = Form(""), image: UploadFile = File(None), token: str = Form("")): | |
| ok, reason = check_write_auth({"username": username, "token": token}) | |
| if not ok: | |
| return {"status": "error", "message": reason} | |
| try: | |
| db = get_db("cloud_playlists.json") | |
| if "admin" not in db: db["admin"] = {} | |
| if "users" not in db: db["users"] = {} | |
| # 🌟 分类歌单取消后，admin 区即网络歌单统一存储区：任何已登录用户均可创建，数量不设上限； | |
| # type=="user" 仅为旧客户端兼容保留（仍有 3 个上限） | |
| if type == "admin": | |
| target_dict = db["admin"] | |
| else: | |
| target_dict = db["users"].setdefault(username, {}) | |
| if len(target_dict) >= 3: return {"status": "error", "message": "上限3个"} | |
| if name in target_dict: return {"status": "error", "message": "歌单已存在"} | |
| cover_name = "" | |
| if image and image.filename: | |
| with tempfile.TemporaryDirectory() as temp_dir: | |
| temp_path, ext = await _save_upload_cover(image, temp_dir) | |
| cover_name = f"pl_{uuid.uuid4().hex}{ext}" | |
| await asyncio.to_thread(upload_file, path_or_fileobj=temp_path, path_in_repo=f"community_images/{cover_name}", repo_id=REPO_ID, repo_type="dataset", token=TOKEN) | |
| target_dict[name] = { | |
| "description": description[:100], "cover": cover_name, "songs": [], "likes": 0, "liked_by": [], "comments": [], | |
| "creator": username # admin 区扁平存储后靠 creator 保留归属，供编辑/删除鉴权与前端展示 | |
| } | |
| save_db(db, "cloud_playlists.json") | |
| return {"status": "success"} | |
| except ValueError as e: | |
| return {"status": "error", "message": str(e)} | |
| except Exception as e: return {"status": "error", "message": str(e)} | |
| 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), token: str = Form("")): | |
| ok, reason = check_write_auth({"username": username, "token": token}) | |
| if not ok: | |
| return {"status": "error", "message": reason} | |
| 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": "歌单不存在"} | |
| # admin 区归属校验：创建者本人或管理员（无 creator 的存量歌单仅管理员可改） | |
| if type == "admin": | |
| creator = target_dict[old_name].get("creator") if isinstance(target_dict[old_name], dict) else None | |
| if creator != username and not is_admin(username): return {"status": "error", "message": "权限不足"} | |
| pl_data = target_dict.pop(old_name) | |
| if image and image.filename: | |
| with tempfile.TemporaryDirectory() as temp_dir: | |
| temp_path, ext = await _save_upload_cover(image, temp_dir) | |
| cover_name = f"pl_{uuid.uuid4().hex}{ext}" | |
| await asyncio.to_thread(upload_file, path_or_fileobj=temp_path, path_in_repo=f"community_images/{cover_name}", repo_id=REPO_ID, repo_type="dataset", token=TOKEN) | |
| if pl_data.get("cover"): | |
| try: asyncio.create_task(asyncio.to_thread(api.delete_file, path_in_repo=f"community_images/{pl_data['cover']}", repo_id=REPO_ID, token=TOKEN)) | |
| except Exception: pass | |
| pl_data["cover"] = cover_name | |
| pl_data["description"] = description[:100] | |
| target_dict[name] = pl_data | |
| save_db(db, "cloud_playlists.json") | |
| return {"status": "success"} | |
| except ValueError as e: | |
| return {"status": "error", "message": str(e)} | |
| except Exception as e: return {"status": "error", "message": str(e)} | |
| async def delete_cloud_playlist(req: PlaylistDeleteReq): | |
| ok, reason = check_write_auth(req.dict()) | |
| if not ok: | |
| return {"status": "error", "message": reason} | |
| 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.name not in target_dict: return {"status": "error", "message": "不存在或无权限"} | |
| # admin 区归属校验：创建者本人或管理员（无 creator 的存量歌单仅管理员可删） | |
| if req.type == "admin": | |
| creator = target_dict[req.name].get("creator") if isinstance(target_dict[req.name], dict) else None | |
| if creator != req.username and not is_admin(req.username): return {"status": "error", "message": "权限不足"} | |
| pl_data = target_dict.pop(req.name) | |
| if pl_data.get("cover"): | |
| try: asyncio.create_task(asyncio.to_thread(api.delete_file, path_in_repo=f"community_images/{pl_data['cover']}", repo_id=REPO_ID, token=TOKEN)) | |
| except Exception: pass | |
| save_db(db, "cloud_playlists.json") | |
| return {"status": "success"} | |
| except Exception as e: return {"status": "error", "message": str(e)} | |
| async def add_to_cloud_playlist(req: PlaylistAddReq): | |
| ok, reason = check_write_auth(req.dict()) | |
| if not ok: | |
| return {"status": "error", "message": reason} | |
| try: | |
| db = get_db("cloud_playlists.json") | |
| if req.type == "admin": | |
| # admin 区即公共网络歌单区：任何已登录用户都可加歌（与旧 users 区语义一致） | |
| target_dict = db.get("admin", {}) | |
| else: | |
| # 使用 author 定位歌单所有者，支持所有已登录用户往任意用户歌单添加歌曲 | |
| author = req.author if req.author else req.username | |
| target_dict = db.get("users", {}).get(author, {}) | |
| if req.playlist_name not in target_dict: return {"status": "error", "message": "歌单不存在"} | |
| songs = target_dict[req.playlist_name].get("songs", []) | |
| if req.filename in songs: return {"status": "error", "message": "歌曲已存在于歌单中"} | |
| songs.append(req.filename) | |
| target_dict[req.playlist_name]["songs"] = songs | |
| save_db(db, "cloud_playlists.json") | |
| return {"status": "success"} | |
| except Exception as e: return {"status": "error", "message": str(e)} | |
| # 🌟 批量添加歌曲到歌单接口 | |
| async def batch_add_to_cloud_playlist(req: PlaylistBatchAddReq): | |
| """批量添加歌曲到歌单（一次请求处理多首）""" | |
| ok, reason = check_write_auth(req.dict()) | |
| if not ok: | |
| return {"status": "error", "message": reason} | |
| try: | |
| db = get_db("cloud_playlists.json") | |
| if req.type == "admin": | |
| # admin 区即公共网络歌单区：任何已登录用户都可批量加歌（与旧 users 区语义一致） | |
| target_dict = db.get("admin", {}) | |
| else: | |
| author = req.author if req.author else req.username | |
| target_dict = db.get("users", {}).get(author, {}) | |
| if req.playlist_name not in target_dict: | |
| return {"status": "error", "message": "歌单不存在"} | |
| songs = target_dict[req.playlist_name].get("songs", []) | |
| existing = set(songs) | |
| added_count = 0 | |
| for filename in req.filenames: | |
| if filename not in existing: | |
| songs.append(filename) | |
| existing.add(filename) | |
| added_count += 1 | |
| target_dict[req.playlist_name]["songs"] = songs | |
| save_db(db, "cloud_playlists.json") | |
| return {"status": "success", "added_count": added_count, "total": len(songs)} | |
| except Exception as e: | |
| return {"status": "error", "message": str(e)} | |
| # 🌟 云端歌单移除单曲接口 | |
| async def remove_from_cloud_playlist(req: PlaylistAddReq): | |
| ok, reason = check_write_auth(req.dict()) | |
| if not ok: | |
| return {"status": "error", "message": reason} | |
| 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.playlist_name not in target_dict: return {"status": "error", "message": "歌单不存在"} | |
| # admin 区移歌权限：创建者本人或管理员（无 creator 的存量歌单仅管理员可移） | |
| if req.type == "admin": | |
| creator = target_dict[req.playlist_name].get("creator") if isinstance(target_dict[req.playlist_name], dict) else None | |
| if creator != req.username and not is_admin(req.username): return {"status": "error", "message": "权限不足"} | |
| songs = target_dict[req.playlist_name].get("songs", []) | |
| if req.filename in songs: | |
| songs.remove(req.filename) | |
| target_dict[req.playlist_name]["songs"] = songs | |
| save_db(db, "cloud_playlists.json") | |
| return {"status": "success"} | |
| return {"status": "error", "message": "未在歌单中找到此歌曲"} | |
| except Exception as e: return {"status": "error", "message": str(e)} | |
| async def like_cloud_playlist(req: PlaylistLikeReq): | |
| ok, reason = check_write_auth(req.dict()) | |
| if not ok: | |
| return {"status": "error", "message": reason} | |
| try: | |
| db = get_db("cloud_playlists.json") | |
| target_dict = db.get("admin", {}) if req.type == "admin" else db.get("users", {}).get(req.author, {}) | |
| pl = target_dict.get(req.name) | |
| if not pl or isinstance(pl, list): return {"status": "error", "message": "格式异常或不存在"} | |
| liked_by = pl.get("liked_by", []) | |
| if req.username in liked_by: | |
| liked_by.remove(req.username) | |
| pl["likes"] = max(0, pl.get("likes", 1) - 1) | |
| else: | |
| liked_by.append(req.username) | |
| pl["likes"] = pl.get("likes", 0) + 1 | |
| pl["liked_by"] = liked_by | |
| save_db(db, "cloud_playlists.json") | |
| return {"status": "success"} | |
| except Exception as e: return {"status": "error", "message": str(e)} | |
| async def comment_cloud_playlist(req: PlaylistCommentReq): | |
| ok, reason = check_write_auth(req.dict()) | |
| if not ok: | |
| return {"status": "error", "message": reason} | |
| try: | |
| if len(req.content) > 99: return {"status": "error", "message": "评论过长"} | |
| db = get_db("cloud_playlists.json") | |
| target_dict = db.get("admin", {}) if req.type == "admin" else db.get("users", {}).get(req.author, {}) | |
| pl = target_dict.get(req.name) | |
| if not pl or isinstance(pl, list): return {"status": "error", "message": "格式异常或不存在"} | |
| if "comments" not in pl: pl["comments"] = [] | |
| new_comment = {"id": uuid.uuid4().hex, "user": req.username, "text": req.content, "time": int(time.time() * 1000)} | |
| if req.parent_id: | |
| parent_found = False | |
| for c in pl["comments"]: | |
| if c.get("id") == req.parent_id: | |
| if "replies" not in c: c["replies"] = [] | |
| c["replies"].append(new_comment) | |
| parent_found = True; break | |
| if not parent_found: return {"status": "error", "message": "原评论不存在"} | |
| else: | |
| pl["comments"].append(new_comment) | |
| save_db(db, "cloud_playlists.json") | |
| return {"status": "success"} | |
| except Exception as e: return {"status": "error", "message": str(e)} | |
| # ========================================== | |
| # 图片取回端点（自 api_share.py 迁入） | |
| # 经验分享模块下线后，community_images/ 目录仍存放歌单封面， | |
| # 路由路径保持 /api/share/image 不变以兼容存量封面数据与客户端旧地址 | |
| # ========================================== | |
| def _safe_image_name(name): | |
| """图片名白名单校验：basename 化、禁止穿越、限定图片扩展名""" | |
| if not name: | |
| return None | |
| name = os.path.basename(str(name)) | |
| if ".." in name or not name or name.startswith("."): | |
| return None | |
| if not name.lower().endswith(IMAGE_EXTS): | |
| return None | |
| return name | |
| async def get_playlist_cover_image(name: str): | |
| safe_name = _safe_image_name(name) | |
| if not safe_name: | |
| return JSONResponse(status_code=403, content={"status": "error", "message": "非法的图片路径"}) | |
| try: | |
| path = await asyncio.to_thread(hf_hub_download, repo_id=REPO_ID, filename=f"community_images/{safe_name}", repo_type="dataset", token=TOKEN) | |
| return FileResponse(path) | |
| except Exception: return {"status": "error"} | |