音乐库整理与播放实战：950+ 首曲库的整理心法、踩坑与 APlayer 落地

补全专辑 → 校准封面 → 全库验收 → 按报告定点修复 → 再验收 → 发布目录 → APlayer 播放

ffmpeg -i input.flac -map_metadata -1 -codec:a libmp3lame \
       -b:a 320k -ar 44100 -ac 2 -id3v2_version 3 output.mp3

cand = old.encode('latin-1').decode('gbk')   # 逐个试 cp437 / cp1252 / latin-1
# 能解出连续汉字，就判定为乱码并写回

_原件备份_FLAC/      ← 无损备份，别进手机
_cover_fix/          ← 换下来的旧封面
__flac_tmp_*.mp3     ← 转码临时文件
*.csv 日志           ← 验收报告、校准日志

补全专辑 → 校准封面 → 全库验收 → 按报告定点修复 → 再验收

ffmpeg -i input.flac -map_metadata -1 -codec:a libmp3lame \
       -b:a 320k -ar 44100 -ac 2 -id3v2_version 3 output.mp3

cand = old.encode('latin-1').decode('gbk')   # 逐个试 cp437 / cp1252 / latin-1
# 能解出连续汉字，就判定为乱码并写回

pip install mutagen pillow requests imageio-ffmpeg

# -*- coding: utf-8 -*-
"""全库最终验收：逐项核对历史全部要求（含 FLAC 自动转码）"""
import os, re, io, csv, unicodedata, hashlib, shutil, subprocess
from collections import defaultdict, Counter
from mutagen.mp3 import MP3
from mutagen.id3 import ID3, ID3NoHeaderError, TIT2, TPE1, TALB, APIC
from mutagen.flac import FLAC
from PIL import Image

ROOT = 'W:/'
OUT = 'W:/最终验收报告.csv'
FLAC_BAK = os.path.join(ROOT, '_原件备份_FLAC')

NAME_RE = re.compile(r'^(.+?) - (.+)\.mp3$', re.I)
BAD_CHARS = re.compile(r'[\\/:*?"<>|\t\r\n]')
MOJI_HINT = re.compile(r'[ÃÂåäæçèéêëìíîïðñòóôõöøùúûüýþ]')
KEEP = {'TIT2', 'TPE1', 'TALB', 'APIC'}

def has_hidden(s):
    if s != s.strip(): return True
    if '\u00a0' in s: return True
    for ch in s:
        if unicodedata.category(ch) in ('Cf', 'Cc', 'Co', 'Cn'):
            return True
    return False

def mojibake(s):
    """判断是否为编码错位文本"""
    if not s: return False
    if MOJI_HINT.search(s): return True
    try:
        fixed = s.encode('cp437').decode('gbk')
        if fixed != s and re.search(r'[\u4e00-\u9fff]', fixed):
            return True
    except Exception:
        pass
    return False

def norm_title(t):
    t = t.lower().strip()
    t = re.sub(r'\(.*?\)|（.*?）|\[.*?\]|【.*?】', '', t)
    t = re.sub(r'\s*(feat\.?|ft\.?|remix|live|伴奏|dj版)\s*.*$', '', t)
    t = re.sub(r'[^\w\u4e00-\u9fff]', '', t)
    return t

def norm_artist(a):
    a = a.lower()
    a = re.sub(r'[^\w\u4e00-\u9fff]', '', a)
    return a

def flac_cover_data(fl):
    """取 FLAC 第一张图，规整为 JPEG 字节"""
    if not fl.pictures:
        return None
    try:
        im = Image.open(io.BytesIO(fl.pictures[0].data)).convert('RGB')
        buf = io.BytesIO(); im.save(buf, 'JPEG', quality=92)
        return buf.getvalue()
    except Exception:
        return None

