自动化业务通报系统实现-平芜编程栈

问题解构：需求核心是构建一个基于Python的自动化业务通报系统，用于从多个.xls报表中提取数据，按团队统计指标完成情况，生成手机适配的通报图片，并通过Web界面展示。系统需支持灵活的配置管理，包括团队信息、日目标、指标映射规则等，并具备落后团队提醒功能。

方案推演：系统将采用Flask构建Web界面，使用Pandas处理Excel数据，Pillow生成图片，并利用JupyterLab进行交互式开发与测试。核心模块包括配置管理、数据解析、统计计算、图片生成和Web展示。

一、系统架构与核心代码

1. 配置文件设计

系统使用YAML格式的配置文件，便于维护和修改。

config.yaml：主配置文件，定义其他配置文件的路径。

# config.yaml data_dir: "./data" # 原始.xls文件存放目录 output_dir: "./output" # 生成的图片输出目录 team_config: "./configs/teams.yaml" # 团队人员配置 daily_target_config: "./configs/daily_targets.yaml" # 日目标配置 indicator_config: "./configs/indicators.yaml" # 统计指标配置 file_mapping_config: "./configs/file_mapping.yaml" # 文件-指标映射配置

teams.yaml：定义团队层级和负责人。

# teams.yaml teams: - name: "团队A" level: 1 leader: "张三" sub_teams: - name: "团队A1" level: 2 leader: "李四" - name: "团队A2" level: 2 leader: "王五" - name: "团队B" level: 1 leader: "赵六"

daily_targets.yaml：定义每个团队每日各指标的目标值。

# daily_targets.yaml targets: - team: "团队A" indicators: 新增用户: 50 业务办理: 30 - team: "团队A1" indicators: 新增用户: 20 业务办理: 15

indicators.yaml：定义需要统计的指标及其显示顺序。

# indicators.yaml indicators: - name: "新增用户" display_name: "新增用户数" - name: "业务办理" display_name: "业务办理量" statistic_level: 2 # 统计层级：1 或 2 team_order: ["团队A1", "团队A2", "团队B"] # 团队输出顺序

file_mapping.yaml：定义每个指标的数据来源和统计规则。

# file_mapping.yaml mappings: - indicator: "新增用户" file: "新增用户报表.xls" sheet_name: "Sheet1" rules: status_column: "受理状态" valid_status: ["正常"] keyword_filter: column: "受理业务名称" keywords: ["新增", "开户"] exclude_keywords: column: "受理业务名称" keywords: ["测试", "演示"] deduplicate_column: "号码" deduplicate: true - indicator: "业务办理" file: "业务办理报表.xls" sheet_name: "Sheet1" rules: status_column: "受理状态" valid_status: ["正常"] keyword_filter: column: "受理业务名称" keywords: ["办理", "开通"] business_code_filter: column: "业务编码" codes: ["1001", "1002"] deduplicate: false

2. 核心数据处理模块

此模块负责读取配置、解析Excel文件，并执行统计计算。

