Skip to content

沈秋雨 / import_hk_songs

Go to a project
Toggle navigation
Toggle navigation pinning
Commit c41b71ef c41b71efbde226ed077cbbe5152a6da8bac78a93 by 沈秋雨
Options

fix(runner): 调整导入资源必填规则并优化选择逻辑 

- 修改 records2 歌曲基础字段要求，封面和歌词允许为空
- 根据导入类型动态设定必需资源，音频始终必需，歌词封面视情况而定
- 调整各平台录音选择函数，支持可选的必需资源参数
- 修改 _require_primary_assets 函数，支持传入必需资源参数校验
- 更新导入流程以使用动态资源需求，避免无必要的拒绝
- 添加单元测试覆盖不同导入表的资源要求和录音候选选择逻辑
- 代码注释及文档更新以反映新的资源校验规则
1 parent bf8d1723
Inline Side-by-side
Showing 4 changed files with 97 additions and 25 deletions
......@@ -56,7 +56,7 @@ if YINYAN_IMPORT_TABLE not in _ALLOWED_YINYAN_IMPORT_TABLES:
f"got {YINYAN_IMPORT_TABLE!r}"
)
BATCH_SIZE = 1000
BATCH_SIZE = 10
BACKFILL_BATCH_SIZE = int(os.environ.get('BACKFILL_BATCH_SIZE', '5000'))
HTTP_POOL_MAXSIZE = int(os.environ.get('HTTP_POOL_MAXSIZE', '128'))
HTTP_TRANSFER_RETRIES = int(os.environ.get('HTTP_TRANSFER_RETRIES', '3'))
......
......@@ -28,8 +28,7 @@ WHERE deleted = '0'
ORDER BY id
"""
# 用于补全 yinyan_song_records2 的歌曲必须具备所有下游所需字段。
# 当前 hk_songs_test 中的 lyrics_url 即歌曲链接字段。
# records2 初始化不要求封面或歌词；正式导入时这两个字段允许为空。
_HK_SONGS_RECORDS2_QUERY = """
SELECT id, source_song_id
FROM hk_songs_test
......@@ -39,7 +38,6 @@ WHERE deleted = '0'
AND name IS NOT NULL AND TRIM(name) != ''
AND song_time IS NOT NULL AND song_time > 0
AND singer IS NOT NULL AND TRIM(singer) != ''
AND lyrics_url IS NOT NULL AND TRIM(lyrics_url) != ''
AND audio_url IS NOT NULL AND TRIM(audio_url) != ''
AND issue_time IS NOT NULL
ORDER BY id
......@@ -150,7 +148,7 @@ def iter_hk_songs_records2_batches(
batch_size: int,
start_after_id: int = 0,
) -> Iterator[list[dict]]:
"""遍历具备 records2 入库必填字段的歌曲。"""
"""遍历具备 records2 基础入库字段的歌曲，封面和歌词可为空。"""
last_id = start_after_id
with conn.cursor() as cur:
while True:
......
......@@ -4,7 +4,10 @@ import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from tqdm import tqdm
from .config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, BATCH_SIZE, BACKFILL_BATCH_SIZE, OSS_CONFIG
from .config import (
PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, BATCH_SIZE, BACKFILL_BATCH_SIZE,
OSS_CONFIG, YINYAN_IMPORT_TABLE,
)
from .connections import get_hk_songs_conn, get_source_conn, get_spider_conn, get_pg_conn, get_oss_bucket, refresh_conn, close_all_pools
from .reader import (
iter_hk_songs_batches,
......@@ -121,14 +124,19 @@ def _safe_upload_lyric(
return ''
def _require_primary_assets(assets: dict) -> None:
"""音频、歌曲封面、歌词均为入库必需资源，任一失败即拒绝整条记录。"""
def _required_import_assets() -> tuple[str, ...]:
"""按导入状态表返回硬性资源要求；音频在两条支线中始终必需。"""
if YINYAN_IMPORT_TABLE == 'yinyan_song_records2':
return ('audio',)
return ('audio', 'lyric')
def _require_primary_assets(assets: dict, required: tuple[str, ...] | None = None) -> None:
"""校验当前导入支线要求的资源，允许非必需资源以空值入库。"""
audio_url, _ = assets.get('audio', ('', ''))
missing = [
name for name, value in (
('audio', audio_url), ('cover', assets.get('cover')), ('lyric', assets.get('lyric')),
) if not value
]
values = {'audio': audio_url, 'cover': assets.get('cover'), 'lyric': assets.get('lyric')}
required = required or ('audio', 'cover', 'lyric')
missing = [name for name in required if not values.get(name)]
if missing:
raise RequiredAssetTransferError(missing)
......@@ -206,14 +214,17 @@ def _select_qq_import_record(
current_record: dict,
candidate_records: list[dict],
songs_by_mid: dict[str, dict],
required_assets: tuple[str, ...] = ('lyric', 'cover'),
) -> tuple[dict | None, str]:
"""为导入选择 QQ 录音；当前录音的歌词或封面不可用时，改选同源的合格候选。
返回 (record, reason)，record 为 None 时 reason 说明失败原因。"""
current_song = songs_by_mid.get(current_record['platform_unique_key'])
if not current_song:
return None, 'spider data missing'
needs_lyric = not _has_usable_qq_lyric(current_song, hk_row)
needs_cover = not _is_usable_qq_cover(_qq_cover_source(hk_row, current_song)) or not current_song.get('album_id')
needs_lyric = 'lyric' in required_assets and not _has_usable_qq_lyric(current_song, hk_row)
needs_cover = 'cover' in required_assets and (
not _is_usable_qq_cover(_qq_cover_source(hk_row, current_song)) or not current_song.get('album_id')
)
if not needs_lyric and not needs_cover:
return current_record, 'ok'
......@@ -239,6 +250,7 @@ def _select_kugou_import_record(
current_record: dict,
candidate_records: list[dict],
songs_by_id: dict[int, dict],
required_assets: tuple[str, ...] = ('lyric', 'cover'),
) -> tuple[dict | None, str]:
"""为导入选择酷狗录音；当前录音的歌词或封面不可用时，改选同源的合格候选。
返回 (record, reason)，record 为 None 时 reason 说明失败原因。"""
......@@ -246,8 +258,8 @@ def _select_kugou_import_record(
current_song = songs_by_id.get(song_id)
if not current_song:
return None, 'spider data missing'
needs_lyric = not _has_kugou_lyric(current_song, hk_row)
needs_cover = not _has_kugou_cover(hk_row, current_song)
needs_lyric = 'lyric' in required_assets and not _has_kugou_lyric(current_song, hk_row)
needs_cover = 'cover' in required_assets and not _has_kugou_cover(hk_row, current_song)
if not needs_lyric and not needs_cover:
return current_record, 'ok'
......@@ -274,6 +286,7 @@ def _select_netease_import_record(
current_record: dict,
candidate_records: list[dict],
songs_by_id: dict[int, dict],
required_assets: tuple[str, ...] = ('lyric', 'cover'),
) -> tuple[dict | None, str]:
"""为导入选择网易录音；当前录音的歌词或封面不可用时，改选同源的合格候选。
返回 (record, reason)，record 为 None 时 reason 说明失败原因。"""
......@@ -281,8 +294,8 @@ def _select_netease_import_record(
current_song = songs_by_id.get(song_id)
if not current_song:
return None, 'spider data missing'
needs_lyric = not _has_netease_lyric(current_song, hk_row)
needs_cover = not _has_netease_cover(hk_row, current_song)
needs_lyric = 'lyric' in required_assets and not _has_netease_lyric(current_song, hk_row)
needs_cover = 'cover' in required_assets and not _has_netease_cover(hk_row, current_song)
if not needs_lyric and not needs_cover:
return current_record, 'ok'
......@@ -332,7 +345,7 @@ def _prepare_qq_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict, sing
)
)
assets = _run_io_tasks(tasks)
_require_primary_assets(assets)
_require_primary_assets(assets, _required_import_assets())
audio_url, audio_md5 = assets['audio']
cover_url = assets['cover']
lyric_url = assets['lyric']
......@@ -461,7 +474,7 @@ def _prepare_kugou_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict, s
)
)
assets = _run_io_tasks(tasks)
_require_primary_assets(assets)
_require_primary_assets(assets, _required_import_assets())
audio_url, audio_md5 = assets['audio']
cover_url = assets['cover']
lyric_url = assets['lyric']
......@@ -573,7 +586,7 @@ def _prepare_netease_payload(hk_row: dict, pr: dict, bucket, base_url, sp: dict,
)
)
assets = _run_io_tasks(tasks)
_require_primary_assets(assets)
_require_primary_assets(assets, _required_import_assets())
audio_url, audio_md5 = assets['audio']
cover_url = assets['cover']
lyric_url = assets['lyric']
......@@ -1846,6 +1859,7 @@ def run(
prepare_inputs = []
rejected_pending: list[dict] = []
required_assets = _required_import_assets()
for pending in pending_records:
src_id = int(pending['song_id'])
platform = str(pending['platform'])
......@@ -1866,15 +1880,15 @@ def run(
continue
if platform == PLATFORM_QQ:
selected, reason = _select_qq_import_record(
hk_row, pr, qq_records_by_song.get(src_id, []), qq_songs_map,
hk_row, pr, qq_records_by_song.get(src_id, []), qq_songs_map, required_assets,
)
elif platform == PLATFORM_KUGOU:
selected, reason = _select_kugou_import_record(
hk_row, pr, kugou_records_by_song.get(src_id, []), kugou_songs_map,
hk_row, pr, kugou_records_by_song.get(src_id, []), kugou_songs_map, required_assets,
)
elif platform == PLATFORM_NETEASE:
selected, reason = _select_netease_import_record(
hk_row, pr, netease_records_by_song.get(src_id, []), netease_songs_map,
hk_row, pr, netease_records_by_song.get(src_id, []), netease_songs_map, required_assets,
)
else:
selected, reason = pr, 'ok'
......
......@@ -78,6 +78,33 @@ def test_required_asset_validation_rejects_partial_transfer():
})
def test_records_import_requires_audio_and_lyric_but_allows_empty_cover(monkeypatch):
monkeypatch.setattr(runner, 'YINYAN_IMPORT_TABLE', 'yinyan_song_records')
runner._require_primary_assets({
'audio': ('https://archive.example/audio.mp3', 'md5'),
'cover': '',
'lyric': 'https://archive.example/lyric.txt',
}, runner._required_import_assets())
with pytest.raises(runner.RequiredAssetTransferError, match='lyric'):
runner._require_primary_assets({
'audio': ('https://archive.example/audio.mp3', 'md5'),
'cover': '',
'lyric': '',
}, runner._required_import_assets())
def test_records2_import_allows_empty_cover_and_lyric(monkeypatch):
monkeypatch.setattr(runner, 'YINYAN_IMPORT_TABLE', 'yinyan_song_records2')
runner._require_primary_assets({
'audio': ('https://archive.example/audio.mp3', 'md5'),
'cover': '',
'lyric': '',
}, runner._required_import_assets())
def test_lyric_upload_prefers_hk_url_and_forces_oss_transfer(monkeypatch):
download = MagicMock(return_value='[00:01.00]HK lyric')
monkeypatch.setattr(runner, 'download_text_url', download)
......@@ -184,6 +211,36 @@ def test_select_qq_import_record_returns_none_when_no_candidate_satisfies_missin
assert 'lyric' in reason and 'cover' in reason
def test_records_selection_does_not_reject_or_replace_for_missing_cover():
current = {'record_id': 10, 'platform_unique_key': 'current-mid'}
selected, reason = runner._select_qq_import_record(
{'cover_url': '', 'lyrics_url': 'https://example.com/hk-lyric.lrc'},
current,
[current],
{'current-mid': {'album_id': None, 'cover': '', 'lyric': ''}},
required_assets=('audio', 'lyric'),
)
assert selected == current
assert reason == 'ok'
def test_records2_selection_accepts_current_record_with_empty_cover_and_lyric():
current = {'record_id': 10, 'platform_unique_key': 'current-mid'}
selected, reason = runner._select_qq_import_record(
{'cover_url': '', 'lyrics_url': ''},
current,
[current],
{'current-mid': {'album_id': None, 'cover': '', 'lyric': ''}},
required_assets=('audio',),
)
assert selected == current
assert reason == 'ok'
def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
pg_conn = _PgConnection()
prepare = MagicMock(return_value={
......@@ -206,6 +263,7 @@ def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
monkeypatch.setattr(runner, 'get_spider_conn', lambda: _Connection())
monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
monkeypatch.setattr(runner, 'fetch_pending_yinyan_song_records', lambda cur, batch_size: [
{'song_id': 10, 'record_id': 200, 'platform': '2'},
] if pg_conn.commits == 0 else [])
......@@ -262,6 +320,7 @@ def test_initialize_yinyan_song_records_inserts_primary_records(monkeypatch):
monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
monkeypatch.setattr(runner, 'fetch_existing_yinyan_song_ids', lambda cur: set())
monkeypatch.setattr(runner, 'iter_hk_songs_batches', lambda conn, batch_size: [[
......@@ -335,6 +394,7 @@ def test_backfill_yinyan_record_platforms_updates_missing_platform_rows(monkeypa
monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
monkeypatch.setattr(runner, 'fetch_yinyan_records_missing_platform', lambda cur, batch_size: [
{'song_id': 10, 'record_id': 100},
{'song_id': 11, 'record_id': 101},
......