def convert_one_flac(src, ff):
    """把单个 FLAC 转成 320k mp3（>13MB 时降为 192k），写标准标签+封面"""
    fn = os.path.basename(src)
    try:
        fl = FLAC(src)
    except Exception as e:
        print('  [跳过] FLAC 解析失败:', fn, e); return None
    tags = fl.tags or {}
    g = lambda k: (tags.get(k) or [''])[0].strip()
    title, artist, album = g('title'), g('artist'), g('album')
    if not (title and artist):
        mfn = re.match(r'^(.+?) - (.+)$', fn[:-5].strip())
        if mfn:
            artist = artist or mfn.group(1).strip()
            title = title or mfn.group(2).strip()
    title = title or fn[:-5]
    artist = artist or '未知歌手'
    out_name = BAD_CHARS.sub('_', '%s - %s.mp3' % (artist, title))
    dst = os.path.join(ROOT, out_name)
    tmp = os.path.join(ROOT, '__flac_tmp_%s.mp3' % hashlib.md5(fn.encode()).hexdigest()[:8])

    br = 320
    cmd = [ff, '-y', '-i', src, '-map_metadata', '-1', '-codec:a', 'libmp3lame',
           '-b:a', '%dk' % br, '-ar', '44100', '-ac', '2', '-id3v2_version', '3', tmp]
    r = subprocess.run(cmd, capture_output=True)
    if r.returncode != 0:
        print('  [跳过] ffmpeg 失败:', fn); return None
    if os.path.getsize(tmp) / 1048576.0 > 13:
        br = 192
        cmd = [ff, '-y', '-i', src, '-map_metadata', '-1', '-codec:a', 'libmp3lame',
               '-b:a', '%dk' % br, '-ar', '44100', '-ac', '2', '-id3v2_version', '3', tmp]
        r = subprocess.run(cmd, capture_output=True)
        if r.returncode != 0:
            print('  [跳过] ffmpeg 失败:', fn); return None

    try:
        t = ID3(tmp)
    except Exception:
        t = ID3()
    t.add(TIT2(encoding=3, text=title))
    t.add(TPE1(encoding=3, text=artist))
    if album:
        t.add(TALB(encoding=3, text=album))
    cov = flac_cover_data(fl)
    if cov:
        t.add(APIC(encoding=3, mime='image/jpeg', type=3, desc='Cover', data=cov))
    t.save(tmp, v2_version=3)

    if os.path.exists(dst):
        base2, ext = os.path.splitext(dst)
        dst = base2 + '_2' + ext
    shutil.move(tmp, dst)
    os.makedirs(FLAC_BAK, exist_ok=True)
    try:
        shutil.move(src, os.path.join(FLAC_BAK, fn))
    except Exception:
        try: os.remove(src)
        except Exception: pass
    print('  [OK] %s -> %s (%.2fMB, %dk)' % (
        fn, os.path.basename(dst), os.path.getsize(dst) / 1048576.0, br))
    return dst

def convert_flac_files(ff):
    flacs = [f for f in sorted(os.listdir(ROOT)) if f.lower().endswith('.flac')]
    if not flacs:
        return
    print('发现 %d 个 FLAC，开始转码为 mp3 并移除原文件...' % len(flacs))
    for fn in flacs:
        convert_one_flac(os.path.join(ROOT, fn), ff)
    print('FLAC 转码完成。')

ff = None
try:
    import imageio_ffmpeg
    ff = imageio_ffmpeg.get_ffmpeg_exe()
except Exception:
    ff = None
if ff:
    convert_flac_files(ff)
else:
    print('警告：未找到 ffmpeg，跳过 FLAC 转码。')

rows = []
files = [f for f in sorted(os.listdir(ROOT)) if f.lower().endswith('.mp3')]
print('files:', len(files))

