import fastapi import os import requests import uvicorn from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from service.userservice import * from service.rank_data import * env = os.environ app = FastAPI() # 配置 CORS 中间件 app.add_middleware( CORSMiddleware, allow_origins=["*"], # 允许所有来源，可以根据需求进行配置 allow_credentials=True, allow_methods=["*"], # 允许所有请求方法 allow_headers=["*"], # 允许所有请求头 ) # 注册 @app.post("/register") async def register(username: str = fastapi.Form(..., description="用户名"), password: str = fastapi.Form(..., description="密码")): """ 注册新用户 :param username: 用户名 :param password: 密码 :return: 注册结果 """ # 读取现有的用户信息 existing_users = read_users() # 检查用户名是否已存在 if username in existing_users: return {"success": False, "message": "用户名已存在"} # 添加新用户信息 new_user = {"username": username, "password": password} existing_users[username] = new_user # 将用户信息保存到CSV文件 save_users(existing_users) return {"success": True, "message": "注册成功"} # 登陆 @app.post("/login") async def login(username: str = fastapi.Form(..., description="用户名"), password: str = fastapi.Form(..., description="密码")): """ 用户登录 :param username: 用户名 :param password: 密码 :return: 登录结果 """ # 读取用户信息 existing_users = read_users() # 验证用户名和密码 user = existing_users.get(username) if user and user["password"] == password: return {"success": True, "message": "登录成功"} return {"success": False, "message": "用户名或密码错误"} # 榜单获取 @app.get("/get_rank_test") async def get_rank( rank_id: str = fastapi.Query(..., description="榜单类型")): """ :param rank_id: 19723756 云音乐飙升榜 3779629 云音乐新歌榜 3778678 云音乐热歌榜 2884035 云音乐原创榜 :return: """ headers = { "Referer": "https://music.163.com/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" " AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.190 Safari/537.36" } url = f"https://music.163.com/api/playlist/detail?id={rank_id}" response = requests.get(url, headers=headers) data = response.json() return {"message": "success", "data": data} @app.get("/get_rank") async def get_rank(rank_id: str = fastapi.Query(..., description="榜单类型")): """ 获取音乐榜单数据 :param rank_id: 榜单ID 19723756 云音乐飙升榜 3779629 云音乐新歌榜 3778678 云音乐热歌榜 2884035 云音乐原创榜 :return: 榜单数据 """ # 检查是否有缓存数据 cached_data = read_cache(rank_id) if cached_data: return {"message": "success", "data": cached_data} headers = { "Referer": "https://music.163.com/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" " AppleWebKit/537.36 (KHTML, like Gecko) Chrome/88.0.4324.190 Safari/537.36" } # 缓存数据不存在，从网络获取榜单数据 url = f"https://music.163.com/api/playlist/detail?id={rank_id}" response = requests.get(url, headers=headers) data = response.json() # 将数据写入缓存文件 write_cache(rank_id, data) return {"message": "success", "data": data} # 根据歌曲名称搜索歌曲 @app.get("/search_song_by_name") async def search_song_by_name( name: str = fastapi.Query(..., description="歌曲名称")): payload = { "input": name, "filter": "name", "type": "netease", # netease, tencent, kugou, xiami, baidu "page": 1 } url = "https://sunpma.com/other/musicss/" # Headers for the POST request headers = { "Accept": "application/json, text/javascript, */*; q=0.01", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7", "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "Origin": "https://sunpma.com", "Referer": f"https://sunpma.com/other/musicss/?name={payload['input']}&type={payload['type']}", "Sec-Ch-Ua": "\"Not.A/Brand\";v=\"8\", \"Chromium\";v=\"114\", \"Google Chrome\";v=\"114\"", "Sec-Ch-Ua-Mobile": "?0", "Sec-Ch-Ua-Platform": "\"macOS\"", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36", "X-Requested-With": "XMLHttpRequest" } # 将所有 headers 的值用 'utf-8' 进行编码 headers = {key: value.encode('utf-8') for key, value in headers.items()} # Making the POST request response = requests.post(url, data=payload, headers=headers) # Check if the request was successful (status code 200) if response.status_code == 200: # Parse the response JSON data data = response.json() return {"message": "success", "data": data} else: print(f"Failed to get data. Status code: {response.status_code}") return {"message": "failed", "data": []} # 根据歌曲id搜索歌曲 @app.get("/search_song_by_id") async def search_song_by_id( id: str = fastapi.Query(..., description="歌曲id")): payload = { "input": id, "filter": "id", "type": "netease", # netease, tencent, kugou, xiami, baidu "page": 1 } headers = { "Accept": "application/json, text/javascript, */*; q=0.01", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7", "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "Origin": "https://sunpma.com", "Referer": f"https://sunpma.com/other/musicss/?id={payload['input']}&type=netease", "Sec-Ch-Ua": "\"Not.A/Brand\";v=\"8\", \"Chromium\";v=\"114\", \"Google Chrome\";v=\"114\"", "Sec-Ch-Ua-Mobile": "?0", "Sec-Ch-Ua-Platform": "\"macOS\"", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36", "X-Requested-With": "XMLHttpRequest" } url = "https://sunpma.com/other/musicss/" # Making the POST request response = requests.post(url, data=payload, headers=headers) # Check if the request was successful (status code 200) if response.status_code == 200: # Parse the response JSON data data = response.json() return {"message": "success", "data": data} else: print(f"Failed to get data. Status code: {response.status_code}") return {"message": "failed", "data": []} if __name__ == '__main__': host = env.get("HOST") if env.get("HOST") is not None else "0.0.0.0" port = int(env.get("PORT")) if env.get("PORT") is not None else 7788 uvicorn.run(app='api:app', host=host, port=port, reload=True)