# core/processor.py import yaml import pandas as pd import os from typing import Dict, List, Any class DataProcessor: def __init__(self, config_path: str = "config.yaml"): with open(config_path, 'r', encoding='utf-8') as f: self.config = yaml.safe_load(f) self._load_all_configs() def _load_all_configs(self): """加载所有配置文件""" with open(self.config['team_config'], 'r', encoding='utf-8') as f: self.team_config = yaml.safe_load(f) with open(self.config['daily_target_config'], 'r', encoding='utf-8') as f: self.target_config = yaml.safe_load(f) with open(self.config['indicator_config'], 'r', encoding='utf-8') as f: self.indicator_config = yaml.safe_load(f) with open(self.config['file_mapping_config'], 'r', encoding='utf-8') as f: self.mapping_config = yaml.safe_load(f) def process_file(self, mapping: Dict) -> pd.DataFrame: """根据映射规则处理单个文件""" file_path = os.path.join(self.config['data_dir'], mapping['file']) df = pd.read_excel(file_path, sheet_name=mapping['sheet_name']) rules = mapping['rules'] # 1. 筛选有效状态 if 'status_column' in rules: df = df[df[rules['status_column']].isin(rules['valid_status'])] # 2. 关键字筛选 if 'keyword_filter' in rules: keyword_condition = df[rules['keyword_filter']['column']].astype(str).apply( lambda x: any(kw in x for kw in rules['keyword_filter']['keywords']) ) df = df[keyword_condition] # 3. 排除关键字 if 'exclude_keywords' in rules: exclude_condition = ~df[rules['exclude_keywords']['column']].astype(str).apply( lambda x: any(kw in x for kw in rules['exclude_keywords']['keywords']) ) df = df[exclude_condition] # 4. 业务编码筛选 if 'business_code_filter' in rules: df = df[df[rules['business_code_filter']['column']].isin(rules['business_code_filter']['codes'])] # 5. 去重 if rules.get('deduplicate', False) and 'deduplicate_column' in rules: df = df.drop_duplicates(subset=[rules['deduplicate_column']]) return df def calculate_statistics(self) -> Dict[str, Any]: """计算所有团队的所有指标统计结果""" results = {} team_stats = {} # 初始化团队统计结构 for team in self._get_all_teams(): team_stats[team] = {ind['name']: 0 for ind in self.indicator_config['indicators']} # 按指标映射处理文件并统计 for mapping in self.mapping_config['mappings']: indicator = mapping['indicator'] df = self.process_file(mapping) # 假设数据中有'团队'列，实际需根据文件结构调整 # 这里简化为按团队名称分组计数 if '团队' in df.columns: team_counts = df.groupby('团队').size().to_dict() for team, count in team_counts.items(): if team in team_stats: team_stats[team][indicator] = count # 计算完成率 for team in team_stats: results[team] = {} for indicator in team_stats[team]: actual = team_stats[team][indicator] target = self._get_target(team, indicator) completion_rate = (actual / target * 100) if target > 0 else 0 results[team][indicator] = { 'target': target, 'actual': actual, 'rate': round(completion_rate, 1) } return results def _get_all_teams(self) -> List[str]: """获取所有需要统计的团队列表""" # 根据配置的统计层级和团队顺序返回 level = self.indicator_config['statistic_level'] teams = [] for team in self.team_config['teams']: if level == 1: teams.append(team['name']) else: for sub in team.get('sub_teams', []): teams.append(sub['name']) # 按配置的顺序排序 ordered_teams = [t for t in self.indicator_config['team_order'] if t in teams] return ordered_teams def _get_target(self, team: str, indicator: str) -> int: """获取指定团队的指标目标值""" for target in self.target_config['targets']: if target['team'] == team: return target['indicators'].get(indicator, 0) return 0 def get_lagging_teams(self, stats: Dict) -> Dict[str, List[str]]: """找出每个指标完成率最低的三个团队""" lagging = {} indicators = [ind['name'] for ind in self.indicator_config['indicators']] for indicator in indicators: # 收集所有团队在该指标上的完成率 team_rates = [] for team in stats: if stats[team][indicator]['actual'] > 0: # 只统计有发展的团队 team_rates.append((team, stats[team][indicator]['rate'])) # 如果所有团队都没有发展，则不计入落后 if not team_rates: continue # 按完成率升序排序，取最后三名 team_rates.sort(key=lambda x: x[1]) lagging[indicator] = [team for team, _ in team_rates[:3]] return lagging

3. 图片生成模块

使用Pillow库生成适配手机屏幕的通报图片。

