305 lines
16 KiB
Python
305 lines
16 KiB
Python
import copy
|
||
import io
|
||
import math
|
||
import posixpath
|
||
import zipfile
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
from xml.etree import ElementTree as XML
|
||
|
||
from PIL import Image
|
||
from domain import material_date, text
|
||
|
||
DEFAULT_SIGNATURES = ['部门长', '剧组出纳', '制片主任', '剧组会计', '执行制片人', '制片人']
|
||
|
||
|
||
def image_for(item):
|
||
return Image.open(item.get('previewPath') or item['path']).convert('RGB')
|
||
|
||
|
||
def export_approved_ppt(records, destination, classified):
|
||
import contextlib
|
||
import tempfile
|
||
import pypdfium2 as pdfium
|
||
|
||
if not records or len(records) > 100:
|
||
raise ValueError('每批请选择 1 至 100 笔已批准申请')
|
||
categories = {'交通差旅': '交通', '住宿费用': '住宿'}
|
||
with tempfile.TemporaryDirectory(prefix='approved-ppt-pages-', dir=Path(destination).parent) as directory:
|
||
matches = []
|
||
for record in records:
|
||
match = dict(invoices=[], payments=[], category=categories.get(record['type'], record['type']))
|
||
match['scope'] = (record.get('groupId', record['team']), record.get('userId', record['name']))
|
||
for material in record['materials']:
|
||
if material['kind'] not in ('invoice', 'payment'):
|
||
continue
|
||
path = Path(material['path'])
|
||
try:
|
||
if path.suffix.lower() == '.pdf':
|
||
with contextlib.closing(pdfium.PdfDocument(str(path))) as document:
|
||
if not len(document):
|
||
raise ValueError('PDF 没有可预览页面')
|
||
with contextlib.closing(document[0]) as page:
|
||
with contextlib.closing(page.render(scale=140 / 72)) as bitmap:
|
||
preview = Path(directory) / f'{len(matches)}-{len(match["invoices"])}-{len(match["payments"])}.png'
|
||
bitmap.to_pil().convert('RGB').save(preview)
|
||
path = preview
|
||
else:
|
||
with Image.open(path) as image:
|
||
image.verify()
|
||
except Exception as error:
|
||
raise ValueError(f"申请 #{record['id']} 的“{material['name']}”无法导出：{error}") from error
|
||
item = dict(path=str(path), ocr=dict(rawText='', dates=[record['date']]))
|
||
match['invoices' if material['kind'] == 'invoice' else 'payments'].append(item)
|
||
if not match['invoices'] and not match['payments']:
|
||
raise ValueError(f"申请 #{record['id']} 没有发票或付款截图，无法按本地贴票格式导出，请取消选择该申请")
|
||
matches.append(match)
|
||
return export_ppt(dict(matches=matches), destination, classified, preserve_order=True)
|
||
|
||
|
||
def export_ppt(state, destination, classified, preserve_order=False):
|
||
if 'approvedRecords' in state:
|
||
return export_approved_ppt(state['approvedRecords'], destination, classified)
|
||
|
||
from pptx import Presentation
|
||
from pptx.util import Pt
|
||
from pptx.dml.color import RGBColor
|
||
from pptx.enum.text import PP_ALIGN, MSO_ANCHOR
|
||
|
||
presentation = Presentation()
|
||
presentation.slide_width, presentation.slide_height = Pt(720), Pt(960)
|
||
|
||
def slide():
|
||
page = presentation.slides.add_slide(presentation.slide_layouts[6])
|
||
page.background.fill.solid()
|
||
page.background.fill.fore_color.rgb = RGBColor(255, 255, 255)
|
||
return page
|
||
|
||
def picture(page, item, box, kind):
|
||
left, top, width, height = box
|
||
with image_for(item) as image:
|
||
scale = min(width / image.width, height / image.height)
|
||
actual_width, actual_height = image.width * scale, image.height * scale
|
||
with io.BytesIO() as content:
|
||
image.save(content, format='PNG')
|
||
content.seek(0)
|
||
shape = page.shapes.add_picture(
|
||
content, Pt(left + (width - actual_width) / 2), Pt(top + (height - actual_height) / 2),
|
||
Pt(actual_width), Pt(actual_height)
|
||
)
|
||
shape.name = f'{kind}-{len(page.shapes)}'
|
||
|
||
def photo_hint(page, box):
|
||
shape = page.shapes.add_textbox(*(Pt(value) for value in box))
|
||
shape.name = 'manual-photo-area'
|
||
frame = shape.text_frame
|
||
frame.clear()
|
||
frame.vertical_anchor = MSO_ANCHOR.MIDDLE
|
||
frame.word_wrap = True
|
||
frame.text = '实物照片粘贴区\n请手动粘贴，可删除此提示'
|
||
for paragraph in frame.paragraphs:
|
||
paragraph.alignment = PP_ALIGN.CENTER
|
||
for run in paragraph.runs:
|
||
run.font.name = 'PingFang SC'
|
||
run.font.size = Pt(14)
|
||
run.font.color.rgb = RGBColor(155, 155, 155)
|
||
|
||
def combined_page(invoice, payments, inline_photo=False):
|
||
page = slide()
|
||
picture(page, invoice, (48, 44, 624, 360), 'invoice')
|
||
for index, item in enumerate(payments):
|
||
picture(page, item, (48 + index * 324, 444, 300, 468), 'payment')
|
||
if inline_photo:
|
||
photo_hint(page, (372, 444, 300, 468))
|
||
|
||
def payment_pages(payments):
|
||
for offset in range(0, len(payments), 6):
|
||
page = slide()
|
||
for index, item in enumerate(payments[offset:offset + 6]):
|
||
box = (48 + index % 3 * 216, 48 + index // 3 * 444, 192, 420)
|
||
picture(page, item, box, 'payment')
|
||
|
||
matches = state['matches'] if preserve_order else sorted(
|
||
state['matches'],
|
||
key=lambda match: (
|
||
min((material_date(item, True) for item in match['invoices']), default='9999-12-31'),
|
||
match['category'] if classified else ''
|
||
)
|
||
)
|
||
if not matches:
|
||
raise ValueError('请先选择要导出的贴票材料')
|
||
for match in matches:
|
||
invoices = sorted(match.get('invoices', []), key=material_date)
|
||
payments = sorted(match.get('payments', []), key=material_date)
|
||
if not invoices and not payments:
|
||
raise ValueError('所选组没有发票或付款截图，无法导出')
|
||
if len(invoices) == 1 and len(payments) == 1:
|
||
combined_page(invoices[0], payments, inline_photo=True)
|
||
continue
|
||
paired_count = len(invoices) - len(invoices) % 2
|
||
for offset in range(0, paired_count, 2):
|
||
page = slide()
|
||
picture(page, invoices[offset], (48, 44, 624, 408), 'invoice')
|
||
picture(page, invoices[offset + 1], (48, 508, 624, 408), 'invoice')
|
||
if len(invoices) % 2:
|
||
combined_page(invoices[-1], payments[:2])
|
||
payments = payments[2:]
|
||
payment_pages(payments)
|
||
photo_hint(slide(), (48, 48, 624, 864))
|
||
presentation.save(destination)
|
||
return dict(pptLayout='portrait-receipts-v1', pptSlideCount=len(presentation.slides))
|
||
|
||
|
||
def export_travel(state, destination):
|
||
import mimetypes
|
||
mimetypes.knownfiles = []
|
||
mimetypes.init()
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Font, PatternFill, Alignment
|
||
from openpyxl.utils import get_column_letter
|
||
invoices = [item for match in state['matches'] for item in match['invoices'] if item['ocr']['travel']['type']]
|
||
if not invoices:
|
||
raise ValueError('已核对材料中没有识别到火车票或航班行程')
|
||
workbook = Workbook()
|
||
sheet = workbook.active
|
||
sheet.title = '火车及航班行程'
|
||
sheet.append(['序号', '交通类型', '姓名', '出发地', '目的地', '出发时间', '车次/航班号', '金额', '发票文件'])
|
||
for cell in sheet[1]:
|
||
cell.font = Font(bold=True, color='FFFFFF')
|
||
cell.fill = PatternFill('solid', fgColor='006633')
|
||
cell.alignment = Alignment(horizontal='center')
|
||
for index, item in enumerate(invoices, 1):
|
||
travel = item['ocr']['travel']
|
||
values = [index, '火车' if travel['type'] == 'train' else '航班', travel['travelerName'], travel['departure'], travel['destination'], travel['departureTime'], travel['transportNumber'], next(iter(item['ocr']['amounts']), ''), item['name']]
|
||
sheet.append(values)
|
||
for cell in sheet[sheet.max_row]:
|
||
if isinstance(cell.value, str):
|
||
cell.data_type = 's'
|
||
for index, width in enumerate([8, 12, 16, 18, 18, 22, 18, 14, 38], 1):
|
||
sheet.column_dimensions[get_column_letter(index)].width = width
|
||
sheet.freeze_panes = 'A2'
|
||
workbook.save(destination)
|
||
|
||
|
||
def expense_rows(matches, purposes, category_purposes=None):
|
||
grouped = {}
|
||
for match in matches:
|
||
category = match.get('category') or '其他'
|
||
row = grouped.setdefault(category, dict(category=category, count=0, amount=Decimal(0), types=[], purposes=[]))
|
||
row['count'] += len(match['invoices'])
|
||
purpose = purposes.get(match['id'], '').strip()
|
||
if purpose and purpose not in row['purposes']:
|
||
row['purposes'].append(purpose)
|
||
for invoice in match['invoices']:
|
||
content = text(invoice)
|
||
if '专用发票' in content:
|
||
invoice_type = '专票'
|
||
elif 'invoice' in content.lower():
|
||
invoice_type = 'Invoice'
|
||
elif '押金' in content and '收据' in content:
|
||
invoice_type = '押金收据'
|
||
else:
|
||
invoice_type = '普票'
|
||
if invoice_type not in row['types']:
|
||
row['types'].append(invoice_type)
|
||
row['amount'] += max((abs(Decimal(value.replace(',', ''))) for value in invoice['ocr']['amounts']), default=Decimal(0))
|
||
for row in grouped.values():
|
||
if category_purposes is not None:
|
||
row['purpose'] = category_purposes.get(row['category'], '').strip() or row['category']
|
||
else:
|
||
row['purpose'] = '；'.join(row['purposes']) or row['category']
|
||
return list(grouped.values())
|
||
|
||
|
||
def export_expense(state, destination, template, purposes, signatures, payee=None, category_purposes=None):
|
||
from lxml import etree as ET
|
||
namespace = 'http://schemas.openxmlformats.org/spreadsheetml/2006/main'
|
||
relationships = 'http://schemas.openxmlformats.org/officeDocument/2006/relationships'
|
||
package = 'http://schemas.openxmlformats.org/package/2006/relationships'
|
||
content_namespace = 'http://schemas.openxmlformats.org/package/2006/content-types'
|
||
tag = lambda name: '{' + namespace + '}' + name
|
||
with zipfile.ZipFile(template) as source:
|
||
parts = {name: source.read(name) for name in source.namelist()}
|
||
workbook = ET.fromstring(parts['xl/workbook.xml'])
|
||
relations = ET.fromstring(parts['xl/_rels/workbook.xml.rels'])
|
||
content_types = ET.fromstring(parts['[Content_Types].xml'])
|
||
sheets = workbook.find(tag('sheets'))
|
||
original = next(sheet for sheet in sheets if sheet.get('name') == '1个人报销单')
|
||
original_index = list(sheets).index(original)
|
||
relation = next(item for item in relations if item.get('Id') == original.get('{' + relationships + '}id'))
|
||
sheet_path = posixpath.normpath(posixpath.join('xl', relation.get('Target'))) if not relation.get('Target').startswith('/') else relation.get('Target').lstrip('/')
|
||
sheet_bytes = parts[sheet_path]
|
||
names = workbook.find(tag('definedNames'))
|
||
retained_names = [copy.deepcopy(item) for item in names if item.get('localSheetId') == str(original_index)] if names is not None else []
|
||
if names is not None:
|
||
names.clear()
|
||
sheets.clear()
|
||
signatures = signatures or DEFAULT_SIGNATURES
|
||
rows = expense_rows(state['matches'], purposes, category_purposes)
|
||
if not rows:
|
||
raise ValueError('请先勾选要导出的已核对材料')
|
||
page_count = math.ceil(len(rows) / 13)
|
||
|
||
def set_cell(document, reference, value):
|
||
cell = document.find('.//' + tag('c') + '[@r="' + reference + '"]')
|
||
if cell is None:
|
||
row_number = ''.join(character for character in reference if character.isdigit())
|
||
row = document.find('.//' + tag('row') + '[@r="' + row_number + '"]')
|
||
if row is None:
|
||
raise ValueError('报销模板缺少明细行：' + row_number)
|
||
cell = ET.SubElement(row, tag('c'), r=reference)
|
||
for child in list(cell):
|
||
cell.remove(child)
|
||
if 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')).text = str(value)
|
||
|
||
for page in range(page_count):
|
||
document = ET.fromstring(sheet_bytes)
|
||
profile = payee or {}
|
||
for reference, field in [('C23', 'recipient'), ('C24', 'bankName'), ('C25', 'accountNumber'), ('H6', 'preparer')]:
|
||
value = str(profile.get(field, '') or '').strip()
|
||
if field == 'preparer' and not value:
|
||
value = str(profile.get('recipient', '') or '').strip()
|
||
set_cell(document, reference, value)
|
||
for row_number in range(9, 22):
|
||
for column in ['B', 'C', 'G', 'H', 'I']:
|
||
set_cell(document, column + str(row_number), '')
|
||
set_cell(document, 'A' + str(row_number), row_number - 8)
|
||
for index, row in enumerate(rows[page * 13:(page + 1) * 13]):
|
||
for column, value in [('B', '/'.join(row['types'])), ('C', row['purpose']), ('G', row['count']), ('H', row['amount'])]:
|
||
set_cell(document, column + str(index + 9), value)
|
||
for index, cell in enumerate(['A30', 'D30', 'A31', 'D31', 'A32', 'D32']):
|
||
value = signatures[index].strip() if index < len(signatures) else ''
|
||
set_cell(document, cell, value + '：' if value else '')
|
||
name = '个人报销单' if page_count == 1 else f'个人报销单-{page + 1}'
|
||
output_path = f'xl/worksheets/native-expense-{page + 1}.xml'
|
||
relation_id = f'rIdNativeExpense{page + 1}'
|
||
ET.SubElement(sheets, tag('sheet'), name=name, sheetId=str(page + 1), attrib={'{' + relationships + '}id': relation_id})
|
||
ET.SubElement(relations, '{' + package + '}Relationship', Id=relation_id, Type=relationships + '/worksheet', Target=output_path[3:])
|
||
ET.SubElement(content_types, '{' + content_namespace + '}Override', PartName='/' + output_path, ContentType='application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml')
|
||
parts[output_path] = ET.tostring(document, xml_declaration=True, encoding='UTF-8', standalone=True)
|
||
rel_path = posixpath.dirname(sheet_path) + '/_rels/' + posixpath.basename(sheet_path) + '.rels'
|
||
if rel_path in parts:
|
||
parts['xl/worksheets/_rels/' + posixpath.basename(output_path) + '.rels'] = parts[rel_path]
|
||
for definition in retained_names:
|
||
cloned = copy.deepcopy(definition)
|
||
cloned.set('localSheetId', str(page))
|
||
cloned.text = (cloned.text or '').replace("'1个人报销单'!", "'" + name + "'!").replace('1个人报销单!', "'" + name + "'!")
|
||
names.append(cloned)
|
||
calc = workbook.find(tag('calcPr'))
|
||
if calc is None:
|
||
calc = ET.SubElement(workbook, tag('calcPr'))
|
||
calc.set('fullCalcOnLoad', '1')
|
||
calc.set('forceFullCalc', '1')
|
||
for key, document in [('xl/workbook.xml', workbook), ('xl/_rels/workbook.xml.rels', relations), ('[Content_Types].xml', content_types)]:
|
||
parts[key] = ET.tostring(document, xml_declaration=True, encoding='UTF-8', standalone=True)
|
||
with zipfile.ZipFile(destination, 'w', zipfile.ZIP_DEFLATED) as output:
|
||
for name, content in parts.items():
|
||
output.writestr(name, content)
|
||
return dict(expenseGrouping='category-v1', expenseRowCount=len(rows))
|