groups = defaultdict(list)
for i, fn in enumerate(files, 1):
    p = os.path.join(ROOT, fn)
    issues = []
    detail = []
    stem = fn[:-4]

    size_mb = os.path.getsize(p) / 1048576.0
    if size_mb > 13:
        issues.append('体积超标'); detail.append('%.1fMB' % size_mb)

    m = NAME_RE.match(fn)
    if not m:
        issues.append('文件名格式')
    else:
        ar_fn, ti_fn = m.group(1).strip(), m.group(2).strip()
    if BAD_CHARS.search(fn): issues.append('非法字符')
    if has_hidden(stem): issues.append('隐藏字符')
    if re.search(r'[\(（\[【]\s*(ost|Ost|OST)\s*[\)）\]】]|\s+-\s*(ost|OST)\s*$', stem):
        issues.append('OST残留')
    if re.search(r'_', stem): issues.append('下划线')

    try:
        tags = ID3(p)
    except ID3NoHeaderError:
        tags = None
    t_title = t_art = t_alb = ''
    has_cover = False
    cover_ok = ''
    extra = []
    if tags is None:
        issues.append('无ID3标签')
    else:
        t_title = str(tags.get('TIT2', ''))
        t_art = str(tags.get('TPE1', ''))
        t_alb = str(tags.get('TALB', ''))
        for k in tags.keys():
            if k.split(':')[0] not in KEEP:
                extra.append(k)
        if extra: issues.append('冗余帧%d' % len(extra))

        aps = tags.getall('APIC')
        if not aps:
            issues.append('无封面')
        else:
            has_cover = True
            ap = aps[0]
            try:
                im = Image.open(io.BytesIO(ap.data))
                w, h = im.size
                cover_ok = '%dx%d/%s/%dKB' % (w, h, im.format, len(ap.data)//1024)
                if min(w, h) < 300: issues.append('封面过小')
                if im.format not in ('JPEG', 'PNG'): issues.append('封面格式%s' % im.format)
            except Exception:
                issues.append('封面损坏')
                cover_ok = 'unreadable'

        for label, val in (('标题', t_title), ('歌手', t_art), ('专辑', t_alb)):
            if val and mojibake(val):
                issues.append('%s乱码' % label)

        if not t_title: issues.append('缺标题')
        if not t_art: issues.append('缺歌手')
        if m:
            if t_title.strip() != ti_fn:
                issues.append('标题≠文件名'); detail.append('T[%s]≠F[%s]' % (t_title, ti_fn))
            if t_art.strip() != ar_fn:
                issues.append('歌手≠文件名'); detail.append('A[%s]≠F[%s]' % (t_art, ar_fn))

    with open(p, 'rb') as f:
        f.seek(-128, 2)
        if f.read(3) == b'TAG':
            issues.append('ID3v1残留')

    br = sr = ch = 0
    dur = 0
    try:
        au = MP3(p)
        br = au.info.bitrate // 1000
        sr = au.info.sample_rate
        ch = au.info.channels
        dur = au.info.length
        if br < 192: issues.append('码率%d' % br)
        if sr < 44100: issues.append('采样率%d' % sr)
    except Exception as e:
        issues.append('音频损坏')

    if m:
        groups[(norm_artist(ar_fn), norm_title(ti_fn))].append(fn)

    rows.append([fn, ' '.join(issues), ' | '.join(detail), br, sr, ch,
                 round(dur, 1), round(size_mb, 2), cover_ok, len(extra), t_alb, t_art, t_title])

dups = {k: v for k, v in groups.items() if len(v) > 1}

with open(OUT, 'w', newline='', encoding='utf-8-sig') as f:
    w = csv.writer(f)
    w.writerow(['文件名', '问题项', '细节', '码率kbps', '采样率', '声道', '时长s', '大小MB', '封面', '冗余帧数', '专辑', '歌手标签', '标题标签'])
    w.writerows(rows)

cnt = Counter()
clean = 0
for r in rows:
    if not r[1]:
        clean += 1
    for x in r[1].split():
        cnt[re.sub(r'\d+$', '', x)] += 1

print('=' * 50)
print('总数 %d  完全合格 %d  有问题 %d' % (len(rows), clean, len(rows) - clean))
for k, v in cnt.most_common():
    print('  %-12s %d' % (k, v))
print('重复组 %d' % len(dups))
for k, v in dups.items():
    print('  DUP:', ' || '.join(v))
print('报告:', OUT)

总数 959  完全合格 930  有问题 29
  码率192       28
  标题乱码      1
  采样率44100   1
重复组 4
报告: W:/最终验收报告.csv

# -*- coding: utf-8 -*-
"""批量查询缺失专辑信息（并发版）：QQ 主力，网易兜底，断点续传"""
import os, re, json, time, threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
from mutagen.id3 import ID3

ROOT = 'W:/'
CKPT = 'C:/tmp/album_ckpt.json'
UA = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}
QQ_H = dict(UA, Referer='https://y.qq.com/', **{'Content-Type': 'application/json'})
NE_H = dict(UA, Referer='https://music.163.com')
lock = threading.Lock()
ne_lock = threading.Lock()

