refactor(etl): 优化音频转存与歌词上传流程，支持音频 MD5 计算和歌词换行处理

- 移除 audit_hk_songs_duplicates.py 中删除零时长歌曲的逻辑
- 新增 utils.py 工具模块，提供音频 MD5 计算、歌词处理、OSS 上传等函数
- etl_to_crawler/oss.py 添加 transfer_url_with_md5 函数，用于下载音频、计算 MD5 并上传 OSS
- etl_to_crawler/lyric.py 增加 ensure_newlines 函数，确保歌词内容行间带换行符
- etl_to_crawler/runner.py 改进音频转存调用，返回音频 MD5 值
- 通过 safe_upload_lyric 函数上传去时间戳的纯文本歌词，失败时使用原始歌词 URL
- etl_to_crawler/writer.py 在歌曲数据写入中新增 audio_md5 字段支持
- 重构 yinyan_song_records 逻辑，新增初始化函数 initialize_yinyan_song_records
- run_etl.py 增加 --init-yinyan-records 命令行参数支持仅初始化状态表
- 优化 ETL 运行流程，从预标记状态表中拉取待处理歌曲逐批处理，提高效率
- 新增相关单元测试以覆盖音频 MD5 计算和任务处理逻辑修改

@@ -558,52 +558,9 @@ def main() -> int:
         help="补处理暂存表中 merge+skipped 记录：existing_into_new 入库+软删旧记录，new_into_existing 补作者合并",
     )
     parser.add_argument("--skip-oss", action="store_true", help="--fix-merge 时跳过歌词 OSS 上传，保留原始文本")
-    parser.add_argument("--dry-run", action="store_true", help="仅打印计划，不写库（配合 --fix-merge / --delete-zero-duration 使用）")
-    parser.add_argument(
-        "--delete-zero-duration",
-        action="store_true",
-        help="软删除 song_time=0 的记录（置 deleted='1'）",
-    )
+    parser.add_argument("--dry-run", action="store_true", help="仅打印计划，不写库（配合 --fix-merge 使用）")
     args = parser.parse_args()
 
-    # --delete-zero-duration 模式
-    if args.delete_zero_duration:
-        conn = pymysql.connect(**TARGET_DB_CONFIG)
-        try:
-            table = quote_identifier(DEFAULT_TARGET_TABLE)
-            with conn.cursor() as cursor:
-                cursor.execute(
-                    f"SELECT id, name, lyricist, composer, song_time "
-                    f"FROM {table} WHERE deleted = '0' AND song_time = 0"
-                )
-                rows = list(cursor.fetchall())
-            print(f"song_time=0 记录数: {len(rows)}")
-            if args.dry_run:
-                for row in rows[:20]:
-                    print(row)
-                if len(rows) > 20:
-                    print(f"... {len(rows) - 20} more rows")
-                return 0
-            if not rows:
-                print("无需处理")
-                return 0
-            ids = [row["id"] for row in rows]
-            placeholders = ",".join(["%s"] * len(ids))
-            with conn.cursor() as cursor:
-                cursor.execute(
-                    f"UPDATE {table} "
-                    f"SET deleted = '1', modify_time = NOW(), "
-                    f"    off_shelf_remark = 'song_time=0 soft-deleted by audit script' "
-                    f"WHERE deleted = '0' AND id IN ({placeholders})",
-                    ids,
-                )
-                affected = cursor.rowcount
-            conn.commit()
-            print(f"soft_deleted={affected}")
-        finally:
-            conn.close()
-        return 0
-
     # --fix-merge 模式：连双库，执行合并方向后处理
     if args.fix_merge:
         target_conn = pymysql.connect(**TARGET_DB_CONFIG)

@@ -4,6 +4,16 @@ _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:
+    """确保歌词每行之间有换行符，保留时间戳等原始内容"""
+    if not text:
+        return ''
+    # 将字面量 \\n / \\r 转为真正换行符
+    text = text.replace('\\n', '\n').replace('\\r', '')
+    lines = [line.strip() for line in text.splitlines() if line.strip()]
+    return '\n'.join(lines)
+
+
 def strip_timestamps(text: str | None) -> str:
     if not text:
         return ''

 import requests
 import oss2
 from urllib.parse import urlparse
