241 lines
9.3 KiB
Plaintext
241 lines
9.3 KiB
Plaintext
|
from __future__ import annotations
|
|||
|
|
import os, json, subprocess, sys
|
|||
|
|
from pathlib import Path
|
|||
|
|
from typing import Any, Callable, Dict, List, Optional
|
|||
|
|
import pandas as pd
|
|||
|
|
from docx import Document
|
|||
|
|
from config_model import AppConfig, PlaceholderConfig, DbConnectionConfig
|
|||
|
|
from influx_service import InfluxConnectionParams, InfluxService
|
|||
|
|
from logger import get_logger
|
|||
|
|
|
|||
|
|
logger = get_logger()
|
|||
|
|
_PROGRESS_CB: Optional[Callable[[str, int, int], None]] = None
|
|||
|
|
|
|||
|
|
def set_progress_callback(cb):
|
|||
|
|
global _PROGRESS_CB; _PROGRESS_CB = cb
|
|||
|
|
def _progress(msg, cur, total):
|
|||
|
|
if _PROGRESS_CB: _PROGRESS_CB(msg, cur, total)
|
|||
|
|
|
|||
|
|
def _build_influx_service(cfg):
|
|||
|
|
return InfluxService(InfluxConnectionParams(url=cfg.influx.url, org=cfg.influx.org, token=cfg.influx.token))
|
|||
|
|
|
|||
|
|
def _execute_db_query(ph, db_cfg):
|
|||
|
|
query = (ph.dbQuery or "").strip()
|
|||
|
|
if not query: return ""
|
|||
|
|
if not db_cfg: db_cfg = DbConnectionConfig()
|
|||
|
|
engine = (db_cfg.engine or "mysql").lower()
|
|||
|
|
|
|||
|
|
if engine in ("sqlite", "sqlite3"):
|
|||
|
|
import sqlite3
|
|||
|
|
conn = sqlite3.connect(db_cfg.database or str(Path(__file__).parent / "experiments.db"))
|
|||
|
|
result = conn.execute(query).fetchone()
|
|||
|
|
conn.close()
|
|||
|
|
return str(result[0]) if result and result[0] else ""
|
|||
|
|
elif engine == "mysql":
|
|||
|
|
import pymysql
|
|||
|
|
conn = pymysql.connect(host=getattr(db_cfg, "host", "localhost"), port=int(getattr(db_cfg, "port", 3306)),
|
|||
|
|
user=getattr(db_cfg, "username", ""), password=getattr(db_cfg, "password", ""),
|
|||
|
|
database=getattr(db_cfg, "database", ""), charset="utf8mb4")
|
|||
|
|
with conn.cursor() as cursor:
|
|||
|
|
cursor.execute(query)
|
|||
|
|
result = cursor.fetchone()
|
|||
|
|
conn.close()
|
|||
|
|
return str(result[0]) if result and result[0] else ""
|
|||
|
|
return ""
|
|||
|
|
|
|||
|
|
def _load_script_data_from_db(experiment_id):
|
|||
|
|
try:
|
|||
|
|
import sqlite3
|
|||
|
|
conn = sqlite3.connect(str(Path(__file__).parent / "experiments.db"))
|
|||
|
|
result = conn.execute("SELECT script_data FROM experiments WHERE id=?", (experiment_id,)).fetchone()
|
|||
|
|
conn.close()
|
|||
|
|
if result and result[0]:
|
|||
|
|
logger.info("从数据库加载脚本数据，实验ID: %d", experiment_id)
|
|||
|
|
return json.loads(result[0])
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error("加载脚本数据失败: %s", e)
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def _load_experiment_info(experiment_id):
|
|||
|
|
try:
|
|||
|
|
import sqlite3
|
|||
|
|
conn = sqlite3.connect(str(Path(__file__).parent / "experiments.db"))
|
|||
|
|
result = conn.execute("SELECT status FROM experiments WHERE id=?", (experiment_id,)).fetchone()
|
|||
|
|
conn.close()
|
|||
|
|
if result:
|
|||
|
|
return {'is_normal': result[0] == 'completed'}
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error("加载实验信息失败: %s", e)
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def _parse_script_tables(script_data):
|
|||
|
|
tables = {}
|
|||
|
|
if isinstance(script_data, dict) and "tables" in script_data:
|
|||
|
|
for item in script_data["tables"]:
|
|||
|
|
key = item.get("token") or item.get("key")
|
|||
|
|
if key: tables[str(key)] = item
|
|||
|
|
return tables
|
|||
|
|
|
|||
|
|
def _replace_global_params(text, cfg):
|
|||
|
|
"""替换文本中的 @参数名 为全局参数的值"""
|
|||
|
|
if not text or '@' not in text: return text
|
|||
|
|
result = text
|
|||
|
|
if hasattr(cfg, 'globalParameters') and hasattr(cfg.globalParameters, 'parameters'):
|
|||
|
|
import re
|
|||
|
|
for param_name in re.findall(r'@(\w+)', text):
|
|||
|
|
if param_name in cfg.globalParameters.parameters:
|
|||
|
|
result = result.replace(f'@{param_name}', cfg.globalParameters.parameters[param_name])
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
def _make_seconds_index(df):
|
|||
|
|
if "_time" in df.columns:
|
|||
|
|
t = pd.to_datetime(df["_time"])
|
|||
|
|
return (t - t.iloc[0]).dt.total_seconds().round().astype(int)
|
|||
|
|
return pd.Series(range(len(df)))
|
|||
|
|
|
|||
|
|
def _format_numeric_columns(df, exclude_cols):
|
|||
|
|
if df is None or df.empty: return df
|
|||
|
|
result = df.copy()
|
|||
|
|
for col in result.columns:
|
|||
|
|
if col not in exclude_cols:
|
|||
|
|
try:
|
|||
|
|
numeric = pd.to_numeric(result[col], errors="coerce")
|
|||
|
|
if numeric.notna().any(): result[col] = numeric.round(2)
|
|||
|
|
except: pass
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
def _to_wide_table(df, fields, first_column, titles_map, first_title=None):
|
|||
|
|
if df.empty: return pd.DataFrame()
|
|||
|
|
work = df.copy()
|
|||
|
|
if "_time" not in work.columns or "_value" not in work.columns: return work
|
|||
|
|
if fields and "_field" in work.columns: work = work[work["_field"].isin(fields)]
|
|||
|
|
|
|||
|
|
if first_column == "seconds":
|
|||
|
|
idx = _make_seconds_index(work)
|
|||
|
|
work = work.assign(__index__=idx)
|
|||
|
|
index_col, index_title = "__index__", first_title or "秒"
|
|||
|
|
else:
|
|||
|
|
index_col, index_title = "_time", first_title or "时间"
|
|||
|
|
|
|||
|
|
if "_field" in work.columns:
|
|||
|
|
wide = work.pivot_table(index=index_col, columns="_field", values="_value", aggfunc="last")
|
|||
|
|
else:
|
|||
|
|
wide = work.set_index(index_col)[["_value"]]
|
|||
|
|
wide.columns = ["value"]
|
|||
|
|
|
|||
|
|
wide = wide.sort_index()
|
|||
|
|
wide.reset_index(inplace=True)
|
|||
|
|
wide.rename(columns={index_col: index_title}, inplace=True)
|
|||
|
|
for f, title in titles_map.items():
|
|||
|
|
if f in wide.columns: wide.rename(columns={f: title}, inplace=True)
|
|||
|
|
return _format_numeric_columns(wide, exclude_cols=[index_title])
|
|||
|
|
|
|||
|
|
def _replace_texts_docx(doc, mapping):
|
|||
|
|
for key, val in mapping.items():
|
|||
|
|
token = '{' + key + '}'
|
|||
|
|
replacement = val or ''
|
|||
|
|
for para in doc.paragraphs:
|
|||
|
|
if token in para.text:
|
|||
|
|
for run in para.runs:
|
|||
|
|
if token in run.text:
|
|||
|
|
run.text = run.text.replace(token, replacement)
|
|||
|
|
for table in doc.tables:
|
|||
|
|
for row in table.rows:
|
|||
|
|
for cell in row.cells:
|
|||
|
|
for para in cell.paragraphs:
|
|||
|
|
if token in para.text:
|
|||
|
|
for run in para.runs:
|
|||
|
|
if token in run.text:
|
|||
|
|
run.text = run.text.replace(token, replacement)
|
|||
|
|
|
|||
|
|
def _fill_script_table_docx(doc, token, table_spec):
|
|||
|
|
cells = table_spec.get("cells") or []
|
|||
|
|
if not cells: return
|
|||
|
|
|
|||
|
|
token_with_braces = '{' + token + '}'
|
|||
|
|
table_found = None
|
|||
|
|
token_row = token_col = 0
|
|||
|
|
|
|||
|
|
for table in doc.tables:
|
|||
|
|
for ri, row in enumerate(table.rows):
|
|||
|
|
for ci, cell in enumerate(row.cells):
|
|||
|
|
if token_with_braces in cell.text:
|
|||
|
|
table_found, token_row, token_col = table, ri, ci
|
|||
|
|
break
|
|||
|
|
if table_found: break
|
|||
|
|
if table_found: break
|
|||
|
|
|
|||
|
|
if not table_found:
|
|||
|
|
logger.warning("未找到token: %s", token_with_braces)
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 清除token
|
|||
|
|
for para in table_found.rows[token_row].cells[token_col].paragraphs:
|
|||
|
|
for run in para.runs:
|
|||
|
|
if token_with_braces in run.text:
|
|||
|
|
run.text = run.text.replace(token_with_braces, '')
|
|||
|
|
|
|||
|
|
# 填充数据 - 使用绝对位置（row/col直接是表格坐标）
|
|||
|
|
for cell_info in cells:
|
|||
|
|
if not isinstance(cell_info, dict): continue
|
|||
|
|
value = cell_info.get("value")
|
|||
|
|
if value is None: continue
|
|||
|
|
|
|||
|
|
abs_row = int(cell_info.get("row", 0))
|
|||
|
|
abs_col = int(cell_info.get("col", 0))
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
if abs_row < len(table_found.rows) and abs_col < len(table_found.rows[abs_row].cells):
|
|||
|
|
cell = table_found.rows[abs_row].cells[abs_col]
|
|||
|
|
if cell.paragraphs and cell.paragraphs[0].runs:
|
|||
|
|
cell.paragraphs[0].runs[0].text = str(value)
|
|||
|
|
else:
|
|||
|
|
cell.text = str(value)
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.warning("填充失败 (%d,%d): %s", abs_row, abs_col, e)
|
|||
|
|
|
|||
|
|
def render_report(template_path, cfg, output_path, experiment_id=None):
|
|||
|
|
logger.info("=== 开始生成报告 ===")
|
|||
|
|
_progress("加载数据", 0, 5)
|
|||
|
|
|
|||
|
|
# 加载脚本数据和实验信息
|
|||
|
|
script_data = _load_script_data_from_db(experiment_id) if experiment_id else None
|
|||
|
|
script_tables = _parse_script_tables(script_data)
|
|||
|
|
logger.info("脚本表格: %s", list(script_tables.keys()))
|
|||
|
|
|
|||
|
|
# 打开模板
|
|||
|
|
doc = Document(str(template_path))
|
|||
|
|
_progress("替换文本", 1, 5)
|
|||
|
|
|
|||
|
|
# 构建文本映射
|
|||
|
|
text_map = {}
|
|||
|
|
if hasattr(cfg, 'placeholders'):
|
|||
|
|
placeholders = cfg.placeholders if isinstance(cfg.placeholders, dict) else {}
|
|||
|
|
for key, ph in placeholders.items():
|
|||
|
|
if hasattr(ph, 'type'):
|
|||
|
|
if ph.type == "text" and hasattr(ph, 'value'):
|
|||
|
|
text_map[key] = _replace_global_params(ph.value or '', cfg)
|
|||
|
|
elif ph.type == "dbText" and hasattr(ph, 'dbQuery'):
|
|||
|
|
text_map[key] = _execute_db_query(ph, getattr(cfg, 'db', None))
|
|||
|
|
|
|||
|
|
# 添加实验信息占位符
|
|||
|
|
if experiment_id:
|
|||
|
|
exp_info = _load_experiment_info(experiment_id)
|
|||
|
|
if exp_info:
|
|||
|
|
text_map['isNormal'] = '√' if exp_info.get('is_normal') else ''
|
|||
|
|
|
|||
|
|
logger.info("文本映射: %d 个", len(text_map))
|
|||
|
|
_replace_texts_docx(doc, text_map)
|
|||
|
|
|
|||
|
|
# 填充表格
|
|||
|
|
_progress("填充表格", 2, 5)
|
|||
|
|
for token, spec in script_tables.items():
|
|||
|
|
_fill_script_table_docx(doc, token, spec)
|
|||
|
|
|
|||
|
|
# 保存
|
|||
|
|
_progress("保存", 4, 5)
|
|||
|
|
doc.save(str(output_path))
|
|||
|
|
_progress("完成", 5, 5)
|
|||
|
|
logger.info("=== 报告生成完成 ===")
|
|||
|
|
return output_path
|