refactor(etl): 优化录音处理及OSS下载URL重写逻辑

- 配置新增公共访问及下载重写基础URL变量，支持下载地址重写
- oss模块新增内部下载地址重写函数，改进下载时的URL处理逻辑
- transfer_url函数判断是否为目标OSS桶URL，避免重复上传
- runner模块改为仅处理待推送yinyan_song_records对应的单条录音
- 精简runner中对录音记录的查询与处理逻辑，提升性能及正确性
- writer模块fetch_pending_yinyan_song_records增加platform字段过滤
- 新增单元测试覆盖OSS下载地址重写和runner的录音导入逻辑调整

@@ -35,7 +35,12 @@ OSS_CONFIG = {
     'access_key_secret': os.environ['OSS_ACCESS_KEY_SECRET'],
     'endpoint': os.environ['OSS_ENDPOINT'],
     'bucket_name': os.environ['OSS_BUCKET_NAME'],
-    'base_url': f"https://{os.environ['OSS_BUCKET_NAME']}.{os.environ['OSS_ENDPOINT']}",
+    'base_url': os.environ.get(
+        'OSS_PUBLIC_BASE_URL',
+        f"https://{os.environ['OSS_BUCKET_NAME']}.{os.environ['OSS_ENDPOINT']}",
+    ).rstrip('/'),
+    'download_base_url': os.environ.get('OSS_DOWNLOAD_BASE_URL', '').rstrip('/'),
+    'download_rewrite_from_base_url': os.environ.get('OSS_DOWNLOAD_REWRITE_FROM_BASE_URL', '').rstrip('/'),
 }
 
 PLATFORM_QQ = '1'

 import requests
 import oss2
 from urllib.parse import urlparse
+from .config import OSS_CONFIG
 from .utils import compute_audio_md5
 
-ARCHIVE_DEV_HOST = 'archive-dev.oss-cn-beijing.aliyuncs.com'
+
+def _host(url: str | None) -> str:
+    return urlparse(url or '').netloc.lower()
+
+
+def _is_target_oss_url(url: str | None, base_url: str) -> bool:
+    return bool(url) and _host(url) == _host(base_url)
+
+
+def _download_url(url: str, base_url: str) -> str:
+    """Optionally rewrite target-bucket downloads to an internal/CNAME base URL."""
+    download_base_url = OSS_CONFIG.get('download_base_url')
+    rewrite_from = OSS_CONFIG.get('download_rewrite_from_base_url') or base_url
+    if not download_base_url or not url.startswith(rewrite_from.rstrip('/') + '/'):
+        return url
+    return f"{download_base_url.rstrip('/')}{url[len(rewrite_from.rstrip('/')):]}"
 
 
 def transfer_url(url: str | None, oss_key: str, bucket: oss2.Bucket, base_url: str) -> str:
     """
     将 url 指向的文件转移到 archive-dev OSS 的 oss_key 路径。
-    若 url 为空或已在 archive-dev，直接返回原 url（不上传）。
-    返回新的 archive-dev URL。
+    若 url 为空或已在目标 bucket，直接返回原 url（不上传）。
+    返回新的公开访问 URL。
     """
     if not url:
         return ''
-    if ARCHIVE_DEV_HOST in url:
+    if _is_target_oss_url(url, base_url):
         return url
-    resp = requests.get(url, timeout=30)
+    resp = requests.get(_download_url(url, base_url), timeout=30)
     resp.raise_for_status()
     bucket.put_object(oss_key, resp.content)
     return f"{base_url.rstrip('/')}/{oss_key}"