+from .utils import compute_audio_md5
 
 ARCHIVE_DEV_HOST = 'archive-dev.oss-cn-beijing.aliyuncs.com'
 
@@ -21,6 +22,22 @@ def transfer_url(url: str | None, oss_key: str, bucket: oss2.Bucket, base_url: s
     return f"{base_url.rstrip('/')}/{oss_key}"
 
 
+def transfer_url_with_md5(url: str | None, oss_key: str, bucket: oss2.Bucket, base_url: str) -> tuple[str, str]:
+    """
+    将音频 URL 转存到 OSS，并基于下载到的音频字节计算 MD5。
+    已在目标 OSS 的 URL 无需重新下载，无法可靠计算 MD5，返回空 MD5。
+    """
+    if not url:
+        return '', ''
+    if ARCHIVE_DEV_HOST in url:
+        return url, ''
+    resp = requests.get(url, timeout=30)
+    resp.raise_for_status()
+    audio_md5 = compute_audio_md5(resp.content)
+    bucket.put_object(oss_key, resp.content)
+    return f"{base_url.rstrip('/')}/{oss_key}", audio_md5
+
+
 def build_oss_key(platform: str, category: str, filename: str) -> str:
     """
     构造 archive-dev 内的存储路径。

@@ -16,6 +16,18 @@ ORDER BY id
 LIMIT %s
 """
 
+_HK_SONGS_BY_SOURCE_IDS_QUERY = """
+SELECT id, name, lyricist, composer, audio_url, lyrics_url,
+       cover_url, singer, issue_time, source_song_id, song_time
+FROM hk_songs_test
+WHERE deleted = '0'
+  AND source_song_id IN ({placeholders})
+  AND name IS NOT NULL AND name != ''
+  AND audio_url IS NOT NULL AND audio_url != ''
+  AND singer IS NOT NULL AND singer != ''
+ORDER BY id
+"""
+
 _PLATFORM_QUERY = """
 SELECT
     sar.song_id AS source_song_id,
@@ -61,6 +73,21 @@ def iter_hk_songs_batches(
                 break
 
 
+def fetch_hk_songs_by_source_ids(conn: pymysql.Connection, source_song_ids: list[int]) -> dict[int, dict]:
+    if not source_song_ids:
+        return {}
+    placeholders = ','.join(['%s'] * len(source_song_ids))
+    query = _HK_SONGS_BY_SOURCE_IDS_QUERY.format(placeholders=placeholders)
+    with conn.cursor() as cur:
+        cur.execute(query, source_song_ids)
+        rows = cur.fetchall()
+    return {
+        int(row['source_song_id']): row
+        for row in rows
+        if row.get('source_song_id') is not None
+    }
+
+
 def fetch_platform_records(source_conn: pymysql.Connection, song_ids: list[int]) -> list[dict]:
     if not song_ids:
         return []

+"""工具函数模块"""
+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 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/{platform}/lyric/{unique_id}.txt"
+    bucket.put_object(oss_key, plain_lyric.encode("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()

 import uuid
 
 
-def upsert_yinyan_song_records(cur, pairs: list[tuple[int, int]]) -> None:
-    """pairs: [(source_song_id, hk_music_record_id), ...]"""
-    if not pairs:
+def insert_yinyan_song_records(cur, records: list[dict]) -> None:
+    """Initialize yinyan_song_records rows before crawler import."""
+    if not records:
         return
     cur.executemany(
-        "DELETE FROM yinyan_song_records WHERE song_id = %s",
-        [(song_id,) for song_id, _ in pairs],
+        """
+        INSERT INTO yinyan_song_records (song_id, record_id, is_yinyan_push)
+        VALUES (%s, %s, FALSE)
+        ON CONFLICT (song_id, record_id) DO NOTHING
+        """,
+        [(r['song_id'], r['record_id']) for r in records],
     )
+
+
+def fetch_pending_yinyan_song_records(cur, limit: int) -> list[dict]:
+    cur.execute(
+        """
+        SELECT song_id, record_id
+        FROM yinyan_song_records
+        WHERE is_yinyan_push = FALSE
+        ORDER BY song_id
+        LIMIT %s
+        """,
+        (limit,),
+    )
+    rows = cur.fetchall()
+    return [{'song_id': row[0], 'record_id': row[1]} for row in rows]
+
+
+def upsert_yinyan_song_records(cur, records: list[dict]) -> None:
+    """Mark pre-initialized yinyan song-record rows as pushed to crawler."""
+    if not records:
+        return
     cur.executemany(
         """
-        INSERT INTO yinyan_song_records (song_id, record_id)
-        VALUES (%s, %s)
+        UPDATE yinyan_song_records
+        SET platform = %s,
+            platform_song_id = %s,
+            is_yinyan_push = TRUE
+        WHERE song_id = %s AND record_id = %s
         """,
-        pairs,
+        [
+            (r['platform'], r['platform_song_id'], r['song_id'], r['record_id'])
+            for r in records
+        ],
     )
 
 
@@ -67,8 +98,8 @@ def upsert_qq_songs(cur, songs: list[dict]) -> None:
             (id, platform_song_id, mid, album_id, cover, title, name, duration,
              lyric, composer_name, lyricist_name, url, lyric_url,
              platform_index_url, published_at, singers, status, created_at, updated_at,
-             provider_name, crawler_source_data)
-        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW(), %s, %s::json)
+             audio_md5, provider_name, crawler_source_data)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s::json)
         ON CONFLICT (platform_song_id) DO NOTHING
     """
     rows = [(
@@ -77,6 +108,7 @@ def upsert_qq_songs(cur, songs: list[dict]) -> None:
         s.get('lyric'), s.get('composer_name'), s.get('lyricist_name'),
         s.get('url', ''), s.get('lyric_url'),
         s.get('platform_index_url'), s.get('published_at'), s.get('singers_json', '[]'),
+        s.get('audio_md5'),
         s.get('provider_name'), s.get('crawler_source_data'),
     ) for s in songs]
     cur.executemany(sql, rows)
@@ -157,8 +189,8 @@ def upsert_kugou_songs(cur, songs: list[dict]) -> None:
             (id, platform_song_id, hash, album_audio_id, album_id, cover, title, name, duration,
              lyric, composer_name, lyricist_name, url, lyric_url,
              platform_index_url, published_at, singers, status, created_at, updated_at,
-             provider_name, crawler_source_data)
-        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW(), %s, %s::json)
+             audio_md5, provider_name, crawler_source_data)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s::json)
         ON CONFLICT (platform_song_id) DO NOTHING
     """
     rows = [(
@@ -168,6 +200,7 @@ def upsert_kugou_songs(cur, songs: list[dict]) -> None:
         s.get('lyric'), s.get('composer_name'), s.get('lyricist_name'),
         s.get('url', ''), s.get('lyric_url'),
         s.get('platform_index_url'), s.get('published_at'), s.get('singers_json', '[]'),
+        s.get('audio_md5'),
         s.get('provider_name'), s.get('crawler_source_data'),
     ) for s in songs]
     cur.executemany(sql, rows)
@@ -244,8 +277,8 @@ def upsert_netease_songs(cur, songs: list[dict]) -> None:
             (id, platform_song_id, album_id, cover, title, name, duration,
              lyric, composer_name, lyricist_name, url, lyric_url,
              platform_index_url, published_at, album, singers, status, created_at, updated_at,
-             provider_name, crawler_source_data)
-        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::json, %s::jsonb, 0, NOW(), NOW(), %s, %s::json)
+             audio_md5, provider_name, crawler_source_data)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::json, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s::json)
         ON CONFLICT (platform_song_id) DO NOTHING
     """
     rows = [(
@@ -254,6 +287,7 @@ def upsert_netease_songs(cur, songs: list[dict]) -> None:
         s.get('lyric'), s.get('composer_name'), s.get('lyricist_name'),
         s.get('url', ''), s.get('lyric_url'),
         s.get('platform_index_url'), s.get('published_at'), s.get('album_json'), s.get('singers_json', '[]'),
+        s.get('audio_md5'),
         s.get('provider_name'), s.get('crawler_source_data'),
     ) for s in songs]
     cur.executemany(sql, rows)

 #!/usr/bin/env python3
 import argparse
 from etl_to_crawler.config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, PLATFORMS