def norm(s):
    return re.sub(r'[^\w\u4e00-\u9fff]', '', s.lower())

def strip_ver(t):
    return re.sub(r'[\(（\[【][^\)）\]】]*[\)）\]】]', '', t).strip()

sess = requests.Session()
sess.trust_env = False

def qq_search(q):
    body = json.dumps({"req_1": {"method": "DoSearchForQQMusicDesktop",
        "module": "music.search.SearchCgiService",
        "param": {"search_type": 0, "query": q, "page_num": 1, "num_per_page": 6}}}).encode()
    for _ in range(2):
        try:
            d = json.loads(sess.post('https://u.y.qq.com/cgi-bin/musicu.fcg',
                                     data=body, headers=QQ_H, timeout=12).text)
            lst = d.get('req_1', {}).get('data', {}).get('body', {}).get('song', {}).get('list', [])
            if lst: return lst
        except Exception:
            pass
        time.sleep(1.5)
    return []

def ne_search(q):
    with ne_lock:
        try:
            d = sess.get('https://music.163.com/api/search/get/web',
                         params={'s': q, 'type': 1, 'limit': 5}, headers=NE_H, timeout=12).json()
            r = d.get('result', {}).get('songs', [])
        except Exception:
            r = []
        time.sleep(1.0)
        return r

def pick_qq(lst, artist, title):
    na, nt = norm(artist), norm(title)
    best = None
    for s in lst:
        singers = norm('/'.join(a.get('name', '') for a in s.get('singer', [])))
        nm = norm(s.get('name', ''))
        if na and na not in singers and singers not in na: continue
        if nt and nt != nm and nt not in nm and nm not in nt: continue
        if s.get('album', {}).get('name'): return s
        best = best or s
    return best

def pick_ne(lst, artist, title):
    na, nt = norm(artist), norm(title)
    for s in lst:
        singers = norm('/'.join(a['name'] for a in s.get('artists', [])))
        nm = norm(s.get('name', ''))
        if na and na not in singers and singers not in na: continue
        if nt and nt != nm and nt not in nm and nm not in nt: continue
        if s.get('album', {}).get('name'): return s
    return None

def work(item):
    f, ar, ti = item
    rec = {'artist': ar, 'title': ti, 'album': '', 'src': '', 'cover': '', 'hit': ''}
    try:
        for q in (f'{ar} {ti}', f'{ar} {strip_ver(ti)}'):
            lst = qq_search(q)
            s = pick_qq(lst, ar, ti) if lst else None
            if s and s.get('album', {}).get('name'):
                al = s['album']
                rec.update(album=al['name'].strip(), src='qq',
                           cover=al.get('mid') or '', hit=s.get('name', ''))
                return f, rec
            time.sleep(0.4)
        for q in (f'{ar} {ti}', f'{ar} {strip_ver(ti)}'):
            lst = ne_search(q)
            s = pick_ne(lst, ar, ti) if lst else None
            if s:
                rec.update(album=s['album']['name'].strip(), src='ne',
                           cover=str(s['id']), hit=s.get('name', ''))
                return f, rec
    except Exception:
        pass
    return f, rec

ck = {}
if os.path.exists(CKPT):
    ck = json.load(open(CKPT, encoding='utf-8'))

files = [f for f in sorted(os.listdir(ROOT)) if f.lower().endswith('.mp3')]
todo = []
for f in files:
    if f in ck: continue
    try:
        t = ID3(os.path.join(ROOT, f))
    except Exception:
        continue
    if 'TALB' in t and str(t['TALB']).strip(): continue
    ar = str(t['TPE1']) if 'TPE1' in t else ''
    ti = str(t['TIT2']) if 'TIT2' in t else ''
    if ar and ti: todo.append((f, ar, ti))

