utils.py
12.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
"""工具函数模块"""
import hashlib
import re
from urllib.parse import urlparse
def extract_plain_lyric(lyric: str) -> str:
"""去除 LRC 歌词中的时间戳和标签,保留纯文本
Args:
lyric: 原始歌词内容(可能含 [mm:ss.xx] 时间戳和 [ti:xxx] 等标签)
Returns:
纯文本歌词,行之间用换行符连接
"""
if not lyric:
return ""
lines = []
for line in lyric.splitlines():
# 去除 LRC 时间戳 [mm:ss.xx] 或 [mm:ss.xxx]
cleaned = re.sub(r"\[\d{2}:\d{2}(?:\.\d{2,3})?\]", "", line)
# 去除标签 [xx:yy]
cleaned = re.sub(r"\[[a-zA-Z]+:[^\]]+\]", "", cleaned)
cleaned = cleaned.strip()
if cleaned:
lines.append(cleaned)
return "\n".join(lines)
def split_title_version(title: str | None) -> tuple[str, str]:
"""拆分歌名末尾括号版本信息。
例如:化风行万里 (DJ默涵版) -> (化风行万里, DJ默涵版)
"""
if not title:
return "", ""
text = title.strip()
match = re.match(r"^(?P<title>.+?)\s*[\((](?P<version>[^()()]+)[\))]\s*$", text)
if not match:
return text, ""
clean_title = match.group("title").strip()
version = match.group("version").strip()
return clean_title or text, version
def upload_plain_lyric_to_bucket(platform: str, unique_id: str, lyric: str, bucket, base_url: str) -> str:
"""将歌词去时间戳后上传到当前 ETL 使用的 OSS bucket"""
plain_lyric = extract_plain_lyric(lyric)
if not plain_lyric:
return ""
oss_key = f"crawler/lyric/{platform}/{unique_id}.txt"
bucket.put_object(oss_key, plain_lyric.encode("utf-8"), headers={'Content-Type': 'text/plain; charset=utf-8'})
return f"{base_url.rstrip('/')}/{oss_key}"
def compute_audio_md5(audio_bytes: bytes) -> str:
"""计算音频字节流的 MD5 值(32 位小写十六进制字符串)
Args:
audio_bytes: 音频文件字节流
Returns:
MD5 字符串,输入为空时返回空字符串
"""
if not audio_bytes:
return ""
return hashlib.md5(audio_bytes).hexdigest()
"""工具函数模块"""
import asyncio
import hashlib
import logging
import re
from datetime import datetime
from urllib.parse import urlparse
from zoneinfo import ZoneInfo
try:
import httpx
except ModuleNotFoundError:
httpx = None
try:
from app.core.config import settings
from app.core.oss_client import oss_client
except ModuleNotFoundError:
settings = None
oss_client = None
CHINA_TZ = ZoneInfo("Asia/Shanghai")
logger = logging.getLogger(__name__)
# OSS 上传默认超时(秒),优先使用全局配置
_OSS_UPLOAD_TIMEOUT = settings.OSS_UPLOAD_TIMEOUT_SECONDS if settings else 30
# 音频下载默认超时(秒),优先使用全局配置
_AUDIO_DOWNLOAD_TIMEOUT = settings.AUDIO_DOWNLOAD_TIMEOUT_SECONDS if settings else 30
# 音频下载最大限制(字节),优先使用全局配置
_MAX_AUDIO_SIZE = settings.MAX_AUDIO_DOWNLOAD_SIZE if settings else 50 * 1024 * 1024
# 歌词中常见的词曲作者匹配规则
_LYRICIST_PATTERNS = [
r"作\s*词\s*[::]\s*([^\r\n//]+)",
r"词\s*[::]\s*([^\r\n//]+)",
r"Lyrics?\s*[::]\s*([^\r\n//]+)",
]
_COMPOSER_PATTERNS = [
r"作\s*曲\s*[::]\s*([^\r\n//]+)",
r"曲\s*[::]\s*([^\r\n//]+)",
r"Composer\s*[::]\s*([^\r\n//]+)",
]
def _search_with_patterns(patterns: list, text: str) -> str:
"""根据多个正则匹配文本,返回首个命中分组"""
if not text:
return ""
for pattern in patterns:
match = re.search(pattern, text)
if match:
return match.group(1).strip()
return ""
def extract_lyricist_composer(lyric: str) -> tuple[str, str]:
"""从歌词文本中提取词作者和曲作者
Returns:
(lyricist_name, composer_name)
"""
lyricist = _search_with_patterns(_LYRICIST_PATTERNS, lyric)
composer = _search_with_patterns(_COMPOSER_PATTERNS, lyric)
return lyricist, composer
def now_cn() -> datetime:
"""返回中国时区的当前时间(无时区信息),保留微秒"""
return datetime.now(CHINA_TZ).replace(tzinfo=None)
def extract_plain_lyric(lyric: str) -> str:
"""去除 LRC 歌词中的时间戳和标签,保留纯文本
Args:
lyric: 原始歌词内容(可能含 [mm:ss.xx] 时间戳和 [ti:xxx] 等标签)
Returns:
纯文本歌词,行之间用换行符连接
"""
if not lyric:
return ""
lines = []
for line in lyric.splitlines():
# 去除 LRC 时间戳 [mm:ss.xx] 或 [mm:ss.xxx]
cleaned = re.sub(r"\[\d{2}:\d{2}(?:\.\d{2,3})?\]", "", line)
# 去除标签 [xx:yy]
cleaned = re.sub(r"\[[a-zA-Z]+:[^\]]+\]", "", cleaned)
cleaned = cleaned.strip()
if cleaned:
lines.append(cleaned)
return "\n".join(lines)
def split_title_version(title: str | None) -> tuple[str, str]:
"""拆分歌名末尾括号版本信息。
例如:化风行万里 (DJ默涵版) -> (化风行万里, DJ默涵版)
"""
if not title:
return "", ""
text = title.strip()
match = re.match(r"^(?P<title>.+?)\s*[\((](?P<version>[^()()]+)[\))]\s*$", text)
if not match:
return text, ""
clean_title = match.group("title").strip()
version = match.group("version").strip()
return clean_title or text, version
def upload_plain_lyric_to_bucket(platform: str, unique_id: str, lyric: str, bucket, base_url: str) -> str:
"""将歌词去时间戳后上传到当前 ETL 使用的 OSS bucket"""
plain_lyric = extract_plain_lyric(lyric)
if not plain_lyric:
return ""
oss_key = f"crawler/lyric/{platform}/{unique_id}.txt"
bucket.put_object(oss_key, plain_lyric.encode("utf-8"), headers={'Content-Type': 'text/plain; charset=utf-8'})
return f"{base_url.rstrip('/')}/{oss_key}"
async def upload_lyric_to_oss(platform: str, unique_id: str, lyric: str) -> str:
"""将歌词去除时间戳后上传为纯文本文件到 OSS
Args:
platform: 平台标识,如 qqmusic/kugou/kuwo/netease/migu
unique_id: 平台唯一标识,如 song_mid/hash/rid 等
lyric: 原始歌词内容
Returns:
OSS 文件访问 URL,上传失败或歌词为空返回空字符串
"""
if not lyric:
return ""
if oss_client is None:
logger.error("OSS client is not configured")
return ""
plain_lyric = extract_plain_lyric(lyric)
if not plain_lyric:
return ""
oss_key = f"lyrics/{platform}/{unique_id}.txt"
try:
file_url = await asyncio.wait_for(
asyncio.to_thread(
oss_client.upload_bytes,
plain_lyric.encode("utf-8"),
oss_key,
"public-read",
"text/plain; charset=utf-8",
),
timeout=_OSS_UPLOAD_TIMEOUT,
)
return file_url or ""
except asyncio.TimeoutError:
logger.exception(f"上传歌词到 OSS 超时 {_OSS_UPLOAD_TIMEOUT}s {platform}/{unique_id}")
return ""
except Exception as e:
logger.exception(f"上传歌词到 OSS 失败 {platform}/{unique_id}: {e}")
return ""
async def download_and_upload_audio(
audio_url: str,
oss_key: str,
*,
headers: dict | None = None,
allow_external_oss_url: bool = False,
) -> tuple[str, str]:
"""流式下载音频并上传到 OSS
内置大小限制、下载超时、上传超时保护,防止大文件或慢网络拖垮 worker。
Args:
audio_url: 音频下载地址
oss_key: OSS 对象键,如 songs/qqmusic/xxxx.mp3
headers: 可选的下载请求头
allow_external_oss_url: 若 audio_url 已是当前 OSS 域名下的地址,是否直接透传
Returns:
(OSS 文件 URL, 音频 MD5),失败返回 ("", "")
"""
if not audio_url:
return "", ""
if allow_external_oss_url and settings.OSS_FILE_BASE_NAME and audio_url.startswith(settings.OSS_FILE_BASE_NAME):
return audio_url, ""
audio_bytes = bytearray()
try:
async with httpx.AsyncClient(timeout=_AUDIO_DOWNLOAD_TIMEOUT) as client:
async with client.stream("GET", audio_url, headers=headers or {}) as response:
response.raise_for_status()
content_length = response.headers.get("Content-Length")
if content_length and int(content_length) > _MAX_AUDIO_SIZE:
logger.warning(
f"音频文件过大,跳过: {oss_key}, "
f"size={int(content_length) / 1024 / 1024:.2f}MB"
)
return "", ""
async for chunk in response.aiter_bytes(chunk_size=64 * 1024):
audio_bytes.extend(chunk)
if len(audio_bytes) > _MAX_AUDIO_SIZE:
logger.warning(
f"音频下载超过大小限制,跳过: {oss_key}, "
f"size>{_MAX_AUDIO_SIZE / 1024 / 1024:.2f}MB"
)
return "", ""
except Exception as e:
logger.error(f"下载音频失败 {oss_key}: {e}")
return "", ""
audio_bytes = bytes(audio_bytes)
audio_md5 = compute_audio_md5(audio_bytes)
try:
file_url = await asyncio.wait_for(
asyncio.to_thread(oss_client.upload_bytes, audio_bytes, oss_key),
timeout=_OSS_UPLOAD_TIMEOUT,
)
except asyncio.TimeoutError:
logger.error(f"上传音频到 OSS 超时 {_OSS_UPLOAD_TIMEOUT}s: {oss_key}")
return "", ""
except Exception as e:
logger.error(f"上传音频到 OSS 失败 {oss_key}: {e}")
return "", ""
return file_url or "", audio_md5
_CONTENT_TYPE_EXT_MAP = {
"image/jpeg": "jpg",
"image/jpg": "jpg",
"image/png": "png",
"image/webp": "webp",
"image/gif": "gif",
}
def _guess_image_ext(content_type: str, url: str) -> str:
"""根据 Content-Type 或 URL 路径推断图片扩展名,默认 jpg"""
if content_type:
mime = content_type.split(";")[0].strip().lower()
ext = _CONTENT_TYPE_EXT_MAP.get(mime)
if ext:
return ext
path = urlparse(url).path
suffix = path.rsplit(".", 1)[-1].lower()
if suffix in {"jpg", "jpeg", "png", "webp", "gif"}:
return "jpg" if suffix == "jpeg" else suffix
return "jpg"
async def upload_cover_to_oss(platform: str, unique_id: str, image_url: str) -> str:
"""下载封面图片并上传到 OSS
Args:
platform: 平台标识,如 qqmusic/kugou/kuwo/netease/migu
unique_id: 平台唯一标识,如 song_mid/hash/rid 等
image_url: 封面图片原始 URL
Returns:
OSS 文件访问 URL,上传失败或 URL 为空返回空字符串
"""
if not image_url:
return ""
try:
async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client:
resp = await client.get(image_url, headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Referer": "https://www.kugou.com/",
})
resp.raise_for_status()
image_bytes = resp.content
content_type = resp.headers.get("content-type", "")
except Exception as e:
logger.exception(f"下载封面图片失败 {platform}/{unique_id} {image_url}: {e}")
return ""
ext = _guess_image_ext(content_type, image_url)
oss_key = f"covers/{platform}/{unique_id}.{ext}"
mime = _CONTENT_TYPE_EXT_MAP.get(content_type.split(";")[0].strip().lower(), f"image/{ext}")
try:
file_url = await asyncio.wait_for(
asyncio.to_thread(
oss_client.upload_bytes,
image_bytes,
oss_key,
"public-read",
mime,
),
timeout=_OSS_UPLOAD_TIMEOUT,
)
return file_url or ""
except asyncio.TimeoutError:
logger.exception(f"上传封面到 OSS 超时 {_OSS_UPLOAD_TIMEOUT}s {platform}/{unique_id}")
return ""
except Exception as e:
logger.exception(f"上传封面到 OSS 失败 {platform}/{unique_id}: {e}")
return ""
def compute_audio_md5(audio_bytes: bytes) -> str:
"""计算音频字节流的 MD5 值(32 位小写十六进制字符串)
Args:
audio_bytes: 音频文件字节流
Returns:
MD5 字符串,输入为空时返回空字符串
"""
if not audio_bytes:
return ""
return hashlib.md5(audio_bytes).hexdigest()