@@ -29,9 +45,9 @@ def transfer_url_with_md5(url: str | None, oss_key: str, bucket: oss2.Bucket, ba
     """
     if not url:
         return '', ''
-    if ARCHIVE_DEV_HOST in url:
+    if _is_target_oss_url(url, base_url):
         return url, ''
-    resp = requests.get(url, timeout=30)
+    resp = requests.get(_download_url(url, base_url), timeout=30)
     resp.raise_for_status()
     audio_md5 = compute_audio_md5(resp.content)
     bucket.put_object(oss_key, resp.content)

@@ -609,58 +609,53 @@ def run(
 
             song_ids = [int(r['song_id']) for r in pending_records]
             hk_by_song = fetch_hk_songs_by_source_ids(hk_conn, song_ids)
-            # 全部录音（不去重），供 singer fallback 遍历
             all_platform_records = fetch_all_platform_records(src_conn, song_ids)
-
-            # 按 (song_id, platform) 分组，每组已按优先级排好序
-            pr_by_song_platform: dict[tuple, list] = {}
+            pr_by_state_key: dict[tuple[int, int, str], dict] = {}
             for pr in all_platform_records:
                 if pr['platform'] in platforms:
-                    key = (int(pr['source_song_id']), pr['platform'])
-                    pr_by_song_platform.setdefault(key, []).append(pr)
-
-            # 每首歌：从各平台各选一条最优录音（有歌手优先）
-            pr_by_song: dict[int, list] = {}
-            for (src_id, _platform), records in pr_by_song_platform.items():
-                chosen = _pick_record_with_singer(records, spider_conn)
-                if chosen:
-                    pr_by_song.setdefault(src_id, []).append(chosen)
+                    key = (int(pr['source_song_id']), int(pr['record_id']), pr['platform'])
+                    pr_by_state_key[key] = pr
 
             with pg_conn.cursor() as pg_cur:
                 pushed_count = 0
-                for src_id in song_ids:
+                for pending in pending_records:
+                    src_id = int(pending['song_id'])
+                    platform = str(pending['platform'])
                     hk_row = hk_by_song.get(src_id)
                     if not hk_row:
                         continue
-                    if not src_id or src_id not in pr_by_song:
+                    if platform not in platforms:
+                        continue
+                    pr = pr_by_state_key.get((src_id, int(pending['record_id']), platform))
+                    if not pr:
+                        log.warning(
+                            "Pending yinyan record not found in source records: song_id=%s record_id=%s platform=%s",
+                            src_id, pending['record_id'], platform,
+                        )
+                        continue
+                    processor = _PROCESSORS.get(platform)
+                    if not processor:
                         continue
-                    wrote_yinyan_record = False
-                    for pr in pr_by_song[src_id]:
-                        processor = _PROCESSORS.get(pr['platform'])
-                        if not processor:
-                            continue
-                        pg_cur.execute('SAVEPOINT sp_song')
-                        try:
-                            result = processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
-                            if result and not wrote_yinyan_record:
-                                upsert_yinyan_song_records(pg_cur, [{
-                                    'song_id': src_id,
-                                    'record_id': int(pr['record_id']),
-                                    'platform': pr['platform'],
-                                    'platform_song_id': int(result['platform_song_id']),
-                                }])
-                                wrote_yinyan_record = True
-                                pushed_count += 1
-                            pg_cur.execute('RELEASE SAVEPOINT sp_song')
-                            total_ok += 1
-                            if result:
-                                imported.append(result)
-                        except Exception as e:
-                            pg_cur.execute('ROLLBACK TO SAVEPOINT sp_song')
-                            pg_cur.execute('RELEASE SAVEPOINT sp_song')
-                            log.error("Error processing song %s platform %s: %s",
-                                      hk_row.get('name'), pr['platform'], e)
-                            total_err += 1
+                    pg_cur.execute('SAVEPOINT sp_song')
+                    try:
+                        result = processor(hk_row, pr, spider_conn, pg_cur, bucket, base_url)
+                        if result:
+                            upsert_yinyan_song_records(pg_cur, [{
+                                'song_id': src_id,
+                                'record_id': int(pr['record_id']),
+                                'platform': platform,
+                                'platform_song_id': int(result['platform_song_id']),
+                            }])
+                            pushed_count += 1
+                            imported.append(result)
+                        pg_cur.execute('RELEASE SAVEPOINT sp_song')
+                        total_ok += 1
+                    except Exception as e:
+                        pg_cur.execute('ROLLBACK TO SAVEPOINT sp_song')
+                        pg_cur.execute('RELEASE SAVEPOINT sp_song')
+                        log.error("Error processing song %s platform %s: %s",
+                                  hk_row.get('name'), platform, e)
+                        total_err += 1
                 pg_conn.commit()
                 if pushed_count == 0:
                     log.error("No yinyan_song_records rows were marked pushed in this batch; stopping to avoid retry loop")

@@ -26,16 +26,17 @@ def fetch_existing_yinyan_song_ids(cur) -> set[int]:
 def fetch_pending_yinyan_song_records(cur, limit: int) -> list[dict]:
     cur.execute(
         """
-        SELECT song_id, record_id
+        SELECT song_id, record_id, platform
         FROM yinyan_song_records
         WHERE is_yinyan_push = FALSE
+          AND platform IS NOT NULL
         ORDER BY song_id
         LIMIT %s
         """,
         (limit,),
     )
     rows = cur.fetchall()
-    return [{'song_id': row[0], 'record_id': row[1]} for row in rows]
+    return [{'song_id': row[0], 'record_id': row[1], 'platform': row[2]} for row in rows]
 
 
 def fetch_yinyan_records_missing_platform(cur, limit: int) -> list[dict]:

@@ -33,6 +33,25 @@ def test_external_url_downloads_and_uploads():
     assert result == f"{BASE_URL}/crawler/qq/audio/abc.mp3"
 
 
+def test_external_url_can_rewrite_download_base_to_internal_endpoint():
+    bucket = MagicMock()
+    public_source = "https://source-bucket.oss-cn-hangzhou.aliyuncs.com/audio/abc.mp3"
+    internal_source_base = "https://source-bucket.oss-cn-hangzhou-internal.aliyuncs.com"
+    fake_content = b"audio_bytes"
+    with patch.dict('etl_to_crawler.oss.OSS_CONFIG', {
+        'download_rewrite_from_base_url': 'https://source-bucket.oss-cn-hangzhou.aliyuncs.com',
+        'download_base_url': internal_source_base,
+    }):
+        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 = transfer_url(public_source, "crawler/qq/audio/abc.mp3", bucket, BASE_URL)
+
+    mock_get.assert_called_once_with(f"{internal_source_base}/audio/abc.mp3", timeout=30)
+    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"

@@ -35,7 +35,7 @@ class _Connection:
         return None
 
 
-def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation(monkeypatch):
+def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
     pg_conn = _PgConnection()
     processors = {
         '1': MagicMock(return_value={'platform': 'qq', 'platform_song_id': 100, 'mid': 'qq-mid', 'title': '歌'}),
@@ -49,7 +49,7 @@ def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation
     monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
     monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
     monkeypatch.setattr(runner, 'fetch_pending_yinyan_song_records', lambda cur, batch_size: [
-        {'song_id': 10, 'record_id': 200},
+        {'song_id': 10, 'record_id': 200, 'platform': '2'},
     ] if pg_conn.commits == 0 else [])
     monkeypatch.setattr(runner, 'fetch_hk_songs_by_source_ids', lambda conn, song_ids: {10: {
         'source_song_id': 10,
@@ -83,19 +83,18 @@ def test_run_imports_all_platform_records_but_writes_one_primary_yinyan_relation
     ]
     monkeypatch.setattr(runner, 'fetch_platform_records', lambda conn, song_ids: platform_records)
     monkeypatch.setattr(runner, 'fetch_all_platform_records', lambda conn, song_ids: platform_records)
-    monkeypatch.setattr(runner, '_pick_record_with_singer', lambda records, spider_conn: records[0])
     monkeypatch.setattr(runner, '_PROCESSORS', processors)
     monkeypatch.setattr(runner, 'upsert_yinyan_song_records', yinyan_writer)
 
     runner.run(['1', '2'])
 
-    processors['1'].assert_called_once()
+    processors['1'].assert_not_called()
     processors['2'].assert_called_once()
     yinyan_writer.assert_called_once_with(pg_conn.cur, [{
         'song_id': 10,
-        'record_id': 100,
-        'platform': '1',
-        'platform_song_id': 100,
+        'record_id': 200,
+        'platform': '2',
+        'platform_song_id': 200,
     }])
     assert pg_conn.commits == 1
 

 from unittest.mock import MagicMock, call
 from etl_to_crawler.writer import (
+    fetch_pending_yinyan_song_records,
     fetch_yinyan_records_missing_platform,
     insert_yinyan_song_records,
     update_yinyan_record_platforms,
@@ -90,6 +91,22 @@ def test_fetch_yinyan_records_missing_platform_reads_null_platform_rows():
     assert rows == [{'song_id': 10, 'record_id': 100}, {'song_id': 11, 'record_id': 101}]
 
 
+def test_fetch_pending_yinyan_song_records_reads_platform_for_single_platform_import():
+    cur = MagicMock()
+    cur.fetchall.return_value = [(10, 100, '1'), (11, 101, '2')]
+
+    rows = fetch_pending_yinyan_song_records(cur, 500)
+
+    sql, params = cur.execute.call_args[0]
+    assert 'SELECT song_id, record_id, platform' in sql
+    assert 'platform IS NOT NULL' in sql
+    assert params == (500,)
+    assert rows == [
+        {'song_id': 10, 'record_id': 100, 'platform': '1'},
+        {'song_id': 11, 'record_id': 101, 'platform': '2'},
+    ]
+
+
 def test_update_yinyan_record_platforms_fills_only_null_platform_rows():
     cur = MagicMock()
     update_yinyan_record_platforms(cur, [