feat(etl): 实现三平台 PostgreSQL 写入函数

+import uuid
+
+
+# ─── QQ Music ────────────────────────────────────────────────────────────────
+
+def upsert_qq_singers(cur, singers: list[dict]) -> None:
+    if not singers:
+        return
+    sql = """
+        INSERT INTO crawler_qqmusic_singers
+            (id, mid, name, avatar, sex, area, "index", intro, home_url, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s::singer_sex, %s::singer_area, %s::singer_index, %s, %s, NOW(), NOW())
+        ON CONFLICT (id) DO NOTHING
+    """
+    rows = [(
+        s['id'], s['mid'], 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'),
+    ) for s in singers]
+    cur.executemany(sql, rows)
+
+
+def upsert_qq_albums(cur, albums: list[dict]) -> None:
+    if not albums:
+        return
+    sql = """
+        INSERT INTO crawler_qqmusic_albums
+            (id, mid, cover, title, intro, type, company_id, company, is_owner, published_at, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
+        ON CONFLICT (id) DO NOTHING
+    """
+    rows = [(
+        a['id'], a.get('mid', ''), 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'),
+    ) for a in albums]
+    cur.executemany(sql, rows)
+
+
+def upsert_qq_songs(cur, songs: list[dict]) -> None:
+    if not songs:
+        return
+    sql = """
+        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)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW())
+        ON CONFLICT (platform_song_id) DO NOTHING
+    """
+    rows = [(
+        s['song_uuid'], s['platform_song_id'], s['mid'], s.get('album_id'),
+        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', '[]'),
+    ) for s in songs]
+    cur.executemany(sql, rows)
+
+
+def upsert_qq_singer_songs(cur, pairs: list[tuple]) -> None:
+    """pairs: [(singer_id, song_uuid), ...]"""
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_qqmusic_singer_songs (id, singer_id, song_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, song_id) DO NOTHING
+    """
+    rows = [(str(uuid.uuid4()), singer_id, song_uuid) for singer_id, song_uuid in pairs]
+    cur.executemany(sql, rows)
+
+
+def upsert_qq_singer_albums(cur, pairs: list[tuple]) -> None:
+    """pairs: [(singer_id, album_id), ...]"""
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_qqmusic_singer_albums (id, singer_id, album_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, album_id) DO NOTHING
+    """
+    rows = [(str(uuid.uuid4()), singer_id, album_id) for singer_id, album_id in pairs]
+    cur.executemany(sql, rows)
+
+
+# ─── Kugou ───────────────────────────────────────────────────────────────────
+
+def upsert_kugou_singers(cur, singers: list[dict]) -> None:
+    if not singers:
+        return
+    sql = """
+        INSERT INTO crawler_kugou_singers
+            (id, name, avatar, sex, area, "index", intro, home_url, created_at, updated_at)
+        VALUES (%s, %s, %s, %s::kugou_singer_sex, %s::kugou_singer_area, %s::kugou_singer_index, %s, %s, NOW(), NOW())
+        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', ''),
+    ) for s in singers]
+    cur.executemany(sql, rows)
+
+
+def upsert_kugou_albums(cur, albums: list[dict]) -> None:
+    if not albums:
+        return
+    sql = """
+        INSERT INTO crawler_kugou_albums
+            (id, cover, title, intro, type, company_id, company, is_owner, published_at, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
+        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'),
+    ) for a in albums]
+    cur.executemany(sql, rows)
+
+
+def upsert_kugou_songs(cur, songs: list[dict]) -> None:
+    if not songs:
+        return
+    sql = """
+        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)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW())
+        ON CONFLICT (platform_song_id) DO NOTHING
+    """
+    rows = [(
+        s['song_uuid'], s['platform_song_id'], s.get('hash', ''),
+        s.get('album_audio_id', 0) or 0, s.get('album_id'),
+        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', '[]'),
+    ) for s in songs]
+    cur.executemany(sql, rows)
+
+
+def upsert_kugou_singer_songs(cur, pairs: list[tuple]) -> None:
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_kugou_singer_songs (id, singer_id, song_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, song_id) DO NOTHING
+    """
+    cur.executemany(sql, [(str(uuid.uuid4()), s, sg) for s, sg in pairs])
+
+
+def upsert_kugou_singer_albums(cur, pairs: list[tuple]) -> None:
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_kugou_singer_albums (id, singer_id, album_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, album_id) DO NOTHING
+    """
+    cur.executemany(sql, [(str(uuid.uuid4()), s, a) for s, a in pairs])
+
+
+# ─── Netease ─────────────────────────────────────────────────────────────────
+
+def upsert_netease_singers(cur, singers: list[dict]) -> None:
+    if not singers:
+        return
+    sql = """
+        INSERT INTO crawler_netease_singers
+            (id, name, avatar, sex, area, "index", intro, home_url, created_at, updated_at)
+        VALUES (%s, %s, %s, %s::netease_singer_sex, %s::netease_singer_area, %s::netease_singer_index, %s, %s, NOW(), NOW())
+        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'),
+    ) for s in singers]
+    cur.executemany(sql, rows)
+
+
+def upsert_netease_albums(cur, albums: list[dict]) -> None:
+    if not albums:
+        return
+    sql = """
+        INSERT INTO crawler_netease_albums
+            (id, cover, title, intro, type, company_id, company, is_owner, published_at, created_at, updated_at)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
+        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'),
+    ) for a in albums]
+    cur.executemany(sql, rows)
+
+
+def upsert_netease_songs(cur, songs: list[dict]) -> None:
+    if not songs:
+        return
+    sql = """
+        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)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, 0, NOW(), NOW())
+        ON CONFLICT (platform_song_id) DO NOTHING
+    """
+    rows = [(
+        s['song_uuid'], s['platform_song_id'], s.get('album_id'),
+        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', '[]'),
+    ) for s in songs]
+    cur.executemany(sql, rows)
+
+
+def upsert_netease_singer_songs(cur, pairs: list[tuple]) -> None:
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_netease_singer_songs (id, singer_id, song_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, song_id) DO NOTHING
+    """
+    cur.executemany(sql, [(str(uuid.uuid4()), s, sg) for s, sg in pairs])
+
+
+def upsert_netease_singer_albums(cur, pairs: list[tuple]) -> None:
+    if not pairs:
+        return
+    sql = """
+        INSERT INTO crawler_netease_singer_albums (id, singer_id, album_id)
+        VALUES (%s, %s, %s)
+        ON CONFLICT (singer_id, album_id) DO NOTHING
+    """
+    cur.executemany(sql, [(str(uuid.uuid4()), s, a) for s, a in pairs])

+from unittest.mock import MagicMock, call
+from etl_to_crawler.writer import upsert_qq_singers, upsert_qq_singer_songs
+
+
+def test_upsert_qq_singers_executes_insert():
+    cur = MagicMock()
+    singers = [{
+        'id': 1, 'mid': 'abc', 'name': '歌手', 'avatar': 'https://archive-dev.oss-cn-beijing.aliyuncs.com/a.jpg',
+        'sex': 'M', 'area': '华语', 'index': 'G', 'intro': None, 'home_url': None
+    }]
+    upsert_qq_singers(cur, singers)
+    assert cur.executemany.called
+    args = cur.executemany.call_args
+    assert 'crawler_qqmusic_singers' in args[0][0]
+    assert 'ON CONFLICT' in args[0][0]
+
+
+def test_upsert_qq_singers_empty_does_nothing():
+    cur = MagicMock()
+    upsert_qq_singers(cur, [])
+    cur.executemany.assert_not_called()
+
+
+def test_upsert_qq_singer_songs():
+    cur = MagicMock()
+    upsert_qq_singer_songs(cur, [(1, 'uuid-abc'), (2, 'uuid-def')])
+    assert cur.executemany.called
+    sql = cur.executemany.call_args[0][0]
+    assert 'crawler_qqmusic_singer_songs' in sql