#!/usr/bin/env python3 """从本地曲库 muse.db 生成"歌曲统计 + 清单"（自建歌曲源支线用）。 产出（默认写入 selfhost-source/inventory/）： song-stats.md 统计报告（总数 / 语言 / 标签 / 歌手 / 覆盖情况） songs-all.csv 全量清单：tid,name,lang,singers,filename,album,hot_score songs-<语言>.csv 按语言拆分（便于分批导入自建服务器） singers-top.csv 歌手榜（按歌曲数） playlists.csv 歌单清单 用法：python3 build_inventory.py [--limit 0] # --limit>0 时只导出前 N 首（试跑用） """ import argparse import csv import json import os import re import sqlite3 import sys import time ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DB = os.path.join(ROOT, "muse.db") OUT = os.path.join(ROOT, "selfhost-source", "inventory") def q(conn, sql, args=()): return conn.execute(sql, args).fetchall() def main(): ap = argparse.ArgumentParser() ap.add_argument("--limit", type=int, default=0) args = ap.parse_args() os.makedirs(OUT, exist_ok=True) for old in os.listdir(OUT): if old.startswith("songs-") or old in ("song-stats.md", "singers-top.csv", "playlists.csv"): os.remove(os.path.join(OUT, old)) conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True) t0 = time.time() L = [] def say(s=""): print(s, flush=True) L.append(s) def has(col): return col in {r[1] for r in q(conn, "PRAGMA table_info(songs)")} def has_table(t): return bool(q(conn, "select 1 from sqlite_master where type='table' and name=?", (t,))) total = q(conn, "select count(*) from songs")[0][0] deleted = q(conn, "select count(*) from songs where deleted_at is not null")[0][0] say("# 麦动KTV 本地曲库 — 歌曲统计与清单") say() say(f"- 数据源：`muse.db`（md5 已核对；版本文件 `db_version.txt`）") say(f"- 生成时间：{time.strftime('%Y-%m-%d %H:%M:%S')}") say(f"- **歌曲总数：{total:,} 首**（其中已删除 deleted_at 非空：{deleted:,}）") say() # ── 语言 ── say("## 一、按语言（归一化到 7 个规范语种）") say() # 源库的 lang 字段有少量脏数据（填了歌名/节目名等）→ 统一归一化，避免清单被拆成几十份 CANON = ["国语", "粤语", "闽南语", "英语", "日语", "韩语", "其它"] ALIAS = {"闽南": "闽南语", "台语": "闽南语", "闽南语 ": "闽南语", "H国语": "国语", "其他": "其它", "西语": "其它", "西班牙语": "其它", "法语": "其它", "俄语": "其它", "泰语": "其它", "印度": "其它", "潮汕语": "其它", "消音": "其它", "原版": "其它"} langcount = {k: 0 for k in CANON} langcount["(未标注)"] = 0 dirty = 0 for (lang, c) in q(conn, "select coalesce(lang,'') lang, count(*) c from songs group by lang"): key = lang.strip() if key in CANON: langcount[key] += c elif key in ALIAS: langcount[ALIAS[key]] += c elif not key: langcount["(未标注)"] += c else: langcount["其它"] += c dirty += c say("| 语言 | 歌曲数 | 占比 |") say("|---|---:|---:|") for name in CANON + ["(未标注)"]: c = langcount[name] if c: say(f"| {name} | {c:,} | {c*100/total:.2f}% |") say(f"\n（另有 **{dirty:,}** 首的 `lang` 字段不是规范语种——填的是歌名/节目名等脏数据，已归入『其它』）") say() # ── 标签（曲风/风格） ── say("## 二、曲风/标签 Top 30（tags_str 为 JSON 数组，取每首首个标签统计）") say() say("| 标签 | 歌曲数 |") say("|---|---:|") tagcount = {} for (tags,) in q(conn, "select tags_str from songs where tags_str is not null and tags_str != ''"): try: arr = json.loads(tags) except Exception: continue if isinstance(arr, list): for t in arr: if t: tagcount[t] = tagcount.get(t, 0) + 1 for t, c in sorted(tagcount.items(), key=lambda kv: -kv[1])[:30]: say(f"| {t} | {c:,} |") say(f"\n（不同标签合计 {len(tagcount):,} 个）\n") # ── 歌手 ── singer_total = q(conn, "select count(*) from singers")[0][0] singer_named = q(conn, "select count(*) from singers where name is not null and name != ''")[0][0] rel_total = q(conn, "select count(*) from song_singer_relations")[0][0] songs_with_singer = q(conn, "select count(distinct song_id) from song_singer_relations")[0][0] say("## 三、歌手") say() say(f"- 歌手表总数：**{singer_total:,}**（其中有名字的 {singer_named:,}）") say(f"- 歌曲-歌手关联：**{rel_total:,}** 条，覆盖 **{songs_with_singer:,}** 首歌" f"（占 {songs_with_singer*100/total:.1f}%）") say() top = q(conn, "select s.name, count(*) c from song_singer_relations r join singers s on s.id=r.singer_id " "where s.name is not null and s.name != '' group by s.id order by c desc limit 30") say("| 歌手 | 歌曲数 |") say("|---|---:|") for n, c in top: say(f"| {n} | {c:,} |") say() # ── 其它维度 ── say("## 四、其它维度") say() say("| 维度 | 统计 |") say("|---|---|") def one(sql): return q(conn, sql)[0][0] # ⚠️ 不同曲库版本的 songs 列不同（261009 版删掉了 cloud_url / block_status / singer_names 等）， # 这里只在"列确实存在"时才统计，避免换库后脚本直接崩。 if has("cloud_url"): say(f"| 有 cloud_url 的歌曲 | {one('select count(*) from songs where cloud_url is not null and cloud_url != \"\"'):,} |") else: say("| 有 cloud_url 的歌曲 | （本版库无该列）|") say(f"| filename 形如 `<编号>.ts` | {one(chr(39).join(['select count(*) from songs where filename like ','%.ts',''])):,} |") say(f"| 可伴奏(accomp=1) | {one('select count(*) from songs where accomp=1'):,} |") say(f"| 有原唱(original=1) | {one('select count(*) from songs where original=1'):,} |") if has("block_status"): say(f"| 已下架(block_status 非空) | {one('select count(*) from songs where block_status is not null'):,} |") say(f"| available=1 | {one('select count(*) from songs where available=1'):,} |") say(f"| 歌单数 | {one('select count(*) from playlists'):,} |") say(f"| 歌单关联歌曲 | {one('select count(*) from playlist_songs'):,} |") say() say("### mv_type 分布") say() say("| mv_type | 歌曲数 |") say("|---|---:|") for v, c in q(conn, "select mv_type, count(*) c from songs group by mv_type order by c desc"): say(f"| {v} | {c:,} |") say() # ── 清单导出 ── say("## 五、导出清单（CSV）") say() limit_sql = f" limit {args.limit}" if args.limit else "" cloud_col = ", coalesce(s.cloud_url, '')" if has("cloud_url") else "" base_sql = f""" select s.tid, s.name, coalesce(nullif(s.lang,''),''), s.filename, coalesce(s.album_name,''), coalesce((select group_concat(sg.name, ' / ') from song_singer_relations r join singers sg on sg.id = r.singer_id where r.song_id = s.id and sg.name is not null and sg.name != ''), ''), coalesce(s.hot_score, 0){cloud_col} from songs s where s.deleted_at is null order by s.tid{limit_sql} """ headers = ["编号", "歌名", "语言", "文件名", "专辑", "歌手", "热度"] if has("cloud_url"): headers.append("cloud_url") all_path = os.path.join(OUT, "songs-all.csv") rows = q(conn, base_sql) with open(all_path, "w", newline="", encoding="utf-8-sig") as fh: w = csv.writer(fh) w.writerow(headers) w.writerows(rows) say(f"- `songs-all.csv`：**{len(rows):,}** 行（{os.path.getsize(all_path)/1048576:.1f} MB）") # 按语言拆分 def norm_lang(v): k = (v or "").strip() if k in CANON: return k if k in ALIAS: return ALIAS[k] return "未标注" if not k else "其它" by_lang = {} for r in rows: by_lang.setdefault(norm_lang(r[2]), []).append(r) for lang, rs in sorted(by_lang.items(), key=lambda kv: -len(kv[1])): safe = re.sub(r"[^\w\u4e00-\u9fff-]", "_", lang) path = os.path.join(OUT, f"songs-{safe}.csv") with open(path, "w", newline="", encoding="utf-8-sig") as fh: w = csv.writer(fh) w.writerow(headers) w.writerows(rs) say(f"- `songs-{safe}.csv`：{len(rs):,} 行") # 歌手榜 sp = os.path.join(OUT, "singers-top.csv") with open(sp, "w", newline="", encoding="utf-8-sig") as fh: w = csv.writer(fh) w.writerow(["歌手", "歌曲数"]) for n, c in q(conn, "select s.name, count(*) c from song_singer_relations r join singers s on s.id=r.singer_id " "where s.name is not null and s.name != '' group by s.id order by c desc"): w.writerow([n, c]) say(f"- `singers-top.csv`：{os.path.getsize(sp)/1024:.0f} KB") # 歌单 pl = os.path.join(OUT, "playlists.csv") with open(pl, "w", newline="", encoding="utf-8-sig") as fh: w = csv.writer(fh) w.writerow(["歌单", "歌曲数", "热度"]) with_songs = "select p.name, count(r.song_id), coalesce(p.hot_score,0) from playlists p " \ "left join playlist_songs r on r.playlist_id = p.id group by p.id order by 2 desc" w.writerows(q(conn, with_songs)) say(f"- `playlists.csv`：{os.path.getsize(pl)/1024:.0f} KB") say() say(f"（耗时 {time.time()-t0:.1f}s）") with open(os.path.join(OUT, "song-stats.md"), "w", encoding="utf-8") as fh: fh.write("\n".join(L) + "\n") print(f"\n报告: {os.path.join(OUT, 'song-stats.md')}") if __name__ == "__main__": main()