feat(crawler): 为酷狗数据新增provider和源数据字段支持

- 新增 PROVIDER_KUGOU 常量用于标识数据提供者 - 增加 _source_json 方法以支持复杂数据的JSON序列化 - 在处理酷狗歌手数据时添加provider_name和crawler_source_data字段 - 在处理酷狗专辑和歌曲数据时添加对应的provider_name和crawler_source_data字段 - netease歌曲数据处理新增album_json字段以存储专辑数据的JSON表示 - 写入数据库时酷狗歌手、专辑、歌曲表增加provider_name和crawler_source_data列及其对应数据 - 网易歌曲写入增加album_json字段支持 - 新增单元测试覆盖酷狗及网易歌曲处理与数据库写入的新增字段和逻辑

feat(crawler): 为酷狗数据新增provider和源数据字段支持
- 新增 PROVIDER_KUGOU 常量用于标识数据提供者 - 增加 _source_json 方法以支持复杂数据的JSON序列化 - 在处理酷狗歌手数据时添加provider_name和crawler_source_data字段 - 在处理酷狗专辑和歌曲数据时添加对应的provider_name和crawler_source_data字段 - netease歌曲数据处理新增album_json字段以存储专辑数据的JSON表示 - 写入数据库时酷狗歌手、专辑、歌曲表增加provider_name和crawler_source_data列及其对应数据 - 网易歌曲写入增加album_json字段支持 - 新增单元测试覆盖酷狗及网易歌曲处理与数据库写入的新增字段和逻辑
沈秋雨
Commit ac91b0fe ... ac91b0fe0e1d6623f8af9968a5a56a1dab40342e authored 2026-07-09 13:56:48 +0800 by 沈秋雨
Showing 4 changed files with 220 additions and 10 deletions
etl_to_crawler/runner.py
etl_to_crawler/writer.py
tests/test_runner.py
tests/test_writer.py
--- a/etl_to_crawler/runner.py
View file @ac91b0f
+++ b/etl_to_crawler/runner.py
View file @ac91b0f
@@ -26,6 +26,7 @@ from .lyric import strip_timestamps
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
+PROVIDER_KUGOU = 'kugou'
 def _safe_transfer(url, oss_key, bucket, base_url):
@@ -36,6 +37,14 @@ def _safe_transfer(url, oss_key, bucket, base_url):
        return url  # 失败时保留原 URL，不阻断流程
+def _json_default(value):
+    return str(value)
+def _source_json(data: dict) -> str:
+    return json.dumps(data, ensure_ascii=False, default=_json_default)
 def _process_qq(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url):
    mid = pr['platform_unique_key']
    songs_map = fetch_qq_songs(spider_conn, [mid])
@@ -165,7 +174,13 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url
            build_oss_key('kugou', 'singer', str(sg['singer_id']) + '.jpg'),
            bucket, base_url
        )
-        singer_rows.append({**sg, 'id': sg['singer_id'], 'avatar': avatar})
+        singer_rows.append({
+            **sg,
+            'id': sg['singer_id'],
+            'avatar': avatar,
+            'provider_name': PROVIDER_KUGOU,
+            'crawler_source_data': _source_json(sg),
+        })
    upsert_kugou_singers(pg_cur, singer_rows)
    singers_json = json.dumps([{
@@ -183,6 +198,18 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url
            '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': sp.get('album_published_at'),
+            'provider_name': PROVIDER_KUGOU,
+            'crawler_source_data': _source_json({
+                'id': sp.get('album_id'),
+                'cover': sp.get('album_cover'),
+                'title': sp.get('album_title'),
+                'intro': sp.get('album_intro'),
+                'type': sp.get('album_type'),
+                'company_id': sp.get('company_id'),
+                'company': sp.get('company'),
+                'is_owner': sp.get('is_owner'),
+                'published_at': sp.get('album_published_at'),
+            }),
        }])
    song_uuid = str(uuid.uuid4())
@@ -204,6 +231,8 @@ def _process_kugou(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_url
        'platform_index_url': sp.get('platform_index_url') or f'http://www.kugou.com/song/#hash={sp.get("hid") or pr.get("platform_mid", "")}',
        'published_at': sp.get('published_at') or hk_row.get('issue_time'),
        'singers_json': singers_json,
+        'provider_name': PROVIDER_KUGOU,
+        'crawler_source_data': _source_json(sp),
    }])
    # 查询实际 UUID（ON CONFLICT DO NOTHING 时使用已有 UUID）
    pg_cur.execute('SELECT id FROM crawler_kugou_songs WHERE platform_song_id = %s', (song_id,))
