Skip to content

沈秋雨 / import_hk_songs

Go to a project
Toggle navigation
Toggle navigation pinning
Commit ae633f56 ae633f56f1aa24cc8606645f550c60decbb50c25 by 沈秋雨
Options

feat(env): 更新目标库配置及添加 PostgreSQL 环境变量 

- 目标库名称由 hk_songs 修改为 hk_songs_test，新增中间表 hk_songs_import_staging
- 添加测试环境 crawler_dev PostgreSQL 连接配置变量
- 添加正式环境 archive_crawler PostgreSQL 连接配置变量

refactor(check_record_titles): 调整平台原始标题核对逻辑

- 变更核对数据库源为 archive_crawler 与 hikoon-data-spider
- 按平台分表批量查询 spider 原始标题替代原录音名查询
- 合并结果中使用 spider_title 与 crawler_title 精确比对
- 输出 Excel 标题及字段相应更新，匹配时整行高亮
- 修改脚本参数说明和默认批处理数量描述
- 日志及信息输出文本调整以反映新数据源

feat(fix_record_titles): 新增基于 spider 标题的归一化批量修复工具

- 支持从 archive_crawler 读取待修复记录的推送关系
- 结合 spider 标题，计算归一化标题和版本号
- 支持对不同平台数据批量生成更新 SQL 预览或应用
- 支持备份更新前数据，支持安全回退操作
- 导出更新报告 Excel，突出显示变化字段
- 支持长文本分批过滤，避免数据库字段长度异常
- 增加命令行参数控制应用范围、批量大小及排除标题列表
- 使用严格的 SQL 语句构造与变更检测优化更新效率
- 实现基于 run_id 的多表回滚机制，保证数据一致性
1 parent 179a08f8
Inline Side-by-side
Showing 14 changed files with 2653 additions and 113 deletions
......@@ -5,13 +5,14 @@ SOURCE_DB_USER=
SOURCE_DB_PASSWORD=
SOURCE_DB_NAME=
# 目标库 - 词曲库
# 目标库 - 词曲测试
TARGET_DB_HOST=
TARGET_DB_PORT=3306
TARGET_DB_USER=
TARGET_DB_PASSWORD=
TARGET_DB_NAME=
TARGET_TABLE_NAME=hk_songs
TARGET_TABLE_NAME=hk_songs_test
TARGET_TABLE_NAME_TMP=hk_songs_import_staging
# run_etl.py 默认读取并回写的导入状态表：yinyan_song_records 或 yinyan_song_records2
YINYAN_IMPORT_TABLE=yinyan_song_records
......@@ -22,3 +23,21 @@ OSS_ACCESS_KEY_SECRET=
OSS_ENDPOINT=
OSS_BUCKET_NAME=
OSS_FILE_BASE_NAME=
# 测试环境 - crawler_dev PostgreSQL
TEST_CRAWLER_DB_HOST=
TEST_CRAWLER_DB_PORT=5432
TEST_CRAWLER_DB_USER=
TEST_CRAWLER_DB_PASSWORD=
TEST_CRAWLER_DB_NAME=crawler_dev
TEST_CRAWLER_DB_SSL=false
TEST_ARCHIVE_DATA_DB_NAME=data_dev
# 正式环境 - archive_crawler PostgreSQL
ARCHIVE_CRAWLER_DB_HOST=
ARCHIVE_CRAWLER_DB_PORT=5432
ARCHIVE_CRAWLER_DB_USER=
ARCHIVE_CRAWLER_DB_PASSWORD=
ARCHIVE_CRAWLER_DB_NAME=archive_crawler
ARCHIVE_CRAWLER_DB_SSL=false
ARCHIVE_DATA_DB_NAME=archive_data
......
#!/usr/bin/env python3
"""核对音眼录音名与 crawler 平台歌曲标题，并输出 Excel。
"""核对 spider 平台原始标题与正式 crawler 歌曲标题，并输出 Excel。
数据链路：
CRAWLER_DB.yinyan_song_records.record_id
-> SOURCE_DB.hk_music_record.id -> record_name
CRAWLER_DB.yinyan_song_records.(platform, platform_song_id)
-> 对应 crawler_*_songs.platform_song_id -> title
archive_crawler.yinyan_song_records.(platform, platform_song_id)
-> hikoon-data-spider.media_*_songs.id -> spider_title
-> archive_crawler.crawler_*_songs.platform_song_id -> crawler_title
用法：
.venv/bin/python check_record_titles.py
......@@ -21,7 +20,7 @@ from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
from etl_to_crawler.connections import close_all_pools, get_pg_conn, get_source_conn
from etl_to_crawler.connections import close_all_pools, get_archive_crawler_conn, get_spider_conn
log = logging.getLogger(__name__)
......@@ -29,6 +28,11 @@ log = logging.getLogger(__name__)
DEFAULT_OUTPUT = "record_title_check.xlsx"
DEFAULT_BATCH_SIZE = 2000
ALLOWED_IMPORT_TABLES = {"yinyan_song_records", "yinyan_song_records2"}
SPIDER_TABLES = {
"1": "media_tencent_songs",
"2": "media_ku_gou_songs",
"4": "media_netease_songs",
}
HEADER_FILL = PatternFill(fill_type="solid", fgColor="4472C4")
MATCH_FILL = PatternFill(fill_type="solid", fgColor="C6EFCE")
......@@ -81,67 +85,77 @@ def fetch_crawler_rows(pg_conn, import_table: str) -> list[dict]:
"platform": row[1],
"platform_song_id": row[2],
"recording_id": row[3],
"title": row[4],
"crawler_title": row[4],
}
for row in rows
]
def fetch_record_names(mysql_conn, record_ids: Iterable[int], batch_size: int) -> dict[int, str | None]:
"""按 record_id 分批读取 hk_music_record.record_name。"""
unique_ids = sorted({int(record_id) for record_id in record_ids if record_id is not None})
names: dict[int, str | None] = {}
for batch in chunked(unique_ids, batch_size):
def fetch_spider_titles(mysql_conn, crawler_rows: list[dict], batch_size: int) -> dict[tuple[str, int], str]:
"""按 platform_song_id 分平台批量读取 spider 原始 title。"""
titles: dict[tuple[str, int], str] = {}
for platform, table in SPIDER_TABLES.items():
ids = sorted({
int(row["platform_song_id"])
for row in crawler_rows
if str(row["platform"]) == platform and row["platform_song_id"] is not None
})
for batch in chunked(ids, batch_size):
placeholders = ", ".join(["%s"] * len(batch))
sql = f"""
SELECT id, record_name
FROM hk_music_record
WHERE id IN ({placeholders})
"""
with mysql_conn.cursor() as cursor:
cursor.execute(sql, tuple(batch))
cursor.execute(
f"SELECT id, title FROM {table} WHERE id IN ({placeholders})",
tuple(batch),
)
for row in cursor.fetchall():
names[int(row["id"])] = row["record_name"]
if row["title"] is not None:
titles[(platform, int(row["id"]))] = str(row["title"])
return titles
return names
def combine_rows(crawler_rows: list[dict], record_names: dict[int, str | None]) -> list[dict]:
def combine_rows(crawler_rows: list[dict], spider_titles: dict[tuple[str, int], str]) -> list[dict]:
"""合并两个数据库的查询结果，并计算是否完全相同。"""
result = []
for row in crawler_rows:
record_id = row["record_id"]
record_name = record_names.get(int(record_id)) if record_id is not None else None
title = row["title"]
platform = str(row["platform"])
platform_song_id = row["platform_song_id"]
spider_title = (
spider_titles.get((platform, int(platform_song_id)))
if platform_song_id is not None else None
)
crawler_title = row["crawler_title"]
result.append(
{
"record_id": record_id,
"record_name": record_name,
"platform": row["platform"],
"platform_song_id": row["platform_song_id"],
"record_id": row["record_id"],
"spider_title": spider_title,
"crawler_title": crawler_title,
"platform": platform,
"platform_song_id": platform_song_id,
"recording_id": row["recording_id"],
"title": title,
# 两边都必须有值；不做 trim、大小写或繁简转换，按库中原值精确比较。
"matched": record_name is not None and title is not None and record_name == title,
"matched": (
spider_title is not None
and crawler_title is not None
and spider_title == crawler_title
),
}
)
return result
def write_excel(rows: list[dict], output_path: Path) -> None:
"""写出三列表格；record_name 与 title 相同时整行标绿。"""
"""写出核对表；spider_title 与 crawler_title 相同时整行标绿。"""
output_path.parent.mkdir(parents=True, exist_ok=True)
workbook = Workbook()
worksheet = workbook.active
worksheet.title = "录音名与标题核对"
worksheet.title = "Spider与Crawler标题核对"
worksheet.freeze_panes = "A2"
headers = (
"record_id",
"record_name",
"title",
"spider_title",
"crawler_title",
"platform",
"platform_song_id",
"recording_id",
......@@ -155,8 +169,8 @@ def write_excel(rows: list[dict], output_path: Path) -> None:
for row_number, row in enumerate(rows, start=2):
values = (
row["record_id"],
row["record_name"],
row["title"],
row["spider_title"],
row["crawler_title"],
row["platform"],
row["platform_song_id"],
row["recording_id"],
......@@ -175,15 +189,15 @@ def write_excel(rows: list[dict], output_path: Path) -> None:
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="核对 hk_music_record.record_name 与 crawler title")
parser = argparse.ArgumentParser(description="核对 spider title 与正式 crawler title")
parser.add_argument("--output", default=DEFAULT_OUTPUT, help=f"Excel 输出路径（默认: {DEFAULT_OUTPUT}）")
parser.add_argument(
"--table",
choices=sorted(ALLOWED_IMPORT_TABLES),
default="yinyan_song_records",
help="待核对的 CRAWLER_DB 关联表",
help="待核对的 ARCHIVE_CRAWLER_DB 关联表",
)
parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE, help="MySQL 单批查询数量")
parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE, help="spider MySQL 单批查询数量")
args = parser.parse_args()
if args.batch_size <= 0:
parser.error("--batch-size 必须大于 0")
......@@ -195,28 +209,24 @@ def main() -> None:
output_path = Path(args.output).expanduser().resolve()
try:
pg_conn = get_pg_conn()
mysql_conn = get_source_conn()
pg_conn = get_archive_crawler_conn()
mysql_conn = get_spider_conn()
crawler_rows = fetch_crawler_rows(pg_conn, args.table)
log.info("从 %s 读取 %d 条关联记录", args.table, len(crawler_rows))
record_names = fetch_record_names(
mysql_conn,
(row["record_id"] for row in crawler_rows),
args.batch_size,
)
log.info("从 hk_music_record 读取 %d 个录音名", len(record_names))
spider_titles = fetch_spider_titles(mysql_conn, crawler_rows, args.batch_size)
log.info("从 hikoon-data-spider 读取 %d 个平台标题", len(spider_titles))
rows = combine_rows(crawler_rows, record_names)
rows = combine_rows(crawler_rows, spider_titles)
write_excel(rows, output_path)
matched = sum(row["matched"] for row in rows)
missing_name = sum(row["record_name"] is None for row in rows)
missing_title = sum(row["title"] is None for row in rows)
missing_spider = sum(row["spider_title"] is None for row in rows)
missing_crawler = sum(row["crawler_title"] is None for row in rows)
log.info(
"核对完成：总数=%d，相同=%d，不同=%d，缺少 record_name=%d，缺少 title=%d",
len(rows), matched, len(rows) - matched, missing_name, missing_title,
"核对完成：总数=%d，相同=%d，不同=%d，缺少 spider_title=%d，缺少 crawler_title=%d",
len(rows), matched, len(rows) - matched, missing_spider, missing_crawler,
)
log.info("Excel 已输出至 %s", output_path)
finally:
......
......@@ -21,13 +21,32 @@ HK_SONGS_DB = {
'charset': 'utf8mb4',
}
CRAWLER_DB = {
'host': os.environ['CRAWLER_DB_HOST'],
'port': int(os.environ.get('CRAWLER_DB_PORT', '5432')),
'user': os.environ['CRAWLER_DB_USER'],
'password': os.environ['CRAWLER_DB_PASSWORD'],
'database': os.environ['CRAWLER_DB_NAME'],
'ssl_context': None if os.environ.get('CRAWLER_DB_SSL', 'false').lower() != 'true' else True,
TEST_CRAWLER_DB = {
'host': os.environ['TEST_CRAWLER_DB_HOST'],
'port': int(os.environ.get('TEST_CRAWLER_DB_PORT', '5432')),
'user': os.environ['TEST_CRAWLER_DB_USER'],
'password': os.environ['TEST_CRAWLER_DB_PASSWORD'],
'database': os.environ['TEST_CRAWLER_DB_NAME'],
'ssl_context': None if os.environ.get('TEST_CRAWLER_DB_SSL', 'false').lower() != 'true' else True,
}
TEST_ARCHIVE_DATA_DB = {
**TEST_CRAWLER_DB,
'database': os.environ.get('TEST_ARCHIVE_DATA_DB_NAME', 'data_dev'),
}
ARCHIVE_CRAWLER_DB = {
'host': os.environ['ARCHIVE_CRAWLER_DB_HOST'],
'port': int(os.environ.get('ARCHIVE_CRAWLER_DB_PORT', '5432')),
'user': os.environ['ARCHIVE_CRAWLER_DB_USER'],
'password': os.environ['ARCHIVE_CRAWLER_DB_PASSWORD'],
'database': os.environ.get('ARCHIVE_CRAWLER_DB_NAME', 'archive_crawler'),
'ssl_context': None if os.environ.get('ARCHIVE_CRAWLER_DB_SSL', 'false').lower() != 'true' else True,
}
ARCHIVE_DATA_DB = {
**ARCHIVE_CRAWLER_DB,
'database': os.environ.get('ARCHIVE_DATA_DB_NAME', 'archive_data'),
}
OSS_CONFIG = {
......
......@@ -3,7 +3,16 @@ import pymysql
import pymysql.cursors
import pg8000
import oss2
from .config import SOURCE_DB, HK_SONGS_DB, CRAWLER_DB, OSS_CONFIG, OSS_CONNECTION_POOL_SIZE
from .config import (
ARCHIVE_CRAWLER_DB,
ARCHIVE_DATA_DB,
SOURCE_DB,
HK_SONGS_DB,
TEST_CRAWLER_DB,
TEST_ARCHIVE_DATA_DB,
OSS_CONFIG,
OSS_CONNECTION_POOL_SIZE,
)
CONN_POOL_SIZE = 4
......@@ -238,7 +247,10 @@ class _PgConnPool:
_source_pool = None
_hk_songs_pool = None
_spider_pool = None
_pg_pool = None
_test_crawler_pool = None
_test_archive_pool = None
_archive_pool = None
_archive_crawler_pool = None
def _get_source_pool() -> _MySqlConnPool:
......@@ -264,11 +276,32 @@ def _get_spider_pool() -> _MySqlConnPool:
return _spider_pool
def _get_pg_pool() -> _PgConnPool:
global _pg_pool
if _pg_pool is None:
_pg_pool = _PgConnPool(CRAWLER_DB)
return _pg_pool
def _get_test_crawler_pool() -> _PgConnPool:
global _test_crawler_pool
if _test_crawler_pool is None:
_test_crawler_pool = _PgConnPool(TEST_CRAWLER_DB)
return _test_crawler_pool
def _get_test_archive_pool() -> _PgConnPool:
global _test_archive_pool
if _test_archive_pool is None:
_test_archive_pool = _PgConnPool(TEST_ARCHIVE_DATA_DB)
return _test_archive_pool
def _get_archive_pool() -> _PgConnPool:
global _archive_pool
if _archive_pool is None:
_archive_pool = _PgConnPool(ARCHIVE_DATA_DB)
return _archive_pool
def _get_archive_crawler_pool() -> _PgConnPool:
global _archive_crawler_pool
if _archive_crawler_pool is None:
_archive_crawler_pool = _PgConnPool(ARCHIVE_CRAWLER_DB)
return _archive_crawler_pool
def get_source_conn() -> pymysql.Connection:
......@@ -283,8 +316,20 @@ def get_hk_songs_conn() -> pymysql.Connection:
return _get_hk_songs_pool().get()
def get_pg_conn() -> _NulSafePgConnection:
return _get_pg_pool().get()
def get_test_crawler_conn() -> _NulSafePgConnection:
return _get_test_crawler_pool().get()
def get_test_archive_conn() -> _NulSafePgConnection:
return _get_test_archive_pool().get()
def get_archive_conn() -> _NulSafePgConnection:
return _get_archive_pool().get()
def get_archive_crawler_conn() -> _NulSafePgConnection:
return _get_archive_crawler_pool().get()
def get_oss_bucket() -> oss2.Bucket:
......@@ -298,18 +343,24 @@ def close_all_pools():
for pool in (_source_pool, _hk_songs_pool, _spider_pool):
if pool is not None:
pool.close_all()
if _pg_pool is not None:
_pg_pool.close_all()
if _test_crawler_pool is not None:
_test_crawler_pool.close_all()
if _test_archive_pool is not None:
_test_archive_pool.close_all()
if _archive_pool is not None:
_archive_pool.close_all()
if _archive_crawler_pool is not None:
_archive_crawler_pool.close_all()
def refresh_conn(conn, pool_type: str):
"""验证连接健康度，断开时返回新连接，正常时原样返回。
用法：在批次循环开头调用，如 ``conn = refresh_conn(conn, 'pg')``。
pool_type 取 'source' | 'hk_songs' | 'spider' | 'pg'。
用法：在批次循环开头调用，如 ``conn = refresh_conn(conn, 'test_crawler')``。
pool_type 取 'source' | 'hk_songs' | 'spider' | 'test_crawler'。
"""
try:
if pool_type == 'pg':
if pool_type == 'test_crawler':
conn.run("SELECT 1")
return conn
# MySQL
......@@ -322,7 +373,7 @@ def refresh_conn(conn, pool_type: str):
'source': _get_source_pool,
'hk_songs': _get_hk_songs_pool,
'spider': _get_spider_pool,
'pg': _get_pg_pool,
'test_crawler': _get_test_crawler_pool,
}
pool = pools.get(pool_type)
if pool is None:
......
......@@ -8,7 +8,7 @@ from .config import (
PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE, BATCH_SIZE, BACKFILL_BATCH_SIZE,
OSS_CONFIG, YINYAN_IMPORT_TABLE,
)
from .connections import get_hk_songs_conn, get_source_conn, get_spider_conn, get_pg_conn, get_oss_bucket, refresh_conn, close_all_pools
from .connections import get_hk_songs_conn, get_source_conn, get_spider_conn, get_test_crawler_conn, get_oss_bucket, refresh_conn, close_all_pools
from .reader import (
iter_hk_songs_batches,
fetch_hk_songs_by_source_ids,
......@@ -1337,7 +1337,7 @@ def _pick_record_with_singer(records: list[dict], spider_conn) -> dict | None:
def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None = None) -> None:
hk_conn = get_hk_songs_conn()
src_conn = get_source_conn()
pg_conn = get_pg_conn()
pg_conn = get_test_crawler_conn()
# 查询已存在的 song_id，用于跳过已处理的记录
with pg_conn.cursor() as pg_cur:
......@@ -1351,7 +1351,7 @@ def initialize_yinyan_song_records(platforms: list[str], max_batches: int | None
# 长任务连接可能断开，每批次开始前刷新
hk_conn = refresh_conn(hk_conn, 'hk_songs')
src_conn = refresh_conn(src_conn, 'source')
pg_conn = refresh_conn(pg_conn, 'pg')
pg_conn = refresh_conn(pg_conn, 'test_crawler')
if max_batches is not None and i >= max_batches:
break
......@@ -1416,7 +1416,7 @@ def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | Non
后续导入允许歌曲的 singers 为空。
"""
src_conn = get_source_conn()
pg_conn = get_pg_conn()
pg_conn = get_test_crawler_conn()
total_inserted = 0
try:
......@@ -1427,7 +1427,7 @@ def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | Non
for index, start in enumerate(tqdm(range(0, len(song_ids), BATCH_SIZE), desc='init-yinyan-records2')):
# 长任务连接可能断开，每批次开始前刷新
src_conn = refresh_conn(src_conn, 'source')
pg_conn = refresh_conn(pg_conn, 'pg')
pg_conn = refresh_conn(pg_conn, 'test_crawler')
if max_batches is not None and index >= max_batches:
break
......@@ -1467,7 +1467,7 @@ def initialize_yinyan_song_records2(platforms: list[str], max_batches: int | Non
def backfill_yinyan_record_platforms(max_batches: int | None = None) -> None:
src_conn = get_source_conn()
pg_conn = get_pg_conn()
pg_conn = get_test_crawler_conn()
total = 0
skipped = 0
......@@ -1477,7 +1477,7 @@ def backfill_yinyan_record_platforms(max_batches: int | None = None) -> None:
while max_batches is None or batch_index < max_batches:
# 长任务连接可能断开，每批次开始前刷新
src_conn = refresh_conn(src_conn, 'source')
pg_conn = refresh_conn(pg_conn, 'pg')
pg_conn = refresh_conn(pg_conn, 'test_crawler')
with pg_conn.cursor() as pg_cur:
rows = fetch_yinyan_records_missing_platform(pg_cur, BACKFILL_BATCH_SIZE)
......@@ -1523,7 +1523,7 @@ def _backfill_qq_singers(spider_conn, pg_conn, bucket, base_url, max_batches):
while max_batches is None or batch_index < max_batches:
# 长任务连接可能断开，每批次开始前刷新
spider_conn = refresh_conn(spider_conn, 'spider')
pg_conn = refresh_conn(pg_conn, 'pg')
pg_conn = refresh_conn(pg_conn, 'test_crawler')
with pg_conn.cursor() as pg_cur:
rows = fetch_qq_songs_missing_singers(pg_cur, BATCH_SIZE)
......@@ -1603,7 +1603,7 @@ def _backfill_kugou_singers(spider_conn, pg_conn, bucket, base_url, max_batches)
while max_batches is None or batch_index < max_batches:
# 长任务连接可能断开，每批次开始前刷新
spider_conn = refresh_conn(spider_conn, 'spider')
pg_conn = refresh_conn(pg_conn, 'pg')
pg_conn = refresh_conn(pg_conn, 'test_crawler')
with pg_conn.cursor() as pg_cur:
rows = fetch_kugou_songs_missing_singers(pg_cur, BATCH_SIZE)
......@@ -1677,7 +1677,7 @@ def _backfill_netease_singers(spider_conn, pg_conn, bucket, base_url, max_batche
while max_batches is None or batch_index < max_batches:
# 长任务连接可能断开，每批次开始前刷新
spider_conn = refresh_conn(spider_conn, 'spider')
pg_conn = refresh_conn(pg_conn, 'pg')
pg_conn = refresh_conn(pg_conn, 'test_crawler')
with pg_conn.cursor() as pg_cur:
rows = fetch_netease_songs_missing_singers(pg_cur, BATCH_SIZE)
......@@ -1746,7 +1746,7 @@ def _backfill_netease_singers(spider_conn, pg_conn, bucket, base_url, max_batche
def backfill_empty_singers(platforms: list[str], max_batches: int | None = None) -> None:
spider_conn = get_spider_conn()
pg_conn = get_pg_conn()
pg_conn = get_test_crawler_conn()
bucket = get_oss_bucket()
base_url = OSS_CONFIG['base_url']
try:
......@@ -1769,7 +1769,7 @@ def backfill_qq_invalid_covers(max_batches: int | None = None) -> None:
hk_conn = get_hk_songs_conn()
src_conn = get_source_conn()
spider_conn = get_spider_conn()
pg_conn = get_pg_conn()
pg_conn = get_test_crawler_conn()
bucket = get_oss_bucket()
base_url = OSS_CONFIG['base_url']
batch_index = replaced = filtered = retry_later = 0
......@@ -1780,7 +1780,7 @@ def backfill_qq_invalid_covers(max_batches: int | None = None) -> None:
hk_conn = refresh_conn(hk_conn, 'hk_songs')
src_conn = refresh_conn(src_conn, 'source')
spider_conn = refresh_conn(spider_conn, 'spider')
pg_conn = refresh_conn(pg_conn, 'pg')
pg_conn = refresh_conn(pg_conn, 'test_crawler')
with pg_conn.cursor() as pg_cur:
invalid_rows = fetch_qq_songs_with_invalid_covers(
......@@ -1879,7 +1879,7 @@ def run(
hk_conn = get_hk_songs_conn()
src_conn = get_source_conn()
spider_conn = get_spider_conn()
pg_conn = get_pg_conn()
pg_conn = get_test_crawler_conn()
bucket = get_oss_bucket()
base_url = OSS_CONFIG['base_url']
total_ok = total_err = 0
......@@ -1905,7 +1905,7 @@ def run(
hk_conn = refresh_conn(hk_conn, 'hk_songs')
src_conn = refresh_conn(src_conn, 'source')
spider_conn = refresh_conn(spider_conn, 'spider')
pg_conn = refresh_conn(pg_conn, 'pg')
pg_conn = refresh_conn(pg_conn, 'test_crawler')
with pg_conn.cursor() as pg_cur:
pending_records = fetch_pending_yinyan_song_records(pg_cur, BATCH_SIZE, platforms)
......
#!/usr/bin/env python3
"""用 hikoon-data-spider 平台原始标题修复 archive/crawler 标题字段。
数据链路：
archive_crawler.yinyan_song_records (is_archive_push = TRUE)
的 (platform, platform_song_id)
-> hikoon-data-spider 对应 media_*_songs.id -> title
-> title_norm.normalize_title(spider_title) -> (title_norm, version)
写入目标：
* archive_data.archive_recording：title/title_norm/version_type
* archive_data.archive_recording_platform_source：title/title_norm/version_type
* archive_crawler.crawler_*_songs：title/version
默认只执行 dry-run 统计，不写库；必须显式添加 ``--apply`` 才会提交。
"""
import argparse
import json
import logging
import os
from datetime import datetime, timezone
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, Sequence
from uuid import uuid4
from openpyxl import Workbook
from openpyxl.cell import WriteOnlyCell
from openpyxl.styles import Font, PatternFill
import title_norm as tn
from etl_to_crawler.connections import (
close_all_pools,
get_archive_conn,
get_archive_crawler_conn,
get_spider_conn,
)
log = logging.getLogger(__name__)
DEFAULT_BATCH_SIZE = 1000
DEFAULT_REPORT = "output/fix_record_titles_report.xlsx"
DEFAULT_BACKUP_DIR = Path("output/backups")
BACKUP_SCHEMA_VERSION = 1
MAX_VARCHAR_LEN = 50 # archive/crawler 表 title/title_norm/version_type 列最大长度
PLATFORMS = {
"1": {
"archive_name": "qqmusic",
"spider_table": "media_tencent_songs",
"crawler_table": "crawler_qqmusic_songs",
},
"2": {
"archive_name": "kugou",
"spider_table": "media_ku_gou_songs",
"crawler_table": "crawler_kugou_songs",
},
"4": {
"archive_name": "netease",
"spider_table": "media_netease_songs",
"crawler_table": "crawler_netease_songs",
},
}
@dataclass(frozen=True)
class InputRow:
record_id: int
platform: str
platform_song_id: int
recording_id: str
@dataclass(frozen=True)
class FixRow(InputRow):
spider_title: str
title_norm: str
version: str
def chunked(rows: Sequence, size: int) -> Iterable[Sequence]:
for start in range(0, len(rows), size):
yield rows[start:start + size]
def fetch_input_rows(crawler_conn, limit: int | None = None) -> tuple[list[InputRow], dict[str, int]]:
"""直接读取正式 yinyan_song_records 的全部 archive 推送关系。"""
sql = """
SELECT record_id, platform, platform_song_id, recording_id
FROM yinyan_song_records
WHERE is_archive_push = TRUE
ORDER BY record_id, platform, platform_song_id
"""
params = None
if limit is not None:
sql += " LIMIT %s"
params = (limit,)
with crawler_conn.cursor() as cursor:
cursor.execute(sql, params)
values_list = cursor.fetchall()
stats = {"total": 0, "missing_key": 0, "unsupported_platform": 0}
rows: list[InputRow] = []
seen_recordings: set[str] = set()
seen_platform_songs: set[tuple[str, int]] = set()
for values in values_list:
stats["total"] += 1
record_id, platform_value, platform_song_id, recording_id = values
platform = str(platform_value)
if platform not in PLATFORMS:
stats["unsupported_platform"] += 1
continue
if None in (platform_song_id, recording_id, record_id):
stats["missing_key"] += 1
continue
row = InputRow(
record_id=int(record_id),
platform=platform,
platform_song_id=int(platform_song_id),
recording_id=str(recording_id),
)
recording_key = row.recording_id
song_key = (row.platform, row.platform_song_id)
if recording_key in seen_recordings:
raise ValueError(f"Excel 中 recording_id 重复: {recording_key}")
if song_key in seen_platform_songs:
raise ValueError(f"Excel 中 platform_song_id 重复: {song_key}")
seen_recordings.add(recording_key)
seen_platform_songs.add(song_key)
rows.append(row)
return rows, stats
def fetch_spider_titles(spider_conn, rows: list[InputRow], batch_size: int) -> dict[tuple[str, int], str]:
"""按平台从 spider 歌曲表读取原始 title；表名严格来自白名单。"""
titles: dict[tuple[str, int], str] = {}
for platform, config in PLATFORMS.items():
ids = sorted({row.platform_song_id for row in rows if row.platform == platform})
table = config["spider_table"]
for batch in chunked(ids, batch_size):
placeholders = ", ".join(["%s"] * len(batch))
with spider_conn.cursor() as cursor:
cursor.execute(
f"SELECT id, title FROM {table} WHERE id IN ({placeholders})",
tuple(batch),
)
for result in cursor.fetchall():
title = result["title"]
if title is not None and str(title).strip() != "":
titles[(platform, int(result["id"]))] = str(title)
return titles
def build_fix_rows(
input_rows: list[InputRow],
spider_titles: dict[tuple[str, int], str],
long_only: bool = False,
exclude_titles: set[str] | None = None,
) -> tuple[list[FixRow], dict[str, int]]:
"""以 spider title 为唯一修复源，并用 title_norm 提取归一标题和版本。
默认模式下超过 ``MAX_VARCHAR_LEN`` 的记录会被跳过，避免写入时 varchar 溢出。
当 ``long_only=True`` 时反转过滤：仅保留超长记录，用于 DBA 扩列后补跑。
"""
fixes: list[FixRow] = []
exclude_titles = exclude_titles or set()
stats = {
"missing_spider": 0,
"excluded_title": 0,
"skip_long_title": 0,
"skip_long_title_norm": 0,
"skip_long_version": 0,
"skip_short": 0,
}
for row in input_rows:
spider_title = spider_titles.get((row.platform, row.platform_song_id))
if spider_title is None:
stats["missing_spider"] += 1
continue
if spider_title in exclude_titles:
stats["excluded_title"] += 1
continue
normalized_title, version = tn.normalize_title(spider_title)
is_long = (
len(spider_title) > MAX_VARCHAR_LEN
or len(normalized_title) > MAX_VARCHAR_LEN
or (version and len(version) > MAX_VARCHAR_LEN)
)
if long_only:
if not is_long:
stats["skip_short"] += 1
continue
else:
if is_long:
if len(spider_title) > MAX_VARCHAR_LEN:
stats["skip_long_title"] += 1
elif len(normalized_title) > MAX_VARCHAR_LEN:
stats["skip_long_title_norm"] += 1
else:
stats["skip_long_version"] += 1
continue
fixes.append(
FixRow(
**row.__dict__,
spider_title=spider_title,
title_norm=normalized_title,
version=version,
)
)
return fixes, stats
def _archive_recording_sql(batch_length: int, apply: bool, lock: bool = False) -> tuple[str, list[str]]:
placeholders = ", ".join(["(%s, %s, %s, %s)"] * batch_length)
values = "(recording_id, new_title, new_title_norm, new_version)"
params = ["recording_id", "spider_title", "title_norm", "archive_version"]
difference = """(
target.title IS DISTINCT FROM source.new_title
OR target.title_norm IS DISTINCT FROM source.new_title_norm
OR target.version_type IS DISTINCT FROM source.new_version
)"""
if apply:
sql = f"""
UPDATE archive_recording AS target
SET title = source.new_title,
title_norm = source.new_title_norm,
version_type = source.new_version,
updated_at = NOW()
FROM (VALUES {placeholders}) AS source{values}
WHERE target.recording_id = source.recording_id
AND {difference}
RETURNING target.recording_id
"""
else:
sql = f"""
SELECT target.recording_id,
target.title, source.new_title,
target.title_norm, source.new_title_norm,
target.version_type, source.new_version,
target.updated_at
FROM archive_recording AS target
JOIN (VALUES {placeholders}) AS source{values}
ON target.recording_id = source.recording_id
WHERE {difference}
{"FOR UPDATE OF target" if lock else ""}
"""
return sql, params
def _archive_source_sql(batch_length: int, apply: bool, lock: bool = False) -> tuple[str, list[str]]:
placeholders = ", ".join(["(%s, %s, %s, %s, %s, %s)"] * batch_length)
values = "(recording_id, platform, platform_song_id, new_title, new_title_norm, new_version)"
params = [
"recording_id", "archive_platform", "platform_song_id_text",
"spider_title", "title_norm", "archive_version",
]
difference = """(
target.title IS DISTINCT FROM source.new_title
OR target.title_norm IS DISTINCT FROM source.new_title_norm
OR target.version_type IS DISTINCT FROM source.new_version
)"""
join = """target.recording_id = source.recording_id
AND target.platform = source.platform
AND target.platform_song_id = source.platform_song_id"""
if apply:
sql = f"""
UPDATE archive_recording_platform_source AS target
SET title = source.new_title,
title_norm = source.new_title_norm,
version_type = source.new_version,
updated_at = NOW()
FROM (VALUES {placeholders}) AS source{values}
WHERE {join} AND {difference}
RETURNING target.recording_id, target.platform, target.platform_song_id
"""
else:
sql = f"""
SELECT target.recording_id, target.platform, target.platform_song_id,
target.title, source.new_title,
target.title_norm, source.new_title_norm,
target.version_type, source.new_version,
target.updated_at
FROM archive_recording_platform_source AS target
JOIN (VALUES {placeholders}) AS source{values} ON {join}
WHERE {difference}
{"FOR UPDATE OF target" if lock else ""}
"""
return sql, params
def _crawler_sql(table: str, batch_length: int, apply: bool, lock: bool = False) -> tuple[str, list[str]]:
placeholders = ", ".join(["(%s, %s, %s)"] * batch_length)
values = "(platform_song_id, new_title, new_version)"
params = ["platform_song_id", "spider_title", "version"]
difference = """(
target.title IS DISTINCT FROM source.new_title
OR target.version IS DISTINCT FROM source.new_version
)"""
if apply:
sql = f"""
UPDATE {table} AS target
SET title = source.new_title,
version = source.new_version,
updated_at = NOW()
FROM (VALUES {placeholders}) AS source{values}
WHERE target.platform_song_id = source.platform_song_id::bigint
AND {difference}
RETURNING target.platform_song_id
"""
else:
sql = f"""
SELECT target.platform_song_id,
target.title, source.new_title,
target.version, source.new_version,
target.updated_at
FROM {table} AS target
JOIN (VALUES {placeholders}) AS source{values}
ON target.platform_song_id = source.platform_song_id::bigint
WHERE {difference}
{"FOR UPDATE OF target" if lock else ""}
"""
return sql, params
def _row_value(row: FixRow, name: str):
values = {
"recording_id": row.recording_id,
"archive_platform": PLATFORMS[row.platform]["archive_name"],
"platform_song_id_text": str(row.platform_song_id),
"platform_song_id": row.platform_song_id,
"spider_title": row.spider_title,
"title_norm": row.title_norm,
"archive_version": row.version or None,
"version": row.version,
}
return values[name]
def execute_batches(conn, rows: list[FixRow], sql_builder, batch_size: int, apply: bool) -> list[tuple]:
if not rows:
return []
matched_keys: list[tuple] = []
for batch in chunked(rows, batch_size):
sql, param_names = sql_builder(len(batch), apply)
params = [_row_value(row, name) for row in batch for name in param_names]
with conn.cursor() as cursor:
cursor.execute(sql, tuple(params))
matched_keys.extend(tuple(result) for result in cursor.fetchall())
return matched_keys
def _changed_fields(pairs: tuple[tuple[str, object, object], ...]) -> str:
return ",".join(field for field, old, new in pairs if old != new)
def build_report_rows(fixes: list[FixRow], preview: dict[str, list[tuple]], apply: bool) -> list[tuple]:
"""把更新前快照和目标值转换成可审计的前后对比明细。"""
by_recording = {row.recording_id: row for row in fixes}
by_platform_song = {(row.platform, row.platform_song_id): row for row in fixes}
action = "updated" if apply else "would_update"
report_rows: list[tuple] = []
for result in preview.get("archive_recording", []):
row = by_recording.get(str(result[0]))
if row:
fields = _changed_fields((
("title", result[1], result[2]),
("title_norm", result[3], result[4]),
("version_type", result[5], result[6]),
))
report_rows.append((
"archive_data", "archive_recording", row.record_id, row.recording_id,
row.platform, row.platform_song_id, fields,
result[1], result[2], result[3], result[4], result[5], result[6], action,
))
for result in preview.get("archive_recording_platform_source", []):
row = by_recording.get(str(result[0]))
if row:
fields = _changed_fields((
("title", result[3], result[4]),
("title_norm", result[5], result[6]),
("version_type", result[7], result[8]),
))
report_rows.append((
"archive_data", "archive_recording_platform_source", row.record_id,
row.recording_id, row.platform, row.platform_song_id,
fields, result[3], result[4], result[5], result[6],
result[7], result[8], action,
))
for platform, config in PLATFORMS.items():
table = config["crawler_table"]
for result in preview.get(table, []):
row = by_platform_song.get((platform, int(result[0])))
if row:
fields = _changed_fields((
("title", result[1], result[2]),
("version", result[3], result[4]),
))
report_rows.append((
"archive_crawler", table, row.record_id, row.recording_id,
row.platform, row.platform_song_id, fields,
result[1], result[2], None, None, result[3], result[4], action,
))
return report_rows
def build_backup_records(
fixes: list[FixRow],
preview: dict[str, list[tuple]],
run_id: str,
) -> list[dict]:
"""将加锁读取的更新前快照转换成可机器回退的记录。"""
by_recording = {row.recording_id: row for row in fixes}
by_platform_song = {(row.platform, row.platform_song_id): row for row in fixes}
records: list[dict] = []
for result in preview.get("archive_recording", []):
row = by_recording[str(result[0])]
records.append({
"type": "change", "schema_version": BACKUP_SCHEMA_VERSION, "run_id": run_id,
"database": "archive_data", "table": "archive_recording",
"keys": {"recording_id": row.recording_id},
"old": {
"title": result[1], "title_norm": result[3],
"version_type": result[5], "updated_at": result[7],
},
"new": {
"title": result[2], "title_norm": result[4], "version_type": result[6],
},
})
for result in preview.get("archive_recording_platform_source", []):
records.append({
"type": "change", "schema_version": BACKUP_SCHEMA_VERSION, "run_id": run_id,
"database": "archive_data", "table": "archive_recording_platform_source",
"keys": {
"recording_id": str(result[0]), "platform": str(result[1]),
"platform_song_id": str(result[2]),
},
"old": {
"title": result[3], "title_norm": result[5],
"version_type": result[7], "updated_at": result[9],
},
"new": {
"title": result[4], "title_norm": result[6], "version_type": result[8],
},
})
for platform, config in PLATFORMS.items():
table = config["crawler_table"]
for result in preview.get(table, []):
row = by_platform_song[(platform, int(result[0]))]
records.append({
"type": "change", "schema_version": BACKUP_SCHEMA_VERSION, "run_id": run_id,
"database": "archive_crawler", "table": table,
"keys": {"platform_song_id": row.platform_song_id},
"old": {
"title": result[1], "version": result[3], "updated_at": result[5],
},
"new": {"title": result[2], "version": result[4]},
})
return records
def _json_default(value):
if isinstance(value, (datetime,)):
return value.isoformat()
raise TypeError(f"无法序列化 {type(value).__name__}")
def write_backup_atomic(path: Path, records: list[dict], run_id: str) -> None:
"""先 fsync 临时文件再无覆盖发布，保证 UPDATE 前备份已持久化。"""
path.parent.mkdir(parents=True, exist_ok=True)
if path.exists():
raise FileExistsError(f"拒绝覆盖已有回退备份: {path}")
temp_path = path.with_name(path.name + f".{uuid4().hex}.tmp")
metadata = {
"type": "metadata",
"schema_version": BACKUP_SCHEMA_VERSION,
"run_id": run_id,
"created_at": datetime.now(timezone.utc).isoformat(),
"change_count": len(records),
}
with temp_path.open("w", encoding="utf-8") as handle:
handle.write(json.dumps(metadata, ensure_ascii=False) + "\n")
for record in records:
handle.write(json.dumps(record, ensure_ascii=False, default=_json_default) + "\n")
handle.flush()
os.fsync(handle.fileno())
try:
# 硬链接发布不会覆盖已存在目标；并发使用同一路径时也会安全失败。
os.link(temp_path, path)
directory_fd = os.open(path.parent, os.O_RDONLY)
try:
os.fsync(directory_fd)
finally:
os.close(directory_fd)
finally:
temp_path.unlink(missing_ok=True)
def load_backup(path: Path) -> tuple[dict, list[dict]]:
with path.open("r", encoding="utf-8") as handle:
lines = [json.loads(line) for line in handle if line.strip()]
if not lines or lines[0].get("type") != "metadata":
raise ValueError("无效的回退备份：缺少 metadata")
metadata = lines[0]
records = lines[1:]
if metadata.get("schema_version") != BACKUP_SCHEMA_VERSION:
raise ValueError("不支持的备份版本")
if metadata.get("change_count") != len(records):
raise ValueError("备份记录数校验失败")
return metadata, records
def _rollback_sql(table: str, batch_length: int) -> tuple[str, list[str]]:
"""生成带新值保护的批量回退 SQL；表名只能来自内部白名单。"""
if table == "archive_recording":
placeholders = ", ".join(["(%s, %s, %s, %s, %s, %s, %s, %s)"] * batch_length)
sql = f"""
UPDATE archive_recording AS target
SET title = source.old_title,
title_norm = source.old_title_norm,
version_type = source.old_version,
updated_at = source.old_updated_at::timestamptz
FROM (VALUES {placeholders}) AS source(
recording_id, old_title, old_title_norm, old_version, old_updated_at,
new_title, new_title_norm, new_version
)
WHERE target.recording_id = source.recording_id
AND target.title IS NOT DISTINCT FROM source.new_title
AND target.title_norm IS NOT DISTINCT FROM source.new_title_norm
AND target.version_type IS NOT DISTINCT FROM source.new_version
RETURNING target.recording_id
"""
names = [
"recording_id", "old_title", "old_title_norm", "old_version", "old_updated_at",
"new_title", "new_title_norm", "new_version",
]
elif table == "archive_recording_platform_source":
placeholders = ", ".join(["(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"] * batch_length)
sql = f"""
UPDATE archive_recording_platform_source AS target
SET title = source.old_title,
title_norm = source.old_title_norm,
version_type = source.old_version,
updated_at = source.old_updated_at::timestamptz
FROM (VALUES {placeholders}) AS source(
recording_id, platform, platform_song_id,
old_title, old_title_norm, old_version, old_updated_at,
new_title, new_title_norm, new_version
)
WHERE target.recording_id = source.recording_id
AND target.platform = source.platform
AND target.platform_song_id = source.platform_song_id
AND target.title IS NOT DISTINCT FROM source.new_title
AND target.title_norm IS NOT DISTINCT FROM source.new_title_norm
AND target.version_type IS NOT DISTINCT FROM source.new_version
RETURNING target.recording_id, target.platform, target.platform_song_id
"""
names = [
"recording_id", "platform", "platform_song_id",
"old_title", "old_title_norm", "old_version", "old_updated_at",
"new_title", "new_title_norm", "new_version",
]
elif table in {config["crawler_table"] for config in PLATFORMS.values()}:
placeholders = ", ".join(["(%s, %s, %s, %s, %s, %s)"] * batch_length)
sql = f"""
UPDATE {table} AS target
SET title = source.old_title,
version = source.old_version,
updated_at = source.old_updated_at::timestamp
FROM (VALUES {placeholders}) AS source(
platform_song_id, old_title, old_version, old_updated_at,
new_title, new_version
)
WHERE target.platform_song_id = source.platform_song_id::bigint
AND target.title IS NOT DISTINCT FROM source.new_title
AND target.version IS NOT DISTINCT FROM source.new_version
RETURNING target.platform_song_id
"""
names = [
"platform_song_id", "old_title", "old_version", "old_updated_at",
"new_title", "new_version",
]
else:
raise ValueError(f"备份包含不允许回退的表: {table}")
return sql, names
def _backup_value(record: dict, name: str):
if name in record["keys"]:
return record["keys"][name]
if name.startswith("old_"):
field = name[4:]
if field == "version":
field = "version_type" if "version_type" in record["old"] else "version"
return record["old"].get(field)
if name.startswith("new_"):
field = name[4:]
if field == "version":
field = "version_type" if "version_type" in record["new"] else "version"
return record["new"].get(field)
raise KeyError(name)
def execute_rollback_batches(conn, table: str, records: list[dict], batch_size: int) -> list[tuple]:
restored: list[tuple] = []
for batch in chunked(records, batch_size):
sql, names = _rollback_sql(table, len(batch))
params = [_backup_value(record, name) for record in batch for name in names]
with conn.cursor() as cursor:
cursor.execute(sql, tuple(params))
restored.extend(tuple(result) for result in cursor.fetchall())
return restored
def rollback_backup(path: Path, batch_size: int) -> dict[str, int]:
"""按备份回退；当前值不再等于本次新值的记录会安全跳过。"""
metadata, records = load_backup(path)
groups: dict[tuple[str, str], list[dict]] = {}
for record in records:
if record.get("type") != "change" or record.get("run_id") != metadata.get("run_id"):
raise ValueError("备份记录 run_id 或类型无效")
key = (record.get("database"), record.get("table"))
groups.setdefault(key, []).append(record)
archive_conn = get_archive_conn()
crawler_conn = get_archive_crawler_conn()
restored: dict[str, int] = {}
try:
for (database, table), table_records in groups.items():
if database == "archive_data":
conn = archive_conn
elif database == "archive_crawler":
conn = crawler_conn
else:
raise ValueError(f"备份包含不允许回退的数据库: {database}")
keys = execute_rollback_batches(conn, table, table_records, batch_size)
restored[table] = len(keys)
archive_conn.commit()
crawler_conn.commit()
except Exception:
archive_conn.rollback()
crawler_conn.rollback()
raise
finally:
close_all_pools()
log.info("已按 run_id=%s 执行回退", metadata["run_id"])
for table, count in restored.items():
log.info("%s：已回退 %d 行", table, count)
return restored
def write_report(report_path: Path, rows: list[tuple]) -> None:
report_path.parent.mkdir(parents=True, exist_ok=True)
workbook = Workbook(write_only=True)
worksheet = workbook.create_sheet()
worksheet.title = "更新明细"
worksheet.freeze_panes = "A2"
headers = [
"database", "table", "record_id", "recording_id", "platform",
"platform_song_id", "changed_fields", "old_title", "new_title",
"old_title_norm", "new_title_norm", "old_version", "new_version", "action",
]
header_fill = PatternFill(fill_type="solid", fgColor="4472C4")
header_font = Font(color="FFFFFF", bold=True)
old_fill = PatternFill(fill_type="solid", fgColor="FFC7CE")
new_fill = PatternFill(fill_type="solid", fgColor="C6EFCE")
header_cells = []
for header in headers:
cell = WriteOnlyCell(worksheet, value=header)
cell.fill = header_fill
cell.font = header_font
header_cells.append(cell)
worksheet.append(header_cells)
for values in rows:
output_cells = []
changed_fields = set(str(values[6]).split(","))
changed_columns = set()
if "title" in changed_fields:
changed_columns.update((8, 9))
if "title_norm" in changed_fields:
changed_columns.update((10, 11))
if "version" in changed_fields or "version_type" in changed_fields:
changed_columns.update((12, 13))
for column, value in enumerate(values, 1):
if value is None:
value = "<NULL>"
elif value == "":
value = "<EMPTY>"
cell = WriteOnlyCell(worksheet, value=value)
if column in changed_columns:
cell.fill = old_fill if column in (8, 10, 12) else new_fill
output_cells.append(cell)
worksheet.append(output_cells)
widths = (18, 42, 15, 24, 10, 20, 34, 48, 48, 40, 40, 24, 24, 16)
for column, width in enumerate(widths, 1):
worksheet.column_dimensions[chr(64 + column)].width = width
worksheet.auto_filter.ref = f"A1:N{len(rows) + 1}"
workbook.save(report_path)
def run(
batch_size: int,
apply: bool,
limit: int | None,
report_path: Path,
backup_path: Path | None = None,
long_only: bool = False,
exclude_titles: set[str] | None = None,
) -> dict[str, int]:
crawler_conn = get_archive_crawler_conn()
input_rows, input_stats = fetch_input_rows(crawler_conn, limit)
spider_conn = get_spider_conn()
spider_titles = fetch_spider_titles(spider_conn, input_rows, batch_size)
fixes, fix_stats = build_fix_rows(
input_rows,
spider_titles,
long_only=long_only,
exclude_titles=exclude_titles,
)
skip_long = fix_stats["skip_long_title"] + fix_stats["skip_long_title_norm"] + fix_stats["skip_long_version"]
if long_only:
log.info(
"[long-only] yinyan archive记录=%d，有效键=%d，超长候选=%d，短标题跳过=%d，spider 缺失=%d",
input_stats["total"], len(input_rows), len(fixes), fix_stats["skip_short"], fix_stats["missing_spider"],
)
else:
log.info(
"yinyan archive记录=%d，有效键=%d，spider 标题=%d，spider 缺失=%d，超长跳过=%d (title=%d, title_norm=%d, version=%d)",
input_stats["total"], len(input_rows), len(fixes), fix_stats["missing_spider"],
skip_long, fix_stats["skip_long_title"], fix_stats["skip_long_title_norm"], fix_stats["skip_long_version"],
)
if fix_stats["excluded_title"]:
log.info("本次按 spider title 精确排除=%d 条，不进入预览、备份或更新", fix_stats["excluded_title"])
archive_conn = get_archive_conn()
preview: dict[str, list[tuple]] = {}
actual: dict[str, list[tuple]] = {}
run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "_" + uuid4().hex[:8]
try:
preview["archive_recording"] = execute_batches(
archive_conn, fixes,
lambda length, _: _archive_recording_sql(length, False, lock=apply),
batch_size, False,
)
preview["archive_recording_platform_source"] = execute_batches(
archive_conn, fixes,
lambda length, _: _archive_source_sql(length, False, lock=apply),
batch_size, False,
)
for platform, config in PLATFORMS.items():
table = config["crawler_table"]
platform_rows = [row for row in fixes if row.platform == platform]
preview[table] = execute_batches(
crawler_conn,
platform_rows,
lambda length, _, table=table: _crawler_sql(table, length, False, lock=apply),
batch_size,
False,
)
if apply:
backup_records = build_backup_records(fixes, preview, run_id)
if backup_path is None:
backup_path = DEFAULT_BACKUP_DIR / f"fix_record_titles_{run_id}.jsonl"
backup_path = backup_path.expanduser().resolve()
write_backup_atomic(backup_path, backup_records, run_id)
log.info("回退备份已持久化至 %s，共 %d 条", backup_path, len(backup_records))
actual["archive_recording"] = execute_batches(
archive_conn, fixes, _archive_recording_sql, batch_size, True,
)
actual["archive_recording_platform_source"] = execute_batches(
archive_conn, fixes, _archive_source_sql, batch_size, True,
)
for platform, config in PLATFORMS.items():
table = config["crawler_table"]
platform_rows = [row for row in fixes if row.platform == platform]
actual[table] = execute_batches(
crawler_conn,
platform_rows,
lambda length, do_apply, table=table: _crawler_sql(table, length, do_apply),
batch_size,
True,
)
archive_conn.commit()
crawler_conn.commit()
else:
actual = preview
archive_conn.rollback()
crawler_conn.rollback()
except Exception:
archive_conn.rollback()
crawler_conn.rollback()
raise
finally:
close_all_pools()
report_rows = build_report_rows(fixes, preview, apply)
write_report(report_path, report_rows)
results = {table: len(keys) for table, keys in actual.items()}
for table, count in results.items():
log.info("%s：%s %d 行", table, "已更新" if apply else "需更新", count)
if not apply:
log.info("当前为 dry-run，未修改数据库；确认后添加 --apply")
log.info("更新明细已输出至 %s，共 %d 行", report_path, len(report_rows))
return results
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="按 spider 原始标题修复 archive/crawler 标题")
parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE)
parser.add_argument("--limit", type=int, help="最多处理多少条 yinyan 关系（默认全部）")
parser.add_argument("--report", default=DEFAULT_REPORT, help=f"更新明细 Excel（默认: {DEFAULT_REPORT}）")
parser.add_argument("--backup", help="--apply 前写入的回退 JSONL 路径（默认自动生成）")
parser.add_argument("--long-only", action="store_true", help="仅处理超长标题记录（DBA 扩列后补跑）")
parser.add_argument(
"--exclude-title",
action="append",
default=[],
help="本次运行不更新的 spider 完整标题（精确匹配，可重复传入）",
)
mode = parser.add_mutually_exclusive_group()
mode.add_argument("--apply", action="store_true", help="备份成功后实际提交更新；默认仅 dry-run")
mode.add_argument("--rollback", help="使用指定 JSONL 备份安全回退")
args = parser.parse_args()
if args.batch_size <= 0:
parser.error("--batch-size 必须大于 0")
if args.limit is not None and args.limit <= 0:
parser.error("--limit 必须大于 0")
if args.rollback and (args.limit is not None or args.backup or args.exclude_title):
parser.error("--rollback 不能与 --limit、--backup 或 --exclude-title 同时使用")
return args
def main() -> None:
args = parse_args()
if args.rollback:
rollback_backup(Path(args.rollback).expanduser().resolve(), args.batch_size)
return
run(
args.batch_size,
args.apply,
args.limit,
Path(args.report).expanduser().resolve(),
Path(args.backup) if args.backup else None,
long_only=args.long_only,
exclude_titles=set(args.exclude_title),
)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
main()
......@@ -15,7 +15,7 @@ import argparse
import logging
from etl_to_crawler.config import PLATFORM_QQ, PLATFORM_KUGOU, PLATFORM_NETEASE
from etl_to_crawler.connections import get_source_conn, get_pg_conn, close_all_pools
from etl_to_crawler.connections import get_source_conn, get_test_crawler_conn, close_all_pools
from etl_to_crawler.utils import split_title_version
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
......@@ -134,7 +134,7 @@ def main():
args = parser.parse_args()
mysql_conn = get_source_conn()
pg_conn = get_pg_conn()
pg_conn = get_test_crawler_conn()
total_updated = 0
total_diff = 0
......
#!/usr/bin/env python3
"""生成港乐词曲资产数据缺口清单 Excel"""
import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.utils import get_column_letter
wb = openpyxl.Workbook()
# ── 样式定义 ──────────────────────────────────────────
title_font = Font(name="Microsoft YaHei", size=12, bold=True, color="FFFFFFFF")
title_fill = PatternFill(start_color="FF4472C4", end_color="FF4472C4", fill_type="solid")
title_align = Alignment(horizontal="center", vertical="center", wrap_text=True)
header_font = Font(name="宋体", size=10.5, bold=True)
header_fill = PatternFill(start_color="FFD9E2F3", end_color="FFD9E2F3", fill_type="solid")
header_align = Alignment(horizontal="center", vertical="center", wrap_text=True)
data_font = Font(name="Microsoft YaHei", size=10)
data_align = Alignment(vertical="top", wrap_text=True)
center_align = Alignment(horizontal="center", vertical="top", wrap_text=True)
# 处理方式颜色
FILL_RED = PatternFill(start_color="FFFFC7CE", end_color="FFFFC7CE", fill_type="solid") # 🔴 人工
FILL_GREEN = PatternFill(start_color="FFC6EFCE", end_color="FFC6EFCE", fill_type="solid") # 🟢 爬虫
FILL_YELLOW = PatternFill(start_color="FFFFEB9C", end_color="FFFFEB9C", fill_type="solid") # 🟡 待扩展
FILL_GRAY = PatternFill(start_color="FFD9D9D9", end_color="FFD9D9D9", fill_type="solid") # ⚪ 暂不可操作
thin_border = Border(
left=Side(style="thin"), right=Side(style="thin"),
top=Side(style="thin"), bottom=Side(style="thin"),
)
# ── Sheet 1: 数据缺口清单 ─────────────────────────────
ws1 = wb.active
ws1.title = "数据缺口清单"
# 标题行
ws1.merge_cells("A1:G1")
c = ws1["A1"]
c.value = "港乐词曲资产导入 — 数据缺口清单（2026-07-14）"
c.font = title_font; c.fill = title_fill; c.alignment = title_align
# 表头
headers = ["序号", "缺口类型", "数量", "处理方式", "涉及数据", "补齐说明", "优先级"]
for i, h in enumerate(headers, 1):
cell = ws1.cell(row=2, column=i, value=h)
cell.font = header_font; cell.fill = header_fill
cell.alignment = header_align; cell.border = thin_border
# 数据
rows = [
# (序号, 缺口类型, 数量, 处理方式, 涉及数据, 补齐说明, 优先级, fill)
(1, "疑似重复，待人工审核", "4,410 首",
"🔴 人工处理", "去重暂存表中的候选记录",
"系统判定与已有歌曲高度相似，需人工逐条判断是同一首歌（合并）还是不同歌曲（新增导入）。已部署 L2 审核看板支持批量操作。",
"P2", FILL_RED),
(2, "歌词为空且词曲作者均不详", "1,167 首",
"🔴 人工处理", "源库中歌词和作者字段同时缺失的歌曲",
"源数据本身缺失，爬虫无法生成不存在的元数据。如需导入，需人工查找并补录歌词和作者信息后重新触发导入。可先抽样检查这批歌曲在平台上是否实际有歌词。",
"P3", FILL_RED),
(3, "主录音导入失败——缺歌手信息", "96 条",
"🟢 爬虫补录", "records1 中因 spider DB 缺歌手关系而失败的记录",
"平台上有歌手信息但尚未同步到本地爬虫数据库。补录歌手关系后系统自动重试。也可调整策略允许空歌手容错（补充录音已支持，主录音暂未支持）。",
"P1", FILL_GREEN),
(4, "主录音导入失败——音频/歌词转存失败", "19 条",
"🟢 自动重试", "records1 中因网络或存储问题转存失败的记录（音频 10 条 + 歌词 9 条）",
"属于临时性故障，源文件仍在平台上。重新触发转存即可，无需重新爬取。",
"P1", FILL_GREEN),
(5, "主录音导入失败——完全无歌词", "3 条",
"🟢 爬虫尝试", "records1 中三个平台均无歌词数据的记录",
"可尝试爬虫从其他来源抓取歌词；如平台确实无歌词，需人工确认是否为纯音乐作品。数量极少，影响可忽略。",
"P2", FILL_GREEN),
(6, "歌曲仅有非目标平台录音", "809 首",
"🟡 扩展平台", "在三平台（QQ/酷狗/网易云）均无有效录音，仅在其他平台有录音的歌曲",
"当前爬虫仅覆盖三个目标平台。扩展平台覆盖后可自动获取录音数据。需业务确认：是否接受这些歌曲暂不入库？",
"P3", FILL_YELLOW),
(7, "已软删歌曲——录音已成功但歌曲不可见", "25 首",
"🔴 人工恢复", "records1 已成功但歌曲状态仍为 deleted 的记录",
"历史遗留的状态不一致，主录音数据已入库但歌曲被标记为已删除导致前端不可见。确认后批量恢复即可，无需重新爬取。",
"P0", FILL_RED),
(8, "已软删歌曲——现具备三平台录音条件", "71 首",
"🔴 人工恢复", "之前因审计/清理被软删除，但当前已具备有效录音条件的歌曲",
"数据已补齐，只需逐条确认删除原因是否已过时，符合条件后恢复歌曲状态。",
"P2", FILL_RED),
(9, "已软删歌曲——当前仍无有效三平台录音", "133 首",
"⚪ 暂不处理", "被软删除且当前仍无法获取三平台录音的歌曲（含 ETL 清理 143 中剩余部分 + song_time=0 + 元数据重复合并）",
"当前无法补齐录音数据。若后续扩展平台覆盖或有其他数据来源，可重新评估。",
"P3", FILL_GRAY),
(10, "原声类歌名", "79 首",
"🔴 人工处理", "歌名形如 '@XXX创作的原声' 或 '用户创作的原声' 的特殊歌曲",
"歌名为平台自动生成的标记文本，通常不具备导入价值。其中 78 首同时满足歌词/作者缺失条件被前置规则先拦截，仅 1 首 dedup_reason 显式标注为原声类。建议直接跳过。",
"P3", FILL_RED),
]
for r_idx, (no, gtype, count, method, scope, desc, priority, fill) in enumerate(rows, 3):
vals = [no, gtype, count, method, scope, desc, priority]
for c_idx, v in enumerate(vals, 1):
cell = ws1.cell(row=r_idx, column=c_idx, value=v)
cell.font = data_font
cell.alignment = center_align if c_idx in (1, 3, 4, 7) else data_align
cell.border = thin_border
# 处理方式列上色
if c_idx == 4:
cell.fill = fill
# 列宽
col_widths = [6, 30, 10, 14, 30, 55, 6]
for i, w in enumerate(col_widths, 1):
ws1.column_dimensions[get_column_letter(i)].width = w
# ── Sheet 2: 总览 ────────────────────────────────────
ws2 = wb.create_sheet("总览")
ws2.merge_cells("A1:C1")
c = ws2["A1"]
c.value = "港乐词曲资产导入 — 缺口处理总览"
c.font = title_font; c.fill = title_fill; c.alignment = title_align
for i, h in enumerate(["处理方式", "涉及数量", "占比"], 1):
cell = ws2.cell(row=2, column=i, value=h)
cell.font = header_font; cell.fill = header_fill
cell.alignment = header_align; cell.border = thin_border
summary = [
("🔴 需人工处理（审核/恢复/补录）", "5,752 首", "84.1%", FILL_RED),
("🟢 可通过爬虫/自动手段解决", "118 条", "1.7%", FILL_GREEN),
("🟡 需扩展平台覆盖后自动导入", "809 首", "11.9%", FILL_YELLOW),
("⚪ 暂不可操作，等待外部条件变化", "133 首", "2.0%", FILL_GRAY),
("缺口合计", "约 6,812 首/条", "—", None),
]
for r_idx, (cat, cnt, pct, fill) in enumerate(summary, 3):
for c_idx, v in enumerate([cat, cnt, pct], 1):
cell = ws2.cell(row=r_idx, column=c_idx, value=v)
cell.font = data_font if r_idx < 7 else Font(name="Microsoft YaHei", size=10, bold=True)
cell.alignment = center_align if c_idx >= 2 else Alignment(vertical="top", wrap_text=True)
cell.border = thin_border
if fill and c_idx == 1:
cell.fill = fill
ws2.column_dimensions["A"].width = 36
ws2.column_dimensions["B"].width = 18
ws2.column_dimensions["C"].width = 10
# ── Sheet 3: 优先级排序 ──────────────────────────────
ws3 = wb.create_sheet("优先级排序")
ws3.merge_cells("A1:E1")
c = ws3["A1"]
c.value = "港乐词曲资产导入 — 补齐优先级排序"
c.font = title_font; c.fill = title_fill; c.alignment = title_align
for i, h in enumerate(["优先级", "处理方式", "涉及数量", "预期效果", "备注"], 1):
cell = ws3.cell(row=2, column=i, value=h)
cell.font = header_font; cell.fill = header_fill
cell.alignment = header_align; cell.border = thin_border
prio_rows = [
("P0", "恢复 25 首状态不一致的歌曲", "25 首", "立即恢复前端可见，零成本", "仅需修改 deleted 字段"),
("P1", "调整策略后自动重试 96 条缺歌手记录", "96 条", "主录音成功率 99.86% → 99.97%", "需开发评估空歌手容错策略"),
("P1", "重试 19 条转存失败记录", "19 条", "自动完成，无额外开发", "临时性故障，重跑即可"),
("P2", "人工审核 4,410 首疑似重复歌曲", "4,410 首", "释放暂存积压，可能新增歌曲或确认合并", "已部署审核看板"),
("P2", "确认恢复 71 首已删除但已有录音的歌曲", "71 首", "增加有效歌曲数", "逐条确认删除原因"),
("P2", "尝试爬取 3 条无歌词记录的歌词", "3 条", "数量极少，可忽略", "如确为纯音乐则标记跳过"),
("P3", "扩展爬虫平台覆盖后自动导入", "809 首", "需平台扩展开发", "依赖技术规划"),
("P3", "人工补录 1,245+1 首歌词/作者后重新导入", "1,246 首", "工作量大，建议先抽样评估价值", "优先检查是否有实际歌词"),
("P3", "等待外部条件后评估 133 首已删歌曲", "133 首", "视平台扩展情况", "当前无可用数据"),
]
for r_idx, (pri, action, cnt, effect, note) in enumerate(prio_rows, 3):
vals = [pri, action, cnt, effect, note]
for c_idx, v in enumerate(vals, 1):
cell = ws3.cell(row=r_idx, column=c_idx, value=v)
cell.font = data_font
cell.alignment = center_align if c_idx in (1, 3) else data_align
cell.border = thin_border
# 优先级上色
if c_idx == 1:
if "P0" in str(v):
cell.fill = PatternFill(start_color="FFFF6B6B", end_color="FFFF6B6B", fill_type="solid")
cell.font = Font(name="Microsoft YaHei", size=10, bold=True, color="FFFFFFFF")
elif "P1" in str(v):
cell.fill = PatternFill(start_color="FFFFD93D", end_color="FFFFD93D", fill_type="solid")
elif "P2" in str(v):
cell.fill = PatternFill(start_color="FF6BCB77", end_color="FF6BCB77", fill_type="solid")
cell.font = Font(name="Microsoft YaHei", size=10, color="FFFFFFFF")
elif "P3" in str(v):
cell.fill = PatternFill(start_color="FF4D96FF", end_color="FF4D96FF", fill_type="solid")
cell.font = Font(name="Microsoft YaHei", size=10, color="FFFFFFFF")
ws3.column_dimensions["A"].width = 8
ws3.column_dimensions["B"].width = 38
ws3.column_dimensions["C"].width = 12
ws3.column_dimensions["D"].width = 38
ws3.column_dimensions["E"].width = 30
# ── Sheet 4: 字段级缺口 ──────────────────────────────
ws4 = wb.create_sheet("字段级缺口")
ws4.merge_cells("A1:G1")
c = ws4["A1"]
c.value = "港乐词曲资产导入 — 字段级缺口（已入库但关键字段为空）"
c.font = title_font; c.fill = title_fill; c.alignment = title_align
for i, h in enumerate(["序号", "缺口类型", "为空数量", "所在表/范围", "平台分布", "补齐方式", "优先级"], 1):
cell = ws4.cell(row=2, column=i, value=h)
cell.font = header_font; cell.fill = header_fill
cell.alignment = header_align; cell.border = thin_border
field_rows = [
(1, "歌曲封面图片缺失", "1,960 首",
"hk_songs_test（86,870 首有效歌曲）", "全平台",
"🟢 爬虫可回补：重新触发封面下载/转存；如平台本身无封面则需人工提供",
"P1", FILL_GREEN),
(2, "曲作者字段为空", "3,232 首",
"hk_songs_test（86,870 首有效歌曲）", "全平台",
"🟢 爬虫可回补：从爬虫数据库 composer_name 批量回写；爬虫也无数据则需人工查找",
"P1", FILL_GREEN),
(3, "词作者字段为空", "1,488 首",
"hk_songs_test（86,870 首有效歌曲）", "全平台",
"🟢 爬虫可回补：从爬虫数据库 lyricist_name 批量回写；爬虫也无数据则需人工查找",
"P1", FILL_GREEN),
(4, "歌手信息为空（singers=[]）", "393 条",
"爬虫库 songs 表（25,369 首已导入歌曲）",
"QQ: 373 / 酷狗: 5 / 网易云: 15",
"🟢 爬虫可回补：系统已内置 backfill-empty-singers 功能，从 spider DB 补录歌手关系",
"P1", FILL_GREEN),
(5, "歌手头像缺失", "4,057 人",
"爬虫库 singers 表（13,142 位歌手）",
"QQ: 3,039 / 酷狗: 549 / 网易云: 469",
"🟢 爬虫可回补：重新爬取歌手页面获取最新头像 URL 并转存 OSS",
"P1", FILL_GREEN),
(6, "歌词文本为空", "366 条",
"爬虫库 songs 表（25,369 首已导入歌曲）",
"QQ: 201 / 酷狗: 157 / 网易云: 8",
"🟢 爬虫可回补：重新爬取歌词页面；如平台确实无歌词则为纯音乐，无需补齐",
"P2", FILL_GREEN),
(7, "歌曲封面为空（爬虫侧）", "139 条",
"爬虫库 songs 表（25,369 首已导入歌曲）",
"QQ: 18 / 酷狗: 58 / 网易云: 63",
"🟢 爬虫可回补：重新触发封面下载/转存",
"P1", FILL_GREEN),
(8, "专辑封面为空", "82 个",
"爬虫库 albums 表（21,081 个专辑）",
"QQ: 0 / 酷狗: 21 / 网易云: 61",
"🟢 爬虫可回补：重新触发专辑封面下载/转存",
"P2", FILL_GREEN),
]
for r_idx, (no, gtype, count, scope, dist, method, priority, fill) in enumerate(field_rows, 3):
vals = [no, gtype, count, scope, dist, method, priority]
for c_idx, v in enumerate(vals, 1):
cell = ws4.cell(row=r_idx, column=c_idx, value=v)
cell.font = data_font
cell.alignment = center_align if c_idx in (1, 3, 7) else data_align
cell.border = thin_border
if c_idx == 6:
cell.fill = fill
# 汇总行
sum_row = 3 + len(field_rows)
ws4.merge_cells(f"A{sum_row}:B{sum_row}")
ws4.cell(row=sum_row, column=1, value="字段级缺口合计").font = Font(name="Microsoft YaHei", size=10, bold=True)
ws4.cell(row=sum_row, column=3, value="约 11,578").font = Font(name="Microsoft YaHei", size=10, bold=True)
ws4.cell(row=sum_row, column=6, value="绝大部分可通过爬虫回补").font = Font(name="Microsoft YaHei", size=10, bold=True)
for c_idx in range(1, 8):
ws4.cell(row=sum_row, column=c_idx).border = thin_border
ws4.column_dimensions["A"].width = 6
ws4.column_dimensions["B"].width = 26
ws4.column_dimensions["C"].width = 12
ws4.column_dimensions["D"].width = 32
ws4.column_dimensions["E"].width = 28
ws4.column_dimensions["F"].width = 52
ws4.column_dimensions["G"].width = 6
# ── 更新 Sheet 3 优先级排序：加入字段级缺口 ──────────────
# 在现有 prio_rows 后面插入字段级缺口的优先级行
field_prio_rows = [
("P1", "爬虫回补 1,960 首封面图片", "1,960 首", "前端展示效果显著改善", "重新触发封面下载/转存"),
("P1", "爬虫回补 4,057 位歌手头像", "4,057 人", "歌手页面展示完整", "重新爬取歌手页面"),
("P1", "爬虫回补词曲作者（3,232 + 1,488 首）", "4,720 首", "元数据完整性提升", "从爬虫库批量回写"),
("P1", "爬虫回补 393 条歌手信息", "393 条", "歌手关联完整性提升", "backfill-empty-singers"),
("P2", "爬虫回补 366 条歌词文本", "366 条", "歌词展示完整", "重新爬取歌词页面"),
("P2", "爬虫回补 139 条歌曲封面 + 82 个专辑封面", "221 条", "爬虫侧数据完整性提升", "重新触发转存"),
]
insert_row = 3 + len(prio_rows)
for i, (pri, action, cnt, effect, note) in enumerate(field_prio_rows):
r = insert_row + i
vals = [pri, action, cnt, effect, note]
for c_idx, v in enumerate(vals, 1):
cell = ws3.cell(row=r, column=c_idx, value=v)
cell.font = data_font
cell.alignment = center_align if c_idx in (1, 3) else data_align
cell.border = thin_border
if c_idx == 1:
if "P1" in str(v):
cell.fill = PatternFill(start_color="FFFFD93D", end_color="FFFFD93D", fill_type="solid")
elif "P2" in str(v):
cell.fill = PatternFill(start_color="FF6BCB77", end_color="FF6BCB77", fill_type="solid")
cell.font = Font(name="Microsoft YaHei", size=10, color="FFFFFFFF")
# ── 更新 Sheet 2 总览：加入字段级缺口分类 ──────────────
# 先取消已有的合并单元格，再插入新行
for merged in list(ws2.merged_cells.ranges):
ws2.unmerge_cells(str(merged))
# 在现有 summary 行后插入字段级缺口行
field_summary = [
("", "", "", None), # 空行分隔
("— 字段级缺口 —", "", "", None),
("🟢 爬虫可回补（封面/歌手/作者/头像/歌词）", "约 11,578 条", "—", FILL_GREEN),
(" 其中：歌曲封面图片", "1,960 首", "—", None),
(" 其中：词曲作者", "4,720 首", "—", None),
(" 其中：歌手信息", "393 条", "—", None),
(" 其中：歌手头像", "4,057 人", "—", None),
(" 其中：歌词文本", "366 条", "—", None),
(" 其中：歌曲/专辑封面（爬虫侧）", "221 条", "—", None),
]
insert_row2 = 3 + len(summary)
for i, (cat, cnt, pct, fill) in enumerate(field_summary):
r = insert_row2 + i
for c_idx, v in enumerate([cat, cnt, pct], 1):
cell = ws2.cell(row=r, column=c_idx, value=v)
cell.font = data_font
cell.alignment = center_align if c_idx >= 2 else Alignment(vertical="top", wrap_text=True)
cell.border = thin_border
if fill and c_idx == 1:
cell.fill = fill
if cat.startswith("—"):
cell.font = Font(name="Microsoft YaHei", size=10, bold=True)
# 重新设置说明区域
notes_start = insert_row2 + len(field_summary) + 1
ws2.merge_cells(f"A{notes_start}:C{notes_start}")
ws2.cell(row=notes_start, column=1, value="📋 处理方式说明：").font = Font(name="Microsoft YaHei", size=10, bold=True)
notes = [
"1. 🔴 人工处理：需业务/版权人员人工审核、恢复或补录源数据，系统仅辅助呈现和标记",
"2. 🟢 爬虫/自动：可通过爬虫补录歌手关系、重试转存失败、或重新触发导入流程自动完成",
"3. 🟡 扩展平台：当前爬虫仅覆盖 QQ/酷狗/网易云三平台，扩展覆盖范围后可自动导入",
"4. ⚪ 暂不处理：当前无法获取录音数据，等待平台扩展或外部数据源接入后再评估",
]
for i, note in enumerate(notes):
cell = ws2.cell(row=notes_start + 1 + i, column=1, value=note)
cell.font = Font(name="Microsoft YaHei", size=9, color="FF666666")
ws2.merge_cells(f"A{notes_start + 1 + i}:C{notes_start + 1 + i}")
# ── Sheet 5: 已同步数据（records1）─────────────────────
ws5 = wb.create_sheet("已同步数据（records1）")
ws5.merge_cells("A1:G1")
c = ws5["A1"]
c.value = "港乐词曲资产导入 — 已同步数据总览（records1 主录音）"
c.font = title_font; c.fill = title_fill; c.alignment = title_align
# ── 5.1 平台分布与导入状态 ──
r = 3
ws5.cell(row=r, column=1, value="一、各平台导入状态").font = Font(name="Microsoft YaHei", size=10.5, bold=True)
r = 4
for i, h in enumerate(["平台", "总量", "已成功", "失败", "成功率"], 1):
cell = ws5.cell(row=r, column=i, value=h)
cell.font = header_font; cell.fill = header_fill
cell.alignment = header_align; cell.border = thin_border
plat_rows = [
("QQ 音乐", "56,132", "56,068", "64", "99.89%"),
("酷狗", "14,716", "14,698", "18", "99.88%"),
("网易云", "15,338", "15,302", "36", "99.77%"),
("合计", "86,186", "86,068", "118", "99.86%"),
]
for i, (plat, total, ok, fail, rate) in enumerate(plat_rows, 5):
vals = [plat, total, ok, fail, rate]
for c_idx, v in enumerate(vals, 1):
cell = ws5.cell(row=i, column=c_idx, value=v)
cell.font = data_font if i < 8 else Font(name="Microsoft YaHei", size=10, bold=True)
cell.alignment = center_align
cell.border = thin_border
if i == 8:
cell.fill = PatternFill(start_color="FFD9E2F3", end_color="FFD9E2F3", fill_type="solid")
# ── 5.2 失败原因分布 ──
r = 10
ws5.cell(row=r, column=1, value="二、导入失败原因分布").font = Font(name="Microsoft YaHei", size=10.5, bold=True)
r = 11
for i, h in enumerate(["失败原因", "失败码", "QQ", "酷狗", "网易云", "合计", "占比"], 1):
cell = ws5.cell(row=r, column=i, value=h)
cell.font = header_font; cell.fill = header_fill
cell.alignment = header_align; cell.border = thin_border
fail_rows = [
("spider DB 缺歌手关系", "-5", "48", "13", "35", "96", "81.4%"),
("音频下载或 OSS 转存失败", "-6", "9", "1", "0", "10", "8.5%"),
("歌词 OSS 转存失败", "-8", "7", "1", "1", "9", "7.6%"),
("无可用歌词", "-2", "0", "3", "0", "3", "2.5%"),
("合计", "", "64", "18", "36", "118", "100%"),
]
for i, (reason, code, qq, kg, ne, total, pct) in enumerate(fail_rows, 12):
vals = [reason, code, qq, kg, ne, total, pct]
for c_idx, v in enumerate(vals, 1):
cell = ws5.cell(row=i, column=c_idx, value=v)
cell.font = data_font if i < 16 else Font(name="Microsoft YaHei", size=10, bold=True)
cell.alignment = center_align
cell.border = thin_border
if i == 16:
cell.fill = PatternFill(start_color="FFD9E2F3", end_color="FFD9E2F3", fill_type="solid")
# ── 5.3 已入库歌曲字段完整度 ──
r = 18
ws5.cell(row=r, column=1, value="三、已入库歌曲字段完整度（records1 严格范围，86,186 首）").font = Font(name="Microsoft YaHei", size=10.5, bold=True)
r = 19
for i, h in enumerate(["字段", "有值", "为空", "完整率", "补齐方式"], 1):
cell = ws5.cell(row=r, column=i, value=h)
cell.font = header_font; cell.fill = header_fill
cell.alignment = header_align; cell.border = thin_border
field_rows = [
("歌名（name）", "86,186", "0", "100%", "—"),
("音频链接（audio_url）", "86,186", "0", "100%", "—"),
("歌词链接（lyrics_url）", "86,183", "3", "100%", "🟢 可忽略"),
("歌手文本（singer）", "86,186", "0", "100%", "—"),
("封面图片（cover_url）", "84,427", "1,759", "98.0%", "🟢 爬虫回补"),
("曲作者（composer）", "84,702", "1,484", "98.3%", "🟢 从爬虫库批量回写"),
("词作者（lyricist）", "82,955", "3,231", "96.3%", "🟢 从爬虫库批量回写"),
("发行时间（issue_time）", "68,797", "17,389", "79.8%", "🟡 爬虫可尝试爬取"),
("BPM 分类（bpm_class）", "35,169", "51,017", "40.8%", "🟡 需 BPM 分析工具计算"),
]
for i, (field, has, empty, rate, method) in enumerate(field_rows, 20):
vals = [field, has, empty, rate, method]
for c_idx, v in enumerate(vals, 1):
cell = ws5.cell(row=i, column=c_idx, value=v)
cell.font = data_font
cell.alignment = center_align if c_idx >= 2 else data_align
cell.border = thin_border
if c_idx == 4 and rate != "—":
pct = float(rate.replace('%', ''))
if pct < 95:
cell.fill = FILL_YELLOW
elif pct < 99:
cell.fill = FILL_GREEN
ws5.column_dimensions["A"].width = 30
ws5.column_dimensions["B"].width = 12
ws5.column_dimensions["C"].width = 12
ws5.column_dimensions["D"].width = 12
ws5.column_dimensions["E"].width = 24
ws5.column_dimensions["F"].width = 10
ws5.column_dimensions["G"].width = 8
# 保存
out = "docs/港乐词曲资产数据缺口清单.xlsx"
wb.save(out)
print(f"✅ 已生成 {out}")
import title_norm as tn
def test_normalize_name():
cases = [
("周杰伦", "周杰伦"),
(" 周杰伦 ", "周杰伦"),
("Ｇｌａｓｓ　Ａｎｉｍａｌｓ", "glass animals"),
("Jay Chou", "jay chou"),
("", ""),
]
for raw, want in cases:
assert tn.normalize_name(raw) == want, raw
def test_singer_identity_key():
assert tn.singer_identity_key("crawler_qqmusic_singers", 162918) == "qqmusic:162918"
def test_author_identity_key():
assert tn.author_identity_key(" 周杰伦 ") == "name:周杰伦"
def test_platform_from_singer_table():
cases = [
("crawler_qqmusic_singers", "qqmusic"),
("crawler_netease_singers", "netease"),
("crawler_kugou_songs", "kugou"),
]
for raw, want in cases:
assert tn.platform_from_singer_table(raw) == want
def test_split_names():
got = tn.split_names("周杰伦/方文山")
assert got == ["周杰伦", "方文山"]
assert tn.split_names("") is None
def test_normalize_title():
cases = [
("七里香", "七里香", ""),
(" 七里香 ", "七里香", ""),
("七里香(Live)", "七里香", "live"),
("七里香（Live版）", "七里香", "live"),
("七里香 [Remix]", "七里香", "remix"),
("七里香（混音版）", "七里香", "remix"),
("七里香（现场版）", "七里香", "live"),
("七里香(DJ版)", "七里香", "dj版"),
("七里香(ass version)", "七里香", "ass version"),
("ANGEL (天使)", "angel (天使)", ""),
("Radio(Dum-Dum)", "radio(dum-dum)", ""),
("七里香(完整版)", "七里香", ""),
("七里香(原版)", "七里香", ""),
("《七里香》", "七里香", ""),
("《七裡香》（Live）", "七里香", "live"),
("七裡香", "七里香", ""),
("七里香 - DJ版", "七里香", "dj版"),
("七里香 - 周杰伦", "七里香 - 周杰伦", ""),
("七里香—Live", "七里香", "live"),
("七里香(天使)(Live)", "七里香(天使)", "live"),
("七里香 现场版", "七里香", "live"),
]
for raw, want_title, want_ver in cases:
got_title, got_ver = tn.normalize_title(raw)
assert (got_title, got_ver) == (want_title, want_ver), raw
def test_normalize_name_set():
a = tn.normalize_name_set("周杰伦,方文山")
b = tn.normalize_name_set("方文山、周杰伦")
assert a == b
assert tn.normalize_name_set("") is None
def test_strip_clip_markers():
cases = [
("父亲写的散文诗-剪辑版", "父亲写的散文诗"),
("人生路漫漫主歌", "人生路漫漫"),
("Stay with me（氛围beat）-全曲", "Stay with me（氛围beat）"),
("咏春（副歌）", "咏春（副歌）"),
("小气鬼（加速版）", "小气鬼"),
("枪火（剪辑一）", "枪火"),
("恋人", "恋人"),
]
for raw, want in cases:
assert tn.strip_clip_markers(raw) == want, raw
def test_is_version_marker():
cases = [
("DJ版", True),
("钢琴版", True),
("女生版", True),
("完整版", False),
("原版", False),
("ass version", True),
("TV Version", True),
("Extended Version", True),
("live", True),
("Live", True),
("伴奏", True),
("对唱", True),
("dj", True),
("remix", True),
("0.8x", True),
("1.1X", True),
("×0.93", True),
("0.8倍", True),
("1.2倍速", True),
("天使", False),
("Dum-Dum", False),
("周杰伦", False),
("", False),
]
for raw, want in cases:
assert tn.is_version_marker(raw) == want, raw
def test_split_names_clip_authors():
cases = [
("唐伯虎Annie、伯爵Johnny", 2),
("梨香JZH & 口古口古", 2),
("派偉俊/mac ova sea", 2),
("李荣浩", 1),
]
for raw, want in cases:
got = tn.split_names(raw)
assert len(got or []) == want, raw
def test_normalize_set():
a = tn.normalize_set(["方文山", "周杰伦"])
b = tn.normalize_set(["周杰伦", "方文山"])
assert a == b
assert tn.normalize_set(None) is None
"""
title_norm — 歌名归一化与版本提取算法。
archive-bridge `internal/subject/normalize.go` 的 Python 移植版本，
用于 title -> (title_norm, version_type) 的独立离线/跨语言复用场景。
移植时保持处理顺序与规则完全一致，覆盖以下能力：
1. NormalizeName : 姓名/文本归一化（全角转半角、繁转简、空白折叠、转小写）
2. NormalizeTitle : 歌名 -> (归一歌名, 版本枚举) 主算法
3. StripClipMarkers: 削去片段标题的切片/版本标记后缀
4. 其余辅助函数：SplitNames / NormalizeNameSet / NormalizeSet /
SingerIdentityKey / AuthorIdentityKey / OrgIdentityKey / PlatformFromSingerTable
不做拼音化、不做去标点。
"""
from __future__ import annotations
import re
from typing import List, Optional, Tuple
try:
from opencc import OpenCC
_t2s = OpenCC("t2s")
except Exception: # pragma: no cover - opencc 加载失败时不影响其余归一逻辑
_t2s = None
__all__ = [
"normalize_name",
"normalize_title",
"strip_clip_markers",
"split_names",
"normalize_name_set",
"normalize_set",
"singer_identity_key",
"author_identity_key",
"org_identity_key",
"platform_from_singer_table",
"is_version_marker",
]
# ---------------------------------------------------------------------------
# 全角转半角（对齐 golang.org/x/text/width.Fold）
# ---------------------------------------------------------------------------
_FULLWIDTH_SPACE = "　"
def _fullwidth_to_halfwidth(s: str) -> str:
"""全角转半角：全角空格 U+3000 -> ASCII 空格；U+FF01-FF5E -> ASCII。"""
out = []
for ch in s:
if ch == _FULLWIDTH_SPACE:
out.append(" ")
elif "！" <= ch <= "～":
out.append(chr(ord(ch) - 0xFEE0))
else:
out.append(ch)
return "".join(out)
_RE_SPACES = re.compile(r"\s+")
def normalize_name(s: str) -> str:
"""归一化姓名/文本：去首尾空白 -> 全角转半角 -> 繁简转简体
-> 折叠连续空白为单空格 -> 转小写。不去标点、不拼音化。
"""
if not s:
return ""
s = _fullwidth_to_halfwidth(s)
if _t2s is not None:
try:
s = _t2s.convert(s)
except Exception:
pass
s = _RE_SPACES.sub(" ", s)
s = s.strip()
s = s.lower()
return s
# ---------------------------------------------------------------------------
# 身份键辅助函数
# ---------------------------------------------------------------------------
def platform_from_singer_table(table: str) -> str:
"""从 crawler 表名解析平台编码。
crawler_qqmusic_singers -> qqmusic
crawler_netease_songs -> netease
"""
s = table
if s.startswith("crawler_"):
s = s[len("crawler_") :]
for suffix in ("_singers", "_songs"):
if s.endswith(suffix):
s = s[: -len(suffix)]
break
return s
def singer_identity_key(singer_table: str, singer_id: int) -> str:
"""歌手身份键：平台:singer_id"""
platform = platform_from_singer_table(singer_table)
return f"{platform}:{singer_id}"
def author_identity_key(name: str) -> str:
"""词曲作者身份键：name:归一名"""
return "name:" + normalize_name(name)
def org_identity_key(name: str) -> str:
"""机构身份键：org:归一名"""
return "org:" + normalize_name(name)
_RE_SEP = re.compile(r"[,/、&＆]")
def split_names(s: str) -> Optional[List[str]]:
"""把逗号/斜杠/顿号/& 分隔的名字串拆成列表，去空白与空项。
split_names("周杰伦/方文山") -> ["周杰伦", "方文山"]
split_names("") -> None
"""
if not s:
return None
parts = _RE_SEP.split(s)
result = [p.strip() for p in parts if p.strip()]
return result or None
def normalize_name_set(raw: str) -> Optional[List[str]]:
"""把多值名字串切割 -> 逐个归一 -> 排序，返回顺序无关的规范集合。空串返回 None。"""
parts = split_names(raw)
if not parts:
return None
out = [n for n in (normalize_name(p) for p in parts) if n]
if not out:
return None
out.sort()
return out
def normalize_set(raw: Optional[List[str]]) -> Optional[List[str]]:
"""把一组原始名字逐个归一并排序（顺序无关比对）。空返回 None。"""
if not raw:
return None
out = [n for n in (normalize_name(p) for p in raw) if n]
if not out:
return None
out.sort()
return out
# ---------------------------------------------------------------------------
# 版本识别词表
# ---------------------------------------------------------------------------
# versionAliases：版本别名 -> 标准枚举映射表。
# 标准枚举：""（原版）/live/remix/cover/accompaniment/instrumental
_VERSION_ALIASES = {
"live": "live",
"live版": "live",
"现场": "live",
"现场版": "live",
"remix": "remix",
"remix版": "remix",
"混音": "remix",
"混音版": "remix",
"翻唱": "cover",
"翻唱版": "cover",
"伴奏": "accompaniment",
"伴奏版": "accompaniment",
"纯音乐": "instrumental",
"纯音乐版": "instrumental",
}
# 按字符长度降序排列的版本后缀词，用于无括号版本后缀识别。
_VERSION_SUFFIX_WORDS = sorted(_VERSION_ALIASES.keys(), key=len, reverse=True)
# fixedVersionKeywords：不含"版"字但明确为版本的关键词集合（全部小写，查询前对输入 normalize_name）。
# 基于 66 万条真实歌曲版本数据整理。
_FIXED_VERSION_KEYWORDS_RAW = [
# 明确版本列表（业务确认）
"伴奏", "未知", "清晰版", "语言版", "恶搞版", "翻唱版",
"dj", "dj版", "dj长音频",
"live", "demo", "remix", "acoustic", "remaster",
"radio edit", "tv version", "lp版", "revival",
"clean", "explicit", "performance", "reprise",
"jam version", "cast version", "extended version",
"纯音乐", "unplugged", "口白", "铃声",
"mr", "试听版", "acappella", "arr", "performod",
"uk version", "ao vivo", "bonus track", "orchestra",
"其他",
# 演唱形式
"对唱", "合唱", "独唱", "清唱", "说唱", "cover",
# 性别/年龄
"女声", "男声", "童声", "女版", "男版",
# 语言
"粤语", "国语", "闽南语", "客家话",
"英文", "日文", "韩文", "cantonese", "mandarin",
# 乐器
"钢琴", "吉他", "古筝", "琵琶", "扬琴",
"二胡", "笛子", "萨克斯", "小提琴", "大提琴",
"piano", "guitar",
# 音乐风格
"摇滚", "民谣", "爵士", "布鲁斯",
"r&b", "rnb", "rock", "jazz", "blues",
"funk", "soul", "disco", "edm", "house",
"techno", "trap", "phonk", "dubstep",
"metal", "punk", "pop", "ballad",
# 片段/结构
"片段", "副歌", "前奏", "间奏", "尾奏", "主歌", "高潮",
# 音效/品质
"环绕", "混响", "立体声",
"3d", "8d", "hifi", "高品质", "无损",
# 制作类型
"edit", "extended", "original", "single",
"album", "ep", "deluxe",
"bootleg", "mashup", "tribute",
# 速度相关（文字）
"加速", "降速", "慢速", "快速",
"slowed", "sped", "speed",
# 其他
"inst", "instrumental", "karaoke",
"rehearsal", "studio",
]
_FIXED_VERSION_KEYWORDS = {normalize_name(w) for w in _FIXED_VERSION_KEYWORDS_RAW}
# versionBlacklist：含"版"字但不当版本处理的词（归一后小写）。
_VERSION_BLACKLIST = {"完整版", "原版"}
# 速度模式：0.8x / 1.1X / ×0.93 / 0.8倍 / 1.2倍速
_RE_SPEED_X = re.compile(r"\d+\.?\d*[xX×]|[xX×]\d+\.?\d*")
_RE_SPEED_BEI = re.compile(r"\d+\.?\d*倍")
# 括号内容分类：0=非版本，1=版本，2=黑名单（剥离但不记版本）
_BRACKET_NOT_VERSION = 0
_BRACKET_VERSION = 1
_BRACKET_BLACKLIST = 2
def _classify_bracket(content: str) -> int:
content = content.strip()
if not content:
return _BRACKET_NOT_VERSION
norm = normalize_name(content)
if "版" in content:
if norm in _VERSION_BLACKLIST:
return _BRACKET_BLACKLIST
return _BRACKET_VERSION
if (
"version" in norm
or norm in _FIXED_VERSION_KEYWORDS
or _RE_SPEED_X.search(content)
or _RE_SPEED_BEI.search(content)
):
return _BRACKET_VERSION
return _BRACKET_NOT_VERSION
def is_version_marker(content: str) -> bool:
"""判定括号内/后缀内容是否为版本标记。
规则（content 应已去除首尾空白）：
1. 含"版"字且不在黑名单 -> 版本
2. 归一后含 "version" 子串 -> 版本
3. 归一后在固定词表 -> 版本
4. 匹配速度模式 -> 版本
5. 否则 -> 非版本
"""
return _classify_bracket(content) == _BRACKET_VERSION
# ---------------------------------------------------------------------------
# NormalizeTitle 主算法
# ---------------------------------------------------------------------------
# 歌名末尾成对括号（全/半角小括号、方括号、中文方括号）。
_RE_TITLE_VERSION = re.compile(r"^(.*?)[（(\[【]([^（(\[【]+)[）)\]】]\s*$")
# 开头书名号《...》（》不必在末尾，如《七里香》（Live））。
_RE_BOOK_TITLE = re.compile(r"^《(.+?)》")
# 末尾空格+任意非空词，用于兜底识别任意尾词是否为版本标记。
_RE_TRAILING_WORD = re.compile(r"\s+(\S+)$")
def _strip_book_title_marks(s: str) -> str:
"""剥离首部书名号《...》（》不必在末尾），循环直到无匹配。"""
while True:
m = _RE_BOOK_TITLE.match(s)
if m is None:
return s
s = m.group(1) + s[m.end() :]
def _dash_split(s: str, idx: int) -> Tuple[str, str]:
"""在破折号分隔符 idx（字符位置）处切分，按版本感知返回。"""
head = s[:idx].strip()
suffix = s[idx + 1 :].strip()
if is_version_marker(suffix):
return head, suffix
return s, ""
def _strip_dash_suffix(s: str) -> Tuple[str, str]:
"""全角/en dash/em dash 或半角 " - " 分隔符后缀的版本感知剥离。"""
for i, ch in enumerate(s):
if ch in "—–":
return _dash_split(s, i)
idx = s.find(" - ")
if idx >= 0:
head = s[:idx].strip()
suffix = s[idx + 3 :].strip()
if is_version_marker(suffix):
return head, suffix
return s, ""
return s, ""
def _strip_trailing_brackets(s: str) -> Tuple[str, str]:
"""循环剥离末尾成对括号（版本感知三态）：
- 版本：剥离并记 version（仅记首个）
- 黑名单：剥离但不记 version
- 非版本：停止剥离，整体保留
"""
version = ""
while True:
m = _RE_TITLE_VERSION.match(s)
if m is None:
break
content = m.group(2).strip()
kind = _classify_bracket(content)
if kind == _BRACKET_NOT_VERSION:
return s, version
if kind == _BRACKET_VERSION:
if version == "":
version = content
# bracket_blacklist：剥离但不记 version
s = m.group(1).strip()
return s, version
def _strip_version_suffix(s: str) -> Tuple[str, str]:
"""识别并剥离无括号版本后缀词（空格+版本词），返回剥离后的标题和版本词。
先查版本别名表（大小写不敏感），再用 is_version_marker 兜底任意尾词。
"""
lower = s.lower()
for vw in _VERSION_SUFFIX_WORDS:
suffix = " " + vw
if lower.endswith(suffix):
return s[: len(s) - len(suffix)].strip(), s[len(s) - len(suffix) + 1 :]
m = _RE_TRAILING_WORD.search(s)
if m:
word = m.group(1)
if is_version_marker(word):
return s[: m.start()].strip(), word
return s, ""
def _normalize_version(v: str) -> str:
"""对版本词做枚举归一：normalize_name 小写后查版本别名表映射到标准枚举。
查不到的版本词保留原文（不误归一）。
"""
v = normalize_name(v)
return _VERSION_ALIASES.get(v, v)
def normalize_title(raw: str) -> Tuple[str, str]:
"""拆分歌名与版本，处理顺序：
1. 书名号《》剥离
2. 破折号后缀版本感知剥离（版本进 version，非版本整体保留）
3. 多重括号循环剥离（首次括号内容为 version）
4. 无括号版本后缀词剥离
5. normalize_name（全角转半角+繁简转简体+折叠空白+小写）
6. version 枚举归一（查版本别名表映射到标准枚举）
返回 (title_norm, version)。
"""
s = raw.strip()
s = _strip_book_title_marks(s)
version = ""
t, v = _strip_dash_suffix(s)
if v:
s = t
version = v
if version == "":
s, bv = _strip_trailing_brackets(s)
if bv:
version = bv
else:
s, _ = _strip_trailing_brackets(s)
t, v = _strip_version_suffix(s)
if v:
s = t
if version == "":
version = v
title_norm = normalize_name(s)
version = _normalize_version(version)
return title_norm, version
# ---------------------------------------------------------------------------
# StripClipMarkers（抖音片段标题削标记）
# ---------------------------------------------------------------------------
# 括号和破折号后缀里视为版本/切片标记的关键词（排除曲段词如主歌/副歌）。
_CLIP_BRACKET_KEYWORDS = [
"剪辑版", "剪辑一", "剪辑二", "剪辑三", "剪辑", "片段", "全曲",
"加速版", "加速", "完整版",
]
# 无分隔符直接拼在末尾时视为标记的关键词（含曲段词）。
_CLIP_SUFFIX_KEYWORDS = [
"剪辑版", "剪辑一", "剪辑二", "剪辑三", "剪辑", "片段", "全曲", "主歌", "副歌",
"加速版", "加速", "完整版", "高潮", "前奏", "间奏", "尾奏",
]
# 末尾成对括号（全/半角小括号）。
_RE_CLIP_BRACKET_TAIL = re.compile(r"^(.*?)[（(]([^（(]+)[）)]\s*$")
def _contains_keywords(s: str, kws: List[str]) -> bool:
return any(kw in s for kw in kws)
def strip_clip_markers(raw: str) -> str:
"""削去抖音片段标题里的切片/版本标记后缀，产出可与全曲匹配的名。
① 末尾括号内容含标记关键词 -> 删整组；否则保留。
② 破折号后缀（半/全角）含标记关键词 -> 删后缀。
③ 无分隔符直接拼在末尾的标记 -> 削去。
"""
s = raw.strip()
m = _RE_CLIP_BRACKET_TAIL.match(s)
if m is not None and _contains_keywords(m.group(2), _CLIP_BRACKET_KEYWORDS):
s = m.group(1).strip()
idx = -1
for i, ch in enumerate(s):
if ch in "-—–":
idx = i
if idx >= 0:
tail = s[idx + 1 :]
if _contains_keywords(tail, _CLIP_BRACKET_KEYWORDS):
s = s[:idx].strip()
for kw in _CLIP_SUFFIX_KEYWORDS:
if s.endswith(kw) and len(s) > len(kw):
s = s[: -len(kw)].strip()
break
return s
#!/usr/bin/env python3
"""将采集库 crawler_*_singers.hot_songs 同步到档案库 archive_subject_account.hot_songs。
支持测试 / 正式环境，采集和档案同时切换。
数据链路：
测试: crawler_dev -> data_dev
正式: archive_crawler -> archive_data
ON archive_subject_account.crawler_singer_id = crawler_*_singers.id
UPDATE archive_subject_account.hot_songs
仅在 crawler_singer_id 能对应上时才更新，且只更新值有变化的行。
默认 dry-run，加 --apply 才会提交。
用法：
.venv/bin/python sync_singer_hot_songs.py # 测试环境 dry-run
.venv/bin/python sync_singer_hot_songs.py --apply # 测试环境执行
.venv/bin/python sync_singer_hot_songs.py --env prod # 正式环境 dry-run
.venv/bin/python sync_singer_hot_songs.py --env prod --apply # 正式环境执行
"""
import argparse
import json
import logging
import sys
from etl_to_crawler.connections import (
close_all_pools,
get_test_crawler_conn,
get_test_archive_conn,
get_archive_crawler_conn,
get_archive_conn,
)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger(__name__)
CRAWLER_SINGER_TABLES = [
"crawler_qqmusic_singers",
"crawler_kugou_singers",
"crawler_netease_singers",
]
BATCH_SIZE = 500
def fetch_crawler_singer_hot_songs(pg_conn) -> list[dict]:
"""从采集库读取所有有 hot_songs 的歌手记录。"""
all_rows: list[dict] = []
for table in CRAWLER_SINGER_TABLES:
sql = f"""
SELECT id, name, hot_songs
FROM {table}
WHERE hot_songs IS NOT NULL
"""
with pg_conn.cursor() as cur:
cur.execute(sql)
rows = cur.fetchall()
for row in rows:
all_rows.append({
"id": str(row[0]), # 统一转 str，与档案库 varchar 类型对齐
"name": row[1],
"hot_songs": row[2],
"source_table": table,
})
log.info(" %s: %d 条有 hot_songs 的歌手", table, len(rows))
return all_rows
def fetch_archive_current_hot_songs(archive_conn, crawler_ids: set) -> dict:
"""查询档案库中匹配记录的当前 hot_songs 值。
返回的 dict 的 key 即为已匹配的 crawler_singer_id，
未出现在 dict 中的 ID 表示档案库中无对应记录。
"""
if not crawler_ids:
return {}
current: dict = {}
id_list = list(crawler_ids)
for i in range(0, len(id_list), BATCH_SIZE):
batch = id_list[i : i + BATCH_SIZE]
sql = """
SELECT crawler_singer_id, hot_songs
FROM archive_subject_account
WHERE crawler_singer_id = ANY(%s)
"""
with archive_conn.cursor() as cur:
cur.execute(sql, (batch,))
for row in cur.fetchall():
current[row[0]] = row[1]
return current
def batch_update_archive(archive_conn, updates: list[tuple], apply: bool) -> int:
"""批量更新 archive_subject_account.hot_songs。"""
if not updates:
return 0
total = 0
for i in range(0, len(updates), BATCH_SIZE):
batch = updates[i : i + BATCH_SIZE]
values_sql_parts = []
params: list = []
for idx, (crawler_singer_id, hot_songs) in enumerate(batch):
values_sql_parts.append(f"(%s, %s::jsonb)")
params.extend([crawler_singer_id, json.dumps(hot_songs, ensure_ascii=False)])
values_sql = ", ".join(values_sql_parts)
sql = f"""
UPDATE archive_subject_account AS a
SET hot_songs = v.hot_songs
FROM (VALUES {values_sql}) AS v(crawler_singer_id, hot_songs)
WHERE a.crawler_singer_id = v.crawler_singer_id
"""
with archive_conn.cursor() as cur:
cur.execute(sql, params)
total += len(batch)
log.info(" 已处理 %d / %d", total, len(updates))
if apply:
archive_conn.commit()
return total
def _json_equal(a, b) -> bool:
"""比较两个 JSON 值是否逻辑相等。"""
if a is None and b is None:
return True
if a is None or b is None:
return False
# 统一用 Python 对象比较（jsonb 反序列化后 key 序无关）
if isinstance(a, str):
a = json.loads(a)
if isinstance(b, str):
b = json.loads(b)
return a == b
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="同步采集库 crawler_*_singers.hot_songs 到档案库 archive_subject_account"
)
parser.add_argument(
"--env",
choices=["test", "prod"],
default="test",
help="运行环境: test=测试(crawler_dev+data_dev), prod=正式(archive_crawler+archive_data)（默认 test）",
)
parser.add_argument(
"--apply",
action="store_true",
help="实际执行更新（默认 dry-run 只预览不写库）",
)
return parser.parse_args()
def main() -> None:
args = parse_args()
mode = "APPLY" if args.apply else "DRY-RUN"
env_label = "正式环境" if args.env == "prod" else "测试环境"
# 根据环境选择连接函数
if args.env == "prod":
get_crawler_conn = get_archive_crawler_conn # 正式采集库: archive_crawler
get_archive_conn_fn = get_archive_conn # 正式档案库: archive_data
crawler_db = "archive_crawler"
archive_db = "archive_data"
else:
get_crawler_conn = get_test_crawler_conn # 测试采集库: crawler_dev
get_archive_conn_fn = get_test_archive_conn # 测试档案库: data_dev
crawler_db = "crawler_dev"
archive_db = "data_dev"
log.info("=== sync_singer_hot_songs [%s] [%s] ===", env_label, mode)
log.info("采集库: %s | 档案库: %s", crawler_db, archive_db)
# 1. 从采集库读取歌手 hot_songs
log.info("步骤 1: 读取采集库 crawler_*_singers.hot_songs (%s)", crawler_db)
crawler_conn = get_crawler_conn()
crawler_rows = fetch_crawler_singer_hot_songs(crawler_conn)
if not crawler_rows:
log.info("采集库无任何 hot_songs 数据，退出")
return
log.info("共读取 %d 条歌手记录", len(crawler_rows))
# 2+3. 一次查询档案库：匹配 + 获取当前值 + 比对变化
crawler_ids = {row["id"] for row in crawler_rows}
log.info("步骤 2: 查询档案库并比对变化 (%s，共 %d 个候选)", archive_db, len(crawler_ids))
archive_conn = get_archive_conn_fn()
current_hot_songs = fetch_archive_current_hot_songs(archive_conn, crawler_ids)
matched_ids = set(current_hot_songs.keys()) # dict 的 key 即为匹配到的 ID
log.info("匹配到 %d 个 crawler_singer_id", len(matched_ids))
if not matched_ids:
log.info("无可匹配的记录，退出")
return
updates: list[tuple] = [] # (crawler_singer_id, new_hot_songs)
changes: list[dict] = [] # 变更明细
updated_ids: set = set() # 已产生变更的 ID（去重）
for row in crawler_rows:
sid = row["id"]
if sid not in matched_ids or sid in updated_ids:
continue
new_val = row["hot_songs"]
old_val = current_hot_songs.get(sid)
if not _json_equal(old_val, new_val):
updated_ids.add(sid)
updates.append((sid, new_val))
changes.append({
"crawler_singer_id": sid,
"name": row["name"],
"source_table": row["source_table"],
"old_hot_songs": old_val,
"new_hot_songs": new_val,
})
skipped = len(matched_ids) - len(updated_ids)
log.info("需更新: %d 条，无变化跳过: %d 条，无法匹配跳过: %d 条",
len(updates), skipped, len(crawler_ids) - len(matched_ids))
if not updates:
print("\n" + "=" * 80)
print("统计摘要")
print("=" * 80)
print(f" 环境 : {env_label}")
print(f" 采集库 : {crawler_db}")
print(f" 档案库 : {archive_db}")
print(f" 采集库歌手总数 : {len(crawler_rows)}")
print(f" 匹配档案数 : {len(matched_ids)}")
print(f" 需更新 : 0")
print(f" 无变化跳过 : {len(matched_ids)}")
print(f" 无法匹配跳过 : {len(crawler_ids) - len(matched_ids)}")
print("=" * 80)
log.info("无变化需要更新，退出")
return
# 4. 执行更新
if args.apply:
log.info("步骤 4: 执行更新 (--apply)")
else:
log.info("步骤 4: DRY-RUN 预览（加 --apply 实际执行）")
if args.apply:
batch_update_archive(archive_conn, updates, apply=True)
log.info("更新完成，已提交")
else:
log.info("DRY-RUN 模式，未写库")
# 5. 输出变更明细
print("\n" + "=" * 80)
print(f"变更明细（共 {len(changes)} 条）")
print("=" * 80)
for c in changes:
print(f"\n crawler_singer_id : {c['crawler_singer_id']}")
print(f" name : {c['name']}")
print(f" source_table : {c['source_table']}")
old_display = json.dumps(c["old_hot_songs"], ensure_ascii=False, indent=2) if c["old_hot_songs"] else "NULL"
new_display = json.dumps(c["new_hot_songs"], ensure_ascii=False, indent=2) if c["new_hot_songs"] else "NULL"
print(f" old_hot_songs : {old_display}")
print(f" new_hot_songs : {new_display}")
# 6. 统计摘要
print("\n" + "=" * 80)
print("统计摘要")
print("=" * 80)
print(f" 环境 : {env_label}")
print(f" 采集库 : {crawler_db}")
print(f" 档案库 : {archive_db}")
print(f" 采集库歌手总数 : {len(crawler_rows)}")
print(f" 匹配档案数 : {len(matched_ids)}")
print(f" 需更新 : {len(updates)}")
print(f" 无变化跳过 : {skipped}")
print(f" 无法匹配跳过 : {len(crawler_ids) - len(matched_ids)}")
print(f" 执行模式 : {'已提交 (--apply)' if args.apply else 'DRY-RUN（未写库）'}")
print("=" * 80)
if __name__ == "__main__":
try:
main()
finally:
close_all_pools()
......@@ -2,7 +2,7 @@ from pathlib import Path
from openpyxl import load_workbook
from check_record_titles import combine_rows, fetch_crawler_rows, fetch_record_names, write_excel
from check_record_titles import combine_rows, fetch_crawler_rows, fetch_spider_titles, write_excel
class Cursor:
......@@ -42,7 +42,7 @@ def test_fetch_crawler_rows_joins_all_platform_tables():
"platform": "1",
"platform_song_id": 101,
"recording_id": "recording-1",
"title": "同名",
"crawler_title": "同名",
}]
assert "WHERE ysr.is_archive_push = TRUE" in cursor.calls[0][0]
......@@ -58,31 +58,35 @@ def test_fetch_crawler_rows_rejects_unknown_table():
raise AssertionError("应拒绝非白名单表")
def test_fetch_record_names_batches_queries():
def test_fetch_spider_titles_batches_queries_by_platform():
connection = Connection([
[{"id": 1, "record_name": "甲"}, {"id": 2, "record_name": "乙"}],
[{"id": 3, "record_name": "丙"}],
[{"id": 101, "title": "甲"}],
[{"id": 102, "title": "乙"}],
])
crawler_rows = [
{"platform": "1", "platform_song_id": 101},
{"platform": "2", "platform_song_id": 102},
]
names = fetch_record_names(connection, [3, 1, 2, 2], batch_size=2)
titles = fetch_spider_titles(connection, crawler_rows, batch_size=2)
assert names == {1: "甲", 2: "乙", 3: "丙"}
assert titles == {("1", 101): "甲", ("2", 102): "乙"}
def test_excel_marks_only_exact_non_null_matches_green(tmp_path: Path):
crawler_rows = [
{"record_id": 1, "platform": "1", "platform_song_id": 101, "recording_id": "r1", "title": "相同"},
{"record_id": 2, "platform": "2", "platform_song_id": 102, "recording_id": "r2", "title": " 名称 "},
{"record_id": 3, "platform": "4", "platform_song_id": 103, "recording_id": None, "title": None},
{"record_id": 1, "platform": "1", "platform_song_id": 101, "recording_id": "r1", "crawler_title": "相同"},
{"record_id": 2, "platform": "2", "platform_song_id": 102, "recording_id": "r2", "crawler_title": " 名称 "},
{"record_id": 3, "platform": "4", "platform_song_id": 103, "recording_id": None, "crawler_title": None},
]
rows = combine_rows(crawler_rows, {1: "相同", 2: "名称", 3: None})
rows = combine_rows(crawler_rows, {("1", 101): "相同", ("2", 102): "名称"})
output = tmp_path / "check.xlsx"
write_excel(rows, output)
worksheet = load_workbook(output).active
assert [worksheet.cell(1, column).value for column in range(1, 7)] == [
"record_id", "record_name", "title", "platform", "platform_song_id", "recording_id",
"record_id", "spider_title", "crawler_title", "platform", "platform_song_id", "recording_id",
]
assert worksheet["A2"].fill.fgColor.rgb == "00C6EFCE"
assert worksheet["A3"].fill.fill_type is None
......
from openpyxl import load_workbook
from datetime import datetime, timezone
from fix_record_titles import (
InputRow,
_archive_recording_sql,
_archive_source_sql,
_crawler_sql,
_rollback_sql,
_row_value,
build_backup_records,
build_report_rows,
build_fix_rows,
fetch_input_rows,
load_backup,
write_backup_atomic,
write_report,
)
class Cursor:
def __init__(self, rows):
self.rows = rows
self.calls = []
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
return False
def execute(self, sql, params=None):
self.calls.append((sql, params))
def fetchall(self):
return self.rows
class Connection:
def __init__(self, rows):
self.raw_cursor = Cursor(rows)
def cursor(self):
return self.raw_cursor
def test_fetch_input_rows_reads_archive_push_relations_directly():
connection = Connection([
(1, "1", 101, "REC-1"),
(2, "2", 102, "REC-2"),
])
rows, stats = fetch_input_rows(connection)
assert rows == [
InputRow(1, "1", 101, "REC-1"),
InputRow(2, "2", 102, "REC-2"),
]
assert stats == {"total": 2, "missing_key": 0, "unsupported_platform": 0}
sql = connection.raw_cursor.calls[0][0]
assert "FROM yinyan_song_records" in sql
assert "WHERE is_archive_push = TRUE" in sql
def test_fetch_input_rows_applies_limit():
connection = Connection([(1, "1", 101, "REC-1")])
fetch_input_rows(connection, limit=10)
sql, params = connection.raw_cursor.calls[0]
assert "LIMIT %s" in sql
assert params == (10,)
def test_build_fix_rows_uses_spider_title_for_normalization():
input_row = InputRow(1, "1", 101, "REC-1")
rows, fix_stats = build_fix_rows(
[input_row],
{("1", 101): "七里香（Live版）"},
)
assert fix_stats["missing_spider"] == 0
assert rows[0].spider_title == "七里香（Live版）"
assert rows[0].title_norm == "七里香"
assert rows[0].version == "live"
assert _row_value(rows[0], "archive_platform") == "qqmusic"
assert _row_value(rows[0], "archive_version") == "live"
def test_build_fix_rows_skips_missing_spider_title():
input_row = InputRow(1, "4", 101, "REC-1")
rows, fix_stats = build_fix_rows([input_row], {})
assert rows == []
assert fix_stats["missing_spider"] == 1
def test_build_fix_rows_temporarily_excludes_exact_spider_title():
excluded = 'Waiting On A Wish (From "Disney\'s Snow White"/Soundtrack Version|Reprise)'
input_rows = [
InputRow(1, "1", 101, "REC-1"),
InputRow(2, "1", 102, "REC-2"),
]
spider_titles = {
("1", 101): excluded,
("1", 102): "Waiting On A Wish",
}
rows, fix_stats = build_fix_rows(
input_rows,
spider_titles,
exclude_titles={excluded},
)
assert [row.platform_song_id for row in rows] == [102]
assert fix_stats["excluded_title"] == 1
def test_update_sql_uses_spider_values_and_only_updates_differences():
recording_sql, recording_params = _archive_recording_sql(1, True)
source_sql, source_params = _archive_source_sql(1, True)
crawler_sql, crawler_params = _crawler_sql("crawler_qqmusic_songs", 1, True)
assert recording_params == ["recording_id", "spider_title", "title_norm", "archive_version"]
assert "title_norm = source.new_title_norm" in recording_sql
assert "IS DISTINCT FROM source.new_title" in recording_sql
assert "target.platform_song_id = source.platform_song_id" in source_sql
assert "spider_title" in source_params
assert crawler_params == ["platform_song_id", "spider_title", "version"]
assert "source.platform_song_id::bigint" in crawler_sql
assert "version = source.new_version" in crawler_sql
assert "RETURNING target.recording_id" in recording_sql
assert "RETURNING target.platform_song_id" in crawler_sql
locked_sql, _ = _archive_recording_sql(1, False, lock=True)
assert "FOR UPDATE OF target" in locked_sql
def test_report_lists_only_returned_target_keys(tmp_path):
fix = build_fix_rows(
[InputRow(1, "1", 101, "REC-1")],
{("1", 101): "七里香（Live版）"},
)[0][0]
matched = {
"archive_recording": [
("REC-1", "旧标题", "七里香（Live版）", "旧标题", "七里香", None, "live"),
],
"archive_recording_platform_source": [
("REC-1", "qqmusic", "101", "旧标题", "七里香（Live版）", None, "七里香", None, "live"),
],
"crawler_qqmusic_songs": [
(101, "旧标题", "七里香（Live版）", "", "live"),
],
"crawler_kugou_songs": [],
"crawler_netease_songs": [],
}
rows = build_report_rows([fix], matched, apply=False)
output = tmp_path / "report.xlsx"
write_report(output, rows)
worksheet = load_workbook(output).active
assert len(rows) == 3
assert worksheet.max_row == 4
assert worksheet["A1"].value == "database"
assert worksheet["H1"].value == "old_title"
assert worksheet["I1"].value == "new_title"
assert worksheet["H2"].value == "旧标题"
assert worksheet["I2"].value == "七里香（Live版）"
assert worksheet["J3"].value == "<NULL>"
assert worksheet["J3"].fill.fgColor.rgb == "00FFC7CE"
assert worksheet["K3"].fill.fgColor.rgb == "00C6EFCE"
assert worksheet["N2"].value == "would_update"
def test_backup_round_trip_preserves_null_empty_and_updated_at(tmp_path):
fix = build_fix_rows(
[InputRow(1, "1", 101, "REC-1")],
{("1", 101): "七里香（Live版）"},
)[0][0]
updated_at = datetime(2026, 7, 15, 10, 0, tzinfo=timezone.utc)
preview = {
"archive_recording": [
("REC-1", "旧标题", "七里香（Live版）", None, "七里香", None, "live", updated_at),
],
"archive_recording_platform_source": [],
"crawler_qqmusic_songs": [
(101, "旧标题", "七里香（Live版）", "", "live", updated_at.replace(tzinfo=None)),
],
"crawler_kugou_songs": [],
"crawler_netease_songs": [],
}
records = build_backup_records([fix], preview, "run-1")
path = tmp_path / "backup.jsonl"
write_backup_atomic(path, records, "run-1")
metadata, loaded = load_backup(path)
assert metadata["change_count"] == 2
assert loaded[0]["old"]["title_norm"] is None
assert loaded[1]["old"]["version"] == ""
assert loaded[0]["old"]["updated_at"] == updated_at.isoformat()
try:
write_backup_atomic(path, records, "run-2")
except FileExistsError:
pass
else:
raise AssertionError("已有备份不应被覆盖")
def test_rollback_sql_restores_old_values_with_new_value_guard():
archive_sql, _ = _rollback_sql("archive_recording", 1)
crawler_sql, _ = _rollback_sql("crawler_qqmusic_songs", 1)
assert "SET title = source.old_title" in archive_sql
assert "updated_at = source.old_updated_at::timestamptz" in archive_sql
assert "target.title IS NOT DISTINCT FROM source.new_title" in archive_sql
assert "target.title_norm IS NOT DISTINCT FROM source.new_title_norm" in archive_sql
assert "updated_at = source.old_updated_at::timestamp" in crawler_sql
assert "target.version IS NOT DISTINCT FROM source.new_version" in crawler_sql
......@@ -375,7 +375,7 @@ def test_run_imports_only_pending_yinyan_platform_record(monkeypatch):
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_test_crawler_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
monkeypatch.setattr(runner, 'fetch_pending_yinyan_song_records', lambda cur, batch_size, platforms: [
......@@ -436,7 +436,7 @@ def test_run_retry_failed_resets_failures_and_restores_hk_sources(monkeypatch):
monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: hk_conn)
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_test_crawler_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'get_oss_bucket', lambda: object())
monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
monkeypatch.setattr(runner, 'reset_failed_yinyan_song_records', reset)
......@@ -456,7 +456,7 @@ def test_initialize_yinyan_song_records_inserts_primary_records(monkeypatch):
monkeypatch.setattr(runner, 'get_hk_songs_conn', lambda: _Connection())
monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'get_test_crawler_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
monkeypatch.setattr(runner, 'fetch_existing_yinyan_song_ids', lambda cur: set())
......@@ -501,7 +501,7 @@ def test_initialize_yinyan_song_records2_writes_all_relations_then_deduplicates(
dedupe_calls = []
monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'get_test_crawler_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
monkeypatch.setattr(runner, 'fetch_yinyan_song_ids', lambda cur: [10])
monkeypatch.setattr(runner, 'fetch_all_song_record_relations', lambda conn, song_ids: [
......@@ -532,7 +532,7 @@ def test_initialize_yinyan_song_records2_keeps_rows_without_valid_singers(monkey
monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
get_spider_conn = MagicMock(side_effect=AssertionError('records2 init must not query spider'))
monkeypatch.setattr(runner, 'get_spider_conn', get_spider_conn)
monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'get_test_crawler_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
monkeypatch.setattr(runner, 'fetch_yinyan_song_ids', lambda cur: [10])
monkeypatch.setattr(runner, 'fetch_all_song_record_relations', lambda conn, song_ids: [
......@@ -561,7 +561,7 @@ def test_backfill_yinyan_record_platforms_updates_missing_platform_rows(monkeypa
updated = []
monkeypatch.setattr(runner, 'get_source_conn', lambda: _Connection())
monkeypatch.setattr(runner, 'get_pg_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'get_test_crawler_conn', lambda: pg_conn)
monkeypatch.setattr(runner, 'refresh_conn', lambda conn, name: conn)
monkeypatch.setattr(runner, 'fetch_yinyan_records_missing_platform', lambda cur, batch_size: [
{'song_id': 10, 'record_id': 100},
......