This commit is contained in:
csj
2026-09-17 14:13:31 +08:00
parent ced5db5473
commit 10697d3dfa
18 changed files with 2097 additions and 26 deletions
+2
View File
@@ -20,6 +20,8 @@ RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_
RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_ppt_portrait.py -k engine_process -v
RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_approved_ppt.py -k engine_process -v
RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_match_explanation.py -k engine_process -v
RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_expense_template.py -k engine_process -v
RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_expense_preview.py -k engine_process -v
if [ "$SOURCE_HASH" != "$(bash "$ROOT/native-engine/engine-fingerprint.sh")" ]; then
echo "error: 打包期间引擎源码发生变化,请重新构建。" >&2
exit 1
+16 -3
View File
@@ -9,6 +9,8 @@ from pathlib import Path
from domain import auto_match, enrich, extract
from exports import export_expense, export_ppt, export_travel
from expense_template import export_custom_expense, inspect_template, validate_template
from expense_preview import prepare_preview
PROTOCOL = sys.stdout
SUPPORTED = {'.jpg', '.jpeg', '.png', '.webp', '.bmp', '.gif', '.pdf'}
@@ -83,6 +85,10 @@ def dispatch(request):
operation = request['operation']
if operation == 'scan':
return scan(request)
if operation == 'inspect-expense-template':
return inspect_template(request['templatePath'])
if operation == 'validate-expense-template':
return validate_template(request['templatePath'], request['templateMapping'])
state = request['state']
if operation == 'refresh':
return enrich(state)
@@ -97,9 +103,16 @@ def dispatch(request):
elif operation == 'travel':
export_travel(state, temporary)
elif operation == 'expense':
template = Path(__file__).parent / 'personal-expense-template.xlsx'
metadata = export_expense(state, temporary, template, request.get('purposes', {}), request.get('signatures', []),
request.get('payee', {}), request.get('categoryPurposes'))
if request.get('templatePath'):
metadata = export_custom_expense(state, temporary, request['templatePath'], request.get('templateMapping'),
request.get('purposes', {}), request.get('signatures', []),
request.get('payee', {}), request.get('categoryPurposes'))
else:
template = Path(__file__).parent / 'personal-expense-template.xlsx'
metadata = export_expense(state, temporary, template, request.get('purposes', {}), request.get('signatures', []),
request.get('payee', {}), request.get('categoryPurposes'))
if request.get('preparePreview'):
metadata.update(prepare_preview(temporary))
else:
raise ValueError('未知操作:' + operation)
os.replace(temporary, destination)
+220
View File
@@ -0,0 +1,220 @@
import ast
import datetime
import re
import zipfile
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from expense_template import Template, column_name, coordinate, encoded, tag
class UnsupportedFormula(ValueError):
pass
RMB_FORMULA = ('SUBSTITUTE(SUBSTITUTE(IF({cell}>-0.5%,,"")&TEXT(INT(ABS({cell})+0.5%),'
'"[dbnum2]G/通用格式元;;")&TEXT(RIGHT(FIXED({cell}),2),"[dbnum2]0角0分;;"&'
'IF(ABS({cell})>1%,"",)),"零角",IF(ABS({cell})<1,,"")),"零分","")')
def chinese_integer(number):
digits = '零壹贰叁肆伍陆柒捌玖'
if not 0 <= number < 10 ** 12:
raise UnsupportedFormula('大写金额超出预览范围')
if number == 0:
return ''
result, pending_zero = '', False
for divisor, suffix in [(10 ** 8, '亿'), (10 ** 4, ''), (1, '')]:
group, number = divmod(number, divisor)
if not group:
if result:
pending_zero = True
continue
if result and (pending_zero or group < 1000):
result += ''
fragment, zero = '', False
for unit, label in [(1000, ''), (100, ''), (10, ''), (1, '')]:
digit, group = divmod(group, unit)
if digit:
if zero:
fragment += ''
fragment += digits[digit] + label
zero = False
elif fragment:
zero = True
result += fragment + suffix
pending_zero = False
return result
def builtin_rmb(value):
absolute = abs(value)
integer = int(absolute + Decimal('0.005'))
cents = int(absolute.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP) * 100) % 100
result = '' if value > Decimal('-0.005') else ''
if integer:
result += chinese_integer(integer) + ''
digits = '零壹贰叁肆伍陆柒捌玖'
result += (digits[cents // 10] + '' + digits[cents % 10] + '') if cents else ('' if absolute > Decimal('0.01') else '')
return result.replace('零角', '' if absolute < 1 else '').replace('零分', '')
class PreviewCalculator:
def __init__(self, document, date_1904=False):
self.cells = {cell.get('r'): cell for cell in document.iter(tag('c'))}
self.values = {}
self.visiting = set()
self.epoch = datetime.date(1904, 1, 1) if date_1904 else datetime.date(1899, 12, 30)
def cell(self, reference):
if reference in self.values:
return self.values[reference]
if reference in self.visiting or len(self.visiting) >= 100:
raise UnsupportedFormula('循环引用或嵌套过深')
cell = self.cells.get(reference)
if cell is None:
return Decimal(0)
self.visiting.add(reference)
try:
formula = cell.find(tag('f'))
if formula is not None:
if formula.get('t') not in (None, 'normal'):
raise UnsupportedFormula('数组或共享公式')
result = self.expression(formula.text or '')
elif cell.get('t') in (None, 'n'):
result = Decimal(cell.findtext(tag('v')) or '0')
elif cell.get('t') == 'inlineStr' and not ''.join(cell.itertext()).strip():
result = Decimal(0)
else:
raise UnsupportedFormula('非数值单元格')
if isinstance(result, Decimal) and not result.is_finite():
raise UnsupportedFormula('非有限数值')
self.values[reference] = result
return result
finally:
self.visiting.remove(reference)
def expression(self, formula):
formula = formula.strip().replace('$', '')
if len(formula) > 2000:
raise UnsupportedFormula('公式过长')
if formula.startswith('SUBSTITUTE(SUBSTITUTE(IF('):
matched = re.match(r'SUBSTITUTE\(SUBSTITUTE\(IF\(([A-Z]+[0-9]+)>', formula)
if matched and formula == RMB_FORMULA.format(cell=matched[1]):
amount = self.cell(matched[1])
if not isinstance(amount, Decimal):
raise UnsupportedFormula('大写金额来源不是数值')
return builtin_rmb(amount)
if formula.upper() == 'TODAY()':
return Decimal((datetime.date.today() - self.epoch).days)
matched = re.fullmatch(r'SUM\(([A-Z]+[0-9]+):([A-Z]+[0-9]+)\)', formula, re.I)
if matched:
left, top = coordinate(matched[1].upper())
right, bottom = coordinate(matched[2].upper())
if right < left or bottom < top or (right - left + 1) * (bottom - top + 1) > 20000:
raise UnsupportedFormula('求和范围过大')
total = Decimal(0)
for row in range(top, bottom + 1):
for column in range(left, right + 1):
reference = column_name(column) + str(row)
cell = self.cells.get(reference)
if cell is not None and cell.find(tag('f')) is None and cell.get('t') not in (None, 'n'):
continue
value = self.cell(reference)
if not isinstance(value, str):
total += value
return total
if formula.upper().startswith('IF(') and formula.endswith(')'):
arguments = []
depth, start = 0, 3
for index in range(3, len(formula) - 1):
character = formula[index]
if character == '(':
depth += 1
elif character == ')':
depth -= 1
elif character == ',' and depth == 0:
arguments.append(formula[start:index])
start = index + 1
arguments.append(formula[start:-1])
if len(arguments) != 3:
raise UnsupportedFormula('不支持的条件公式')
return self.expression(arguments[1] if self.expression(arguments[0]) else arguments[2])
formula = formula.replace('<>', '!=')
formula = re.sub(r'(?<![<>=!])=(?!=)', '==', formula)
tree = ast.parse(formula, mode='eval')
return self.node(tree.body)
def node(self, node):
if isinstance(node, ast.Constant) and type(node.value) in (int, float):
return Decimal(str(node.value))
if isinstance(node, ast.Name) and re.fullmatch(r'[A-Z]{1,3}[1-9][0-9]{0,3}', node.id, re.I):
return self.cell(node.id.upper())
if isinstance(node, ast.UnaryOp) and isinstance(node.op, (ast.UAdd, ast.USub)):
value = self.node(node.operand)
if isinstance(value, str):
raise UnsupportedFormula('非数值运算')
return -value if isinstance(node.op, ast.USub) else value
if isinstance(node, ast.BinOp):
left, right = self.node(node.left), self.node(node.right)
if isinstance(left, str) or isinstance(right, str):
raise UnsupportedFormula('非数值运算')
if isinstance(node.op, ast.Add):
return left + right
if isinstance(node.op, ast.Sub):
return left - right
if isinstance(node.op, ast.Mult):
return left * right
if isinstance(node.op, ast.Div):
return left / right
if isinstance(node, ast.Compare) and len(node.ops) == 1:
left, right = self.node(node.left), self.node(node.comparators[0])
if isinstance(left, str) or isinstance(right, str):
raise UnsupportedFormula('非数值比较')
operator = node.ops[0]
if isinstance(operator, ast.Gt):
return left > right
if isinstance(operator, ast.GtE):
return left >= right
if isinstance(operator, ast.Lt):
return left < right
if isinstance(operator, ast.LtE):
return left <= right
if isinstance(operator, ast.Eq):
return left == right
if isinstance(operator, ast.NotEq):
return left != right
raise UnsupportedFormula('自定义公式需由 Excel / WPS 计算')
def prepare_preview(path):
from lxml import etree as ET
template = Template(path)
properties = template.workbook.find(tag('workbookPr'))
date_1904 = properties is not None and properties.get('date1904') in ('1', 'true')
uncalculated = []
for sheet in template.sheets:
if sheet.get('state', 'visible') != 'visible':
continue
_, sheet_path, document = template.sheet(sheet.get('name'))
calculator = PreviewCalculator(document, date_1904)
for cell in document.iter(tag('c')):
if cell.find(tag('f')) is None:
continue
for cached in cell.findall(tag('v')):
cell.remove(cached)
try:
value = calculator.cell(cell.get('r'))
cell.set('t', 'str' if isinstance(value, str) else 'b' if isinstance(value, bool) else 'n')
ET.SubElement(cell, tag('v')).text = str(int(value)) if isinstance(value, bool) else str(value)
except (ValueError, SyntaxError, ArithmeticError, InvalidOperation, RecursionError):
uncalculated.append(sheet.get('name') + '!' + cell.get('r'))
template.parts[sheet_path] = encoded(document)
for name, document in [('xl/workbook.xml', template.workbook),
('xl/_rels/workbook.xml.rels', template.relations),
('[Content_Types].xml', template.types)]:
template.parts[name] = encoded(document)
with zipfile.ZipFile(path, 'w', zipfile.ZIP_DEFLATED) as output:
for name, data in template.parts.items():
output.writestr(name, data)
return dict(expensePreview='xlsx-quicklook-v1', previewUncalculatedCells=uncalculated)
+440
View File
@@ -0,0 +1,440 @@
import copy
import hashlib
import json
import math
import posixpath
import re
import zipfile
from decimal import Decimal
from pathlib import Path
from lxml import etree as ET
MAIN = 'http://schemas.openxmlformats.org/spreadsheetml/2006/main'
REL = 'http://schemas.openxmlformats.org/officeDocument/2006/relationships'
PACKAGE = 'http://schemas.openxmlformats.org/package/2006/relationships'
CONTENT = 'http://schemas.openxmlformats.org/package/2006/content-types'
def tag(name):
return '{' + MAIN + '}' + name
def xml(data):
return ET.fromstring(data, ET.XMLParser(resolve_entities=False, no_network=True))
def encoded(document):
return ET.tostring(document, xml_declaration=True, encoding='UTF-8', standalone=True)
def coordinate(reference):
match = re.fullmatch(r'([A-Z]{1,3})([1-9][0-9]{0,3})', reference)
if not match:
raise ValueError('单元格地址不正确:' + reference)
column = 0
for character in match[1]:
column = column * 26 + ord(character) - 64
row = int(match[2])
if column > 256 or row > 2000:
raise ValueError('模板填充范围限于前 256 列、2000 行')
return column, row
def column_name(number):
result = ''
while number:
number, remainder = divmod(number - 1, 26)
result = chr(65 + remainder) + result
return result
def cell_values(document, shared):
result = {}
for cell in document.iter(tag('c')):
value = cell.findtext(tag('v'), '')
if cell.get('t') == 's':
value = shared[int(value)] if value else ''
elif cell.get('t') == 'inlineStr':
value = ''.join(cell.itertext()) if cell.find(tag('is')) is not None else ''
if cell.find(tag('f')) is not None:
value = '=' + cell.findtext(tag('f'), '')
if value:
result[cell.get('r')] = value
return result
class Template:
def __init__(self, path):
path = Path(path)
if path.suffix.lower() != '.xlsx' or not path.is_file():
raise ValueError('请选择有效的 .xlsx 报销单模板;旧版 .xls 请先另存为 .xlsx')
if path.stat().st_size > 20 * 1024 * 1024:
raise ValueError('模板不能超过 20 MB')
self.digest = hashlib.sha256(path.read_bytes()).hexdigest()
try:
with zipfile.ZipFile(path) as archive:
entries = archive.infolist()
if len(entries) > 5000 or sum(entry.file_size for entry in entries) > 100 * 1024 * 1024:
raise ValueError('模板内容过大,请使用精简的报销单模板')
if len({entry.filename for entry in entries}) != len(entries):
raise ValueError('模板包含重复文件项,请重新另存为 .xlsx')
self.parts = {entry.filename: archive.read(entry) for entry in entries}
if any('vbaproject' in name.lower() for name in self.parts):
raise ValueError('不支持带宏的模板,请移除后重试')
self.workbook = xml(self.parts['xl/workbook.xml'])
self.relations = xml(self.parts['xl/_rels/workbook.xml.rels'])
self.types = xml(self.parts['[Content_Types].xml'])
shared_xml = xml(self.parts['xl/sharedStrings.xml']) if 'xl/sharedStrings.xml' in self.parts else []
self.shared = [''.join(item.itertext()) for item in shared_xml]
self.sheets = list(self.workbook.find(tag('sheets')))
active_paths = set()
for sheet in self.sheets:
relation = next(item for item in self.relations if item.get('Id') == sheet.get('{' + REL + '}id'))
target = relation.get('Target')
active_paths.add(target.lstrip('/') if target.startswith('/') else posixpath.normpath('xl/' + target))
active_documents = [xml(self.parts[path]) for path in active_paths] + [self.workbook]
for document in active_documents:
for node in document.iter():
if ET.QName(node).localname in {'f', 'formula', 'formula1', 'formula2', 'definedName'}:
if re.search(r'\[[^\]]+\].*!|(?:WEBSERVICE|RTD)\s*\(|\|[^!]+!', node.text or '', re.I):
raise ValueError('模板含有正在使用的外部工作簿链接或外部数据公式,请移除后重试')
orphan_sheets = {item.get('Target').lstrip('/') if item.get('Target').startswith('/') else
posixpath.normpath('xl/' + item.get('Target'))
for item in self.relations if item.get('Type', '').endswith('/worksheet')} - active_paths
for path in orphan_sheets:
self.parts.pop(path, None)
self.parts.pop(posixpath.dirname(path) + '/_rels/' + posixpath.basename(path) + '.rels', None)
for name in list(self.parts):
if name.startswith('xl/externalLinks/'):
self.parts.pop(name)
external = self.workbook.find(tag('externalReferences'))
if external is not None:
self.workbook.remove(external)
for item in list(self.relations):
target = item.get('Target', '')
path = target.lstrip('/') if target.startswith('/') else posixpath.normpath('xl/' + target)
if item.get('Type', '').endswith('/externalLink') or path in orphan_sheets:
self.relations.remove(item)
for item in list(self.types):
path = item.get('PartName', '').lstrip('/')
if path.startswith('xl/externalLinks/') or path in orphan_sheets:
self.types.remove(item)
except (zipfile.BadZipFile, KeyError, ET.XMLSyntaxError, RuntimeError) as error:
raise ValueError('无法读取模板;请确认它是未加密、未损坏的 .xlsx 文件') from error
def sheet(self, name):
sheet = next((item for item in self.sheets if item.get('name') == name), None)
if sheet is None:
raise ValueError('模板中找不到工作表:' + name)
if sheet.get('state', 'visible') != 'visible':
raise ValueError('请选择可见工作表作为报销单')
relation = next(item for item in self.relations if item.get('Id') == sheet.get('{' + REL + '}id'))
target = relation.get('Target')
path = target.lstrip('/') if target.startswith('/') else posixpath.normpath('xl/' + target)
return sheet, path, xml(self.parts[path])
def merged_ranges(document):
result = []
for merged in document.iter(tag('mergeCell')):
start, end = merged.get('ref').split(':')
left, top = coordinate(start)
right, bottom = coordinate(end)
result.append((start, left, top, right, bottom))
return result
def anchor(reference, merges):
column, row = coordinate(reference)
return next((start for start, left, top, right, bottom in merges
if left <= column <= right and top <= row <= bottom), reference)
def suggest(document, shared, name):
values = cell_values(document, shared)
merges = merged_ranges(document)
normalized = {reference: re.sub(r'\s+', '', value).lower() for reference, value in values.items()}
columns = {}
start_row, end_row = 9, 21
headers = {
'purpose': ['支出项目', '报销内容', '费用项目', '用途', '事由', '摘要'],
'amount': ['金额', 'amount'],
'count': ['单据数量', '单据张数', '票据张数', '附件张数', '张数'],
'invoiceType': ['票据', '发票类型', '票据类型'],
'sequence': ['序号'],
'remarks': ['备注', '说明/'],
'category': ['费用类型', '费用类别'],
}
for row in range(1, 101):
candidate = {}
for field, labels in headers.items():
found = [reference for reference, value in normalized.items()
if coordinate(reference)[1] == row and any(label in value for label in labels)]
if len(found) == 1:
candidate[field] = re.sub(r'\d+', '', found[0])
if 'purpose' in candidate and 'amount' in candidate:
columns = candidate
if columns.get('invoiceType') == columns.get('count'):
columns.pop('invoiceType', None)
start_row = row + 1
end_row = start_row
amount_column = columns['amount']
for value in values.values():
match = re.fullmatch(r'=SUM\(\$?' + amount_column + r'\$?' + str(start_row) + r':\$?' + amount_column + r'\$?(\d+)\)', value, re.I)
if match:
end_row = int(match[1])
break
break
cells = {}
labels = {'recipient': ['收款人'], 'bankName': ['开户行', '开户银行'],
'accountNumber': ['账号', '银行账号', '银行账户'], 'preparer': ['制单人', '报销人'],
'total': ['本次报销金额', '报销合计', '合计金额', '合计']}
for field, options in labels.items():
found = [reference for reference, value in normalized.items()
if value.rstrip(':') in options]
if len(found) == 1:
reference = found[0]
column, row = coordinate(reference)
merged = next((entry for entry in merges if entry[0] == reference), None)
if merged:
column = merged[3]
cells[field] = anchor(column_name(column + 1) + str(row), merges)
signature_cells = []
if columns.get('purpose') == 'C' and columns.get('amount') == 'H' and start_row == 9 and end_row == 21:
signature_cells = [reference for reference in ['A30', 'D30', 'A31', 'D31', 'A32', 'D32']
if anchor(reference, merges) == reference]
return dict(sheetName=name, startRow=start_row, endRow=end_row, columns=columns, cells=cells,
signatureCells=','.join(signature_cells), clearCells='')
def inspect_template(path):
template = Template(path)
sheets = []
for sheet in template.sheets:
if sheet.get('state', 'visible') != 'visible':
continue
name = sheet.get('name')
_, _, document = template.sheet(name)
try:
mapping = suggest(document, template.shared, name)
except ValueError:
mapping = dict(sheetName=name, startRow=9, endRow=21, columns={}, cells={},
signatureCells='', clearCells='')
values = cell_values(document, template.shared)
rows = {}
for reference, value in values.items():
if len(rows) >= 80 and int(re.sub('[A-Z]', '', reference)) not in rows:
continue
rows.setdefault(int(re.sub('[A-Z]', '', reference)), []).append(reference + ' ' + value[:160])
sheets.append(dict(name=name, mapping=mapping,
preview=[f"{row} 行 · " + ' | '.join(entries) for row, entries in sorted(rows.items())]))
if not sheets:
raise ValueError('模板没有可见工作表')
return dict(templateVersion='mapped-xlsx-v1', fingerprint=template.digest, sheets=sheets)
def references(value):
return [item.upper() for item in re.split(r'[,;\s]+', value.strip()) if item]
def validate_mapping(template, mapping):
if not isinstance(mapping, dict):
raise ValueError('请先配置报销单模板的填充位置')
_, _, document = template.sheet(mapping.get('sheetName', ''))
if document.find(tag('sheetProtection')) is not None:
raise ValueError('模板工作表已保护,请先取消保护后导入')
start, end = mapping.get('startRow'), mapping.get('endRow')
if type(start) is not int or type(end) is not int or not 1 <= start <= end <= 2000 or end - start >= 200:
raise ValueError('明细行范围不正确;每页最多支持 200 条明细')
columns = {key: value.strip().upper() for key, value in mapping.get('columns', {}).items() if value.strip()}
fields = {'sequence', 'invoiceType', 'purpose', 'count', 'amount', 'remarks', 'category'}
if set(columns) - fields or not all(field in columns for field in ['purpose', 'amount']):
raise ValueError('至少需要指定“用途”和“金额”两列')
if len(set(columns.values())) != len(columns):
raise ValueError('不同明细字段不能使用同一列')
cells = {key: value.strip().upper() for key, value in mapping.get('cells', {}).items() if value.strip()}
if set(cells) - {'recipient', 'bankName', 'accountNumber', 'preparer', 'total'}:
raise ValueError('模板含有不支持的收款字段')
signatures = references(mapping.get('signatureCells', ''))
clears = references(mapping.get('clearCells', ''))
if len(signatures) > 6:
raise ValueError('最多支持 6 个签字岗位单元格')
merges = merged_ranges(document)
targets = []
for column in columns.values():
coordinate(column + str(start))
targets.extend(column + str(row) for row in range(start, end + 1))
for reference in list(cells.values()) + signatures + clears:
_, row = coordinate(reference)
if start <= row <= end:
raise ValueError('收款、合计、签字或额外清空位置不能放在明细行内:' + reference)
targets.append(reference)
if len(set(targets)) != len(targets):
raise ValueError('填充位置或清空位置重复,请检查配置')
for reference in targets:
if anchor(reference, merges) != reference:
raise ValueError(f'{reference} 位于合并单元格内,请填写左上角 {anchor(reference, merges)}')
column, row = coordinate(reference)
merged = next((entry for entry in merges if entry[0] == reference), None)
if start <= row <= end and merged and merged[4] != row:
raise ValueError('明细区不能跨行合并:' + reference)
normalized = dict(sheetName=mapping['sheetName'], startRow=start, endRow=end, columns=columns,
cells=cells, signatureCells=','.join(signatures), clearCells=','.join(clears))
if 'total' not in cells and not any(cell.find(tag('f')) is not None for cell in document.iter(tag('c'))):
raise ValueError('请指定合计单元格,或在模板中设置合计公式')
return normalized
def mapping_digest(mapping):
return hashlib.sha256(json.dumps(mapping, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
def validate_template(path, mapping):
template = Template(path)
normalized = validate_mapping(template, mapping)
return dict(templateVersion='mapped-xlsx-v1', fingerprint=template.digest,
mapping=normalized, mappingDigest=mapping_digest(normalized))
def set_cell(document, reference, value, formula=False):
column, row_number = coordinate(reference)
data = document.find(tag('sheetData'))
row = next((entry for entry in data if entry.get('r') == str(row_number)), None)
if row is None:
row = ET.Element(tag('row'), r=str(row_number))
index = next((index for index, entry in enumerate(data) if int(entry.get('r')) > row_number), len(data))
data.insert(index, row)
cell = next((entry for entry in row if entry.get('r') == reference), None)
if cell is None:
cell = ET.Element(tag('c'), r=reference)
index = next((index for index, entry in enumerate(row) if coordinate(entry.get('r'))[0] > column), len(row))
row.insert(index, cell)
for child in list(cell):
cell.remove(child)
if formula:
cell.attrib.pop('t', None)
ET.SubElement(cell, tag('f')).text = value
elif isinstance(value, (int, float, Decimal)):
cell.set('t', 'n')
ET.SubElement(cell, tag('v')).text = str(value)
else:
cell.set('t', 'inlineStr')
inline = ET.SubElement(cell, tag('is'))
ET.SubElement(inline, tag('t'), attrib={'{http://www.w3.org/XML/1998/namespace}space': 'preserve'}).text = str(value)
def export_custom_expense(state, destination, path, mapping, purposes, signatures, payee, category_purposes):
from exports import DEFAULT_SIGNATURES, expense_rows
template = Template(path)
signatures = signatures or DEFAULT_SIGNATURES
mapping = validate_mapping(template, mapping)
rows = expense_rows(state['matches'], purposes, category_purposes)
if not rows:
raise ValueError('请先勾选要导出的已核对材料')
sheet, sheet_path, original = template.sheet(mapping['sheetName'])
start, end = mapping['startRow'], mapping['endRow']
capacity = end - start + 1
pages = math.ceil(len(rows) / capacity)
if pages > 1 and original.find(tag('tableParts')) is not None:
raise ValueError('模板明细区使用了 Excel 表对象,不能自动复制分页;请增加明细行容量或先转换为普通区域')
parts = template.parts
sheets = template.workbook.find(tag('sheets'))
original_index = list(sheets).index(sheet)
names = template.workbook.find(tag('definedNames'))
local_names = [copy.deepcopy(item) for item in names if item.get('localSheetId') == str(original_index)] if names is not None else []
used_names = {item.get('name') for item in sheets}
maximum_id = max(int(item.get('sheetId')) for item in sheets)
source_rels = posixpath.dirname(sheet_path) + '/_rels/' + posixpath.basename(sheet_path) + '.rels'
for page in range(pages):
document = copy.deepcopy(original)
output_path = sheet_path
name = mapping['sheetName']
if page:
suffix = f'-续{page + 1}'
name = mapping['sheetName'][:31 - len(suffix)] + suffix
collision = 1
while name in used_names:
name = mapping['sheetName'][:23] + f'-续{page + 1}-{collision}'
collision += 1
used_names.add(name)
output_path = f'xl/worksheets/receipt-custom-{page + 1}.xml'
while output_path in parts:
output_path = output_path.replace('.xml', '-copy.xml')
relation_id = 'rIdReceiptCustom' + str(page)
while any(entry.get('Id') == relation_id for entry in template.relations):
relation_id += 'x'
new_index = len(sheets)
cloned_sheet = ET.SubElement(sheets, tag('sheet'), name=name, sheetId=str(maximum_id + page),
attrib={'{' + REL + '}id': relation_id})
if sheet.get('state'):
cloned_sheet.set('state', sheet.get('state'))
ET.SubElement(template.relations, '{' + PACKAGE + '}Relationship', Id=relation_id,
Type=REL + '/worksheet', Target=output_path[3:])
ET.SubElement(template.types, '{' + CONTENT + '}Override', PartName='/' + output_path,
ContentType='application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml')
if source_rels in parts:
parts['xl/worksheets/_rels/' + posixpath.basename(output_path) + '.rels'] = parts[source_rels]
old_ref = "'" + mapping['sheetName'].replace("'", "''") + "'!"
new_ref = "'" + name.replace("'", "''") + "'!"
for definition in local_names:
cloned = copy.deepcopy(definition)
cloned.set('localSheetId', str(new_index))
cloned.text = (cloned.text or '').replace(old_ref, new_ref).replace(mapping['sheetName'] + '!', new_ref)
names.append(cloned)
for formula in document.iter(tag('f')):
formula.text = (formula.text or '').replace(old_ref, new_ref).replace(mapping['sheetName'] + '!', new_ref)
for row_number in range(start, end + 1):
for column in mapping['columns'].values():
set_cell(document, column + str(row_number), '')
page_rows = rows[page * capacity:(page + 1) * capacity]
for index, row in enumerate(page_rows):
fields = dict(sequence=page * capacity + index + 1, invoiceType='/'.join(row['types']),
purpose=row['purpose'], category=row['category'], count=row['count'],
amount=row['amount'], remarks='')
for field, column in mapping['columns'].items():
set_cell(document, column + str(start + index), fields[field])
for field, reference in mapping['cells'].items():
if field == 'total':
set_cell(document, reference, f"SUM({mapping['columns']['amount']}{start}:{mapping['columns']['amount']}{end})", formula=True)
else:
value = str(payee.get(field, '') or '').strip()
if field == 'preparer' and not value:
value = str(payee.get('recipient', '') or '').strip()
set_cell(document, reference, value)
for reference in references(mapping['clearCells']):
set_cell(document, reference, '')
for index, reference in enumerate(references(mapping['signatureCells'])):
value = signatures[index].strip() if index < len(signatures) else ''
set_cell(document, reference, value + '' if value else '')
parts[output_path] = encoded(document)
for name in list(parts):
if name.startswith('xl/worksheets/') and name.endswith('.xml'):
document = xml(parts[name])
for cell in document.iter(tag('c')):
if cell.find(tag('f')) is not None:
for value in cell.findall(tag('v')):
cell.remove(value)
parts[name] = encoded(document)
parts.pop('xl/calcChain.xml', None)
for relation in list(template.relations):
if relation.get('Type', '').endswith('/calcChain'):
template.relations.remove(relation)
for override in list(template.types):
if override.get('PartName') == '/xl/calcChain.xml':
template.types.remove(override)
calc = template.workbook.find(tag('calcPr'))
if calc is None:
calc = ET.SubElement(template.workbook, tag('calcPr'))
calc.set('calcMode', 'auto')
calc.set('fullCalcOnLoad', '1')
calc.set('forceFullCalc', '1')
for name, document in [('xl/workbook.xml', template.workbook), ('xl/_rels/workbook.xml.rels', template.relations),
('[Content_Types].xml', template.types)]:
parts[name] = encoded(document)
with zipfile.ZipFile(destination, 'w', zipfile.ZIP_DEFLATED) as output:
for name, data in parts.items():
output.writestr(name, data)
return dict(expenseGrouping='category-v1', expenseRowCount=len(rows), templateVersion='mapped-xlsx-v1',
templateFingerprint=template.digest, templateMappingDigest=mapping_digest(mapping), expensePageCount=pages)
+118
View File
@@ -0,0 +1,118 @@
import datetime
import json
import os
import subprocess
import sys
import unittest
from decimal import Decimal
from pathlib import Path
from openpyxl import load_workbook
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import test_expense_template as fixtures
from expense_preview import PreviewCalculator, UnsupportedFormula, builtin_rmb, prepare_preview
from expense_template import Template, encoded, set_cell, tag
class ExpensePreviewTests(unittest.TestCase):
def setUp(self):
self.fixture = fixtures.ExpenseTemplateTests()
self.fixture.setUp()
self.addCleanup(self.fixture.tearDown)
def test_filled_workbook_previews_current_totals_and_preserves_formulas(self):
output, _ = self.fixture.export()
result = prepare_preview(output)
values = load_workbook(output, data_only=True)
formulas = load_workbook(output)
sheet = values.active
self.assertEqual(sheet['C9'].value, '交通')
self.assertEqual(sheet['H9'].value, 100.3)
self.assertEqual(sheet['H10'].value, 444)
self.assertEqual(sheet['F27'].value, 544.3)
self.assertEqual(sheet['H29'].value, 544.3)
self.assertEqual(formulas.active['F27'].value, '=SUM(H9:H21)')
self.assertEqual(formulas.active['H29'].value, '=IF(F27>C27,F27-C27,0)')
self.assertEqual(result['previewUncalculatedCells'], [])
self.assertEqual(sheet['D29'].value, '伍佰肆拾肆元叁角整')
self.assertEqual(result['expensePreview'], 'xlsx-quicklook-v1')
values.close()
formulas.close()
def test_builtin_uppercase_amount_formula(self):
for amount, expected in [('3491.86', '叁仟肆佰玖拾壹元捌角陆分'), ('10001.01', '壹万零壹元零壹分'),
('100000001', '壹亿零壹元整'), ('100010000', '壹亿零壹万元整'),
('0.01', '壹分'), ('0.1', '壹角整'), ('-50.02', '负伍拾元零贰分'),
('0', ''), ('1.999', '贰元整'), ('999999999999.99', '玖仟玖佰玖拾玖亿玖仟玖佰玖拾玖万玖仟玖佰玖拾玖元玖角玖分')]:
with self.subTest(amount=amount):
self.assertEqual(builtin_rmb(Decimal(amount)), expected)
def test_caches_each_custom_page_not_whole_report(self):
source, mapping = self.fixture.alternate()
state = dict(matches=self.fixture.state['matches'] + [self.fixture.match('third', '住宿', '50')])
output, _ = self.fixture.export(mapping=mapping, template=source, state=state)
prepare_preview(output)
workbook = load_workbook(output, data_only=True)
self.assertEqual(workbook['组 B 报销']['F8'].value, 544.3)
self.assertEqual(workbook['组 B 报销-续2']['F8'].value, 50)
self.assertIn('保留说明', workbook.sheetnames)
workbook.close()
def test_clears_unknown_cache_and_does_not_execute_formula(self):
output, _ = self.fixture.export()
template = Template(output)
_, path, document = template.sheet('个人报销单')
set_cell(document, 'I35', '__import__("os").system("touch /tmp/should-not-run")', formula=True)
from lxml import etree as ET
ET.SubElement(document.find('.//' + tag('c') + '[@r="I35"]'), tag('v')).text = '9999'
template.parts[path] = encoded(document)
self.fixture.rewrite(output, template.parts)
result = prepare_preview(output)
self.assertIn('个人报销单!I35', result['previewUncalculatedCells'])
workbook = load_workbook(output, data_only=True)
self.assertIsNone(workbook.active['I35'].value)
workbook.close()
def test_calculator_cycles_text_dates_and_conditionals(self):
template = Template(self.fixture.template)
_, _, sheet = template.sheet('个人报销单')
set_cell(sheet, 'J1', 10)
set_cell(sheet, 'J2', '文字')
set_cell(sheet, 'J3', '-J1*2+5', formula=True)
set_cell(sheet, 'J4', 'J5', formula=True)
set_cell(sheet, 'J5', 'J4', formula=True)
calculator = PreviewCalculator(sheet)
self.assertEqual(calculator.cell('J3'), -15)
self.assertEqual(calculator.expression('SUM(J1:J3)'), -5)
self.assertEqual(calculator.expression('IF(J1>0,J1/2,1/0)'), 5)
self.assertEqual(calculator.expression('TODAY()'), (datetime.date.today() - datetime.date(1899, 12, 30)).days)
self.assertEqual(PreviewCalculator(sheet, True).expression('TODAY()'),
(datetime.date.today() - datetime.date(1904, 1, 1)).days)
with self.assertRaises(UnsupportedFormula):
calculator.cell('J4')
with self.assertRaises(UnsupportedFormula):
calculator.expression('2**10000000')
def test_engine_process_preview_builtin_and_imported_templates(self):
for custom in [False, True]:
with self.subTest(custom=custom):
output = self.fixture.root / ('custom-preview.xlsx' if custom else 'builtin-preview.xlsx')
request = dict(operation='expense', state=self.fixture.state, destination=str(output), preparePreview=True)
if custom:
request.update(templatePath=str(self.fixture.template), templateMapping=self.fixture.mapping)
binary = os.environ.get('RECEIPT_ENGINE_BINARY')
command = [binary] if binary else [sys.executable, str(Path(__file__).resolve().parents[1] / 'engine.py')]
process = subprocess.run(command, input=json.dumps(request) + '\n', text=True, capture_output=True, timeout=60)
self.assertEqual(process.returncode, 0, process.stderr + process.stdout)
result = next(json.loads(line)['result'] for line in process.stdout.splitlines() if json.loads(line).get('event') == 'result')
self.assertEqual(result['expensePreview'], 'xlsx-quicklook-v1')
self.assertEqual(result['expenseRowCount'], 2)
workbook = load_workbook(output, data_only=True)
self.assertEqual(workbook.active['F27'].value, 544.3)
self.assertEqual(workbook.active['H29'].value, 544.3)
self.assertIn(workbook.active['C23'].value, ('', None))
workbook.close()
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,288 @@
import copy
import hashlib
import json
import os
import subprocess
import sys
import tempfile
import unittest
import zipfile
from pathlib import Path
from lxml import etree as ET
from openpyxl import load_workbook
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from expense_template import (CONTENT, MAIN, PACKAGE, REL, Template, encoded, export_custom_expense,
inspect_template, set_cell, tag, validate_template, xml)
from exports import export_expense
from test_domain import material, workspace
class ExpenseTemplateTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.root = Path(self.temporary.name)
self.builtin = Path(__file__).resolve().parents[1] / 'personal-expense-template.xlsx'
self.state = workspace([], [])
self.state['matches'] = [self.match('traffic-1', '交通', '30.10'), self.match('traffic-2', '交通', '70.20'),
self.match('office', '办公用品', '444.00')]
self.template = self.root / '旧报销单.xlsx'
old = dict(matches=[self.match('old-' + str(index), '旧类别' + str(index), '999.00') for index in range(7)])
export_expense(old, self.template, self.builtin, {}, ['旧岗位'],
dict(recipient='旧收款人', bankName='旧开户行', accountNumber='9999999', preparer='旧制单人'))
self.mapping = inspect_template(self.template)['sheets'][0]['mapping']
def tearDown(self):
self.temporary.cleanup()
def match(self, identifier, category, amount):
return dict(id=identifier, category=category, invoices=[material(identifier, 'invoice', amount)], payments=[])
def export(self, mapping=None, template=None, state=None, payee=None):
output = self.root / 'output.xlsx'
metadata = export_custom_expense(state or self.state, output, template or self.template, mapping or self.mapping,
{}, ['经办人', '财务'], payee or {},
{'交通': '交通', '办公用品': '=办公用品'})
return output, metadata
def rewrite(self, path, parts):
with zipfile.ZipFile(path, 'w', zipfile.ZIP_DEFLATED) as archive:
for name, data in parts.items():
archive.writestr(name, data)
def alternate(self):
template = Template(self.template)
sheet, path, _ = template.sheet('个人报销单')
sheet.set('name', '组 B 报销')
document = ET.Element(tag('worksheet'), nsmap={None: MAIN})
ET.SubElement(document, tag('sheetData'))
values = {'A1': '自定义组 B 报销单', 'A4': '序号', 'B4': '用途', 'D4': '单据数量', 'F4': '金额',
'G4': '备注', 'B5': '旧内容1', 'F5': 500, 'G5': '旧备注', 'B6': '旧内容2', 'F6': 800,
'A10': '收款人', 'B10': '旧收款人', 'A11': '开户行', 'B11': '旧银行',
'A12': '账号', 'B12': '旧账号', 'E2': '制单人', 'F2': '旧人', 'E8': '报销合计',
'A14': '要清空的旧值', 'A15': '旧签字', 'F15': '旧岗位', 'A16': '保留的自定义说明'}
for reference, value in values.items():
set_cell(document, reference, value)
set_cell(document, 'F8', 'SUM(F5:F6)', formula=True)
merges = ET.SubElement(document, tag('mergeCells'), count='3')
for reference in ['B4:C4', 'B5:C5', 'B6:C6']:
ET.SubElement(merges, tag('mergeCell'), ref=reference)
ET.SubElement(document, tag('pageMargins'), left='0.2', right='0.2', top='0.3', bottom='0.3', header='0', footer='0')
ET.SubElement(document, tag('pageSetup'), paperSize='9', orientation='landscape', fitToWidth='1')
template.parts[path] = encoded(document)
names = template.workbook.find(tag('definedNames'))
for item in list(names):
names.remove(item)
ET.SubElement(names, tag('definedName'), name='_xlnm.Print_Area', localSheetId='0').text = "'组 B 报销'!$A$1:$G$16"
sheet_list = template.workbook.find(tag('sheets'))
ET.SubElement(sheet_list, tag('sheet'), name='保留说明', sheetId='2', attrib={'{' + REL + '}id': 'rIdNotes'})
ET.SubElement(template.relations, '{' + PACKAGE + '}Relationship', Id='rIdNotes', Type=REL + '/worksheet',
Target='worksheets/notes.xml')
ET.SubElement(template.types, '{' + CONTENT + '}Override', PartName='/xl/worksheets/notes.xml',
ContentType='application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml')
notes = ET.Element(tag('worksheet'), nsmap={None: MAIN})
ET.SubElement(notes, tag('sheetData'))
set_cell(notes, 'A1', '保持这个工作表不变')
template.parts['xl/worksheets/notes.xml'] = encoded(notes)
template.parts['xl/workbook.xml'] = encoded(template.workbook)
template.parts['xl/_rels/workbook.xml.rels'] = encoded(template.relations)
template.parts['[Content_Types].xml'] = encoded(template.types)
target = self.root / 'different-layout.xlsx'
self.rewrite(target, template.parts)
mapping = inspect_template(target)['sheets'][0]['mapping']
mapping['signatureCells'] = 'A15,F15'
mapping['clearCells'] = 'A14'
return target, mapping
def test_auto_detect_uploaded_style_and_merged_signature_slots(self):
self.assertEqual(self.mapping['sheetName'], '个人报销单')
self.assertEqual((self.mapping['startRow'], self.mapping['endRow']), (9, 21))
self.assertEqual(self.mapping['columns'], dict(purpose='C', amount='H', count='G', invoiceType='B', sequence='A', remarks='I'))
self.assertEqual(self.mapping['cells'], dict(recipient='C23', bankName='C24', accountNumber='C25', preparer='H6', total='F27'))
self.assertNotIn('D32', self.mapping['signatureCells'])
validation = validate_template(self.template, self.mapping)
self.assertEqual(validation['mapping'], self.mapping)
self.assertEqual(len(validation['fingerprint']), 64)
def test_replaces_old_seven_rows_with_two_aggregated_categories(self):
original = self.template.read_bytes()
output, metadata = self.export()
self.assertEqual(metadata['expenseRowCount'], 2)
self.assertEqual(metadata['expensePageCount'], 1)
workbook = load_workbook(output)
sheet = workbook.active
self.assertEqual(sheet['C9'].value, '交通')
self.assertEqual(sheet['H9'].value, 100.30)
self.assertEqual(sheet['G9'].value, 2)
self.assertEqual(sheet['C10'].value, '=办公用品')
self.assertEqual(sheet['C10'].data_type, 's')
self.assertEqual(sheet['H10'].value, 444)
for row in range(11, 22):
for column in ['A', 'B', 'C', 'G', 'H', 'I']:
self.assertIn(sheet[f'{column}{row}'].value, ('', None))
for reference in ['C23', 'C24', 'C25', 'H6']:
self.assertIn(sheet[reference].value, ('', None))
self.assertEqual(sheet['F27'].value, '=SUM(H9:H21)')
self.assertEqual(sheet['A30'].value, '经办人:')
self.assertEqual(sheet['D30'].value, '财务:')
self.assertEqual(self.template.read_bytes(), original)
workbook.close()
def test_preserves_styles_merges_print_settings_and_media(self):
output, _ = self.export()
with zipfile.ZipFile(self.template) as original, zipfile.ZipFile(output) as generated:
for name in original.namelist():
if name == 'xl/styles.xml' or name.startswith('xl/media/') or name.startswith('xl/drawings/'):
self.assertEqual(original.read(name), generated.read(name), name)
old, new = load_workbook(self.template), load_workbook(output)
for reference in ['A1', 'A3', 'A4', 'C9', 'H9', 'C25']:
self.assertEqual(old.active[reference].style_id, new.active[reference].style_id)
self.assertEqual(str(old.active.merged_cells), str(new.active.merged_cells))
self.assertEqual(str(old.active.print_area), str(new.active.print_area))
self.assertEqual(old.active.page_setup, new.active.page_setup)
self.assertEqual(old.active['A3'].value, new.active['A3'].value)
self.assertEqual(old.active['A4'].value, new.active['A4'].value)
self.assertEqual(old.active['D29'].value, new.active['D29'].value)
old.close()
new.close()
def test_alternate_layout_preserves_other_sheet_and_paginates(self):
template, mapping = self.alternate()
self.assertEqual(mapping['columns']['purpose'], 'B')
self.assertEqual((mapping['startRow'], mapping['endRow']), (5, 6))
self.assertEqual(mapping['cells']['total'], 'F8')
state = dict(matches=self.state['matches'] + [self.match('hotel', '住宿', '9.99')])
output, result = self.export(mapping, template, state,
dict(recipient='新收款人', bankName='=新银行', accountNumber='0001234567890123456789'))
self.assertEqual(result['expensePageCount'], 2)
workbook = load_workbook(output)
self.assertEqual(workbook.sheetnames, ['组 B 报销', '保留说明', '组 B 报销-续2'])
self.assertEqual(workbook['保留说明']['A1'].value, '保持这个工作表不变')
for sheet in [workbook['组 B 报销'], workbook['组 B 报销-续2']]:
self.assertEqual(sheet['B10'].value, '新收款人')
self.assertEqual(sheet['F2'].value, '新收款人')
self.assertEqual(sheet['B11'].data_type, 's')
self.assertEqual(sheet['B12'].value, '0001234567890123456789')
self.assertEqual(sheet['B12'].data_type, 's')
self.assertEqual(sheet['F8'].value, '=SUM(F5:F6)')
self.assertEqual(sheet['A16'].value, '保留的自定义说明')
self.assertIn(sheet['A14'].value, ('', None))
self.assertEqual(sheet['A15'].value, '经办人:')
self.assertEqual(sheet.page_setup.orientation, 'landscape')
self.assertTrue(str(sheet.print_area).endswith('!$A$1:$G$16'))
self.assertEqual(workbook['组 B 报销-续2']['B5'].value, '住宿')
self.assertEqual(workbook['组 B 报销-续2']['A5'].value, 3)
self.assertIn(workbook['组 B 报销-续2']['B6'].value, ('', None))
workbook.close()
def test_unmapped_signature_cells_preserve_template(self):
mapping = copy.deepcopy(self.mapping)
mapping['signatureCells'] = ''
output, _ = self.export(mapping)
book = load_workbook(output)
self.assertEqual(book.active['A30'].value, '旧岗位:')
book.close()
def test_mapping_normalizes_lowercase(self):
mapping = copy.deepcopy(self.mapping)
mapping['columns']['purpose'] = ' c '
mapping['cells']['recipient'] = ' c23 '
mapping['signatureCells'] = 'a30 d30'
result = validate_template(self.template, mapping)
self.assertEqual(result['mapping']['columns']['purpose'], 'C')
self.assertEqual(result['mapping']['cells']['recipient'], 'C23')
self.assertEqual(result['mapping']['signatureCells'], 'A30,D30')
def test_invalid_mappings_do_not_write_output(self):
mutations = [
lambda mapping: mapping.update(startRow=22, endRow=9),
lambda mapping: mapping.update(startRow=0),
lambda mapping: mapping.update(endRow=300),
lambda mapping: mapping.update(sheetName='不存在'),
lambda mapping: mapping['columns'].update(purpose='D'),
lambda mapping: mapping['columns'].update(count='H'),
lambda mapping: mapping['columns'].update(amount=''),
lambda mapping: mapping['columns'].update(amount='A1'),
lambda mapping: mapping['cells'].update(recipient='C9'),
lambda mapping: mapping['cells'].update(bankName='C23'),
lambda mapping: mapping.update(signatureCells='D32'),
lambda mapping: mapping.update(clearCells='D23'),
lambda mapping: mapping.update(clearCells='C9'),
]
for mutate in mutations:
mapping = copy.deepcopy(self.mapping)
mutate(mapping)
with self.subTest(mapping=mapping), self.assertRaises(ValueError):
self.export(mapping)
self.assertFalse((self.root / 'output.xlsx').exists())
def test_corrupt_and_non_xlsx_rejected(self):
for name in ['bad.xlsx', 'old.xls', 'macro.xlsm']:
path = self.root / name
path.write_bytes(b'not an xlsx')
with self.assertRaises(ValueError):
inspect_template(path)
def test_live_external_formula_rejected_but_orphans_removed(self):
template = Template(self.template)
_, path, document = template.sheet('个人报销单')
set_cell(document, 'J1', "'[1]Sheet1'!A1", formula=True)
template.parts[path] = encoded(document)
live = self.root / 'external.xlsx'
self.rewrite(live, template.parts)
with self.assertRaisesRegex(ValueError, '外部'):
inspect_template(live)
output, _ = self.export()
with zipfile.ZipFile(output) as archive:
self.assertFalse(any(name.startswith('xl/externalLinks/') for name in archive.namelist()))
def test_formula_caches_are_invalidated(self):
output, _ = self.export()
template = Template(output)
_, _, document = template.sheet('个人报销单')
self.assertTrue(list(document.iter(tag('f'))))
for cell in document.iter(tag('c')):
if cell.find(tag('f')) is not None:
self.assertIsNone(cell.find(tag('v')))
self.assertEqual(template.workbook.find(tag('calcPr')).get('calcMode'), 'auto')
self.assertNotIn('xl/calcChain.xml', template.parts)
def engine(self, request):
binary = os.environ.get('RECEIPT_ENGINE_BINARY')
command = [binary] if binary else [sys.executable, str(Path(__file__).resolve().parents[1] / 'engine.py')]
process = subprocess.run(command, input=json.dumps(request), capture_output=True, text=True, timeout=60)
return process, json.loads(process.stdout.splitlines()[-1])
def test_engine_process_inspects_validates_and_uses_custom_template(self):
process, inspected = self.engine(dict(operation='inspect-expense-template', templatePath=str(self.template)))
self.assertEqual(process.returncode, 0, process.stderr)
self.assertEqual(inspected['result']['sheets'][0]['mapping'], self.mapping)
process, validated = self.engine(dict(operation='validate-expense-template', templatePath=str(self.template), templateMapping=self.mapping))
self.assertEqual(process.returncode, 0, process.stderr)
output = self.root / 'protocol.xlsx'
process, result = self.engine(dict(operation='expense', state=self.state, destination=str(output),
templatePath=str(self.template), templateMapping=self.mapping,
payee=dict(recipient='新用户', accountNumber='0009876543210987654321')))
self.assertEqual(process.returncode, 0, process.stderr)
self.assertEqual(result['result']['templateFingerprint'], hashlib.sha256(self.template.read_bytes()).hexdigest())
self.assertEqual(result['result']['templateMappingDigest'], validated['result']['mappingDigest'])
self.assertEqual(result['result']['expenseRowCount'], 2)
book = load_workbook(output)
self.assertEqual(book.active['C23'].value, '新用户')
self.assertEqual(book.active['H9'].value, 100.30)
book.close()
def test_engine_process_missing_custom_template_never_falls_back(self):
output = self.root / 'existing.xlsx'
output.write_bytes(b'keep existing output')
process, result = self.engine(dict(operation='expense', state=self.state, destination=str(output),
templatePath=str(self.root / 'missing.xlsx'), templateMapping=self.mapping))
self.assertNotEqual(process.returncode, 0)
self.assertEqual(result['event'], 'error')
self.assertEqual(output.read_bytes(), b'keep existing output')
self.assertFalse(list(self.root.glob('.*.xlsx')))
if __name__ == '__main__':
unittest.main()