#!/usr/bin/env python3 """歌曲表 CSV 导入的**编码 / 分隔符 / 表头 / 编号**容错测试（纯内存字节夹具，不联网、不碰真实数据）。 背景（真实用户事故）：用户把服务器导出的 `keeps.csv` 用 **Office 2019 中文版**编辑后另存， 编码变成 GB18030（无 BOM）、分隔符从逗号变成 TAB、行尾 CRLF；导入端当时固定按 `utf-8-sig + 逗号` 解析 → 整行被当成一个字段、中文全乱码，这就是用户看到的"表格错乱"。 本文件用**用户那份文件的真实字节**钉死修复（base64 内嵌，逐字节比对见 `KEEPS_GBK_TAB_B64`）：不经任何手工转换也必须能直接导入。 覆盖点： 1) GB18030 + TAB + CRLF + 无 BOM（用户真实字节）； 2) 分号 `;` + UTF-8 无 BOM； 3) UTF-16（带 BOM，LE/BE）+ TAB； 4) 没有表头（首行就是编号）； 5) 带引号字段里内嵌分隔符 / 换行； 6) 科学计数法 / 非整数编号 → 进 `errors` 且有中文提示，其它行照常导入； 7) 空文件 / 只有表头 → merge 不报错、replace 返回 `replace_cancelled=True`； 8) 表头别名（编号/tid/TID/ID/id/歌曲编号、全半角、空白、BOM、零宽字符）； 9) HTTP 层 `/api/keeps/import` 直接 POST 原始 GBK 字节也能导入（**不能在 HTTP 层丢字节**）。 """ import base64 import codecs import json import os import sys import tempfile import unittest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from ktv import admin from ktv.config import Config from ktv.store import CSV_TID_EXCEL_HINT, Store, decode_csv_bytes, detect_csv_delimiter, parse_keep_tid from ktv.web import Request # ★ 用户上传的那份 keeps.csv 的真实字节（224 字节）： # GB18030/GBK 无 BOM + TAB 分隔 + CRLF 行尾，4 个编号 4100081 / 4100293 / 9999999 / 8897358。 KEEPS_GBK_TAB_B64 = ( "seC6xQm46MP7CdPv0dQJ16i8rQm46MrWCc7EvP7D+wnIyLbICc/C1NjXtMysCdfWvdoNCjQxMDAw" "ODEJucPE78Tj1NrO0tDEyc8JufrT7wkJzfu6o7jfuOgJNDEwMDA4MS50cwkyMTQ3NDgzNjQ3CbT9" "z8LU2AkwDQo0MTAwMjkzCcHos78JufrT7wkJzfu6o7jfuOgJNDEwMDI5My50cwkyMTQ3NDgzNjQ3" "CbT9z8LU2AkwDQo5OTk5OTk5CQkJCQkJMAm0/c/C1NgJMA0KODg5NzM1OAkJCQkJCQkJDQo=" ) # 原始附件位置：本机还在时逐字节比对（证明内嵌夹具就是用户那份文件，不在时跳过） REAL_ATTACHMENT = ("/home/zijun/.dsh/attachments/v1/files/15/" "152b1268054c52292a943bc962ca829f602ae9b99c8ac6b07fd202b5490a8c37/keeps (1).csv") def req(method, path, query=None, body=b""): return Request(method, path, query or {}, {}, body) def jbody(data: dict) -> bytes: return json.dumps(data, ensure_ascii=False).encode() class ImportCase(unittest.TestCase): """公共夹具：空库 + 一个进程内 SQLite（全在临时目录，不碰真实数据目录）。""" def setUp(self): self.tmp = tempfile.mkdtemp() self.cfg = Config(os.path.join(self.tmp, "c.json")) self.cfg.update({"data_dir": self.tmp, "media_dir": os.path.join(self.tmp, "media"), "tmp_dir": os.path.join(self.tmp, "t")}) self.store = Store(os.path.join(self.tmp, "t.db")) def tearDown(self): try: self.store.conn.close() except Exception: pass def lib(self, *specs): """往曲库放几首（(编号, 歌名, 语言, 歌手, 热度)），用于验证导入后字段能自动补全。""" self.store.import_library([(tid, name, lang, "", singers, f"{tid}.ts", hot, "") for tid, name, lang, singers, hot in specs], "v1") def tids(self) -> list[int]: return sorted(r["tid"] for r in self.store.keep_rows(page_size=100)["items"]) def keeps_state(self, tid: int) -> str: row = self.store.q1("SELECT state FROM keeps WHERE tid=?", (tid,)) return row["state"] if row else "" def note_of(self, tid: int) -> str: return self.store.q1("SELECT note FROM keeps WHERE tid=?", (tid,))["note"] class TestUserExcelFile(ImportCase): """1) 用户那份文件的**真实字节**：GB18030 + TAB + CRLF + 无 BOM。""" def test_fixture_matches_real_attachment_when_present(self): raw = base64.b64decode(KEEPS_GBK_TAB_B64) self.assertEqual(len(raw), 224) self.assertFalse(raw.startswith(codecs.BOM_UTF8), "用户文件没有 BOM") with self.assertRaises(UnicodeDecodeError): raw.decode("utf-8") # 严格 UTF-8 解不了 → 必须回退 gb18030 self.assertEqual((raw[0:4], b"\t" in raw, b"\r\n" in raw), (b"\xb1\xe0\xba\xc5", True, True)) if os.path.isfile(REAL_ATTACHMENT): # 本机留着原始附件 → 逐字节比对 with open(REAL_ATTACHMENT, "rb") as fh: self.assertEqual(fh.read(), raw, "内嵌夹具必须与用户上传的原始文件完全一致") def test_gbk_tab_crlf_imports_without_manual_conversion(self): raw = base64.b64decode(KEEPS_GBK_TAB_B64) # 编码识别：中文歌名/歌手要从 GBK 字节里正确解出来（不是 U+FFFD） text, enc = decode_csv_bytes(raw) self.assertEqual(enc, "gb18030") self.assertIn("姑娘你在我心上", text) self.assertIn("望海高歌", text) self.assertNotIn("\ufffd", text) self.assertEqual(detect_csv_delimiter(text), "\t") # 导入：4 个编号全在，没有跳过、没有错误 res = self.store.keep_import_csv(raw, mode="merge") self.assertEqual(res["detected"], {"encoding": "gb18030", "delimiter": "\t"}) self.assertEqual((res["updated"], res["skipped"], res["errors"], res["replace_cancelled"]), (4, 0, [], False)) self.assertEqual(self.tids(), [4100081, 4100293, 8897358, 9999999]) for tid in (4100081, 4100293, 9999999, 8897358): self.assertEqual(self.keeps_state(tid), "want") # 「待下载」/空状态都按待下载 # 曲库有这两个编号时，中文歌名立刻补全（证明 TAB 被正确切列、GBK 被正确解码） self.lib((4100081, "姑娘你在我心上", "国语", "望海高歌", 2147483647), (4100293, "凌晨", "国语", "望海高歌", 2147483647)) items = {r["tid"]: r for r in self.store.keep_rows(page_size=100)["items"]} self.assertEqual(items[4100081]["name"], "姑娘你在我心上") self.assertEqual(items[4100081]["singers"], "望海高歌") self.assertEqual(items[4100081]["lang"], "国语") self.assertEqual(items[4100293]["name"], "凌晨") self.assertIs(items[4100081]["matched"], True) self.assertIs(items[9999999]["matched"], False) # 导出格式**保持不变**（utf-8-sig + 逗号由 HTTP 层负责；这里是逗号 + 中文表头） self.assertEqual(self.store.keep_csv().splitlines()[0], "编号,歌名,语言,专辑,歌手,文件名,热度,下载状态,字节") def test_admin_endpoint_keeps_raw_bytes(self): """★ 回归：HTTP 层若先按 UTF-8 `errors='replace'` 解码，GBK 字节就永久丢了。""" router = admin.build_router(self.cfg, self.store, None) raw = base64.b64decode(KEEPS_GBK_TAB_B64) resp = router.dispatch(req("POST", "/api/keeps/import", {"mode": "merge"}, raw)) d = json.loads(resp.body) self.assertTrue(d["ok"]) self.assertEqual(d["detected"], {"encoding": "gb18030", "delimiter": "\t"}) self.assertEqual((d["updated"], d["errors"]), (4, [])) self.assertEqual(d["stats"]["total"], 4) # JSON 信封（管理台/脚本）仍然可用，返回里同样带 detected d = json.loads(router.dispatch(req("POST", "/api/keeps/import", {"mode": "replace"}, jbody({"text": "编号\n7\n", "mode": "replace"}))).body) self.assertEqual((d["mode"], d["removed"], d["updated"]), ("replace", 4, 1)) self.assertEqual(d["detected"], {"encoding": "utf-8", "delimiter": ","}) class TestDelimiterAndEncoding(ImportCase): """2)3) 分隔符与编码的其它组合。""" def test_semicolon_utf8_without_bom(self): raw = "编号;下载状态\n1;待下载\n2;已下\n".encode("utf-8") res = self.store.keep_import_csv(raw, mode="merge") self.assertEqual(res["detected"], {"encoding": "utf-8", "delimiter": ";"}) self.assertEqual((res["updated"], res["errors"]), (2, [])) self.assertEqual((self.keeps_state(1), self.keeps_state(2)), ("want", "done")) def test_pipe_utf8_sig(self): raw = "\ufeff编号|下载状态\r\n1|已下\r\n".encode("utf-8") res = self.store.keep_import_csv(raw, mode="merge") self.assertEqual(res["detected"], {"encoding": "utf-8-sig", "delimiter": "|"}) self.assertEqual((res["updated"], self.keeps_state(1)), (1, "done")) def test_utf16_tab_with_bom(self): text = "编号\t下载状态\n1\t待下载\n2\t已下\n" for enc, name in (("utf-16", "utf-16"), ("utf-16-le", "utf-16"), ("utf-16-be", "utf-16")): with self.subTest(enc=enc): self.store.ex("DELETE FROM keeps") if enc == "utf-16": raw = text.encode(enc) # 自带 BOM elif enc == "utf-16-le": raw = codecs.BOM_UTF16_LE + text.encode(enc) else: raw = codecs.BOM_UTF16_BE + text.encode(enc) res = self.store.keep_import_csv(raw, mode="replace") self.assertEqual(res["detected"], {"encoding": name, "delimiter": "\t"}) self.assertEqual(res["updated"], 2) self.assertEqual(self.keeps_state(2), "done") def test_gb18030_comma_and_gbk_semicolon(self): res = self.store.keep_import_csv("编号,下载状态\n11,已下\n".encode("gb18030"), mode="merge") self.assertEqual(res["detected"], {"encoding": "gb18030", "delimiter": ","}) res = self.store.keep_import_csv("编号;下载状态\n12;已下\n".encode("gbk"), mode="merge") self.assertEqual(res["detected"], {"encoding": "gb18030", "delimiter": ";"}) self.assertEqual(sorted(self.tids()), [11, 12]) def test_line_endings_cr_lf_crlf(self): for text in ("编号\n1\n2\n", "编号\r\n1\r\n2\r\n", "编号\r1\r2\r"): with self.subTest(text=text[-6:]): self.store.ex("DELETE FROM keeps") self.assertEqual(self.store.keep_import_csv(text, mode="replace")["updated"], 2) class TestHeaderAndTidTolerance(ImportCase): """4)8) 无表头 / 表头别名 / 不可见字符；编号的容错解析。""" def test_no_header_first_row_is_tid(self): res = self.store.keep_import_csv(b"1\n2\n3\n", mode="replace") self.assertEqual((res["detected"]["delimiter"], res["updated"], res["errors"]), (",", 3, [])) self.assertEqual(self.tids(), [1, 2, 3]) # 无表头 + TAB + 多余列：第 1 列是编号，其余列忽略（无表头时无法知道哪列是备注） self.store.ex("DELETE FROM keeps") res = self.store.keep_import_csv("5\t备注A\t多余列\n6\t备注B\n".encode("utf-8"), mode="merge") self.assertEqual((res["detected"]["delimiter"], res["updated"], res["errors"]), ("\t", 2, [])) self.assertEqual(self.tids(), [5, 6]) # 无表头 + 最前面留了空行：仍然要能认出来（Excel 另存偶尔会多一个空行） self.store.ex("DELETE FROM keeps") res = self.store.keep_import_csv(b"\n\n7\n8\n", mode="merge") self.assertEqual((res["updated"], res["errors"]), (2, [])) self.assertEqual(self.tids(), [7, 8]) def test_no_header_with_decimal_tid(self): res = self.store.keep_import_csv("4100081.0\n4.100293E+06\n".encode("utf-8"), mode="merge") self.assertEqual((res["updated"], res["errors"]), (2, [])) self.assertEqual(self.tids(), [4100081, 4100293]) def test_header_aliases_and_invisible_chars(self): for header in ("\ufeff编号", " 编号 ", "TID", "tid", "ID", "id", "歌曲编号", "ＴＩＤ", "\u200b编号\u3000"): with self.subTest(header=header): self.store.ex("DELETE FROM keeps") raw = f"{header},下载状态\n 1 ,待下载\n".encode("utf-8") res = self.store.keep_import_csv(raw, mode="merge") self.assertEqual((res["updated"], res["errors"]), (1, []), f"表头 {header!r} 没认出来") self.assertEqual(self.tids(), [1]) # 状态/备注/用户列的别名 + 字段前后空白与不可见字符清理 self.store.ex("DELETE FROM keeps") raw = "编号,状态,备注,用户\n\u200b1\u3000,已下, 客户点的 ,\ufeff前台\n".encode("utf-8") res = self.store.keep_import_csv(raw, mode="merge") self.assertEqual(res["errors"], []) self.assertEqual((self.keeps_state(1), self.note_of(1)), ("done", "客户点的")) self.assertEqual(self.store.q1("SELECT user FROM keeps WHERE tid=1")["user"], "前台") def test_header_only_and_empty_file(self): res = self.store.keep_import_csv(b"", mode="merge") self.assertEqual(res["updated"], 0) self.assertTrue(res["errors"]) res = self.store.keep_import_csv(b"\xef\xbb\xbf", mode="merge") # 只有 BOM self.assertEqual(res["updated"], 0) self.assertTrue(res["errors"]) res = self.store.keep_import_csv("编号\n", mode="replace") # 只有表头 self.assertTrue(res["replace_cancelled"]) self.assertEqual((res["updated"], res["removed"]), (0, 0)) def test_only_header_with_extra_columns_is_still_cancelled(self): self.store.keep_add([1, 2]) res = self.store.keep_import_csv("编号\t歌名\t歌手\n".encode("gb18030"), mode="replace") self.assertTrue(res["replace_cancelled"]) self.assertEqual(self.store.keep_stats()["total"], 2) # 一行都没丢 class TestQuotedFields(ImportCase): """5) 带引号字段里内嵌分隔符 / 换行。""" def test_quoted_embedded_delimiter_and_newline(self): raw = '编号,备注\n1,"含,逗号"\n2,"跨\n行备注"\n3,"含""引号"""\n'.encode("utf-8") res = self.store.keep_import_csv(raw, mode="merge") self.assertEqual((res["detected"]["delimiter"], res["updated"], res["skipped"], res["errors"]), (",", 3, 0, [])) self.assertEqual(self.tids(), [1, 2, 3]) self.assertEqual(self.note_of(1), "含,逗号") self.assertEqual(self.note_of(2), "跨\n行备注") self.assertEqual(self.note_of(3), '含"引号"') def test_quoted_embedded_tab_when_delimiter_is_tab(self): raw = '编号\t备注\n1\t"含\tTAB 的备注"\n'.encode("utf-8") res = self.store.keep_import_csv(raw, mode="merge") self.assertEqual((res["detected"]["delimiter"], res["updated"]), ("\t", 1)) self.assertEqual(self.note_of(1), "含\tTAB 的备注") class TestTidNumericTolerance(ImportCase): """6) 科学计数法 / 非整数编号：进 errors 且有中文提示，其它行照常导入。""" def test_excel_numeric_rewrite(self): raw = ("编号,下载状态\n" "1.0,待下载\n" # 整数值 → 接受 "2,已下\n" # 纯数字 → 接受 "4.100081E+06,待下载\n" # 有效数字 7 位 → 精确整数，接受 "4.10008E+06,待下载\n" # Excel 按显示精度改写 → 拒绝 + 中文提示 "1.5,待下载\n" # 非整数 → 拒绝 "abc,待下载\n" # 文本 → 拒绝 "3,啥玩意\n").encode("utf-8") # 状态无法识别 → 拒绝 res = self.store.keep_import_csv(raw, mode="merge") self.assertEqual((res["updated"], res["skipped"]), (3, 4)) self.assertEqual(self.tids(), [1, 2, 4100081]) self.assertEqual((self.keeps_state(1), self.keeps_state(2)), ("want", "done")) self.assertTrue(any(CSV_TID_EXCEL_HINT in e for e in res["errors"]), f"缺少 Excel 改写的中文提示：{res['errors']}") self.assertTrue(any("第 5 行" in e and "4.10008E+06" in e for e in res["errors"])) self.assertTrue(any("第 6 行" in e and "不是整数" in e for e in res["errors"])) self.assertTrue(any("第 7 行" in e and "不是数字" in e for e in res["errors"])) self.assertTrue(any("第 8 行" in e and "状态无法识别" in e for e in res["errors"])) # 提示里必须告诉用户怎么办（把该列设为"文本"后重新另存） self.assertIn("文本", CSV_TID_EXCEL_HINT) def test_error_list_is_capped_but_other_rows_still_import(self): raw = ("编号\n" + "".join(f"x{i}\n" for i in range(20)) + "123\n").encode("utf-8") res = self.store.keep_import_csv(raw, mode="merge") self.assertEqual((res["updated"], res["skipped"]), (1, 20)) self.assertLessEqual(len(res["errors"]), 5) def test_replace_refused_when_all_tids_unusable(self): self.store.keep_add([1, 2]) res = self.store.keep_import_csv("编号\n4.10008E+06\n".encode("utf-8"), mode="replace") self.assertTrue(res["replace_cancelled"]) self.assertTrue(any(CSV_TID_EXCEL_HINT in e for e in res["errors"])) self.assertEqual(sorted(self.tids()), [1, 2]) # 旧行保留 class TestDetectionHelpers(unittest.TestCase): """识别函数本身的行为（模块级，便于单独回归）。""" def test_decode_csv_bytes_strict_then_gb18030(self): self.assertEqual(decode_csv_bytes("编号".encode("gb18030")), ("编号", "gb18030")) self.assertEqual(decode_csv_bytes("编号".encode("utf-8")), ("编号", "utf-8")) self.assertEqual(decode_csv_bytes("\ufeff编号".encode("utf-8")), ("编号", "utf-8-sig")) self.assertEqual(decode_csv_bytes("编号".encode("utf-16")), ("编号", "utf-16")) self.assertEqual(decode_csv_bytes(b""), ("", "utf-8")) self.assertEqual(decode_csv_bytes(b"\xff\xfe\x00"), ("\ufffd", "utf-16")) # BOM + 残字节 → 兜底 def test_detect_csv_delimiter_defaults_to_comma(self): self.assertEqual(detect_csv_delimiter("编号\n1\n"), ",") self.assertEqual(detect_csv_delimiter(""), ",") self.assertEqual(detect_csv_delimiter("编号;歌名;语言\n"), ";") self.assertEqual(detect_csv_delimiter("编号\t歌名\n"), "\t") self.assertEqual(detect_csv_delimiter("编号|歌名\n"), "|") self.assertEqual(detect_csv_delimiter("\n\n编号,歌名;语言\n"), ",") # 跳过空行；并列取逗号 def test_parse_keep_tid(self): self.assertEqual(parse_keep_tid("4100081"), (4100081, "")) self.assertEqual(parse_keep_tid(" 4100081.0 "), (4100081, "")) self.assertEqual(parse_keep_tid("4.100081E+06"), (4100081, "")) self.assertEqual(parse_keep_tid("4100081e0"), (4100081, "")) self.assertEqual(parse_keep_tid("4.10008E+06")[0], None) self.assertEqual(parse_keep_tid("4.10008E+06")[1], CSV_TID_EXCEL_HINT) self.assertEqual(parse_keep_tid("1.5")[0], None) self.assertEqual(parse_keep_tid("abc")[0], None) self.assertEqual(parse_keep_tid("")[0], None) self.assertEqual(parse_keep_tid("nan")[0], None) self.assertEqual(parse_keep_tid("inf")[0], None) self.assertEqual(parse_keep_tid("\u200b100\u3000"), (100, "")) # 不可见字符 class TestUiImportHint(unittest.TestCase): """6) 前端提示：导入区说明支持的编码/分隔符；导入成功提示显示识别结果 + 上传原始字节。""" def test_import_area_documents_encodings_and_delimiters(self): from ktv import ui self.assertIn("支持 UTF-8/GBK 等编码，逗号/分号/TAB 分隔均可（Excel 另存后可直接导入）", ui.PAGE) def test_import_uploads_raw_bytes_and_shows_detected(self): from ktv import ui js = ui.PAGE fn = js[js.index("async function importKeeps(el)"):] fn = fn[:fn.index("// ── 下载队列 ──")] code = "\n".join(l for l in fn.splitlines() if not l.strip().startswith("//")) # 去掉注释行 self.assertIn("f.arrayBuffer()", code) # ★ 必须传原始字节：f.text() 会按 UTF-8 吞掉 GBK 字节 self.assertNotIn("f.text()", code) self.assertIn("raw:true", code) self.assertIn("r.detected", fn) self.assertIn("识别到编码", fn) self.assertIn("raw === true", js) # api() 支持原始字节上传 self.assertIn("'text/csv'", js) if __name__ == "__main__": unittest.main(verbosity=2)