# core/image_generator.py from PIL import Image, ImageDraw, ImageFont import os from typing import Dict, List class ReportImageGenerator: def __init__(self, output_dir: str = "./output"): self.output_dir = output_dir os.makedirs(output_dir, exist_ok=True) # 使用系统字体，确保支持中文 self.font_path = "/System/Library/Fonts/PingFang.ttc" # macOS # Windows可使用：r"C:\Windows\Fonts\msyh.ttc" def generate_image(self, stats: Dict, lagging_teams: Dict, date_str: str) -> str: """生成业务通报图片""" # 图片尺寸适配手机屏幕（1080x1920） width, height = 1080, 1920 image = Image.new('RGB', (width, height), color='white') draw = ImageDraw.Draw(image) # 加载字体 try: title_font = ImageFont.truetype(self.font_path, 60) header_font = ImageFont.truetype(self.font_path, 40) text_font = ImageFont.truetype(self.font_path, 35) except: # 备选字体 title_font = ImageFont.load_default() header_font = ImageFont.load_default() text_font = ImageFont.load_default() # 绘制标题 title = f"业务发展通报 ({date_str})" title_bbox = draw.textbbox((0, 0), title, font=title_font) title_width = title_bbox[2] - title_bbox[0] draw.text(((width - title_width) / 2, 80), title, fill='black', font=title_font) # 绘制表格 y_offset = 200 row_height = 80 col_widths = [300, 200, 200, 200] # 团队名称、目标、完成量、完成率 # 表头 headers = ["团队", "目标", "完成量", "完成率"] for i, header in enumerate(headers): x = sum(col_widths[:i]) + 50 draw.rectangle([x, y_offset, x + col_widths[i], y_offset + row_height], outline='black', width=2) draw.text((x + 20, y_offset + 20), header, fill='black', font=header_font) y_offset += row_height # 数据行 teams = list(stats.keys()) indicators = list(next(iter(stats.values())).keys()) for team in teams: for idx, indicator in enumerate(indicators): data = stats[team][indicator] # 团队和指标名 if idx == 0: draw.text((60, y_offset + 20), team, fill='black', font=text_font) # 数据单元格 cells = [ str(data['target']), str(data['actual']), f"{data['rate']}%" ] for i, cell in enumerate(cells): x = sum(col_widths[:i+1]) + 50 draw.rectangle([x, y_offset, x + col_widths[i+1], y_offset + row_height], outline='gray', width=1) # 高亮落后团队 if indicator in lagging_teams and team in lagging_teams[indicator]: draw.rectangle([x, y_offset, x + col_widths[i+1], y_offset + row_height], fill='#FFCCCC') draw.text((x + 20, y_offset + 20), cell, fill='black', font=text_font) y_offset += row_height y_offset += 10 # 团队间间隔 # 保存图片 filename = f"business_report_{date_str}.png" filepath = os.path.join(self.output_dir, filename) image.save(filepath, quality=95) return filepath

4. Web界面模块

使用Flask构建简单的Web界面，展示统计结果和生成的图片。

# app.py from flask import Flask, render_template, send_file, jsonify from core.processor import DataProcessor from core.image_generator import ReportImageGenerator from datetime import datetime import os app = Flask(__name__) processor = DataProcessor() image_gen = ReportImageGenerator() @app.route('/') def index(): """主页面：显示最新统计结果""" stats = processor.calculate_statistics() lagging = processor.get_lagging_teams(stats) date_str = datetime.now().strftime("%Y-%m-%d") # 生成图片 img_path = image_gen.generate_image(stats, lagging, date_str) return render_template('index.html', stats=stats, lagging_teams=lagging, date_str=date_str, image_url=f'/report_image/{date_str}') @app.route('/report_image/<date>') def get_report_image(date): """获取通报图片""" img_path = os.path.join(image_gen.output_dir, f"business_report_{date}.png") if os.path.exists(img_path): return send_file(img_path, mimetype='image/png') else: return "Image not found", 404 @app.route('/api/stats') def get_stats(): """API接口：获取统计结果（JSON格式）""" stats = processor.calculate_statistics() lagging = processor.get_lagging_teams(stats) return jsonify({ 'stats': stats, 'lagging_teams': lagging, 'timestamp': datetime.now().isoformat() }) if __name__ == '__main__': app.run(debug=True, port=5000)

5. HTML模板

