thief_music_api/api.py

import fastapi
import os
import requests
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from service.userservice 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")
async def get_rank(
        rank_id: str = fastapi.Query(..., description="榜单类型")):
    """
    :param rank_id:
    19723756 云音乐飙升榜
    3779629 云音乐新歌榜
    3778678 云音乐热歌榜
    2884035 云音乐原创榜
    :return:
    """
    url = f"https://music.163.com/api/playlist/detail?id={rank_id}"
    response = requests.get(url)
    data = response.json()

    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)