-from etl_to_crawler.runner import run
+from etl_to_crawler.runner import initialize_yinyan_song_records, run
 
 PLATFORM_MAP = {
     'qq': PLATFORM_QQ,
@@ -16,14 +16,8 @@ if __name__ == '__main__':
                         help='要导入的平台（默认 all）')
     parser.add_argument('--max-batches', type=int, default=None,
                         help='最多处理多少批次（冒烟测试用）')
-    parser.add_argument('--resume', action='store_true',
-                        help='启用断点续传，按 hk_songs_test.id 从上次成功提交的批次后继续')
-    parser.add_argument('--state-file', default='output/etl_to_crawler_state.json',
-                        help='断点状态文件路径')
-    parser.add_argument('--start-after-id', type=int, default=None,
-                        help='手动指定从哪个 hk_songs_test.id 之后开始读取')
-    parser.add_argument('--reset-state', action='store_true',
-                        help='忽略已有断点状态，从头或 --start-after-id 指定位置重新开始')
+    parser.add_argument('--init-yinyan-records', action='store_true',
+                        help='只初始化 yinyan_song_records 待导入状态表，不执行 crawler 导入')
     args = parser.parse_args()
 
     if args.platform == 'all':
@@ -32,12 +26,7 @@ if __name__ == '__main__':
         platforms = [PLATFORM_MAP[args.platform]]
 
     print(f"Starting ETL for platforms: {platforms}")
