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)