#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 南通新东方租赁梳理 — 批量 Excel Builder ======================================== 从所有已完成的 nantong-lease-audit workflow threads 中提取数据， 按校区分 sheet，合并到一个工作簿。 用法： python3 batch-excel-builder.py <输出xlsx路径> 例： python3 batch-excel-builder.py /tmp/南通-workflow-batch.xlsx 前置：所有thread已跑完（status=end）。 """ import subprocess, json, re, sys, os UWF = "/home/maggie/.hermes/node/bin/uwf" def run(cmd): r = subprocess.run(cmd, capture_output=True, text=True) return r.stdout def get_thread_read(thread_id): """获取thread完整markdown输出（需要大quota）""" return run([UWF, "thread", "read", thread_id, "--quota", "200000", "--start"]) def get_all_nantong_threads(): """Get all completed nantong-lease-audit threads""" r = subprocess.run([UWF, "workflow", "list"], capture_output=True, text=True) # Find nantong-lease-audit hash(es) wf_hashes = [] for line in r.stdout.strip().split('\n'): if 'nantong-lease' in line.lower(): parts = line.split() if len(parts) >= 2: wf_hashes.append(parts[1]) if not wf_hashes: print("No nantong-lease-audit workflow found") return [] r = subprocess.run([UWF, "thread", "list", "--all"], capture_output=True, text=True) threads = [] for line in r.stdout.strip().split('\n'): for h in wf_hashes: if h in line: parts = line.split() if len(parts) >= 3 and parts[2] == 'end': threads.append(parts[0]) return threads def get_thread_info(thread_id): """Get campus and filename from thread prompt""" text = run([UWF, "thread", "read", thread_id, "--quota", "500"]) campus_m = re.search(r'校区：(\S+)', text) file_m = re.search(r'合同文件：([^|]+)', text) ocr_m = re.search(r'OCR文本路径：(\S+)', text) return { 'campus': campus_m.group(1) if campus_m else '', 'filename': file_m.group(1).strip() if file_m else '', 'ocr_path': ocr_m.group(1) if ocr_m else '', } def extract_step_output(thread_text, role_name): """Extract content from a specific role step""" pattern = rf'## Step \d+: {role_name}.*?\n\n(.*?)\n' m = re.search(pattern, thread_text, re.DOTALL) return m.group(1) if m else "" def extract_frontmatter(output_text): """Extract YAML frontmatter fields from step output""" m = re.search(r'^---\n(.*?)\n---', output_text, re.DOTALL) if not m: return {}, output_text fm_text = m.group(1) body = output_text[m.end():].strip() result = {} current_key = None current_val = [] is_multiline = False for line in fm_text.split('\n'): if not line.strip(): if is_multiline: current_val.append('') continue km = re.match(r'^(\w[\w_]*):\s*(.*)', line) if km and not line.startswith(' '): if current_key: result[current_key] = '\n'.join(current_val).strip() current_key = km.group(1) val = km.group(2).strip() if val == '|' or val == '': is_multiline = True current_val = [] else: is_multiline = False current_val = [val] elif is_multiline and current_key: current_val.append(line.strip()) if current_key: result[current_key] = '\n'.join(current_val).strip() return result, body def build_campus_sheet(wb, campus, rows): """Build a campus sheet with all contract rows""" import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side F = Font(name='微软雅黑', size=10) FB = Font(name='微软雅黑', size=10, bold=True) TITLE_FONT = Font(name='微软雅黑', size=14, bold=True, color='FFFFFFFF') fill_title = PatternFill(start_color='FF8B1A2B', fill_type='solid') fill_sec = PatternFill(start_color='FFC0504D', fill_type='solid') fill_hdr = PatternFill(start_color='FFE2EFDA', fill_type='solid') fill_sub = PatternFill(start_color='FFF2F2F2', fill_type='solid') thin = Side(style='thin') border = Border(left=thin, right=thin, top=thin, bottom=thin) AL = Alignment(horizontal='left', vertical='top', wrap_text=True) AC = Alignment(horizontal='center', vertical='center', wrap_text=True) ws = wb.create_sheet(campus[:31]) widths = dict(A=5, B=20, C=17, D=27, E=23, F=9, G=21, H=36.33, I=34, J=8, K=50, L=35) for c, w in widths.items(): ws.column_dimensions[c].width = w def merge_row(r, text, font, fill, h=None, al=AC): ws.merge_cells(f'A{r}:L{r}') cell = ws.cell(r, 1, text); cell.font = font; cell.fill = fill; cell.alignment = al for col in range(1, 13): ws.cell(r, col).fill = fill; ws.cell(r, col).border = border if h: ws.row_dimensions[r].height = h r = 1 merge_row(r, f'{campus}校区 — 租赁合同梳理', TITLE_FONT, fill_title, 30); r += 1 merge_row(r, f'共{len(rows)}份合同', Font(name='微软雅黑', size=10, bold=True, color='FF404040'), fill_sub, 76, AL); r += 1 HDR = ['序号','文件名称','合同类型','合同当事人','租赁标的/服务范围','面积(㎡)', '合同期限','金额/费用','核心内容','当前状态','法律风险（站乙方立场）','与07标准模版差异'] for ci, h in enumerate(HDR, 1): c = ws.cell(r, ci, h); c.font = FB; c.fill = fill_hdr; c.alignment = AC; c.border = border ws.row_dimensions[r].height = 30; r += 1 for idx, row in enumerate(rows, 1): ws.cell(r, 1, str(idx)).font = F; ws.cell(r, 1).alignment = AC; ws.cell(r, 1).border = border for ci, col in enumerate('BCDEFGHIJKL', 2): c = ws.cell(r, ci, row.get(col, '')); c.font = F; c.alignment = AL; c.border = border ws.row_dimensions[r].height = 300; r += 1 ws.freeze_panes = 'A4' return ws def main(): out_path = sys.argv[1] if len(sys.argv) > 1 else '/tmp/南通-workflow-batch.xlsx' print("=== 获取所有已完成的nantong-lease-audit threads ===") threads = get_all_nantong_threads() print(f"找到 {len(threads)} 个已完成的thread") if not threads: print("No completed threads found. Exiting.") sys.exit(1) campus_data = {} for tid in threads: info = get_thread_info(tid) campus = info['campus'] if not campus: print(f" SKIP {tid}: no campus info") continue print(f" 处理 {tid}: {campus} / {info['filename']}") thread_text = get_thread_read(tid) cls_output = extract_step_output(thread_text, 'classifier') cls_fm, _ = extract_frontmatter(cls_output) td_output = extract_step_output(thread_text, 'template-d') td_fm, td_body = extract_frontmatter(td_output) ra_output = extract_step_output(thread_text, 'rule-analy') ra_fm, ra_body = extract_frontmatter(ra_output) de_output = extract_step_output(thread_text, 'data-extra') de_fm, de_body = extract_frontmatter(de_output) row = {} row['B'] = info['filename'] + '.pdf' if info['filename'] else cls_fm.get('contract_title', '') row['C'] = cls_fm.get('contract_type', '') row['D'] = f"甲方：{cls_fm.get('party_a', '')}\n乙方：{cls_fm.get('party_b', '')}" row['E'] = cls_fm.get('property_address', '') row['F'] = cls_fm.get('area_sqm', '') start = cls_fm.get('term_start', '') end = cls_fm.get('term_end', '') free = cls_fm.get('rent_free_period', '') row['G'] = f"{start}至{end}\n免租期：{free}" if start else '' for col in 'HIJ': pattern = rf'{col}\.\s+[^：:]+[：:]\s*(.*?)(?=\n[A-L]\.\s|\nh_column|\Z)' m = re.search(pattern, de_body, re.DOTALL) row[col] = m.group(1).strip() if m else '' h_pattern = r'H\.\s+金额/费用[：:]\s*(.*?)(?=\nI\.\s|\Z)' hm = re.search(h_pattern, de_body, re.DOTALL) if hm: row['H'] = hm.group(1).strip() risk = ra_fm.get('risk_detail', ra_body[:5000]) term = ra_fm.get('termination_analysis', '') row['K'] = risk + ('\n\n' + term if term else '') row['L'] = td_fm.get('diff_detail', td_body[:3000]) if campus not in campus_data: campus_data[campus] = [] campus_data[campus].append(row) import openpyxl wb = openpyxl.Workbook() wb.remove(wb.active) for campus, rows in campus_data.items(): print(f" {campus}: {len(rows)} rows") build_campus_sheet(wb, campus, rows) os.makedirs(os.path.dirname(out_path) or '.', exist_ok=True) wb.save(out_path) print(f"\n✅ 已保存: {out_path}") print(f"\n=== 汇总 ===") for campus, rows in campus_data.items(): print(f" {campus}: {len(rows)}份合同") if __name__ == "__main__": main()