-    run(
-        platforms,
-        max_batches=args.max_batches,
-        resume=args.resume,
-        state_file=args.state_file,
-        state_key=args.platform,
-        start_after_id=args.start_after_id,
-        reset_state=args.reset_state,
-    )
+    if args.init_yinyan_records:
+        initialize_yinyan_song_records(platforms, max_batches=args.max_batches)
+    else:
+        run(platforms, max_batches=args.max_batches)

 from unittest.mock import MagicMock, patch
-from etl_to_crawler.oss import transfer_url
+from etl_to_crawler.oss import transfer_url, transfer_url_with_md5
 
 ARCHIVE_URL = "https://archive-dev.oss-cn-beijing.aliyuncs.com/some/path.mp3"
 OTHER_URL = "https://hikoon-data-platform.oss-cn-beijing.aliyuncs.com/qq-audio/abc.mp3"
@@ -31,3 +31,15 @@ def test_external_url_downloads_and_uploads():
         result = transfer_url(OTHER_URL, "crawler/qq/audio/abc.mp3", bucket, BASE_URL)
     bucket.put_object.assert_called_once_with("crawler/qq/audio/abc.mp3", fake_content)
     assert result == f"{BASE_URL}/crawler/qq/audio/abc.mp3"
+
+
+def test_transfer_url_with_md5_hashes_downloaded_content_before_upload():
+    bucket = MagicMock()
+    fake_content = b"audio_bytes"
+    with patch('etl_to_crawler.oss.requests.get') as mock_get:
+        mock_get.return_value.content = fake_content
+        mock_get.return_value.raise_for_status = MagicMock()
+        result, audio_md5 = transfer_url_with_md5(OTHER_URL, "crawler/qq/audio/abc.mp3", bucket, BASE_URL)
+    bucket.put_object.assert_called_once_with("crawler/qq/audio/abc.mp3", fake_content)
+    assert result == f"{BASE_URL}/crawler/qq/audio/abc.mp3"
+    assert audio_md5 == "04d43544b267629d9089eaed3b847a99"

