zijunchiang
87aa979abd
版本号：
- 1.5.11 越界（规则 3 规定 patch ≤ 10）→ 修正为 1.6.0（进位到 minor、patch 归 0）
- versionCode 43 不变：43 从未对外发布，本次修正只是把命名拉回约定
合并支线（原在工作区根目录、App 仓库之外，DESIGN.md 自称"支线"）：
- selfhost-source/server/ 自建歌曲源服务器（纯 Python 标准库、零 pip 依赖）
- selfhost-source/deploy/、DESIGN.md、LIBRARY.md、build_inventory.py
- selfhost-source/inventory/ 小清单（playlists.csv、song-stats.md 等）
- ⚠️ inventory 生成物（songs-*.csv / singers-top.csv，共 84MB）不入库，
已写入 .gitignore；需要时用 build_inventory.py 依据 muse.db 重新生成
同时入库的 App 侧支线改动（此前为工作树未提交状态）：
- Whoami.kt（whoami 设备心跳）、DeviceIdentity.kt（设备身份伪装）
- PushDownload.kt / PushDownloadClient.kt（点歌推送）
- AdVideoFilter 带 query/fragment 的地址闸门；KTV_BRIDGE_API 11 → 13
- 修 ktv_api.js 的 global.KtvBridgeApiVersion（12 → 13，与注释及 Kotlin 两处校验一致）
- 对应单测：WhoamiTest / DeviceIdentityTest / PushDownloadTest / AdVideoFilterTest（共 39 例）
文档：
- DESIGN.md 的"在 App 仓库之外"约定改为"已并入仓库统一版本化"
- AGENTS.md 版本号行更新为 1.6.0/43，并新增支线并入说明
验证：assembleRelease + testReleaseUnitTest 全过（39 例）
373 lines
21 KiB
Python
373 lines
21 KiB
Python
#!/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)
|