vent_agent_2.0/runtime/test_lab/spreadsheets.py

"""应用内的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='''<?xml version="1.0" encoding="UTF-8"?><styleSheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"><fonts count="3"><font><sz val="11"/><name val="Microsoft YaHei"/></font><font><b/><color rgb="FFFFFFFF"/><sz val="11"/><name val="Microsoft YaHei"/></font><font><b/><sz val="16"/><color rgb="FF173E65"/><name val="Microsoft YaHei"/></font></fonts><fills count="3"><fill><patternFill patternType="none"/></fill><fill><patternFill patternType="gray125"/></fill><fill><patternFill patternType="solid"><fgColor rgb="FF173E65"/><bgColor indexed="64"/></patternFill></fill></fills><borders count="1"><border/></borders><cellStyleXfs count="1"><xf/></cellStyleXfs><cellXfs count="5"><xf fontId="0" fillId="0" borderId="0" xfId="0"/><xf fontId="1" fillId="2" borderId="0" xfId="0" applyAlignment="1"><alignment vertical="center" wrapText="1"/></xf><xf fontId="2" fillId="0" borderId="0" xfId="0"/><xf fontId="0" fillId="0" borderId="0" numFmtId="2" xfId="0" applyNumberFormat="1"/><xf fontId="0" fillId="0" borderId="0" xfId="0" applyAlignment="1"><alignment vertical="top" wrapText="1"/></xf></cellXfs><cellStyles count="1"><cellStyle name="Normal" xfId="0" builtinId="0"/></cellStyles></styleSheet>'''
    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'<Relationships xmlns="{PKG}"><Relationship Id="rId1" Type="{REL}/officeDocument" Target="xl/workbook.xml"/></Relationships>')
        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_id<len(fonts):
                        font=fonts[font_id]
                        bold=font.find(f'{{{NS}}}b')
                        if bold is not None and bold.attrib.get('val','1')!='0':
                            style['bold']=True
                        size=font.find(f'{{{NS}}}sz')
                        if size is not None:
                            style['font_size']=float(size.attrib['val'])
                        color=font.find(f'{{{NS}}}color')
                        if color is not None and re.fullmatch(r'(?:[0-9A-Fa-f]{2})?[0-9A-Fa-f]{6}',color.attrib.get('rgb','')):
                            style['color']='#'+color.attrib['rgb'][-6:]
                    if fill_id<len(fills):
                        fill=fills[fill_id].find(f'{{{NS}}}patternFill')
                        color=fill.find(f'{{{NS}}}fgColor') if fill is not None and fill.attrib.get('patternType')=='solid' else None
                        if color is not None and re.fullmatch(r'(?:[0-9A-Fa-f]{2})?[0-9A-Fa-f]{6}',color.attrib.get('rgb','')):
                            style['background']='#'+color.attrib['rgb'][-6:]
                    fmt_id=int(xf.attrib.get('numFmtId','0'))
                    style['number_format']=custom_formats.get(fmt_id,builtin_formats.get(fmt_id,'General'))
                    styles.append(style)
            if 'xl/sharedStrings.xml' in archive.namelist():
                shared=ET.fromstring(archive.read('xl/sharedStrings.xml'))
                strings=[''.join(si.itertext()) for si in shared.findall(f'{{{NS}}}si')]
            relationships=ET.fromstring(archive.read('xl/_rels/workbook.xml.rels'))
            targets={r.attrib['Id']:r.attrib['Target'] for r in relationships if r.attrib.get('TargetMode')!='External'}
            workbook=ET.fromstring(archive.read('xl/workbook.xml'))
            sheets=[]
            for record in workbook.findall(f'{{{NS}}}sheets/{{{NS}}}sheet')[:30]:
                target=targets[record.attrib[f'{{{REL}}}id']]
                filename=posixpath.normpath(target.lstrip('/') if target.startswith('/') else 'xl/'+target)
                if not filename.startswith('xl/'):
                    raise ValueError('工作簿引用的工作表路径异常')
                sheet=ET.fromstring(archive.read(filename))
                rows=[]
                cell_styles={}
                total_rows=0
                truncated=False
                for row in sheet.findall(f'{{{NS}}}sheetData/{{{NS}}}row'):
                    number=int(row.attrib.get('r',total_rows+1))
                    total_rows=max(total_rows,number)
                    if number>max_rows:
                        truncated=True
                        continue
                    while len(rows)<number:
                        rows.append([])
                    values=rows[number-1]
                    for cell in row.findall(f'{{{NS}}}c'):
                        column=re.match(r'([A-Z]+)',cell.attrib.get('r','A1'))
                        index=0
                        for ch in column[1]:
                            index=index*26+ord(ch)-64
                        if index>max_cols:
                            truncated=True
                            continue
                        while len(values)<index:
                            values.append(None)
                        cell_styles[f'{number-1}:{index-1}']=int(cell.attrib.get('s','0'))
                        kind=cell.attrib.get('t')
                        value=cell.findtext(f'{{{NS}}}v')
                        if kind=='inlineStr':
                            value=''.join(cell.find(f'{{{NS}}}is').itertext())
                        elif kind=='s':
                            value=strings[int(value)]
                        elif kind=='b':
                            value=value=='1'
                        elif kind not in {'str','e','d'} and value is not None:
                            value=float(value)
                            if not math.isfinite(value):
                                raise ValueError('工作簿包含非有限数值')
                            if value.is_integer():
                                value=int(value)
                        if value is None and cell.find(f'{{{NS}}}f') is not None:
                            value='='+cell.findtext(f'{{{NS}}}f')
                        values[index-1]=value
                merges=[]
                for merge in sheet.findall(f'{{{NS}}}mergeCells/{{{NS}}}mergeCell'):
                    match=re.fullmatch(r'([A-Z]+)([1-9][0-9]*):([A-Z]+)([1-9][0-9]*)',merge.attrib.get('ref',''))
                    if not match:
                        continue
                    left,top,right,bottom=match.groups()
                    col=lambda text:sum((ord(c)-64)*26**i for i,c in enumerate(reversed(text)))-1
                    start_col,end_col=col(left),col(right)
                    start_row,end_row=int(top)-1,int(bottom)-1
                    if start_row<max_rows and start_col<max_cols and end_col>=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