print('已有 %d 条，待查 %d 首' % (len(ck), len(todo)), flush=True)
n = 0
with ThreadPoolExecutor(max_workers=5) as ex:
    futs = [ex.submit(work, it) for it in todo]
    for fu in as_completed(futs):
        f, rec = fu.result()
        ck[f] = rec
        n += 1
        if n % 20 == 0:
            with lock:
                json.dump(ck, open(CKPT, 'w', encoding='utf-8'), ensure_ascii=False)
            print('[%d/%d] 命中 %d' % (n, len(todo), sum(1 for v in ck.values() if v['album'])), flush=True)

json.dump(ck, open(CKPT, 'w', encoding='utf-8'), ensure_ascii=False)
print('DONE 总 %d，命中 %d' % (len(ck), sum(1 for v in ck.values() if v['album'])), flush=True)

# -*- coding: utf-8 -*-
"""按官方专辑信息批量校准封面：下载 -> 感知哈希比对 -> 差异大则替换"""
import os, io, json, time, csv, re
import requests
from mutagen.id3 import ID3, APIC
from PIL import Image

ROOT = 'W:/'
CKPT = 'C:/tmp/album_ckpt.json'
LOG = 'C:/tmp/cover_batch_ckpt.json'
UA = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}
QQ_H = dict(UA, Referer='https://y.qq.com/')
NE_H = dict(UA, Referer='https://music.163.com')

def dhash(im):
    im = im.convert('L').resize((9, 8), Image.LANCZOS)
    px = list(im.getdata())
    bits = []
    for r in range(8):
        for c in range(8):
            bits.append(px[r*9+c] < px[r*9+c+1])
    return bits

def ham(a, b):
    return sum(1 for x, y in zip(a, b) if x != y)

def get_qq(mid):
    for size in ('T002R500x500M000', 'T002R800x800M000'):
        try:
            b = requests.get('https://y.gtimg.cn/music/photo_new/%s%s.jpg' % (size, mid),
                             headers=QQ_H, timeout=20).content
            im = Image.open(io.BytesIO(b))
            if min(im.size) >= 400:
                return im, b
        except Exception:
            pass
    return None, None

def get_ne(song_id):
    try:
        d = requests.get('https://music.163.com/api/song/detail/?id=%s&ids=[%s]' % (song_id, song_id),
                         headers=NE_H, timeout=20).json()
        pic = d['songs'][0]['album'].get('picUrl')
        if not pic: return None, None
        b = requests.get(pic + '?param=700y700', headers=NE_H, timeout=25).content
        return Image.open(io.BytesIO(b)), b
    except Exception:
        return None, None

ck = json.load(open(CKPT, encoding='utf-8'))
done = {}
if os.path.exists(LOG):
    done = json.load(open(LOG, encoding='utf-8'))

rows = []
todo = [(f, r) for f, r in ck.items() if (r.get('album') or '').strip() and f not in done]
print('待校封面 %d 首' % len(todo), flush=True)

for i, (f, rec) in enumerate(todo, 1):
    p = os.path.join(ROOT, f)
    if not os.path.exists(p):
        continue
    try:
        t = ID3(p)
    except Exception:
        continue
    old = t.getall('APIC')
    old_im = None
    if old:
        try: old_im = Image.open(io.BytesIO(old[0].data))
        except Exception: old_im = None

    im = b = None
    src = rec.get('src')
    if src == 'qq' and rec.get('cover'):
        im, b = get_qq(rec['cover'])
    if im is None and rec.get('cover'):
        im, b = get_ne(rec['cover'])
    time.sleep(0.7)

    if im is None:
        done[f] = 'no_download'
        rows.append([f, rec.get('album'), '下载失败', '', ''])
    else:
        acted = 'kept'
        if old_im is None:
            acted = 'added'
        else:
            dh = ham(dhash(old_im), dhash(im))
            if dh > 8:
                acted = 'replaced'
            else:
                acted = 'same'
        if acted in ('added', 'replaced'):
            buf = io.BytesIO()
            im.convert('RGB').save(buf, 'JPEG', quality=92)
            nj = buf.getvalue()
            for k in list(t.keys()):
                if k.startswith('APIC'): del t[k]
            t.add(APIC(encoding=3, mime='image/jpeg', type=3, desc='Cover', data=nj))
            t.save(p, v2_version=3)
        done[f] = acted
        rows.append([f, rec.get('album'), acted,
                     '%dx%d' % im.size, dh if old_im else ''])

    if i % 10 == 0 or i == len(todo):
        json.dump(done, open(LOG, 'w', encoding='utf-8'), ensure_ascii=False)
        print('[%d/%d]' % (i, len(todo)), flush=True)