@@ -48,12 +48,15 @@ def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation
     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, 'iter_hk_songs_batches', lambda conn, batch_size, start_after_id=0: [[{
+    monkeypatch.setattr(runner, 'fetch_pending_yinyan_song_records', lambda cur, batch_size: [
+        {'song_id': 10, 'record_id': 200},
+    ] if pg_conn.commits == 0 else [])
+    monkeypatch.setattr(runner, 'fetch_hk_songs_by_source_ids', lambda conn, song_ids: {10: {
         'source_song_id': 10,
         'name': '歌',
         'audio_url': 'https://example.com/a.mp3',
         'singer': '歌手',
-    }]])
+    }})
     monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: [
         {
             'source_song_id': 10,
@@ -85,37 +88,56 @@ def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation
 
     processors['1'].assert_called_once()
     processors['2'].assert_called_once()
-    yinyan_writer.assert_called_once_with(pg_conn.cur, [(10, 200)])
+    yinyan_writer.assert_called_once_with(pg_conn.cur, [{
+        'song_id': 10,
+        'record_id': 200,
+        'platform': 'kugou',
+        'platform_song_id': 200,
+    }])
     assert pg_conn.commits == 1
 
 
-def test_run_resume_starts_after_saved_id_and_updates_state_after_commit(monkeypatch, tmp_path):
+def test_initialize_yinyan_song_records_inserts_primary_records(monkeypatch):
     pg_conn = _PgConnection()
-    state_file = tmp_path / 'etl_state.json'
-    state_file.write_text('{"all": {"last_hk_songs_id": 40}}', encoding='utf-8')
-    seen_start_ids = []
+    inserted = []
 
     monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
     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())
 
-    def fake_batches(conn, batch_size, start_after_id=0):
-        seen_start_ids.append(start_after_id)
-        return iter([[
+    monkeypatch.setattr(runner, 'iter_hk_songs_batches', lambda conn, batch_size: [[
         {'id': 50, 'source_song_id': 10, 'name': '歌', 'audio_url': 'https://example.com/a.mp3', 'singer': '歌手'},
-            {'id': 60, 'source_song_id': 20, 'name': '歌2', 'audio_url': 'https://example.com/b.mp3', 'singer': '歌手2'},
     ]])
+    monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: [
+        {
+            'source_song_id': 10,
+            'record_id': 100,
+            'platform': '1',
+            'platform_unique_key': 'qq-mid',
+            'platform_mid': '100',
+            'album_audio_id': None,
+            'is_main_version': 0,
+            'is_high': 0,
+            'pub_time': '2020-01-01',
+        },
+        {
+            'source_song_id': 10,
+            'record_id': 200,
+            'platform': '2',
+            'platform_unique_key': '200',
+            'platform_mid': 'kg-hash',
+            'album_audio_id': None,
+            'is_main_version': 1,
+            'is_high': 0,
+            'pub_time': '2021-01-01',
+        },
+    ])
+    monkeypatch.setattr(runner, 'insert_yinyan_song_records', lambda cur, rows: inserted.extend(rows))
 
-    monkeypatch.setattr(runner, 'iter_hk_songs_batches', fake_batches)
-    monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: [])
-
-    runner.run(['1', '2', '4'], resume=True, state_file=state_file, state_key='all')
+    runner.initialize_yinyan_song_records(['1', '2'])
 
-    assert seen_start_ids == [40]
+    assert inserted == [{'song_id': 10, 'record_id': 200}]
     assert pg_conn.commits == 1
-    assert '"last_hk_songs_id": 60' in state_file.read_text(encoding='utf-8')
 
 
 def test_process_netease_builds_album_json_for_song_insert(monkeypatch):
@@ -147,6 +169,7 @@ def test_process_netease_builds_album_json_for_song_insert(monkeypatch):
     })
     monkeypatch.setattr(runner, 'fetch_netease_singers', lambda conn, song_ids: {})
     monkeypatch.setattr(runner, '_safe_transfer', lambda url, oss_key, bucket, base_url: url)
