chore: bump version to 0.7.1
This commit is contained in:
ethan.chen
435
backend/modules/lyrics.py
Normal file
435
backend/modules/lyrics.py
Normal file
@@ -0,0 +1,435 @@
|
||||
"""
|
||||
歌词处理模块
|
||||
负责歌词的解析、格式化、存储和获取
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import logging
|
||||
import requests
|
||||
import sqlite3
|
||||
from urllib.parse import quote
|
||||
from Crypto.Cipher import AES
|
||||
import base64
|
||||
import random
|
||||
import codecs
|
||||
import json
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 配置应用支持目录
|
||||
APP_SUPPORT_DIR = os.path.expanduser('~/Library/Application Support/lyroc')
|
||||
os.makedirs(APP_SUPPORT_DIR, exist_ok=True)
|
||||
|
||||
# SQLite数据库配置
|
||||
DB_PATH = os.path.join(APP_SUPPORT_DIR, 'lyroc.db')
|
||||
|
||||
def init_db():
|
||||
"""初始化SQLite数据库"""
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 创建歌词表
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS lyrics (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
track_name TEXT NOT NULL,
|
||||
artist TEXT NOT NULL,
|
||||
lyrics_content TEXT NOT NULL,
|
||||
source TEXT DEFAULT 'netease',
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE(track_name, artist)
|
||||
)
|
||||
''')
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
logger.debug(f"数据库初始化完成: {DB_PATH}")
|
||||
|
||||
# 初始化数据库
|
||||
init_db()
|
||||
|
||||
def clear_lyrics_cache():
|
||||
"""清空歌词缓存"""
|
||||
global _lyrics_cache
|
||||
_lyrics_cache = {
|
||||
"track_name": None,
|
||||
"artist": None,
|
||||
"lyrics_text": None,
|
||||
"source": None,
|
||||
"deleted": False
|
||||
}
|
||||
|
||||
# 歌词缓存
|
||||
clear_lyrics_cache()
|
||||
|
||||
# 网易云音乐API配置
|
||||
NETEASE_API_BASE = "https://music.163.com/weapi"
|
||||
NETEASE_SEARCH_URL = f"{NETEASE_API_BASE}/search/get"
|
||||
NETEASE_LYRIC_URL = f"{NETEASE_API_BASE}/song/lyric"
|
||||
|
||||
# 加密相关配置
|
||||
MODULUS = '00e0b509f6259df8642dbc35662901477df22677ec152b5ff68ace615bb7b725152b3ab17a876aea8a5aa76d2e417629ec4ee341f56135fccf695280104e0312ecbda92557c93870114af6c9d05c4f7f0c3685b7a46bee255932575cce10b424d813cfe4875d3e82047b97ddef52741d546b8e289dc6935b3ece0462db0a22b8e7'
|
||||
NONCE = '0CoJUm6Qyw8W8jud'
|
||||
PUBKEY = '010001'
|
||||
IV = '0102030405060708'
|
||||
|
||||
class LyricLine:
|
||||
def __init__(self, time, text):
|
||||
self.time = time # 时间（秒）
|
||||
self.text = text # 歌词文本
|
||||
|
||||
def load_lyrics_from_db(track_name, artist):
|
||||
"""从数据库加载歌词"""
|
||||
try:
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
SELECT lyrics_content FROM lyrics WHERE track_name = ? AND artist = ?
|
||||
''', (track_name, artist))
|
||||
result = cursor.fetchone()
|
||||
conn.close()
|
||||
if result:
|
||||
return result[0]
|
||||
except Exception as e:
|
||||
logger.error(f"加载数据库歌词时出错: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
def parse_time(time_str):
|
||||
"""解析时间标记 [mm:ss.xxx]"""
|
||||
try:
|
||||
# 移除方括号
|
||||
time_str = time_str.strip('[]')
|
||||
# 分离分钟和秒
|
||||
minutes, seconds = time_str.split(':')
|
||||
# 转换为秒
|
||||
total_seconds = float(minutes) * 60 + float(seconds)
|
||||
return total_seconds
|
||||
except:
|
||||
return 0
|
||||
|
||||
def format_lrc_lyrics(lyrics_text, current_position=None):
|
||||
"""格式化 LRC 歌词，找出当前、下一句和下下句歌词"""
|
||||
if not lyrics_text:
|
||||
return None, None, None, None
|
||||
|
||||
# 解析歌词行
|
||||
lyric_lines = []
|
||||
for line in lyrics_text.split('\n'):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
|
||||
# 查找所有时间标记
|
||||
time_tags = re.findall(r'\[([0-9:.]+)\]', line)
|
||||
if not time_tags:
|
||||
continue
|
||||
|
||||
text = re.sub(r'\[[0-9:.]+\]', '', line).strip()
|
||||
if not text:
|
||||
continue
|
||||
|
||||
# 一行可能有多个时间标记
|
||||
for time_tag in time_tags:
|
||||
try:
|
||||
time_seconds = parse_time(time_tag)
|
||||
lyric_lines.append({"time": time_seconds, "text": text})
|
||||
except Exception as e:
|
||||
logger.error(f"解析时间标记出错: {time_tag}, 错误: {e}")
|
||||
|
||||
# 按时间排序
|
||||
lyric_lines.sort(key=lambda x: x["time"])
|
||||
|
||||
# 如果没有有效歌词，返回空
|
||||
if not lyric_lines:
|
||||
return None, None, None, None
|
||||
|
||||
if not current_position:
|
||||
# 如果没有提供当前位置，返回前三句歌词
|
||||
current = lyric_lines[0]["text"] if len(lyric_lines) > 0 else None
|
||||
next_ = lyric_lines[1]["text"] if len(lyric_lines) > 1 else None
|
||||
next_next = lyric_lines[2]["text"] if len(lyric_lines) > 2 else None
|
||||
return current, next_, next_next, lyric_lines[0]["time"]
|
||||
|
||||
# 找出当前歌词
|
||||
current_index = -1
|
||||
for i, line in enumerate(lyric_lines):
|
||||
if line["time"] > current_position:
|
||||
if i > 0:
|
||||
current_index = i - 1
|
||||
break
|
||||
|
||||
# 处理当前位置超过最后一行歌词的情况
|
||||
if current_index == -1 and lyric_lines:
|
||||
# 如果遍历完所有歌词都没找到大于当前位置的，说明当前位置超过了最后一行歌词
|
||||
if current_position >= lyric_lines[-1]["time"]:
|
||||
# 返回最后三句歌词，将最后一句作为当前歌词
|
||||
last_index = len(lyric_lines) - 1
|
||||
if last_index >= 2:
|
||||
return lyric_lines[last_index]["text"], None, None, lyric_lines[last_index]["time"]
|
||||
elif last_index == 1:
|
||||
return lyric_lines[1]["text"], None, None, lyric_lines[1]["time"]
|
||||
else:
|
||||
return lyric_lines[0]["text"], None, None, lyric_lines[0]["time"]
|
||||
|
||||
# 如果找到了当前歌词
|
||||
if current_index >= 0:
|
||||
current = lyric_lines[current_index]["text"]
|
||||
next_ = lyric_lines[current_index + 1]["text"] if current_index + 1 < len(lyric_lines) else None
|
||||
next_next = lyric_lines[current_index + 2]["text"] if current_index + 2 < len(lyric_lines) else None
|
||||
return current, next_, next_next, lyric_lines[current_index]["time"]
|
||||
|
||||
# 如果当前位置在所有歌词之前
|
||||
if lyric_lines:
|
||||
return None, lyric_lines[0]["text"], lyric_lines[1]["text"] if len(lyric_lines) > 1 else None, lyric_lines[0]["time"]
|
||||
|
||||
return None, None, None, None
|
||||
|
||||
def get_lyrics_data(status):
|
||||
"""获取当前歌词（先获取状态，再从缓存或存储中获取歌词）"""
|
||||
global _lyrics_cache
|
||||
|
||||
# 如果没有播放，清空缓存并返回空
|
||||
if status.get("status") not in ["playing", "paused"]:
|
||||
clear_lyrics_cache()
|
||||
return {
|
||||
"current_lyric_time": None,
|
||||
"current_lyric": None,
|
||||
"next_lyric": None,
|
||||
"next_next_lyric": None,
|
||||
"track_name": None,
|
||||
"artist_name": None
|
||||
}
|
||||
|
||||
track_name = status.get("track_name")
|
||||
artist = status.get("artist")
|
||||
position = status.get("position", 0)
|
||||
|
||||
# 检查是否切换了歌曲
|
||||
if _lyrics_cache["track_name"] != track_name or _lyrics_cache["artist"] != artist:
|
||||
logger.debug(f"歌曲切换，重新获取歌词: {track_name} - {artist}")
|
||||
# 清空缓存
|
||||
_lyrics_cache["track_name"] = track_name
|
||||
_lyrics_cache["artist"] = artist
|
||||
_lyrics_cache["lyrics_text"] = None
|
||||
_lyrics_cache["deleted"] = False
|
||||
|
||||
# 1. 先尝试从数据库获取
|
||||
lyrics_text = load_lyrics_from_db(track_name, artist)
|
||||
if lyrics_text:
|
||||
_lyrics_cache["lyrics_text"] = lyrics_text
|
||||
_lyrics_cache["source"] = "db"
|
||||
|
||||
# 2. 如果本地没有，且不是被删除的歌词，尝试从网易云音乐API获取歌词
|
||||
elif not _lyrics_cache["lyrics_text"] and not _lyrics_cache.get("deleted", False):
|
||||
lyrics_text = search_lyrics(track_name, artist)
|
||||
if lyrics_text:
|
||||
# 保存到数据库
|
||||
save_lyrics(track_name, artist, lyrics_text)
|
||||
_lyrics_cache["lyrics_text"] = lyrics_text
|
||||
_lyrics_cache["source"] = "netease"
|
||||
|
||||
# 如果有缓存的歌词，直接解析
|
||||
if _lyrics_cache["lyrics_text"]:
|
||||
if _lyrics_cache.get("deleted", False):
|
||||
return {
|
||||
"current_lyric_time": None,
|
||||
"current_lyric": None,
|
||||
"next_lyric": None,
|
||||
"next_next_lyric": None,
|
||||
}
|
||||
# 根据播放位置解析当前歌词
|
||||
current, next_, next_next, current_time = format_lrc_lyrics(_lyrics_cache["lyrics_text"], position)
|
||||
if current or next_ or next_next:
|
||||
return {
|
||||
"current_lyric_time": current_time,
|
||||
"current_lyric": current,
|
||||
"next_lyric": next_,
|
||||
"next_next_lyric": next_next,
|
||||
"track_name": track_name,
|
||||
"artist_name": artist,
|
||||
"lyrics_source": _lyrics_cache["source"] # 标记歌词来源
|
||||
}
|
||||
|
||||
|
||||
# 都没找到返回空
|
||||
return {
|
||||
"current_lyric_time": None,
|
||||
"current_lyric": None,
|
||||
"next_lyric": None,
|
||||
"next_next_lyric": None,
|
||||
"track_name": track_name,
|
||||
"artist_name": artist,
|
||||
"lyrics_source": "none" # 标记歌词来源
|
||||
}
|
||||
|
||||
def create_secret_key(size):
|
||||
"""生成随机密钥"""
|
||||
return ''.join(random.choice('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789') for _ in range(size))
|
||||
|
||||
def aes_encrypt(text, key):
|
||||
"""AES加密"""
|
||||
pad = 16 - len(text) % 16
|
||||
text = text + chr(pad) * pad
|
||||
encryptor = AES.new(key.encode('utf-8'), AES.MODE_CBC, IV.encode('utf-8'))
|
||||
encrypt_text = encryptor.encrypt(text.encode('utf-8'))
|
||||
encrypt_text = base64.b64encode(encrypt_text).decode('utf-8')
|
||||
return encrypt_text
|
||||
|
||||
def rsa_encrypt(text, pubkey, modulus):
|
||||
"""RSA加密"""
|
||||
text = text[::-1]
|
||||
rs = pow(int(codecs.encode(text.encode('utf-8'), 'hex'), 16), int(pubkey, 16), int(modulus, 16))
|
||||
return format(rs, 'x').zfill(256)
|
||||
|
||||
headers = {
|
||||
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36',
|
||||
'Referer': 'https://music.163.com/',
|
||||
'Origin': 'https://music.163.com',
|
||||
'Content-Type': 'application/x-www-form-urlencoded'
|
||||
}
|
||||
def encrypt_request(text):
|
||||
"""加密请求数据"""
|
||||
secret_key = create_secret_key(16)
|
||||
params = aes_encrypt(text, NONCE)
|
||||
params = aes_encrypt(params, secret_key)
|
||||
encSecKey = rsa_encrypt(secret_key, PUBKEY, MODULUS)
|
||||
return {
|
||||
'params': params,
|
||||
'encSecKey': encSecKey
|
||||
}
|
||||
|
||||
def get_lyrics_from_netease(track_name, artist_name):
|
||||
# 构建搜索关键词
|
||||
search_keyword = f"{track_name} {artist_name}"
|
||||
|
||||
# 搜索歌曲
|
||||
data = {
|
||||
's': search_keyword,
|
||||
'type': 1, # 1: 单曲, 10: 专辑, 100: 歌手, 1000: 歌单
|
||||
'limit': 5,
|
||||
'offset': 0,
|
||||
'hlpretag': '<span class="s-fc7">',
|
||||
'hlposttag': '</span>',
|
||||
'total': True,
|
||||
'csrf_token': ''
|
||||
}
|
||||
|
||||
encrypted_data = encrypt_request(json.dumps(data))
|
||||
response = requests.post(NETEASE_SEARCH_URL, data=encrypted_data, headers=headers)
|
||||
logger.debug(f"网易云音乐搜索响应: {response.text}")
|
||||
response.raise_for_status()
|
||||
|
||||
result = response.json()
|
||||
return result
|
||||
|
||||
def get_lyrics_from_id(id):
|
||||
"""从网易云音乐获取歌词"""
|
||||
# 获取歌词
|
||||
data = {
|
||||
'id': id,
|
||||
'lv': 1, # 获取歌词
|
||||
'kv': 1, # 获取翻译
|
||||
'tv': -1, # 不获取罗马音
|
||||
'csrf_token': ''
|
||||
}
|
||||
|
||||
encrypted_data = encrypt_request(json.dumps(data))
|
||||
response = requests.post(NETEASE_LYRIC_URL, data=encrypted_data, headers=headers)
|
||||
response.raise_for_status()
|
||||
|
||||
result = response.json()
|
||||
return result
|
||||
|
||||
|
||||
def search_lyrics(track_name, artist_name):
|
||||
"""从网易云音乐搜索并获取歌词"""
|
||||
try:
|
||||
result = get_lyrics_from_netease(track_name, artist_name)
|
||||
if result['code'] == 200 and result['result']['songCount'] > 0:
|
||||
song = result['result']['songs'][0]
|
||||
id = song['id']
|
||||
lrc_result = get_lyrics_from_id(id)
|
||||
if lrc_result['code'] == 200 and 'lrc' in lrc_result:
|
||||
logger.debug(f"网易云音乐歌词: {lrc_result['lrc']['lyric']}")
|
||||
return lrc_result['lrc']['lyric']
|
||||
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"从网易云音乐获取歌词时出错: {e}", exc_info=True)
|
||||
return None
|
||||
def format_lyrics(lyrics):
|
||||
"""格式化歌词文本，保留时间信息"""
|
||||
if not lyrics:
|
||||
logger.debug("歌词内容为空")
|
||||
return []
|
||||
|
||||
lyric_lines = []
|
||||
for line in lyrics.split('\n'):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
|
||||
# 查找所有时间标记
|
||||
time_tags = re.findall(r'\[([0-9:.]+)\]', line)
|
||||
if not time_tags:
|
||||
continue
|
||||
|
||||
text = re.sub(r'\[[0-9:.]+\]', '', line).strip()
|
||||
if not text:
|
||||
continue
|
||||
|
||||
# 可能一行有多个时间标记
|
||||
for time_tag in time_tags:
|
||||
try:
|
||||
time_seconds = parse_time(time_tag)
|
||||
lyric_lines.append(LyricLine(time_seconds, text))
|
||||
except Exception as e:
|
||||
logger.error(f"解析时间标记出错: {time_tag}, 错误: {e}")
|
||||
|
||||
# 按时间排序
|
||||
sorted_lyrics = sorted(lyric_lines, key=lambda x: x.time)
|
||||
logger.debug(f"解析完成，共 {len(sorted_lyrics)} 行歌词")
|
||||
return sorted_lyrics
|
||||
|
||||
def save_lyrics(track_name, artist, lyrics):
|
||||
"""保存歌词到本地文件，如果已存在则更新"""
|
||||
try:
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
INSERT INTO lyrics (track_name, artist, lyrics_content)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(track_name, artist)
|
||||
DO UPDATE SET
|
||||
lyrics_content = excluded.lyrics_content,
|
||||
created_at = CURRENT_TIMESTAMP
|
||||
''', (track_name, artist, lyrics))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
logger.debug(f"歌词已保存/更新到数据库: {track_name} - {artist}")
|
||||
except Exception as e:
|
||||
logger.error(f"保存歌词时出错: {e}", exc_info=True)
|
||||
|
||||
def delete_lyrics(track_name, artist):
|
||||
"""从数据库中删除歌词"""
|
||||
try:
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
DELETE FROM lyrics WHERE track_name = ? AND artist = ?
|
||||
''', (track_name, artist))
|
||||
conn.commit()
|
||||
deleted = cursor.rowcount > 0
|
||||
conn.close()
|
||||
|
||||
if deleted:
|
||||
logger.info(f"已删除歌词: {track_name} - {artist}")
|
||||
# 标记这首歌的歌词已被删除，避免重新获取
|
||||
_lyrics_cache["deleted"] = True
|
||||
return True
|
||||
else:
|
||||
logger.warning(f"未找到要删除的歌词: {track_name} - {artist}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"删除歌词时出错: {e}", exc_info=True)
|
||||
return False
|
||||
Reference in New Issue
Block a user