feat(etl): 支持重试失败导入并修复相关逻辑

@@ -56,6 +56,8 @@ if YINYAN_IMPORT_TABLE not in _ALLOWED_YINYAN_IMPORT_TABLES:
         f"got {YINYAN_IMPORT_TABLE!r}"
         f"got {YINYAN_IMPORT_TABLE!r}"
     )
     )
 
 
+TARGET_TABLE_NAME = os.environ.get('TARGET_TABLE_NAME', 'hk_songs').strip()
+
 BATCH_SIZE = 10
 BATCH_SIZE = 10
 BACKFILL_BATCH_SIZE = int(os.environ.get('BACKFILL_BATCH_SIZE', '5000'))
 BACKFILL_BATCH_SIZE = int(os.environ.get('BACKFILL_BATCH_SIZE', '5000'))
 HTTP_POOL_MAXSIZE = int(os.environ.get('HTTP_POOL_MAXSIZE', '128'))
 HTTP_POOL_MAXSIZE = int(os.environ.get('HTTP_POOL_MAXSIZE', '128'))

@@ -328,6 +328,4 @@ def refresh_conn(conn, pool_type: str):
     if pool is None:
     if pool is None:
         raise ValueError(f"unknown pool_type: {pool_type}")
         raise ValueError(f"unknown pool_type: {pool_type}")
 
 
-    if pool_type == 'pg':
-        return pool().check(conn)
     return pool().check(conn)
     return pool().check(conn)

-import re
-
-_TIMESTAMP_RE = re.compile(r'\[\d{2}:\d{2}\.\d{2,3}\]')
-_META_TAG_RE = re.compile(r'\[[a-zA-Z]+:[^\]]*\]')
-
-
 def ensure_newlines(text: str | None) -> str:
 def ensure_newlines(text: str | None) -> str:
     """确保歌词每行之间有换行符，保留时间戳等原始内容"""
     """确保歌词每行之间有换行符，保留时间戳等原始内容"""
     if not text:
     if not text:
@@ -13,11 +7,3 @@ def ensure_newlines(text: str | None) -> str:
     lines = [line.strip() for line in text.splitlines() if line.strip()]
     lines = [line.strip() for line in text.splitlines() if line.strip()]
     return '\n'.join(lines)
     return '\n'.join(lines)
 
 
-
-def strip_timestamps(text: str | None) -> str:
-    if not text:
-        return ''
-    text = _META_TAG_RE.sub('', text)
-    text = _TIMESTAMP_RE.sub('', text)
-    lines = [line.strip() for line in text.splitlines() if line.strip()]
-    return '\n'.join(lines)

 from typing import Iterator
 from typing import Iterator
 import pymysql
 import pymysql
-from .config import PLATFORMS
+from .config import PLATFORMS, TARGET_TABLE_NAME
 
 
 
 
-_HK_SONGS_QUERY = """
+_HK_SONGS_QUERY = f"""
 SELECT id, name, lyricist, composer, audio_url, lyrics_url,
 SELECT id, name, lyricist, composer, audio_url, lyrics_url,
        cover_url, singer, issue_time, source_song_id, song_time
        cover_url, singer, issue_time, source_song_id, song_time
-FROM hk_songs_test
+FROM {TARGET_TABLE_NAME}
 WHERE deleted = '0'
 WHERE deleted = '0'
   AND id > %s
   AND id > %s
   AND name IS NOT NULL AND name != ''
   AND name IS NOT NULL AND name != ''
@@ -16,34 +16,18 @@ ORDER BY id
 LIMIT %s
 LIMIT %s
 """
 """
 
 
-_HK_SONGS_BY_SOURCE_IDS_QUERY = """
+_HK_SONGS_BY_SOURCE_IDS_QUERY = f"""
 SELECT id, name, lyricist, composer, audio_url, lyrics_url,
 SELECT id, name, lyricist, composer, audio_url, lyrics_url,
        cover_url, singer, issue_time, source_song_id, song_time
        cover_url, singer, issue_time, source_song_id, song_time
-FROM hk_songs_test
+FROM {TARGET_TABLE_NAME}
 WHERE deleted = '0'
 WHERE deleted = '0'