json.dump(done, open(LOG, 'w', encoding='utf-8'), ensure_ascii=False)
with open('W:/_封面校准日志.csv', 'w', newline='', encoding='utf-8-sig') as fp:
    w = csv.writer(fp)
    w.writerow(['文件', '专辑', '动作', '新图尺寸', '汉明距离'])
    w.writerows(rows)

from collections import Counter
print(Counter(done.values()))

def strip_v1(p):
    with open(p, 'rb') as f:
        f.seek(0, 2); size = f.tell()
        if size < 128: return False
        f.seek(-128, 2)
        if f.read(3) != b'TAG': return False
    with open(p, 'r+b') as f:
        f.truncate(size - 128)
    return True

def clean_frames(p, KEEP={'TIT2','TPE1','TALB','APIC'}):
    t = ID3(p); removed = []
    for k in list(t.keys()):
        if k.split(':')[0] not in KEEP:
            removed.append(k); del t[k]
    if removed: t.save(p, v2_version=3)
    return removed

old = str(t['TALB'])
fixed = None
for enc in ('latin-1', 'cp1252', 'cp437'):
    try:
        cand = old.encode(enc, errors='strict').decode('gbk')
        if re.search(r'[\u4e00-\u9fff]', cand):
            fixed = cand; break
    except Exception:
        continue
if fixed:
    t['TALB'] = TALB(encoding=3, text=fixed)
    t.save(p, v2_version=3)

# 转码
ff = imageio_ffmpeg.get_ffmpeg_exe()
cmd = [ff, '-y', '-i', SRC, '-codec:a', 'libmp3lame', '-b:a', '320k',
       '-ac', '2', '-ar', '44100', '-map_metadata', '-1', '-id3v2_version', '3', TMP]
subprocess.run(cmd, capture_output=True)

# 下载官方封面（QQ）
UA = {'User-Agent': 'Mozilla/5.0', 'Referer': 'https://y.qq.com/'}
b = requests.get('https://y.gtimg.cn/music/photo_new/T002R500x500M000%s.jpg' % MID,
                 headers=UA, timeout=20).content

# 写标准四帧
t = ID3()
t.add(TIT2(encoding=3, text=TITLE))
t.add(TPE1(encoding=3, text=ARTIST))
t.add(TALB(encoding=3, text=ALBUM))
t.add(APIC(encoding=3, mime='image/jpeg', type=3, desc='Cover', data=cover))
t.save(TMP, v2_version=3)

# 规范命名入库，原 FLAC 移备份
os.replace(TMP, FINAL)
shutil.move(SRC, os.path.join(bak, os.path.basename(SRC)))

1. 把所有音乐（含 FLAC）丢进 ROOT 目录，平铺。
2. 跑 album_query：补全专辑名（QQ 主力 + 网易兜底 + 断点续传）。
3. 跑 cover_batch：按官方图校准封面（感知哈希比对）。
4. 跑 final_audit：自动转码 FLAC + 全库逐项验收，生成 CSV 报告。
5. 看报告，按"问题项"列分批跑 fix_final：删 ID3v1、清冗余帧、修乱码、改命名。
6. 复查 final_audit，直到"完全合格"接近 100%。
7. 重复组人工确认，再决定是否删。