@@ -262,8 +291,21 @@ def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_u
    } for sg in singer_list], ensure_ascii=False)
    album_id = None
+    album_json = None
    if sp.get('album_id'):
        album_id = sp['album_id']
+        album_payload = {
+            'id': sp['album_id'],
+            '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_netease_albums(pg_cur, [{
            'id': sp['album_id'], 'cover': album_cover,
            'title': sp.get('album_title') or '', 'intro': sp.get('album_intro'),
@@ -277,6 +319,7 @@ def _process_netease(hk_row: dict, pr: dict, spider_conn, pg_cur, bucket, base_u
        'song_uuid': song_uuid,
        'platform_song_id': song_id,
        'album_id': album_id,
+        'album_json': album_json,
        'cover': cover_url,
        'title': sp.get('title', hk_row['name']),
        'name': hk_row['name'],
--- a/etl_to_crawler/writer.py
View file @ac91b0f
+++ b/etl_to_crawler/writer.py
View file @ac91b0f
@@ -108,14 +108,17 @@ def upsert_kugou_singers(cur, singers: list[dict]) -> None:
        return
    sql = """
        INSERT INTO crawler_kugou_singers
-            (id, name, avatar, sex, area, "index", intro, home_url, created_at, updated_at)
+            (id, name, avatar, sex, area, "index", intro, home_url,
-        VALUES (%s, %s, %s, %s::kugou_singer_sex, %s::kugou_singer_area, %s::kugou_singer_index, %s, %s, NOW(), NOW())
+             created_at, updated_at, provider_name, crawler_source_data)
+        VALUES (%s, %s, %s, %s::kugou_singer_sex, %s::kugou_singer_area, %s::kugou_singer_index, %s, %s,
+                NOW(), NOW(), %s, %s::json)
        ON CONFLICT (id) DO NOTHING
    """
    rows = [(
        s['id'], s['name'], s.get('avatar', ''),
        s.get('sex') or 'U', s.get('area') or '其他', s.get('index') or '#',
        s.get('intro'), s.get('home_url', ''),
+        s.get('provider_name'), s.get('crawler_source_data'),
    ) for s in singers]
    cur.executemany(sql, rows)
@@ -125,14 +128,16 @@ def upsert_kugou_albums(cur, albums: list[dict]) -> None:
        return
    sql = """
        INSERT INTO crawler_kugou_albums
-            (id, cover, title, intro, type, company_id, company, is_owner, published_at, created_at, updated_at)
+            (id, cover, title, intro, type, company_id, company, is_owner, published_at,
-        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
+             created_at, updated_at, provider_name, crawler_source_data)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s, %s::json)
        ON CONFLICT (id) DO NOTHING
    """
    rows = [(
        a['id'], a.get('cover', ''), a.get('title', ''), a.get('intro'),
        a.get('type', ''), a.get('company_id', 0) or 0,
        a.get('company', ''), a.get('is_owner', 0) or 0, a.get('published_at'),
+        a.get('provider_name'), a.get('crawler_source_data'),
    ) for a in albums]
    cur.executemany(sql, rows)
@@ -144,8 +149,9 @@ def upsert_kugou_songs(cur, songs: list[dict]) -> None:
        INSERT INTO crawler_kugou_songs
            (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)
+             platform_index_url, published_at, singers, status, created_at, updated_at,
-        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW())
+             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)
        ON CONFLICT (platform_song_id) DO NOTHING
    """
    rows = [(
@@ -155,6 +161,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('provider_name'), s.get('crawler_source_data'),
    ) for s in songs]
    cur.executemany(sql, rows)
@@ -224,8 +231,8 @@ def upsert_netease_songs(cur, songs: list[dict]) -> None:
        INSERT INTO crawler_netease_songs
            (id, platform_song_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)
+             platform_index_url, published_at, album, singers, status, created_at, updated_at)
-        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW())
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::json, %s::jsonb, 0, NOW(), NOW())
        ON CONFLICT (platform_song_id) DO NOTHING
    """
    rows = [(
@@ -233,7 +240,7 @@ def upsert_netease_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', '[]'),
    ) for s in songs]
    cur.executemany(sql, rows)
