mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
配方导入只把 .csv / .tsv 当 CSV 读;网页、XML 冒充的 Excel 照旧拒收并说清楚
复核发现(C-csv-sniff-any-text):配方导入的扫描入口 scan_any 只看内容,不是 zip、不是 OLE、前 64KB 没有 NUL 就当 CSV 读,不看扩展名。企业系统常见的「HTML 表格冒充 .xls」、误传的文本文件,会被拆成一张叫「CSV」的工作表去 起草、核对;给已有 xlsx 配方传新一期时还会报找不到工作表、给出 rename_sheet 修复,都误导用户。 改法: - 和简单导入(tabular._CSV)一样按扩展名定:scan_any(原件, 文件名) 只把 .csv / .tsv(且内容是文字)按 CSV 读, 别的照旧交给 xlsx_scan 拒收。内容以「<」开头的(网页、Excel 2003 的 XML 表格,含带 BOM 的 UTF-8 / UTF-16)另给 一句「它其实是网页或 XML 格式……请用 Excel 打开后另存为 .xlsx 或 .csv」。扩展名是 CSV、内容却不是文字的(图片 改了扩展名)照实说;内容其实是 xlsx 的照工作簿读。 - .tsv 一律按制表符分隔(同 tabular):每行逗号、制表符个数都一样时,csv.Sniffer 优先挑逗号,会把 TSV 拆错。 - 扫描时定下的分隔符记进工作表的 path("csv:," / "csv:\t"),读格子(xlsx_cells.sheet_rows)用同一个,重放、 试运行不再各嗅各的;这一版之前只有「csv:」的照旧再嗅一次。 - Pipeline.scan 改成 (原件, 文件名),recipe_imports 的四处调用都传上暂存区记下的文件名。 测试:test_csv_grid 加 .txt 的 CSV 内容拒收、四种网页 / XML 冒充 .xls 的拒收说法、.csv 扩展名的图片、TSV 按制表符 并且分隔符经扫描 JSON 往返后读格子照用、老暂存区的 path;改回只看内容回跑两条失败。
This commit is contained in:
@@ -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)
|
||||
|
||||
|
||||
|
||||
@@ -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)}"
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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 = "<html><body><table><tr><td>订单号</td><td>金额</td></tr></table></body></html>".encode("utf-8")
|
||||
for raw in (html, b"\xef\xbb\xbf \r\n" + html, b"\xff\xfe" + "<table><tr><td>a</td></tr></table>".encode("utf-16-le"),
|
||||
b'<?xml version="1.0"?><Workbook xmlns="urn:schemas-microsoft-com:office:spreadsheet"/>'):
|
||||
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():
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user