自动化业务通报系统实现

问题解构需求核心是构建一个基于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 业务办理: 15indicators.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: false2. 核心数据处理模块此模块负责读取配置、解析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, encodingutf-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, encodingutf-8) as f: self.team_config yaml.safe_load(f) with open(self.config[daily_target_config], r, encodingutf-8) as f: self.target_config yaml.safe_load(f) with open(self.config[indicator_config], r, encodingutf-8) as f: self.indicator_config yaml.safe_load(f) with open(self.config[file_mapping_config], r, encodingutf-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_namemapping[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(keylambda x: x[1]) lagging[indicator] [team for team, _ in team_rates[:3]] return lagging3. 图片生成模块使用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_okTrue) # 使用系统字体确保支持中文 self.font_path /System/Library/Fonts/PingFang.ttc # macOS # Windows可使用rC:\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), colorwhite) 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, fonttitle_font) title_width title_bbox[2] - title_bbox[0] draw.text(((width - title_width) / 2, 80), title, fillblack, fonttitle_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], outlineblack, width2) draw.text((x 20, y_offset 20), header, fillblack, fontheader_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, fillblack, fonttext_font) # 数据单元格 cells [ str(data[target]), str(data[actual]), f{data[rate]}% ] for i, cell in enumerate(cells): x sum(col_widths[:i1]) 50 draw.rectangle([x, y_offset, x col_widths[i1], y_offset row_height], outlinegray, width1) # 高亮落后团队 if indicator in lagging_teams and team in lagging_teams[indicator]: draw.rectangle([x, y_offset, x col_widths[i1], y_offset row_height], fill#FFCCCC) draw.text((x 20, y_offset 20), cell, fillblack, fonttext_font) y_offset row_height y_offset 10 # 团队间间隔 # 保存图片 filename fbusiness_report_{date_str}.png filepath os.path.join(self.output_dir, filename) image.save(filepath, quality95) return filepath4. 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, statsstats, lagging_teamslagging, date_strdate_str, image_urlf/report_image/{date_str}) app.route(/report_image/date) def get_report_image(date): 获取通报图片 img_path os.path.join(image_gen.output_dir, fbusiness_report_{date}.png) if os.path.exists(img_path): return send_file(img_path, mimetypeimage/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(debugTrue, port5000)5. HTML模板!-- templates/index.html -- !DOCTYPE html html head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.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 classcontainer div classheader h1业务发展通报系统/h1 p统计日期: {{ date_str }}/p button onclicklocation.reload()刷新数据/button /div div classalert-box h3落后团队提醒/h3 {% for indicator, teams in lagging_teams.items() %} p{{ indicator }}{{ teams|join(, ) }}/p {% endfor %} /div h2详细统计结果/h2 table classstats-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] %}classlagging{% 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业务通报 classreport-image pa href{{ image_url }} download下载图片/a/p /div /body /html二、使用文档1. 环境准备# 1. 创建并激活conda环境 conda create -n business_report python3.8 conda activate business_report # 2. 安装依赖包 pip install pandas openpyxl pillow flask pyyaml # 3. 启动JupyterLab jupyter lab2. 项目目录结构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.pyWeb展示提供可视化界面和API接口app.py7. 自定义扩展新增统计规则在processor.py的process_file方法中添加新的规则处理逻辑。修改图片样式调整image_generator.py中的字体、颜色和布局参数。添加新指标在indicators.yaml中定义新指标并在file_mapping.yaml中配置数据源。调整提醒逻辑修改get_lagging_teams方法中的排序和筛选条件。8. 注意事项文件编码确保所有配置文件使用UTF-8编码避免中文乱码。Excel格式系统使用openpyxl引擎读取.xls文件确保文件格式正确。字体支持如需生成中文图片请确保系统安装中文字体或修改image_generator.py中的字体路径。数据更新每次统计前请确保data/目录中的Excel文件为最新版本。配置验证修改配置文件后建议重启Flask应用以确保配置生效。该系统通过模块化设计实现了业务通报的全自动化处理结合JupyterLab的交互特性和Flask的Web展示能力提供了灵活、可配置的解决方案。用户只需按规范准备配置文件和原始数据即可一键生成符合移动端展示需求的业务通报。