--- a/tests/test_runner.py
View file @ac91b0f
+++ b/tests/test_runner.py
View file @ac91b0f
@@ -116,3 +116,61 @@ def test_run_resume_starts_after_saved_id_and_updates_state_after_commit(monkeyp
    assert seen_start_ids == [40]
    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):
+    pg_cur = MagicMock()
+    pg_cur.fetchone.return_value = ('song-uuid',)
+    inserted_songs = []
+    monkeypatch.setattr(runner, 'fetch_netease_songs', lambda conn, song_ids: {
+        300: {
+            'id': 300,
+            'album_id': 20,
+            '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': '录音标题',
+            'duration': 180,
+            'lyric': '[00:01.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, '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)
+    monkeypatch.setattr(runner, 'upsert_netease_singer_albums', lambda cur, pairs: None)
+    runner._process_netease(
+        {
+            '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': '300'},
+        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 '"title": "专辑"' in inserted_songs[0]['album_json']
--- a/tests/test_writer.py
View file @ac91b0f
+++ b/tests/test_writer.py
View file @ac91b0f
 from unittest.mock import MagicMock, call
 from etl_to_crawler.writer import (
+    upsert_kugou_albums,
+    upsert_kugou_singers,
+    upsert_kugou_songs,
+    upsert_netease_songs,
    upsert_qq_singers,
    upsert_qq_singer_songs,
    upsert_yinyan_song_records,
@@ -45,3 +49,101 @@ def test_upsert_yinyan_song_records_replaces_existing_song_relation():
    assert delete_rows == [(10,), (11,)]
    assert 'INSERT INTO yinyan_song_records' in insert_sql
    assert insert_rows == [(10, 100), (11, 101)]
+def test_upsert_netease_songs_writes_album_json_column():
+    cur = MagicMock()
+    upsert_netease_songs(cur, [{
+        'song_uuid': 'song-uuid',
+        'platform_song_id': 300,
+        '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://music.163.com/#/song?id=300',
+        '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][-2] == '{"id": 20, "title": "专辑"}'
+def test_upsert_kugou_songs_writes_provider_and_source_data():
+    cur = MagicMock()
+    upsert_kugou_songs(cur, [{
+        'song_uuid': 'song-uuid',
+        'platform_song_id': 200,
+        'hash': 'hash',
+        'album_audio_id': 300,
+        'album_id': 20,
+        '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://www.kugou.com/song/#hash=hash',
+        'published_at': '2020-01-01',
+        'singers_json': '[]',
+        'provider_name': 'kugou',
+        'crawler_source_data': '{"id": 200}',
+    }])
+    sql, rows = cur.executemany.call_args[0]
+    assert 'provider_name, crawler_source_data' in sql
+    assert '%s::json' in sql
+    assert rows[0][-2:] == ('kugou', '{"id": 200}')
+def test_upsert_kugou_singers_writes_provider_and_source_data():
+    cur = MagicMock()
+    upsert_kugou_singers(cur, [{
+        'id': 1,
+        'name': '歌手',
+        'avatar': 'https://example.com/avatar.jpg',
+        'sex': 'M',
+        'area': '华语',
+        'index': 'G',
+        'intro': '简介',
+        'home_url': 'https://example.com/singer',
+        'provider_name': 'kugou',
+        'crawler_source_data': '{"id": 1}',
+    }])
+    sql, rows = cur.executemany.call_args[0]
+    assert 'provider_name, crawler_source_data' in sql
+    assert rows[0][-2:] == ('kugou', '{"id": 1}')
+def test_upsert_kugou_albums_writes_provider_and_source_data():
+    cur = MagicMock()
+    upsert_kugou_albums(cur, [{
+        'id': 20,
+        'cover': 'https://example.com/album.jpg',
+        'title': '专辑',
+        'intro': '简介',
+        'type': '专辑类型',
+        'company_id': 7,
+        'company': '唱片公司',
+        'is_owner': 1,
+        'published_at': '2020-01-01',
+        'provider_name': 'kugou',
+        'crawler_source_data': '{"id": 20}',
+    }])
+    sql, rows = cur.executemany.call_args[0]
+    assert 'provider_name, crawler_source_data' in sql
+    assert rows[0][-2:] == ('kugou', '{"id": 20}')