diff --git a/backend/app/data/csv_grid.py b/backend/app/data/csv_grid.py index 83e6177c..7fb8ece7 100644 --- a/backend/app/data/csv_grid.py +++ b/backend/app/data/csv_grid.py @@ -2,9 +2,12 @@ 和网格(Grid / 逐行流),之后的执行器、核对、说明、溯源、按期累积全部复用,不另写一套 CSV 的解析(否则会和配方 语言分叉,违背「一种说法」)。 -**怎么认出是 CSV。** 看内容不看扩展名(配方的几个入口只拿得到原件字节):不是 zip(xlsx)、不是旧版 .xls 的 OLE -文件头、前 64 KB 里没有 NUL 字节(图片、PDF 这类二进制文件都有),就按 CSV 读。编码识别、分隔符嗅探复用 tabular -读 CSV 的那一段(UTF-8 / GB18030 / 带 BOM 的 UTF-16,逗号、分号、制表符、竖线)。 +**怎么认出是 CSV。** 和简单导入(tabular._CSV)一样按扩展名定:只有 .csv / .tsv 按 CSV 读,并且内容得是文字 +(不是 zip、不是旧版 .xls 的 OLE 文件头、前 64 KB 里没有 NUL 字节)。以前只看内容,企业系统常见的「HTML 表格冒充 +.xls」、误传的文本文件都会被拆成一张叫「CSV」的工作表去起草、核对,不再像以前那样明确拒收;现在别的扩展名照旧交给 +xlsx_scan 拒收,内容是网页或 XML 的另给一句能照着做的话。编码识别复用 tabular 读 CSV 的那一段(UTF-8 / GB18030 / +带 BOM 的 UTF-16);.tsv 一律按制表符分隔(同 tabular),.csv 嗅探逗号、分号、制表符、竖线。扫描时定下的分隔符 +记进工作表的 path(CSV_PATH 后面跟分隔符),之后读格子用同一个,不再各嗅各的。 **坐标沿用「工作表!A1」。** 细案原写「格子坐标记成 R{行}C{列}」,这里没这么做(见交付说明的偏离项):坐标在执行器、 核对、修复按钮、框选、溯源、裁判摘录、前端网格里一共有十来处解析和拼写,引入第二种写法就要处处认两种,漏一处就是 @@ -28,13 +31,23 @@ from typing import Any, Iterator from app.data.recipe_types import GridCell from app.data.xlsx_scan import Bounds, SheetScan, UnsupportedTable, WorkbookScan -__all__ = ["CSV_PATH", "CSV_SHEET", "MAX_COLS", "MAX_ROWS", "CsvTable", "cell_value", "looks_like_csv", "parse", - "rows", "scan", "scan_any"] +__all__ = ["CSV_PATH", "CSV_SHEET", "CSV_SUFFIXES", "MAX_COLS", "MAX_ROWS", "CsvTable", "cell_value", "is_csv_sheet", + "looks_like_csv", "looks_like_markup", "parse", "rows", "scan", "scan_any", "sheet_delimiter"] #: CSV 的「工作表名」:配方按它认(SheetMatch.name),每月的文件名不同也认得出 CSV_SHEET = "CSV" -#: SheetScan.path 的取值:读取层据此知道这张「工作表」要从 CSV 里读,不去压缩包里找 XML +#: SheetScan.path 的开头:读取层据此知道这张「工作表」要从 CSV 里读,不去压缩包里找 XML。后面跟着扫描时定下的分隔符 +#: ("csv:,"、"csv:\t");只有「csv:」的(这一版之前的暂存区)读的时候再嗅一次 CSV_PATH = "csv:" +#: 按 CSV 读的扩展名,和简单导入(tabular._CSV)一致 +CSV_SUFFIXES = (".csv", ".tsv") +#: 嗅探认的分隔符 +_DELIMITERS = ",;\t|" +#: 内容是网页或 XML(「HTML 表格冒充 .xls」、Excel 2003 的 XML 表格)时的说法 +MARKUP_MESSAGE = ("无法打开该文件:它其实是网页或 XML 格式(有的系统导出的「Excel」文件是这样),不是 Excel 工作簿。" + "请用 Excel 打开后另存为 .xlsx 或 .csv 再上传") +NOT_TEXT_MESSAGE = ("无法按 CSV 读取这个文件:扩展名是 CSV,内容却不是文字(可能是图片、PDF 之类的文件改了扩展名)。" + "请确认上传的文件,或在 Excel 中另存为 .csv 后重新上传") #: 和 Excel 的上限一致:超过的 CSV 用 Excel 打开也装不下,交给模型的格子坐标也写不出来 MAX_ROWS = 1_048_576 MAX_COLS = 16_384 @@ -57,6 +70,24 @@ def looks_like_csv(raw: bytes) -> bool: return b"\x00" not in raw[:_SNIFF] +def looks_like_markup(raw: bytes) -> bool: + """内容是网页或 XML:去掉 BOM 和开头的空白后以「<」开头。真的工作簿以 zip(PK)或 OLE 文件头开头,不会是这样。""" + head = raw[:4096] + if head[:2] in (b"\xff\xfe", b"\xfe\xff"): + try: + text = head.decode("utf-16", errors="ignore") + except UnicodeDecodeError: + return False + else: + text = head.decode("latin-1") + return text.lstrip("\ufeff\xef\xbb\xbf \t\r\n").startswith("<") + + +def _suffix(filename: str | None) -> str: + name = str(filename or "").rsplit("/", 1)[-1] + return "." + name.rsplit(".", 1)[-1].lower() if "." in name else "" + + def cell_value(text: str) -> Any: """CSV 的一个字段 → 格子的值:空 / 只有空白是 None(空格),写法毫无歧义的整数、小数转成数,其余原样(文字)。""" t = text.strip() @@ -79,15 +110,17 @@ class CsvTable: delimiter: str -def parse(raw: bytes) -> CsvTable: - """原件 → 记录。编码、分隔符照 tabular 读 CSV 的做法;字段太长、引号不配对读不了时抛 UnsupportedTable。""" +def parse(raw: bytes, delimiter: str | None = None) -> CsvTable: + """原件 → 记录。编码照 tabular 读 CSV 的做法;delimiter 不给时嗅探(同 tabular)。字段太长、引号不配对读不了时抛 + UnsupportedTable。""" from app.data import tabular text = tabular._decode_csv(raw) - try: - delimiter = csv.Sniffer().sniff(text[:8192], delimiters=",;\t|").delimiter - except csv.Error: - delimiter = "," # 嗅不出来就按最常见的来 + if delimiter is None: + try: + delimiter = csv.Sniffer().sniff(text[:8192], delimiters=_DELIMITERS).delimiter + except csv.Error: + delimiter = "," # 嗅不出来就按最常见的来 records: list[list[str]] = [] try: for record in csv.reader(io.StringIO(text, newline=""), delimiter=delimiter): @@ -109,11 +142,22 @@ def _cells(table: CsvTable) -> Iterator[tuple[int, list[tuple[int, GridCell]]]]: yield r, out -def scan(raw: bytes) -> WorkbookScan: - """CSV 原件 → 扫描结果:一张可见工作表(名字 CSV_SHEET、路径 CSV_PATH),边界、非空格数、逐行非空格数的游程。 - 没有合并区、隐藏行列、公式、错误值、离群格(CSV 里每一格都是写出来的,没有残留格式)。""" - table = parse(raw) - sheet = SheetScan(name=CSV_SHEET, state="visible", path=CSV_PATH) +def is_csv_sheet(sheet: Any) -> bool: + """这张「工作表」是 CSV 适配出来的(path 以 CSV_PATH 开头)。""" + return str(getattr(sheet, "path", "") or "").startswith(CSV_PATH) + + +def sheet_delimiter(sheet: Any) -> str | None: + """扫描时定下的分隔符(记在 path 里);这一版之前的暂存区没记,返回 None(读的时候再嗅)。""" + tail = str(getattr(sheet, "path", "") or "")[len(CSV_PATH):] + return tail if len(tail) == 1 and tail in _DELIMITERS else None + + +def scan(raw: bytes, *, delimiter: str | None = None) -> WorkbookScan: + """CSV 原件 → 扫描结果:一张可见工作表(名字 CSV_SHEET、路径 CSV_PATH 加分隔符),边界、非空格数、逐行非空格数的 + 游程。没有合并区、隐藏行列、公式、错误值、离群格(CSV 里每一格都是写出来的,没有残留格式)。""" + table = parse(raw, delimiter) + sheet = SheetScan(name=CSV_SHEET, state="visible", path=CSV_PATH + table.delimiter) lo_r = lo_c = hi_r = hi_c = 0 runs: list[tuple[int, int, int]] = [] for r, cells in _cells(table): @@ -134,12 +178,20 @@ def scan(raw: bytes) -> WorkbookScan: return WorkbookScan(sheets=[sheet], format="csv") -def scan_any(raw: bytes) -> WorkbookScan: - """配方导入的扫描入口:CSV 走这里的适配,其余照旧交给 xlsx_scan(读不了照旧拒收,文案不变)。""" +def scan_any(raw: bytes, filename: str | None) -> WorkbookScan: + """配方导入的扫描入口:.csv / .tsv 走这里的适配(.tsv 按制表符分隔),其余照旧交给 xlsx_scan(读不了照旧拒收, + 文案不变);内容是网页或 XML 的另给一句能照着做的话(MARKUP_MESSAGE)。扩展名是 CSV、内容却是工作簿(xlsx 改了 + 扩展名)的照工作簿读。""" from app.data import xlsx_scan - if looks_like_csv(raw): - return scan(raw) + suffix = _suffix(filename) + if suffix in CSV_SUFFIXES: + if looks_like_csv(raw): + return scan(raw, delimiter="\t" if suffix == ".tsv" else None) + if raw[:4] != b"PK\x03\x04": + raise UnsupportedTable(NOT_TEXT_MESSAGE) + elif looks_like_markup(raw): + raise UnsupportedTable(MARKUP_MESSAGE) return xlsx_scan.scan(raw) diff --git a/backend/app/data/recipe_imports.py b/backend/app/data/recipe_imports.py index e715a08c..2901abba 100644 --- a/backend/app/data/recipe_imports.py +++ b/backend/app/data/recipe_imports.py @@ -177,7 +177,8 @@ class Pipeline: recipe_sha256: Callable[[Recipe], str] canonical: Callable[[Recipe], dict[str, Any]] apply_patch: Callable[[dict[str, Any], list[dict[str, Any]]], dict[str, Any]] - scan: Callable[[bytes], WorkbookScan] + #: (原件, 文件名) → 扫描结果。文件名定按 CSV 读还是按工作簿读(csv_grid.scan_any) + scan: Callable[[bytes, str], WorkbookScan] scan_to_json: Callable[[WorkbookScan], dict[str, Any]] scan_from_json: Callable[[dict[str, Any]], WorkbookScan] read_grid: Callable[..., Grid] @@ -474,7 +475,7 @@ def _open_book(p: Pipeline, raw: bytes, scan: WorkbookScan) -> _Book: def _book_of(p: Pipeline, snap: _Snap) -> _Book: raw = _read_raw(snap.raw_sha256) - scan = p.scan_from_json(snap.scan) if snap.scan else p.scan(raw) + scan = p.scan_from_json(snap.scan) if snap.scan else p.scan(raw, snap.file_name) return _open_book(p, raw, scan) @@ -772,7 +773,7 @@ class _Staged: def _stage_work(p: Pipeline, raw: bytes, filename: str) -> _Staged: """首次导入(或切换)的暂存:扫描、网格预览、规则起草(带干跑)、对草稿做一次静态校验。""" - scan = p.scan(raw) + scan = p.scan(raw, filename) book = _open_book(p, raw, scan) dry = _Capture(_dry_fn(p, book, filename, {})) d = p.draft(scan, book.grids, filename, validate=lambda data, f, o: p.validate(data, facts=f, origin=o), @@ -860,7 +861,7 @@ async def stage_upload(session: AsyncSession, *, raw: bytes, filename: str, name def _replay_work(p: Pipeline, raw: bytes, current: dict[str, Any], filename: str, snap: _Snap, *, evaluate: bool) -> tuple[_Staged, _Eval | None]: """上传新一期 / 修改配方:扫描(或取存下的扫描)、按现行配方的写法取系统发现,修改配方时再评估一遍。""" - scan = p.scan_from_json(snap.scan) if snap.scan else p.scan(raw) + scan = p.scan_from_json(snap.scan) if snap.scan else p.scan(raw, filename) book = _open_book(p, raw, scan) facts = p.detect_facts(book.grids, scan, recipe=current) staged = _Staged(scan_json=p.scan_to_json(scan), previews=[p.preview(g) for g in book.grids.values()], @@ -1860,7 +1861,7 @@ def _trial_work(p: Pipeline, snap: _Snap, work: dict[str, Any], facts: DraftFact if parsed is None or problems: return _TrialRun(parsed=parsed, problems=list(problems)) raw = _read_raw(snap.raw_sha256) - scan = p.scan_from_json(snap.scan) if snap.scan else p.scan(raw) + scan = p.scan_from_json(snap.scan) if snap.scan else p.scan(raw, snap.file_name) path.parent.mkdir(parents=True, exist_ok=True) # 先写 .tmp- 再改名:半成品归启动清理(_sweep_tmp)管,发布、回收不会把半个库当成试运行库 tmp = path.parent / f"{path.name}{raw_store.TMP_MARK}{secrets.token_hex(6)}" diff --git a/backend/app/data/xlsx_cells.py b/backend/app/data/xlsx_cells.py index c270f8bb..69f93fad 100644 --- a/backend/app/data/xlsx_cells.py +++ b/backend/app/data/xlsx_cells.py @@ -72,10 +72,11 @@ class Workbooks: self._formulas: Any = None self._csv: Any = None - def csv(self) -> Any: - """CSV 原件解析好的记录(csv_grid.CsvTable),第一次要读时才解析,同一次执行里复用。""" + def csv(self, delimiter: str | None = None) -> Any: + """CSV 原件解析好的记录(csv_grid.CsvTable),第一次要读时才解析,同一次执行里复用。delimiter 是扫描时定下的 + 分隔符(csv_grid.sheet_delimiter),不给时再嗅一次。""" if self._csv is None: - self._csv = csv_grid.parse(self.raw) + self._csv = csv_grid.parse(self.raw, delimiter) return self._csv def values(self) -> Any: @@ -133,9 +134,10 @@ def sheet_rows(books: Workbooks, sheet: SheetScan, *, min_row: int, max_row: int 行数和每行的宽度都一样。文件内容写坏了(见 _READ_ERRORS)抛 UnsupportedTable,日志只记异常类型。 """ try: - if sheet.path == csv_grid.CSV_PATH: - # CSV 适配出来的「工作表」:从解析好的记录里取,口径同下(只给非空格、整行为空的不给) - yield from csv_grid.rows(books.csv(), min_row=min_row, max_row=max_row, min_col=min_col, max_col=max_col) + if csv_grid.is_csv_sheet(sheet): + # CSV 适配出来的「工作表」:从解析好的记录里取,口径同下(只给非空格、整行为空的不给)。分隔符用扫描时定下的 + yield from csv_grid.rows(books.csv(csv_grid.sheet_delimiter(sheet)), min_row=min_row, max_row=max_row, + min_col=min_col, max_col=max_col) return yield from _sheet_rows(books, sheet, min_row=min_row, max_row=max_row, min_col=min_col, max_col=max_col) except UnsupportedTable: diff --git a/backend/tests/test_csv_grid.py b/backend/tests/test_csv_grid.py index 30a5f78a..dca02df7 100644 --- a/backend/tests/test_csv_grid.py +++ b/backend/tests/test_csv_grid.py @@ -8,7 +8,7 @@ import pytest from app.data import csv_grid, recipe_engine, xlsx_cells from app.data.recipe_types import Recipe -from app.data.xlsx_scan import UnsupportedTable, scan_from_json, scan_to_json +from app.data.xlsx_scan import SheetScan, UnsupportedTable, scan_from_json, scan_to_json CSV = ("订单号,下单日期,门店,金额,备注\n" "A001,2026-08-01,一号店,120,\n" @@ -31,13 +31,13 @@ def test_scan_gives_one_visible_sheet_with_bounds_and_row_counts(): scan = csv_grid.scan(CSV) assert scan.format == "csv" and len(scan.sheets) == 1 sh = scan.sheets[0] - assert sh.name == csv_grid.CSV_SHEET and sh.state == "visible" and sh.path == csv_grid.CSV_PATH + assert sh.name == csv_grid.CSV_SHEET and sh.state == "visible" and sh.path == csv_grid.CSV_PATH + "," assert (sh.bounds.min_row, sh.bounds.min_col, sh.bounds.max_row, sh.bounds.max_col) == (1, 1, 5, 6) # 空字段不算非空格;第 4 行是空行 assert sh.row_runs == [(1, 1, 5), (2, 2, 4), (3, 3, 5), (5, 5, 6)] assert sh.nonempty == 20 again = scan_from_json(scan_to_json(scan)) - assert again.format == "csv" and again.sheets[0].path == csv_grid.CSV_PATH + assert again.format == "csv" and again.sheets[0].path == csv_grid.CSV_PATH + "," def test_cell_values_are_typed_only_when_unambiguous(): @@ -71,9 +71,47 @@ def test_semicolon_and_tab_delimiters_and_gbk(): def test_scan_any_dispatches_and_keeps_the_xlsx_refusal(): - assert csv_grid.scan_any(CSV).format == "csv" + assert csv_grid.scan_any(CSV, "orders.csv").format == "csv" + assert csv_grid.scan_any(CSV, "ORDERS.CSV").format == "csv" with pytest.raises(UnsupportedTable): - csv_grid.scan_any(b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + b"\x00" * 100) + csv_grid.scan_any(b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + b"\x00" * 100, "old.xls") + + +def test_only_csv_and_tsv_names_are_read_as_csv(): + """复核反例:以前只看内容,不是 zip / OLE、前 64KB 没有 NUL 就当 CSV 读。系统导出的「HTML 表格冒充 .xls」、 + 误传的文本文件,都被拆成一张叫「CSV」的工作表去起草。现在和简单导入(tabular)一样按扩展名定:只有 .csv / .tsv + 按 CSV 读;别的照旧交给 xlsx_scan 拒收。""" + with pytest.raises(UnsupportedTable) as e: + csv_grid.scan_any(CSV, "orders.txt") + assert "不是有效的 .xlsx" in str(e.value) + html = "
订单号金额
".encode("utf-8") + for raw in (html, b"\xef\xbb\xbf \r\n" + html, b"\xff\xfe" + "
a
".encode("utf-16-le"), + b''): + with pytest.raises(UnsupportedTable) as e: + csv_grid.scan_any(raw, "report.xls") + assert "网页或 XML" in str(e.value) and "另存为 .xlsx 或 .csv" in str(e.value) + # 扩展名是 .csv、内容却不是文字(图片之类改了扩展名):按 CSV 读不了,照实说 + with pytest.raises(UnsupportedTable) as e: + csv_grid.scan_any(b"\x89PNG\r\n\x1a\n\x00\x00", "orders.csv") + assert "不是文字" in str(e.value) + + +def test_tsv_is_always_tab_separated_and_the_delimiter_is_kept_for_reading(): + """.tsv 一律按制表符分隔(同 tabular):每行逗号个数一样的 TSV,嗅探会挑逗号(csv.Sniffer 优先逗号)。 + 扫描时定下的分隔符记在工作表的 path 里,之后读格子(重放、试运行)用同一个,不再各嗅各的。""" + raw = "名称,规格\t金额\n甲,1kg\t10\n乙,2kg\t20\n".encode("utf-8") # 每行一个逗号、一个制表符 + scan = csv_grid.scan_any(raw, "goods.tsv") + assert scan.sheets[0].path == csv_grid.CSV_PATH + "\t" + grid = xlsx_cells.read_grid(raw, scan_from_json(scan_to_json(scan)), csv_grid.CSV_SHEET) + assert grid.get(2, 1).value == "甲,1kg" and grid.get(2, 2).value == 10 + # 这一版之前的暂存区:path 只有「csv:」,读的时候再嗅 + assert csv_grid.is_csv_sheet(SheetScan(name="CSV", state="visible", path=csv_grid.CSV_PATH)) + assert csv_grid.sheet_delimiter(SheetScan(name="CSV", state="visible", path=csv_grid.CSV_PATH)) is None + assert not csv_grid.is_csv_sheet(SheetScan(name="明细", state="visible", path="xl/worksheets/sheet1.xml")) + sniffed = csv_grid.scan_any(raw, "goods.csv") + assert sniffed.sheets[0].path == csv_grid.CSV_PATH + "," # 同样的内容叫 .csv:照嗅探的来(逗号) + grid = xlsx_cells.read_grid(raw, sniffed, csv_grid.CSV_SHEET) + assert grid.get(2, 1).value == "甲" and grid.get(2, 2).value == "1kg\t10" def test_too_many_columns_is_refused(): diff --git a/backend/tests/test_recipe_imports.py b/backend/tests/test_recipe_imports.py index cdfed879..74b54535 100644 --- a/backend/tests/test_recipe_imports.py +++ b/backend/tests/test_recipe_imports.py @@ -260,7 +260,7 @@ def fake_pipeline(state: FakeState) -> Pipeline: return Pipeline( validate=validate, recipe_sha256=recipe.recipe_sha256, canonical=recipe.canonical_recipe, - apply_patch=recipe.apply_patch, scan=lambda raw: fake_scan(raw, state), + apply_patch=recipe.apply_patch, scan=lambda raw, _filename: fake_scan(raw, state), scan_to_json=xlsx_scan.scan_to_json, scan_from_json=xlsx_scan.scan_from_json, read_grid=fake_grid, preview=xlsx_cells.preview, execute=execute, run_checks=run_checks, column_null_counts=null_counts, build_notes=build_notes, apply_notes=recipe_notes.apply_notes, detect_facts=detect_facts, draft=draft, diff --git a/backend/tests/test_recipe_p5.py b/backend/tests/test_recipe_p5.py index 9361fd42..163a7bf7 100644 --- a/backend/tests/test_recipe_p5.py +++ b/backend/tests/test_recipe_p5.py @@ -43,7 +43,7 @@ def _book(rows: list[tuple[Any, ...]], *, sheet: str = "明细", top: list[tuple def _run(raw: bytes, recipe: dict[str, Any], db: str | None = None, filename: str = "明细.xlsx", **kw: Any): - scan = csv_grid.scan_any(raw) + scan = csv_grid.scan_any(raw, filename) return recipe_engine.execute(Recipe.model_validate(recipe), raw, filename, scan, db, **kw) @@ -314,7 +314,7 @@ def _ok_validate(data, facts, origin): def _draft(raw: bytes, filename: str = "明细.xlsx"): - scan = csv_grid.scan_any(raw) + scan = csv_grid.scan_any(raw, filename) grids = {s.name: xlsx_cells.read_grid(raw, scan, s.name, max_rows=recipe_suggest.DRAFT_MAX_ROWS) for s in scan.sheets if s.state == "visible" and s.nonempty}