Merge branch 'main' of https://repos.hackathon.bz.it/2025-summer/team-1
This commit is contained in:
commit
0d1a3b8c09
4 changed files with 56 additions and 10 deletions
|
@ -6,9 +6,8 @@ from flask_socketio import SocketIO, emit
|
|||
from state import State
|
||||
|
||||
from .room import Room
|
||||
from .song import init_db
|
||||
|
||||
# from .song_fetch import *
|
||||
from .song import Song, init_db, get_song_by_title_artist, add_song_in_db
|
||||
from .song_fetch import lastfm_query_search, download_song_mp3
|
||||
|
||||
dotenv.load_dotenv()
|
||||
|
||||
|
@ -123,5 +122,32 @@ def room():
|
|||
]
|
||||
|
||||
|
||||
@app.post("/api/addsong")
|
||||
def add_song():
|
||||
if (room_id := request.args.get("room")) is None:
|
||||
return error("Missing room id")
|
||||
|
||||
if (room := state.rooms.get(int(room_id))) is None:
|
||||
return error("Invalid room")
|
||||
|
||||
if (query := request.args.get("query")) is None:
|
||||
return error("Missing query")
|
||||
|
||||
info = lastfm_query_search(query)
|
||||
|
||||
if (song := get_song_by_title_artist(info.title, info.artist)) is None:
|
||||
res = download_song_mp3(info.title, info.artist)
|
||||
if res is None:
|
||||
return error("Cannot get info from YT")
|
||||
yt_id, _ = res
|
||||
## song not found, downolad from YT
|
||||
## add in DB
|
||||
add_song_in_db(info.artist, info.title, info.tags, info.img_id, yt_id)
|
||||
|
||||
return {"artist": info.artist, "title": info.title, "tags": info.tags, "image_id": info.img_id}
|
||||
|
||||
|
||||
init_db()
|
||||
|
||||
if __name__ == "__main__":
|
||||
socketio.run(app, debug=True)
|
||||
|
|
|
@ -1,4 +1,5 @@
|
|||
from dataclasses import dataclass
|
||||
import uuid
|
||||
from sqlite3 import Cursor
|
||||
|
||||
from .connect import get_connection
|
||||
|
@ -55,7 +56,7 @@ def get_song_by_title_artist(title: str, artist: str) -> Song | None:
|
|||
return song
|
||||
|
||||
|
||||
def add_song(song: Song):
|
||||
def add_song_in_db(title: str, artist: str, tags: list[str], image_id: str, yt_id: str) -> None:
|
||||
conn = get_connection()
|
||||
cursor = conn.cursor()
|
||||
|
||||
|
@ -63,8 +64,8 @@ def add_song(song: Song):
|
|||
"""
|
||||
INSERT OR REPLACE INTO songs (uuid, title, artist, tags, lastfm_image_id, youtube_id)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(song.uuid, song.title, song.artist, ",".join(song.tags), song.image_id, song.youtube_id),
|
||||
""",
|
||||
(uuid.uuid4(), title, artist, ",".join(tags), image_id, yt_id),
|
||||
) # Updates song info if it already exists
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
|
|
@ -3,12 +3,21 @@ import urllib.parse
|
|||
import os.path
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
|
||||
sys.path.append("/yt-dlp")
|
||||
import yt_dlp
|
||||
|
||||
|
||||
def lastfm_search(query: str) -> tuple[str, str]:
|
||||
@dataclass
|
||||
class SongInfo:
|
||||
artist: str
|
||||
title: str
|
||||
img_id: str
|
||||
tags: list[str]
|
||||
|
||||
|
||||
def _lastfm_search(query: str) -> tuple[str, str]:
|
||||
response = requests.get(
|
||||
url="https://ws.audioscrobbler.com/2.0/?method=track.search&format=json",
|
||||
params={"limit": 5, "track": query, "api_key": os.environ["LASTFM_API_KEY"]},
|
||||
|
@ -21,7 +30,7 @@ def lastfm_search(query: str) -> tuple[str, str]:
|
|||
return track_info["name"], track_info["artist"]
|
||||
|
||||
|
||||
def lastfm_getinfo(name: str, artist: str) -> tuple[str, str, str, str, list[str]]: # ( id, image_id, tags )
|
||||
def _lastfm_getinfo(name: str, artist: str) -> tuple[str, list[str]]: # ( image_id, tags )
|
||||
response = requests.get(
|
||||
url="https://ws.audioscrobbler.com/2.0/?method=track.getInfo&format=json",
|
||||
params={
|
||||
|
@ -36,12 +45,20 @@ def lastfm_getinfo(name: str, artist: str) -> tuple[str, str, str, str, list[str
|
|||
image_url = urllib.parse.urlparse(track_info["album"]["image"][0]["#text"])
|
||||
|
||||
return (
|
||||
track_info["mbid"],
|
||||
[t["name"] for t in track_info["toptags"]["tag"]],
|
||||
# track_info["mbid"],
|
||||
os.path.splitext(os.path.basename(image_url.path))[0],
|
||||
[t["name"] for t in track_info["toptags"]["tag"]],
|
||||
)
|
||||
|
||||
|
||||
def lastfm_query_search(query: str) -> SongInfo:
|
||||
name, artist = _lastfm_search(query)
|
||||
|
||||
img_id, tags = _lastfm_getinfo(name, artist)
|
||||
|
||||
return SongInfo(artist=artist, title=name, img_id=img_id, tags=tags)
|
||||
|
||||
|
||||
def download_song_mp3(name: str, artist: str) -> tuple[str, str] | None: # ( id, audio )
|
||||
ydl_opts = {
|
||||
"format": "bestaudio",
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue