refactor(etl_to_crawler): 优化QQ音乐入库流程并新增album_json字段

- 将BATCH_SIZE由4调整为1以支持更细粒度的数据处理
- 在runner中为QQ音乐歌曲构建完整album_json对象并序列化
- 修改upsert_qq_songs接口，新增album_json参数的传递和入库支持
- 在writer模块SQL插入语句中添加album字段，支持json类型数据写入
- 编写单元测试验证album_json字段正确生成与写入逻辑
- 兼容旧逻辑，album_json字段为空时不影响其他数据写入

@@ -43,4 +43,4 @@ PLATFORM_KUGOU = '2'
 PLATFORM_NETEASE = '4'
 PLATFORMS = [PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE]
 
-BATCH_SIZE = 4
+BATCH_SIZE = 1

@@ -133,8 +133,22 @@ def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
 
     # 写入 album
     album_id = None
+    album_json = None
     if sp.get('album_id'):
         album_id = sp['album_id']
+        album_payload = {
+            'id': sp['album_id'],
+            'mid': sp.get('album_mid') or '',
+            'cover': album_cover,
+            'title': sp.get('album_title') or '',
+            'intro': sp.get('album_intro'),
+            'type': sp.get('album_type') or '',
+            'company_id': sp.get('company_id') or 0,
+            'company': sp.get('company') or '',
+            'is_owner': sp.get('is_owner') or 0,
+            'published_at': str(sp.get('album_published_at')) if sp.get('album_published_at') else None,
+        }
+        album_json = json.dumps(album_payload, ensure_ascii=False)
         upsert_qq_albums(pg_cur, [{
             'id': sp['album_id'], 'mid': sp.get('album_mid') or '',
             'cover': album_cover, 'title': sp.get('album_title') or '',
@@ -167,6 +181,7 @@ def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
         'platform_song_id': platform_song_id,
         'mid': mid,
         'album_id': album_id,
+        'album_json': album_json,
         'cover': cover_url,
         'title': title,
         'version': version,

@@ -138,9 +138,9 @@ def upsert_qq_songs(cur, songs: list[dict]) -> None:
         INSERT INTO crawler_qqmusic_songs
             (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,
+             platform_index_url, published_at, album, singers, status, created_at, updated_at,
              version, 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, %s::json)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::json, %s::jsonb, 0, NOW(), NOW(), %s, %s, %s, %s::json)
         ON CONFLICT (platform_song_id) DO NOTHING
     """
     rows = [(
@@ -148,7 +148,7 @@ def upsert_qq_songs(cur, songs: list[dict]) -> None:
         s.get('cover', ''), s.get('title', ''), s.get('name', ''), s.get('duration', 0) or 0,
         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('platform_index_url'), s.get('published_at'), s.get('album_json'), s.get('singers_json', '[]'),
         s.get('version'),
         s.get('audio_md5'),
         s.get('provider_name'), s.get('crawler_source_data'),

@@ -169,6 +169,70 @@ def test_backfill_yinyan_record_platforms_updates_missing_platform_rows(monkeypa
     assert pg_conn.commits == 1
 
 
+def test_process_qq_builds_album_json_for_song_insert(monkeypatch):
+    pg_cur = MagicMock()
+    pg_cur.fetchone.return_value = ('song-uuid',)
+    inserted_songs = []
+
+    monkeypatch.setattr(runner, 'fetch_qq_songs', lambda conn, mids: {
+        'qq-mid': {
+            'id': 200,
+            'mid': 'qq-mid',
+            'album_id': 20,
+            'album_mid': 'album-mid',
+            'album_cover': 'https://example.com/album.jpg',
+            'album_title': '专辑',
+            'album_intro': '简介',
+            'album_type': '专辑类型',
+            'company_id': 7,
+            'company': '唱片公司',
+            'is_owner': 1,
+            'album_published_at': '2020-01-01',
+            'cover': 'https://example.com/cover.jpg',
+            'title': '化风行万里 (DJ默涵版)',
+            'duration': 180,
+            'lyric': '[00:01.00]歌词',
+            'composer_name': '曲作者',
+            'lyricist_name': '词作者',
+            'platform_index_url': None,
+            'published_at': '2020-01-02',
+        },
+    })
+    monkeypatch.setattr(runner, 'fetch_qq_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_qq_singers', lambda cur, singers: None)
+    monkeypatch.setattr(runner, 'upsert_qq_albums', lambda cur, albums: None)
+    monkeypatch.setattr(runner, 'upsert_qq_songs', lambda cur, songs: inserted_songs.extend(songs))
+    monkeypatch.setattr(runner, 'upsert_qq_singer_songs', lambda cur, pairs: None)
+    monkeypatch.setattr(runner, 'upsert_qq_singer_albums', lambda cur, pairs: None)
+
+    runner._process_qq(
+        {
+            'name': '词曲名',
+            'audio_url': 'https://example.com/audio.mp3',
+            'lyrics_url': 'https://example.com/lyric.lrc',
+            'cover_url': '',
+            'composer': '词曲曲作者',
+            'lyricist': '词曲词作者',
+            'issue_time': '2019-01-01',
+            'song_time': 120,
+        },
+        {'platform_unique_key': 'qq-mid', 'platform_mid': '200'},
+        spider_conn=object(),
+        pg_cur=pg_cur,
+        bucket=object(),
+        base_url='https://bucket.example.com',
+    )
+
+    assert inserted_songs[0]['album_json']
+    assert '"id": 20' in inserted_songs[0]['album_json']
+    assert '"mid": "album-mid"' in inserted_songs[0]['album_json']
+    assert '"title": "专辑"' in inserted_songs[0]['album_json']
+    assert inserted_songs[0]['title'] == '化风行万里'
+    assert inserted_songs[0]['version'] == 'DJ默涵版'
+
+
 def test_process_netease_builds_album_json_for_song_insert(monkeypatch):
     pg_cur = MagicMock()
     pg_cur.fetchone.return_value = ('song-uuid',)

@@ -130,6 +130,34 @@ def test_upsert_netease_songs_writes_album_json_column():
     assert rows[0][14] == '{"id": 20, "title": "专辑"}'
 
 
+def test_upsert_qq_songs_writes_album_json_column():
+    cur = MagicMock()
+    upsert_qq_songs(cur, [{
+        'song_uuid': 'song-uuid',
+        'platform_song_id': 200,
+        'mid': 'song-mid',
+        'album_id': 20,
+        'album_json': '{"id": 20, "title": "专辑"}',
+        'cover': 'https://example.com/cover.jpg',
+        'title': '标题',
+        'name': '词曲名',
+        'duration': 180,
+        'lyric': '歌词',
+        'composer_name': '曲作者',
+        'lyricist_name': '词作者',
+        'url': 'https://example.com/audio.mp3',
+        'lyric_url': 'https://example.com/lyric.lrc',
+        'platform_index_url': 'https://y.qq.com/n/ryqq/songDetail/song-mid',
+        'published_at': '2020-01-01',
+        'singers_json': '[]',
+    }])
+
+    sql, rows = cur.executemany.call_args[0]
+    assert 'album, singers' in sql
+    assert '%s::json, %s::jsonb' in sql
+    assert rows[0][15] == '{"id": 20, "title": "专辑"}'
+
+
 def test_upsert_kugou_songs_writes_provider_and_source_data():
     cur = MagicMock()
     upsert_kugou_songs(cur, [{