Python自动化办公工具开发实践：打造智能报表生成系统的心得与洞见

报表生成系统
├── 数据层
│   ├── 数据库连接器 (SQLAlchemy)
│   ├── API客户端 (Requests)
│   └── 文件解析器 (Pandas)
├── 逻辑层
│   ├── 数据清洗引擎
│   ├── 业务规则处理器
│   └── 计算引擎
└── 输出层
    ├── Excel生成器 (Openpyxl)
    ├── PDF生成器 (ReportLab)
    └── 邮件发送器 (smtplib)

class DataUnifier:
    def __init__(self, config):
        self.sources = config['data_sources']
        
    def fetch_data(self):
        unified_data = pd.DataFrame()
        for source in self.sources:
            if source['type'] == 'api':
                data = self._fetch_api(source)
            elif source['type'] == 'database':
                data = self._fetch_db(source)
            elif source['type'] == 'excel':
                data = self._parse_excel(source)
                
            # 统一字段映射
            data.rename(columns=source['field_mapping'], inplace=True)
            unified_data = pd.concat([unified_data, data], ignore_index=True)
        return unified_data

    def _fetch_api(self, source):
        # 带认证的API请求
        session = requests.Session()
        session.auth = (source['user'], source['token'])
        response = session.get(source['url'], params=source['params'])
        return pd.DataFrame(response.json()['data'])

def apply_template(data, report_type):
    # 加载对应业务类型的模板
    template_file = f"templates/{report_type}_template.xlsx"
    wb = load_workbook(template_file)
    ws = wb.active
    
    # 获取模板中的占位符映射
    placeholder_map = {}
    for row in ws.iter_rows():
        for cell in row:
            if cell.value and str(cell.value).startswith("${"):
                key = cell.value[2:-1]
                placeholder_map[key] = cell.coordinate
    
    # 填充数据
    for field, coord in placeholder_map.items():
        if field in data.columns:
            value = data[field].iloc[0] if not data.empty else "N/A"
            ws[coord] = value
            
    # 应用条件格式
    self._apply_conditional_formatting(ws, report_type)
    return wb

def data_cleaning_pipeline(df):
    # 异常值检测与修复
    for col in df.select_dtypes(include=np.number):
        # 检测离群值
        q1 = df[col].quantile(0.25)
        q3 = df[col].quantile(0.75)
        iqr = q3 - q1
        
        # 构建修复掩码
        outlier_mask = (df[col] < q1 - 1.5*iqr) | (df[col] > q3 + 1.5*iqr)
        
        if outlier_mask.any():
            # 首次尝试：用中位数替换
            median_val = df[col].median()
            df.loc[outlier_mask, col] = median_val
            
            # 记录修复日志
            logging.warning(f"Replaced {outlier_mask.sum()} outliers in {col} with median {median_val}")
            
            # 二次验证
            if (df[col] == 0).all():
                # 极端情况处理
                send_alert(f"Column {col} may have critical issues after cleaning")
    
    # 处理缺失值
    df.fillna(method='ffill', inplace=True)
    return df

# 使用分块处理
chunk_size = 5000
results = []
for chunk in pd.read_sql_query(query, conn, chunksize=chunk_size):
    processed = process_chunk(chunk)  # 逐块处理
    results.append(processed)
    
final_df = pd.concat(results)

# 使用Dask并行计算
import dask.dataframe as dd
ddf = dd.from_pandas(df, npartitions=8)
result = ddf.map_partitions(process_partition).compute()

def excel_injection(template_path, output_path, data):
    # 复制模板
    shutil.copyfile(template_path, output_path)
    
    # 打开复制的文件
    wb = load_workbook(output_path)
    ws = wb.active
    
    # 数据注入
    for idx, row in data.iterrows():
        for col_idx, value in enumerate(row):
            cell = ws.cell(row=idx+2, column=col_idx+1)
            cell.value = value
            
            # 保留原有样式
            if idx == 0:
                source_cell = ws.cell(row=1, column=col_idx+1)
                cell.font = copy(source_cell.font)
                cell.fill = copy(source_cell.fill)
                cell.border = copy(source_cell.border)
    
    wb.save(output_path)

from apscheduler.schedulers.blocking import BlockingScheduler

def job_with_heartbeat():
    try:
        # 开始前更新状态
        update_job_status('running')
        
        # 核心业务逻辑
        generate_reports()
        
        # 成功后更新状态
        update_job_status('completed')
    except Exception as e:
        update_job_status(f'failed: {str(e)}')
        send_alert(f"Job failed: {traceback.format_exc()}")

scheduler = BlockingScheduler()
scheduler.add_job(job_with_heartbeat, 'cron', day_of_week='mon-fri', hour=3)

# 心跳监测线程
def monitor():
    while True:
        status = get_job_status()
        if status.startswith('running'):
            last_update = get_last_update_time()
            if (datetime.now() - last_update) > timedelta(hours=2):
                restart_job()
        time.sleep(300)

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "properties": {
    "data_sources": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "name": {"type": "string"},
          "type": {"enum": ["api", "database", "excel"]},
          "refresh_interval": {"type": "integer"}
        },
        "required": ["name", "type"]
      }
    }
  },
  "required": ["data_sources"]
}

import structlog

structlog.configure(
    processors=[
        structlog.processors.JSONRenderer(indent=2)
    ],
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory()
)

logger = structlog.get_logger()
logger.info("report_generated", report_type="sales", duration_sec=42.7)

@pytest.fixture
def mock_data_sources():
    # 创建模拟API响应
    responses.add(
        responses.GET,
        'https://api.example.com/data',
        json={'data': [{'id': 1, 'value': 100}]},
        status=200
    )
    
    # 创建模拟数据库
    conn = sqlite3.connect(':memory:')
    conn.execute('CREATE TABLE sales (id INT, amount REAL)')
    conn.execute('INSERT INTO sales VALUES (1, 100.0)')
    yield conn
    conn.close()

def test_report_generation(mock_data_sources):
    config = load_test_config()
    report = generate_report(config)
    assert report.total_sales == 100.0