+    monkeypatch.setattr(runner, '_safe_transfer_audio', lambda url, oss_key, bucket, base_url: (url, 'audio-md5'))
     monkeypatch.setattr(runner, 'upsert_netease_singers', lambda cur, singers: None)
     monkeypatch.setattr(runner, 'upsert_netease_albums', lambda cur, albums: None)
     monkeypatch.setattr(runner, 'upsert_netease_songs', lambda cur, songs: inserted_songs.extend(songs))
@@ -174,3 +197,60 @@ def test_process_netease_builds_album_json_for_song_insert(monkeypatch):
     assert inserted_songs[0]['album_json']
     assert '"id": 20' in inserted_songs[0]['album_json']
     assert '"title": "专辑"' in inserted_songs[0]['album_json']
+
+
+def test_process_netease_keeps_timestamped_lyric_and_uploads_plain_lyric(monkeypatch):
+    pg_cur = MagicMock()
+    pg_cur.fetchone.return_value = ('song-uuid',)
+    inserted_songs = []
+    uploaded = {}
+
+    class Bucket:
+        def put_object(self, key, body):
+            uploaded['key'] = key
+            uploaded['body'] = body
+
+    monkeypatch.setattr(runner, 'fetch_netease_songs', lambda conn, song_ids: {
+        300: {
+            'id': 300,
+            'album_id': None,
+            'cover': 'https://example.com/cover.jpg',
+            'title': '录音标题',
+            'duration': 180,
+            'lyric': '[ti:歌名]\n[00:01.00]第一句\n[00:02.00]第二句',
+            'composer_name': '曲作者',
+            'lyricist_name': '词作者',
+            'platform_index_url': None,
+            'published_at': '2020-01-02',
+        },
+    })
+    monkeypatch.setattr(runner, 'fetch_netease_singers', lambda conn, song_ids: {})
+    monkeypatch.setattr(runner, '_safe_transfer', lambda url, oss_key, bucket, base_url: url)
+    monkeypatch.setattr(runner, '_safe_transfer_audio', lambda url, oss_key, bucket, base_url: (url, 'audio-md5'))
+    monkeypatch.setattr(runner, 'upsert_netease_singers', lambda cur, singers: None)
+    monkeypatch.setattr(runner, 'upsert_netease_albums', lambda cur, albums: None)
+    monkeypatch.setattr(runner, 'upsert_netease_songs', lambda cur, songs: inserted_songs.extend(songs))
+    monkeypatch.setattr(runner, 'upsert_netease_singer_songs', lambda cur, pairs: None)
+
+    runner._process_netease(
+        {
+            'name': '词曲名',
+            'audio_url': 'https://example.com/audio.mp3',
+            'lyrics_url': 'https://example.com/original.lrc',
+            'cover_url': '',
+            'composer': '词曲曲作者',
+            'lyricist': '词曲词作者',
+            'issue_time': '2019-01-01',
+            'song_time': 120,
+        },
+        {'platform_unique_key': '300'},
+        spider_conn=object(),
+        pg_cur=pg_cur,
+        bucket=Bucket(),
+        base_url='https://bucket.example.com',
+    )
+
+    assert inserted_songs[0]['lyric'] == '[ti:歌名]\n[00:01.00]第一句\n[00:02.00]第二句'
+    assert inserted_songs[0]['audio_md5'] == 'audio-md5'
+    assert inserted_songs[0]['lyric_url'] == 'https://bucket.example.com/crawler/netease/lyric/300.txt'
+    assert uploaded['body'].decode('utf-8') == '第一句\n第二句'

 from unittest.mock import MagicMock, call
 from etl_to_crawler.writer import (
+    insert_yinyan_song_records,
     upsert_kugou_albums,
     upsert_kugou_singers,
     upsert_kugou_songs,
@@ -41,18 +42,35 @@ def test_upsert_qq_singer_songs():
     assert 'crawler_qqmusic_singer_songs' in sql
 
 
-def test_upsert_yinyan_song_records_replaces_existing_song_relation():
+def test_upsert_yinyan_song_records_marks_existing_relation_as_pushed():
     cur = MagicMock()
-    upsert_yinyan_song_records(cur, [(10, 100), (11, 101)])
+    upsert_yinyan_song_records(cur, [
+        {'song_id': 10, 'record_id': 100, 'platform': 'qq', 'platform_song_id': 1000},
+        {'song_id': 11, 'record_id': 101, 'platform': 'kugou', 'platform_song_id': 1001},
+    ])
 
-    assert cur.executemany.call_count == 2
-    delete_sql, delete_rows = cur.executemany.call_args_list[0][0]
-    insert_sql, insert_rows = cur.executemany.call_args_list[1][0]
-    assert 'DELETE FROM yinyan_song_records' in delete_sql
-    assert 'WHERE song_id = %s' in delete_sql
-    assert delete_rows == [(10,), (11,)]
-    assert 'INSERT INTO yinyan_song_records' in insert_sql
-    assert insert_rows == [(10, 100), (11, 101)]
+    sql, rows = cur.executemany.call_args[0]
+    assert 'UPDATE yinyan_song_records' in sql
+    assert 'platform = %s' in sql
+    assert 'platform_song_id = %s' in sql
+    assert 'is_yinyan_push = TRUE' in sql
+    assert 'WHERE song_id = %s AND record_id = %s' in sql
+    assert rows == [('qq', 1000, 10, 100), ('kugou', 1001, 11, 101)]
+
+
+def test_insert_yinyan_song_records_initializes_unpushed_rows():
+    cur = MagicMock()
+    insert_yinyan_song_records(cur, [
+        {'song_id': 10, 'record_id': 100},
+        {'song_id': 11, 'record_id': 101},
+    ])
+
+    sql, rows = cur.executemany.call_args[0]
+    assert 'INSERT INTO yinyan_song_records' in sql
+    assert 'is_yinyan_push' in sql
+    assert 'FALSE' in sql
+    assert 'ON CONFLICT (song_id, record_id) DO NOTHING' in sql
+    assert rows == [(10, 100), (11, 101)]
 
 
 def test_upsert_netease_songs_writes_album_json_column():
@@ -104,12 +122,14 @@ def test_upsert_kugou_songs_writes_provider_and_source_data():
         'singers_json': '[]',
         'provider_name': 'yinyan',
         'crawler_source_data': '{"id": 200}',
+        'audio_md5': 'md5-200',
     }])
 
     sql, rows = cur.executemany.call_args[0]
