"""应用内的XLSX结果导出与只读预览；只写显式数值，不执行上传文件中的公式。"""
import csv
from io import StringIO
import json
import math
from pathlib import PurePosixPath
import posixpath
import re
from xml.etree import ElementTree as ET
from zipfile import BadZipFile, ZIP_DEFLATED, ZipFile
NS='http://schemas.openxmlformats.org/spreadsheetml/2006/main'
REL='http://schemas.openxmlformats.org/officeDocument/2006/relationships'
PKG='http://schemas.openxmlformats.org/package/2006/relationships'
ET.register_namespace('',NS)
def _plain(value):
if isinstance(value, dict) and 'components_kw' in value:
return '+'.join(f'{v:g}' for v in value['components_kw'])+f" kW（合计{value['total_kw']:g} kW）"
return json.dumps(value,ensure_ascii=False) if isinstance(value,(list,dict)) else value
def result_tables(result):
payload=result.get('platform_payload') or {}
judgment=payload.get('judgment') or {}
assessment=result.get('assessment') or {}
names=[f['fan_name'] for f in assessment.get('fan_pair',[])]
names=names if len(names)==2 else ['设备1','设备2']
rows=[[result.get('result_title') or '智能体研判结果'],['矿井',result.get('mine_name','')],['研判对象',result.get('check_object','')],
['研判时间',payload.get('checkedAt','')],['结果编码',judgment.get('resultCode','')],['综合结论',judgment.get('result','')],[],
['判识项目','实际值','单位','正常上限','分项结论']]
if assessment.get('mode') == 'sandbox_exploration':
# 只导出 Agent 的实际交付，不再填充旧七阶段的固定业务行或替它生成计算结果。
agent_result = assessment['agent_result']
rows = [*rows[:7], ['数据范围', agent_result.get('data_detail', '')], [], ['来源', '物理页', '字段路径', '原文']]
rows.extend([[item['source'], item.get('page'), item.get('pointer', ''), item.get('excerpt', '')]
for item in assessment.get('evidence', [])])
details = [['字段', 'Agent 交付值']]
details.extend([[key, _plain(value)] for key, value in agent_result.get('details', {}).items()])
return [{'name': '研判结果', 'rows': rows, 'headers': [10], 'widths': [32, 46, 36, 20, 24]},
{'name': 'Agent分析明细', 'rows': details, 'headers': [1], 'widths': [38, 100]}]
labels={'pass':'通过','fail':'不通过','insufficient':'未形成有效对照'}
decision=assessment.get('decision',{})
if assessment:
rows.append(['两台风机基础参数一致性','','','',labels.get(decision.get('items',{}).get('parameters'),'')])
for event in assessment.get('switch_comparison',{}).get('events',[]):
rows.extend([['停机至启动间隔',event['metrics'].get('switch_interval_minutes'),'分钟',event['timing']['limit_minutes'],labels.get(event['checks'].get('switch_interval'),'')],
['切换前后稳定风量差',event['metrics'].get('flow_deviation_percent'),'%',event['limits']['flow_deviation_percent'],labels.get(event['checks'].get('flow_change'),'')]])
basic=assessment.get('basic_comparison',{}).get('rows',[])
headers=[8]
if basic:
rows.extend([[],['基础参数',*(names or ['设备1','设备2']),'比较结果']])
headers.append(len(rows))
units={'rated_speed_rpm':'r/min','impeller_diameter_m':'m','rated_flow_range_m3_min':'m³/min','rated_pressure_range_pa':'Pa'}
rows.extend([[r['label']+(f"（{units[r['field']]}）" if r['field'] in units else ''),
*[_plain(v) for v in r['values']],'一致' if r['status']=='match' else '不一致'] for r in basic])
tables=[{'name':'研判结果','rows':rows,'headers':headers,'widths':[32,46,36,20,24]}]
curves=assessment.get('performance_comparison',{}).get('results',[])
if curves:
rows=[['频率（Hz）','风量（m³/min）',f'{names[0]}静压（Pa）',f'{names[1]}静压（Pa）',
f'{names[0]}轴功率（kW）',f'{names[1]}轴功率（kW）',f'{names[0]}静压效率（%）',f'{names[1]}静压效率（%）']]
for curve in curves:
rows.extend([[curve['frequency_hz'],r['flow_m3_min'],r['fan_1'][0],r['fan_2'][0],r['fan_1'][1],r['fan_2'][1],r['fan_1'][2],r['fan_2'][2]] for r in curve['curve_samples']])
tables.append({'name':'性能曲线比较','rows':rows,'headers':[1],'widths':[15,23,23,23,23,23,25,25]})
return tables
def _column(index):
text=''
while index:
index,remainder=divmod(index-1,26)
text=chr(65+remainder)+text
return text
def _xml(element):
return ET.tostring(element,encoding='utf-8',xml_declaration=True)
def write_workbook(path, tables):
"""服务运行时生成标准OOXML；预览直接读取这个下载文件。"""
content=ET.Element('Types',xmlns='http://schemas.openxmlformats.org/package/2006/content-types')
for extension,kind in [('rels','application/vnd.openxmlformats-package.relationships+xml'),('xml','application/xml')]:
ET.SubElement(content,'Default',Extension=extension,ContentType=kind)
for part,kind in [('workbook','sheet.main'),('styles','styles')]:
ET.SubElement(content,'Override',PartName=f'/xl/{part}.xml',ContentType=f'application/vnd.openxmlformats-officedocument.spreadsheetml.{kind}+xml')
workbook=ET.Element(f'{{{NS}}}workbook')
sheets=ET.SubElement(workbook,f'{{{NS}}}sheets')
relationships=ET.Element('Relationships',xmlns=PKG)
styles=''''''
style_root=ET.fromstring(styles)
for xf in style_root.findall(f'{{{NS}}}cellXfs/{{{NS}}}xf'):
alignment=xf.find(f'{{{NS}}}alignment')
if alignment is None:
alignment=ET.SubElement(xf,f'{{{NS}}}alignment')
alignment.set('vertical','center')
xf.set('applyAlignment','1')
styles=_xml(style_root)
with ZipFile(path,'w',ZIP_DEFLATED) as archive:
archive.writestr('_rels/.rels',f'')
archive.writestr('xl/styles.xml',styles)
for number,table in enumerate(tables,1):
ET.SubElement(sheets,f'{{{NS}}}sheet',name=table['name'],sheetId=str(number),attrib={f'{{{REL}}}id':f'rId{number}'})
ET.SubElement(relationships,'Relationship',Id=f'rId{number}',Type=f'{REL}/worksheet',Target=f'worksheets/sheet{number}.xml')
ET.SubElement(content,'Override',PartName=f'/xl/worksheets/sheet{number}.xml',ContentType='application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml')
sheet=ET.Element(f'{{{NS}}}worksheet')
views=ET.SubElement(sheet,f'{{{NS}}}sheetViews')
view=ET.SubElement(views,f'{{{NS}}}sheetView',workbookViewId='0',showGridLines='0')
ET.SubElement(view,f'{{{NS}}}pane',ySplit='1',topLeftCell='A2',activePane='bottomLeft',state='frozen')
cols=ET.SubElement(sheet,f'{{{NS}}}cols')
for i,width in enumerate(table['widths'],1):
ET.SubElement(cols,f'{{{NS}}}col',min=str(i),max=str(i),width=str(width),customWidth='1')
body=ET.SubElement(sheet,f'{{{NS}}}sheetData')
for r,values in enumerate(table['rows'],1):
row=ET.SubElement(body,f'{{{NS}}}row',r=str(r),ht='54' if r==6 and number==1 else '30',customHeight='1')
for c,value in enumerate(values,1):
if value is None:
continue
numeric=isinstance(value,(int,float)) and not isinstance(value,bool) and math.isfinite(value)
style='1' if r in table.get('headers',[]) else '2' if r==1 and number==1 else '3' if isinstance(value,float) and numeric else '0' if numeric else '4'
cell=ET.SubElement(row,f'{{{NS}}}c',r=f'{_column(c)}{r}',s=style,t='n' if numeric else 'inlineStr')
if numeric:
ET.SubElement(cell,f'{{{NS}}}v').text=str(value)
else:
text=ET.SubElement(ET.SubElement(cell,f'{{{NS}}}is'),f'{{{NS}}}t',attrib={'{http://www.w3.org/XML/1998/namespace}space':'preserve'})
text.text=re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]','',str(value))[:32767]
if number==1:
merges=ET.SubElement(sheet,f'{{{NS}}}mergeCells',count='2')
ET.SubElement(merges,f'{{{NS}}}mergeCell',ref='A1:E1')
ET.SubElement(merges,f'{{{NS}}}mergeCell',ref='B6:E6')
archive.writestr(f'xl/worksheets/sheet{number}.xml',_xml(sheet))
ET.SubElement(relationships,'Relationship',Id='rIdStyles',Type=f'{REL}/styles',Target='styles.xml')
archive.writestr('xl/workbook.xml',_xml(workbook))
archive.writestr('xl/_rels/workbook.xml.rels',_xml(relationships))
archive.writestr('[Content_Types].xml',_xml(content))
def preview_workbook(path, *, max_rows=300, max_cols=60):
if path.suffix.lower()=='.csv':
text=path.read_bytes()
if len(text)>10*1024*1024:
raise ValueError('CSV预览最大10MB')
try:
value=text.decode('utf-8-sig')
except UnicodeDecodeError:
value=text.decode('gb18030')
rows,total_rows,truncated=[],0,False
try:
for row in csv.reader(StringIO(value)):
total_rows+=1
if total_rows<=max_rows:
rows.append(row[:max_cols])
truncated=truncated or total_rows>max_rows or len(row)>max_cols
except csv.Error as exc:
raise ValueError('CSV内容格式异常，或单个字段过长') from exc
return {'sheets':[{'name':path.stem,'rows':rows,'total_rows':total_rows,'truncated':truncated}]}
try:
with ZipFile(path) as archive:
entries=archive.infolist()
if len(entries)>10000 or len({e.filename for e in entries})!=len(entries) or sum(e.file_size for e in entries)>100*1024*1024:
raise ValueError('工作簿展开内容过大或格式异常')
strings=[]
styles=[]
if 'xl/styles.xml' in archive.namelist():
style_root=ET.fromstring(archive.read('xl/styles.xml'))
fonts=style_root.findall(f'{{{NS}}}fonts/{{{NS}}}font')
fills=style_root.findall(f'{{{NS}}}fills/{{{NS}}}fill')
custom_formats={int(item.attrib['numFmtId']):item.attrib['formatCode'] for item in style_root.findall(f'{{{NS}}}numFmts/{{{NS}}}numFmt')}
builtin_formats={0:'General',1:'0',2:'0.00',3:'#,##0',4:'#,##0.00',9:'0%',10:'0.00%',14:'yyyy-mm-dd',22:'yyyy-mm-dd hh:mm'}
for xf in style_root.findall(f'{{{NS}}}cellXfs/{{{NS}}}xf'):
style={}
font_id,fill_id=int(xf.attrib.get('fontId','0')),int(xf.attrib.get('fillId','0'))
if font_idmax_rows:
truncated=True
continue
while len(rows)max_cols:
truncated=True
continue
while len(values)=start_col and end_row>=start_row:
merges.append({'start_row':start_row,'start_col':start_col,'row_span':min(end_row+1,max_rows)-start_row,'col_span':min(end_col+1,max_cols)-start_col})
sheets.append({'name':record.attrib['name'],'rows':rows,'total_rows':total_rows,'truncated':truncated,'merges':merges,'cell_styles':cell_styles})
return {'sheets':sheets,'styles':styles,'total_sheets':len(workbook.findall(f'{{{NS}}}sheets/{{{NS}}}sheet'))}
except (BadZipFile, ET.ParseError,KeyError,IndexError,TypeError,AttributeError,RuntimeError) as exc:
raise ValueError('不是可预览的XLSX文件，或工作簿内容已损坏') from exc