-  AND source_song_id IN ({placeholders})
+  AND source_song_id IN ({{placeholders}})
   AND name IS NOT NULL AND name != ''
   AND name IS NOT NULL AND name != ''
   AND audio_url IS NOT NULL AND audio_url != ''
   AND audio_url IS NOT NULL AND audio_url != ''
   AND singer IS NOT NULL AND singer != ''
   AND singer IS NOT NULL AND singer != ''
 ORDER BY id
 ORDER BY id
 """
 """
 
 
-# records2 初始化不要求封面或歌词；正式导入时这两个字段允许为空。
-_HK_SONGS_RECORDS2_QUERY = """
-SELECT id, source_song_id
-FROM hk_songs_test
-WHERE deleted = '0'
-  AND id > %s
-  AND source_song_id IS NOT NULL
-  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 audio_url IS NOT NULL AND TRIM(audio_url) != ''
-  AND issue_time IS NOT NULL
-ORDER BY id
-LIMIT %s
-"""
-
 _PLATFORM_QUERY = """
 _PLATFORM_QUERY = """
 SELECT
 SELECT
     sar.song_id AS source_song_id,
     sar.song_id AS source_song_id,
@@ -143,30 +127,17 @@ def iter_hk_songs_batches(
                 break
                 break
 
 
 
 
-def iter_hk_songs_records2_batches(
+def fetch_hk_songs_by_source_ids(
     conn: pymysql.Connection,
     conn: pymysql.Connection,
-    batch_size: int,
-    start_after_id: int = 0,
-) -> Iterator[list[dict]]:
-    """遍历具备 records2 基础入库字段的歌曲，封面和歌词可为空。"""
-    last_id = start_after_id
-    with conn.cursor() as cur:
-        while True:
-            cur.execute(_HK_SONGS_RECORDS2_QUERY, (last_id, batch_size))
-            rows = cur.fetchall()
-            if not rows:
-                break
-            yield rows
-            last_id = int(rows[-1]['id'])
-            if len(rows) < batch_size:
-                break
-
-
-def fetch_hk_songs_by_source_ids(conn: pymysql.Connection, source_song_ids: list[int]) -> dict[int, dict]:
+    source_song_ids: list[int],
+    include_deleted: bool = False,
+) -> dict[int, dict]:
     if not source_song_ids:
     if not source_song_ids:
         return {}
         return {}
     placeholders = ','.join(['%s'] * len(source_song_ids))
     placeholders = ','.join(['%s'] * len(source_song_ids))
     query = _HK_SONGS_BY_SOURCE_IDS_QUERY.format(placeholders=placeholders)
     query = _HK_SONGS_BY_SOURCE_IDS_QUERY.format(placeholders=placeholders)
+    if include_deleted:
+        query = query.replace("WHERE deleted = '0'", 'WHERE 1 = 1', 1)
     with conn.cursor() as cur:
     with conn.cursor() as cur:
         cur.execute(query, source_song_ids)
         cur.execute(query, source_song_ids)
         rows = cur.fetchall()
         rows = cur.fetchall()
@@ -178,20 +149,35 @@ def fetch_hk_songs_by_source_ids(conn: pymysql.Connection, source_song_ids: list
 
 
 
 
 def mark_hk_songs_deleted(conn: pymysql.Connection, source_song_ids: list[int]) -> int:
 def mark_hk_songs_deleted(conn: pymysql.Connection, source_song_ids: list[int]) -> int:
-    """将无法导入的源歌曲在 hk_songs_test 中软删除。"""
+    """将无法导入的源歌曲在目标表中软删除。"""
     if not source_song_ids:
     if not source_song_ids:
         return 0
         return 0
     unique_ids = list(dict.fromkeys(int(song_id) for song_id in source_song_ids))
     unique_ids = list(dict.fromkeys(int(song_id) for song_id in source_song_ids))
     placeholders = ','.join(['%s'] * len(unique_ids))
     placeholders = ','.join(['%s'] * len(unique_ids))
     with conn.cursor() as cur:
     with conn.cursor() as cur:
         cur.execute(
         cur.execute(
-            f"UPDATE hk_songs_test SET deleted = '1' "
+            f"UPDATE {TARGET_TABLE_NAME} SET deleted = '1' "
             f"WHERE source_song_id IN ({placeholders}) AND deleted = '0'",
             f"WHERE source_song_id IN ({placeholders}) AND deleted = '0'",
             unique_ids,
             unique_ids,
         )
         )
         return cur.rowcount
         return cur.rowcount
 
 
 
 
+def mark_hk_songs_active(conn: pymysql.Connection, source_song_ids: list[int]) -> int:
+    """恢复需要重新导入的 HK 源歌曲。"""
+    if not source_song_ids:
+        return 0
+    unique_ids = list(dict.fromkeys(int(song_id) for song_id in source_song_ids))
+    placeholders = ','.join(['%s'] * len(unique_ids))
+    with conn.cursor() as cur:
+        cur.execute(
+            f"UPDATE {TARGET_TABLE_NAME} SET deleted = '0' "
+            f"WHERE source_song_id IN ({placeholders}) AND deleted = '1'",
+            unique_ids,
+        )
+        return cur.rowcount
+
+
 def fetch_platform_records(source_conn: pymysql.Connection, song_ids: list[int]) -> list[dict]:
 def fetch_platform_records(source_conn: pymysql.Connection, song_ids: list[int]) -> list[dict]:
     if not song_ids:
     if not song_ids:
         return []
         return []

@@ -13,6 +13,7 @@ from .reader import (
     iter_hk_songs_batches,
     iter_hk_songs_batches,
     fetch_hk_songs_by_source_ids,
     fetch_hk_songs_by_source_ids,
     mark_hk_songs_deleted,
     mark_hk_songs_deleted,
+    mark_hk_songs_active,
     fetch_platform_records,
     fetch_platform_records,
     fetch_all_platform_records,
     fetch_all_platform_records,
     fetch_all_song_record_relations,
     fetch_all_song_record_relations,
@@ -28,6 +29,7 @@ from .spider import (
 )
 )
 from .writer import (
 from .writer import (
     fetch_pending_yinyan_song_records,
     fetch_pending_yinyan_song_records,
+    reset_failed_yinyan_song_records,
     fetch_yinyan_records_missing_platform,
     fetch_yinyan_records_missing_platform,
     insert_yinyan_song_records,
     insert_yinyan_song_records,
     insert_yinyan_song_records2,
     insert_yinyan_song_records2,
@@ -1384,7 +1386,7 @@ def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None
                 pg_conn.commit()
                 pg_conn.commit()
                 total += len(init_rows)
                 total += len(init_rows)
 
 
-            # 本批次中无法匹配到任何录音的歌曲 → 软删 hk_songs_test
+            # 本批次中无法匹配到任何录音的歌曲 → 软删目标表
             matched_src_ids = {int(hk_row['source_song_id']) for hk_row in batch
             matched_src_ids = {int(hk_row['source_song_id']) for hk_row in batch
                                if hk_row.get('source_song_id') and int(hk_row['source_song_id']) in pr_by_song}
                                if hk_row.get('source_song_id') and int(hk_row['source_song_id']) in pr_by_song}
             unmatched_ids = [
             unmatched_ids = [
@@ -1804,7 +1806,7 @@ def backfill_qq_invalid_covers(max_batches: int | None = None) -> None:
                     removals.append(row)
                     removals.append(row)
                     continue
                     continue
                 if not hk_row:
                 if not hk_row:
-                    log.warning('QQ cover backfill cannot find hk_songs_test row: source_song_id=%s', row['song_id'])
+                    log.warning('QQ cover backfill cannot find target table row: source_song_id=%s', row['song_id'])
                     retry_later += 1
                     retry_later += 1
                     continue
                     continue
 
 
@@ -1863,6 +1865,7 @@ def backfill_qq_invalid_covers(max_batches: int | None = None) -> None:
 def run(
 def run(
     platforms: list[str],
     platforms: list[str],
     max_batches: int | None = None,
     max_batches: int | None = None,
+    retry_failed: bool = False,
 ) -> None:
 ) -> None:
     hk_conn = get_hk_songs_conn()
     hk_conn = get_hk_songs_conn()
     src_conn = get_source_conn()
     src_conn = get_source_conn()
@@ -1874,6 +1877,18 @@ def run(
     imported: list[dict] = []
     imported: list[dict] = []
 
 
     try:
     try:
+        if retry_failed:
+            with pg_conn.cursor() as pg_cur:
+                reset_count, retry_song_ids = reset_failed_yinyan_song_records(pg_cur, platforms)
+            pg_conn.commit()
+            if retry_song_ids:
+                restored = mark_hk_songs_active(hk_conn, retry_song_ids)
+                hk_conn.commit()
+                log.info(
+                    'Reset failed imports for retry: rows=%d songs=%d restored_hk_rows=%d',
+                    reset_count, len(retry_song_ids), restored,
+                )
+
         batch_index = 0
         batch_index = 0
         pbar = tqdm(desc='batches')
         pbar = tqdm(desc='batches')
         while max_batches is None or batch_index < max_batches:
         while max_batches is None or batch_index < max_batches:
@@ -1884,12 +1899,15 @@ def run(
             pg_conn = refresh_conn(pg_conn, 'pg')
             pg_conn = refresh_conn(pg_conn, 'pg')
 
 
             with pg_conn.cursor() as pg_cur:
             with pg_conn.cursor() as pg_cur:
-                pending_records = fetch_pending_yinyan_song_records(pg_cur, BATCH_SIZE)
+                pending_records = fetch_pending_yinyan_song_records(pg_cur, BATCH_SIZE, platforms)
             if not pending_records:
             if not pending_records:
                 break
                 break
 
 
             song_ids = [int(r['song_id']) for r in pending_records]
             song_ids = [int(r['song_id']) for r in pending_records]
             record_ids = [int(r['record_id']) for r in pending_records]
             record_ids = [int(r['record_id']) for r in pending_records]
+            if retry_failed:
+                hk_by_song = fetch_hk_songs_by_source_ids(hk_conn, song_ids, include_deleted=True)
+            else:
                 hk_by_song = fetch_hk_songs_by_source_ids(hk_conn, song_ids)
                 hk_by_song = fetch_hk_songs_by_source_ids(hk_conn, song_ids)
             platform_records = fetch_platform_records_by_record_ids(src_conn, record_ids)
             platform_records = fetch_platform_records_by_record_ids(src_conn, record_ids)
             pr_by_state_key: dict[tuple[int, int, str], dict] = {}
             pr_by_state_key: dict[tuple[int, int, str], dict] = {}

 """工具函数模块"""
 """工具函数模块"""
+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 asyncio
 import hashlib
 import hashlib
 import logging
 import logging

@@ -112,7 +112,18 @@ def fetch_existing_yinyan_song_ids(cur) -> set[int]:
     return {row[0] for row in cur.fetchall()}
     return {row[0] for row in cur.fetchall()}
 
 
 
 
-def fetch_pending_yinyan_song_records(cur, limit: int) -> list[dict]:
+def fetch_pending_yinyan_song_records(
+    cur,
+    limit: int,
+    platforms: list[str] | None = None,
+) -> list[dict]:
+    platform_filter = ''
+    params: list = []
+    if platforms:
+        placeholders = ', '.join(['%s'] * len(platforms))
+        platform_filter = f'AND platform IN ({placeholders})'
+        params.extend(platforms)
+    params.append(limit)
     cur.execute(
     cur.execute(
         f"""
         f"""
         SELECT song_id, record_id, platform
         SELECT song_id, record_id, platform
@@ -120,15 +131,36 @@ def fetch_pending_yinyan_song_records(cur, limit: int) -> list[dict]:
         WHERE is_yinyan_push = FALSE
         WHERE is_yinyan_push = FALSE
           AND platform IS NOT NULL
           AND platform IS NOT NULL
           AND platform_song_id IS NULL
           AND platform_song_id IS NULL
+          {platform_filter}
         ORDER BY song_id
         ORDER BY song_id
         LIMIT %s
         LIMIT %s
         """,
         """,
-        (limit,),
+        tuple(params),
     )
     )
     rows = cur.fetchall()
     rows = cur.fetchall()
     return [{'song_id': row[0], 'record_id': row[1], 'platform': row[2]} for row in rows]
     return [{'song_id': row[0], 'record_id': row[1], 'platform': row[2]} for row in rows]
 
 
 
 
+def reset_failed_yinyan_song_records(cur, platforms: list[str]) -> tuple[int, list[int]]:
+    """将指定平台的失败行恢复为待导入，返回 (行数, HK song_ids)。"""
+    if not platforms:
+        return 0, []
+    placeholders = ', '.join(['%s'] * len(platforms))
+    cur.execute(
+        f"""
+        UPDATE {YINYAN_IMPORT_TABLE}
+        SET platform_song_id = NULL
+        WHERE is_yinyan_push = FALSE
+          AND platform_song_id < 0
+          AND platform IN ({placeholders})
+        RETURNING song_id
+        """,
+        tuple(platforms),
+    )
+    rows = cur.fetchall()
+    return len(rows), list(dict.fromkeys(int(row[0]) for row in rows))
+
+
 def fetch_yinyan_records_missing_platform(cur, limit: int) -> list[dict]:
 def fetch_yinyan_records_missing_platform(cur, limit: int) -> list[dict]:
     cur.execute(
     cur.execute(
         """
         """

@@ -16,6 +16,8 @@ if __name__ == '__main__':
                         help='要导入的平台（默认 all）')
                         help='要导入的平台（默认 all）')
     parser.add_argument('--max-batches', type=int, default=None,
     parser.add_argument('--max-batches', type=int, default=None,
                         help='最多处理多少批次（冒烟测试用）')
                         help='最多处理多少批次（冒烟测试用）')
+    parser.add_argument('--retry-failed', action='store_true',
+                        help='将当前导入状态表中的失败记录恢复为待处理并重新导入')
     parser.add_argument('--init-yinyan-records', action='store_true',
     parser.add_argument('--init-yinyan-records', action='store_true',
                         help='只初始化 yinyan_song_records 待导入状态表，不执行 crawler 导入')
                         help='只初始化 yinyan_song_records 待导入状态表，不执行 crawler 导入')
     parser.add_argument('--init-yinyan-records2', action='store_true',
     parser.add_argument('--init-yinyan-records2', action='store_true',
@@ -45,4 +47,4 @@ if __name__ == '__main__':
     elif args.backfill_qq_invalid_covers:
     elif args.backfill_qq_invalid_covers:
         backfill_qq_invalid_covers(max_batches=args.max_batches)
         backfill_qq_invalid_covers(max_batches=args.max_batches)
     else:
     else:
-        run(platforms, max_batches=args.max_batches)
+        run(platforms, max_batches=args.max_batches, retry_failed=args.retry_failed)

-from etl_to_crawler.lyric import strip_timestamps
-
-def test_strips_standard_timestamps():
-    lrc = "[00:12.34]第一行歌词\n[01:23.45]第二行歌词"
-    assert strip_timestamps(lrc) == "第一行歌词\n第二行歌词"
-
-def test_strips_millisecond_timestamps():
-    lrc = "[00:12.345]带三位毫秒"
-    assert strip_timestamps(lrc) == "带三位毫秒"
-
-def test_strips_meta_tags():
-    lrc = "[ti:歌曲名]\n[ar:歌手]\n[00:01.00]歌词"
-    result = strip_timestamps(lrc)
-    assert "歌词" in result
-    assert "[ti:" not in result
-
-def test_empty_and_none():
-    assert strip_timestamps(None) == ''
-    assert strip_timestamps('') == ''
-
-def test_plain_text_unchanged():
-    assert strip_timestamps("没有时间戳的歌词") == "没有时间戳的歌词"

 from unittest.mock import MagicMock
 from unittest.mock import MagicMock
+import etl_to_crawler.reader as reader
 
 
 from etl_to_crawler.reader import (
 from etl_to_crawler.reader import (
     fetch_all_song_record_relations,
     fetch_all_song_record_relations,
@@ -87,6 +88,32 @@ def test_fetch_platform_records_by_record_ids_queries_exact_records_once():
     assert result[0]['platform'] == '2'
     assert result[0]['platform'] == '2'
 
 
 
 
+def test_retry_fetch_can_include_soft_deleted_hk_songs():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([])
+
+    reader.fetch_hk_songs_by_source_ids(conn, [10], include_deleted=True)
+
+    sql, params = conn.cursor.return_value.execute.call_args.args
+    assert "deleted = '0'" not in sql
+    assert 'WHERE 1 = 1' in sql
+    assert params == [10]
+
+
+def test_mark_hk_songs_active_restores_retry_sources():
+    conn = MagicMock()
+    conn.cursor.return_value = _make_cursor([])
+    conn.cursor.return_value.rowcount = 2
+
+    restored = reader.mark_hk_songs_active(conn, [10, 10, 11])
+
+    sql, params = conn.cursor.return_value.execute.call_args.args
+    assert "SET deleted = '0'" in sql
+    assert "AND deleted = '1'" in sql
+    assert params == [10, 11]
+    assert restored == 2
+
+
 def test_fetch_all_song_record_relations_keeps_all_records_for_a_song():
 def test_fetch_all_song_record_relations_keeps_all_records_for_a_song():
     conn = MagicMock()
     conn = MagicMock()
     conn.cursor.return_value = _make_cursor([
     conn.cursor.return_value = _make_cursor([

@@ -340,7 +340,7 @@ def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
     monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
     monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
     monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
     monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
     monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
     monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
-    monkeypatch.setattr(runner, 'fetch_pending_yinyan_song_records', lambda cur, batch_size: [
+    monkeypatch.setattr(runner, 'fetch_pending_yinyan_song_records', lambda cur, batch_size, platforms: [
         {'song_id': 10, 'record_id': 200, 'platform': '2'},
         {'song_id': 10, 'record_id': 200, 'platform': '2'},
     ] if pg_conn.commits == 0 else [])
     ] if pg_conn.commits == 0 else [])
     monkeypatch.setattr(runner, 'fetch_hk_songs_by_source_ids', lambda conn, song_ids: {10: {
     monkeypatch.setattr(runner, 'fetch_hk_songs_by_source_ids', lambda conn, song_ids: {10: {
@@ -389,6 +389,29 @@ def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
     assert pg_conn.commits == 1
     assert pg_conn.commits == 1
 
 
 
 
+def test_run_retry_failed_resets_failures_and_restores_hk_sources(monkeypatch):
+    pg_conn = _PgConnection()
+    hk_conn = _Connection()
+    reset = MagicMock(return_value=(2, [10, 11]))
+    restore = MagicMock(return_value=2)
+
+    monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: hk_conn)
+    monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
+    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, 'reset_failed_yinyan_song_records', reset)
+    monkeypatch.setattr(runner, 'mark_hk_songs_active', restore)
+    monkeypatch.setattr(runner, 'fetch_pending_yinyan_song_records', lambda cur, batch_size, platforms: [])
+
+    runner.run(['1'], retry_failed=True)
+
+    reset.assert_called_once_with(pg_conn.cur, ['1'])
+    restore.assert_called_once_with(hk_conn, [10, 11])
+    assert pg_conn.commits == 1
+
+
 def test_initialize_yinyan_song_records_inserts_primary_records(monkeypatch):
 def test_initialize_yinyan_song_records_inserts_primary_records(monkeypatch):
     pg_conn = _PgConnection()
     pg_conn = _PgConnection()
     inserted = []
     inserted = []

@@ -208,6 +208,33 @@ def test_fetch_pending_yinyan_song_records_reads_platform_for_single_platform_im
     ]
     ]
 
 
 
 
+def test_fetch_pending_yinyan_song_records_filters_requested_platforms():
+    cur = MagicMock()
+    cur.fetchall.return_value = []
+
+    fetch_pending_yinyan_song_records(cur, 10, ['4'])
+
+    sql, params = cur.execute.call_args.args
+    assert 'AND platform IN (%s)' in sql
+    assert params == ('4', 10)
+
+
+def test_reset_failed_yinyan_song_records_resets_selected_platforms_once():
+    cur = MagicMock()
+    cur.fetchall.return_value = [(10,), (10,), (11,)]
+
+    reset_count, song_ids = writer.reset_failed_yinyan_song_records(cur, ['1', '2'])
+
+    sql, params = cur.execute.call_args.args
+    assert f'UPDATE {writer.YINYAN_IMPORT_TABLE}' in sql
+    assert 'platform_song_id = NULL' in sql
+    assert 'platform_song_id < 0' in sql
+    assert 'RETURNING song_id' in sql
+    assert params == ('1', '2')
+    assert reset_count == 3
+    assert song_ids == [10, 11]
+
+
 def test_import_state_queries_can_target_records2(monkeypatch):
 def test_import_state_queries_can_target_records2(monkeypatch):
     monkeypatch.setattr(writer, 'YINYAN_IMPORT_TABLE', 'yinyan_song_records2')
     monkeypatch.setattr(writer, 'YINYAN_IMPORT_TABLE', 'yinyan_song_records2')
     cur = MagicMock()
     cur = MagicMock()