+    assert 'audio_md5' in sql
     assert 'provider_name, crawler_source_data' in sql
     assert '%s::json' in sql
-    assert rows[0][-2:] == ('yinyan', '{"id": 200}')
+    assert rows[0][-3:] == ('md5-200', 'yinyan', '{"id": 200}')
 
 
 def test_upsert_kugou_singers_writes_provider_and_source_data():
@@ -211,10 +231,12 @@ def test_upsert_qq_entities_write_provider_and_source_data():
         'singers_json': '[]',
         'provider_name': 'yinyan',
         'crawler_source_data': '{"id": 200}',
+        'audio_md5': 'md5-200',
     }])
     sql, rows = cur.executemany.call_args[0]
+    assert 'audio_md5' in sql
     assert 'provider_name, crawler_source_data' in sql
-    assert rows[0][-2:] == ('yinyan', '{"id": 200}')
+    assert rows[0][-3:] == ('md5-200', 'yinyan', '{"id": 200}')
 
 
 def test_upsert_netease_entities_write_provider_and_source_data():
@@ -273,7 +295,9 @@ def test_upsert_netease_entities_write_provider_and_source_data():
         'singers_json': '[]',
         'provider_name': 'yinyan',
         'crawler_source_data': '{"id": 300}',
+        'audio_md5': 'md5-300',
     }])
     sql, rows = cur.executemany.call_args[0]
+    assert 'audio_md5' in sql
     assert 'provider_name, crawler_source_data' in sql
-    assert rows[0][-2:] == ('yinyan', '{"id": 300}')
+    assert rows[0][-3:] == ('md5-300', 'yinyan', '{"id": 300}')