<!-- templates/index.html --> <!DOCTYPE html> <html> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>业务发展通报系统</title> <style> body { font-family: Arial, sans-serif; margin: 20px; } .container { max-width: 1200px; margin: 0 auto; } .header { text-align: center; margin-bottom: 30px; } .stats-table { width: 100%; border-collapse: collapse; margin: 20px 0; } .stats-table th, .stats-table td { border: 1px solid #ddd; padding: 12px; text-align: center; } .stats-table th { background-color: #f4f4f4; } .lagging { background-color: #ffcccc; } .report-image { max-width: 100%; margin: 20px 0; border: 1px solid #ddd; } .alert-box { background-color: #fff3cd; border: 1px solid #ffeaa7; padding: 15px; margin: 20px 0; } </style> </head> <body> <div class="container"> <div class="header"> <h1>业务发展通报系统</h1> <p>统计日期: {{ date_str }}</p> <button onclick="location.reload()">刷新数据</button> </div> <div class="alert-box"> <h3>落后团队提醒</h3> {% for indicator, teams in lagging_teams.items() %} <p>{{ indicator }}：@{{ teams|join(', @') }}</p> {% endfor %} </div> <h2>详细统计结果</h2> <table class="stats-table"> <thead> <tr> <th>团队</th> <th>指标</th> <th>目标</th> <th>完成量</th> <th>完成率</th> </tr> </thead> <tbody> {% for team, indicators in stats.items() %} {% for indicator_name, values in indicators.items() %} <tr {% if indicator_name in lagging_teams and team in lagging_teams[indicator_name] %}class="lagging"{% endif %}> <td>{{ team }}</td> <td>{{ indicator_name }}</td> <td>{{ values.target }}</td> <td>{{ values.actual }}</td> <td>{{ values.rate }}%</td> </tr> {% endfor %} {% endfor %} </tbody> </table> <h2>通报图片</h2> <img src="{{ image_url }}" alt="业务通报" class="report-image"> <p><a href="{{ image_url }}" download>下载图片</a></p> </div> </body> </html>

二、使用文档

1. 环境准备

# 1. 创建并激活conda环境 conda create -n business_report python=3.8 conda activate business_report # 2. 安装依赖包 pip install pandas openpyxl pillow flask pyyaml # 3. 启动JupyterLab jupyter lab

2. 项目目录结构

business_report/ ├── app.py # Flask主应用 ├── config.yaml # 主配置文件 ├── configs/ # 配置文件夹 │ ├── teams.yaml │ ├── daily_targets.yaml │ ├── indicators.yaml │ └── file_mapping.yaml ├── core/ # 核心模块 │ ├── __init__.py │ ├── processor.py # 数据处理 │ └── image_generator.py # 图片生成 ├── templates/ # HTML模板 │ └── index.html ├── data/ # 原始Excel文件 │ ├── 新增用户报表.xls │ └── 业务办理报表.xls └── output/ # 生成的图片

3. 配置步骤

准备团队配置(configs/teams.yaml)：按照YAML格式定义团队层级和负责人。
设置日目标(configs/daily_targets.yaml)：为每个团队配置每日各指标的目标值。
定义统计指标(configs/indicators.yaml)：确定需要统计的指标及其显示顺序。
配置文件映射(configs/file_mapping.yaml)：为每个指标指定数据源文件和统计规则。
放置数据文件：将下载的.xls报表文件放入data/目录，确保文件名与映射配置一致。

4. 运行系统

# 方法1：在JupyterLab中运行 # 新建Notebook，执行： %run app.py # 方法2：命令行运行 python app.py

访问http://localhost:5000查看Web界面。

5. 数据更新流程

下载最新报表：将最新的.xls文件放入data/目录，覆盖旧文件。
刷新页面：访问Web界面，点击“刷新数据”按钮。
获取通报图片：页面会自动生成并显示最新的通报图片，可下载用于微信发送。

6. 关键功能说明

功能模块	说明	配置文件
团队管理	定义团队层级结构，支持一级/二级团队	`teams.yaml`
目标管理	设置各团队每日指标目标值	`daily_targets.yaml`
指标配置	控制统计指标和显示顺序	`indicators.yaml`
数据映射	定义指标的数据来源和统计规则	`file_mapping.yaml`
统计计算	自动处理Excel文件，应用去重、关键字过滤等规则	`processor.py`
图片生成	生成手机适配的通报图片，高亮落后团队	`image_generator.py`
Web展示	提供可视化界面和API接口	`app.py`