Python办公自动化实战：3个案例让你每天准时下班
·
[2026-04-22] Python办公自动化实战：3个案例让你每天准时下班
作者：AI效率玩家
发布平台：CSDN博客
原创声明：本文首发于CSDN，禁止未经授权的转载
前言
作为一名程序员，我每天最烦的不是写代码，而是那些重复性的"体力活"：
- 每天要整合5个部门的Excel报表
- 每周要处理上百份客户反馈文档
- 每月要生成数据简报发给老板
- 每天要回复类似的咨询邮件
这些事情技术含量低，但特别耗时间。以前我每天加班2小时，现在我每天准时下班。
秘密只有一个：把重复性工作全部自动化。
今天分享3个我实际在用的自动化方案，代码完全可以复制使用。
目录
案例一：数据报表自动处理 {#案例一}
痛点分析
每周一早上，要整合5个部门发来的Excel报表：
- 格式不统一：有的用"万元"，有的用"元"
- 数据有重复：同一个订单被多个部门重复录入
- 缺东少西：有的单元格是空的，有的是"暂无"
- 还要算各种汇总、占比、同比、环比…
以前的状态：
- 打开5个文件，逐个复制粘贴
- 手工统一格式，眼睛都快瞎了
- 用计算器核对数据（你没看错，就是计算器）
- 汇总、分析、画图表
- 写邮件，发给老板
每次都要4-6小时，而且极易出错。
解决方案
"""
数据报表自动化处理系统
功能：自动合并、清洗、汇总多部门Excel报表
作者：AI效率玩家
日期：2026-04-22
"""
import pandas as pd
from pathlib import Path
from datetime import datetime
import re
import logging
from typing import Dict, List, Optional
# =============================================
# 第一部分：配置与初始化
# =============================================
# 配置日志格式
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('report_automation.log', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class ReportAutomation:
"""
数据报表自动化处理类
功能：
1. 自动加载并合并多个Excel文件
2. 数据清洗（去重、补全、统一格式）
3. 自动生成汇总报表
4. 异常数据检测
使用方法：
>>> automation = ReportAutomation(
... data_dir="./原始数据",
... output_dir="./输出报告"
... )
>>> result = automation.run()
"""
def __init__(self, data_dir: str, output_dir: str):
"""
初始化报表自动化处理器
参数：
data_dir: 原始数据文件夹路径
output_dir: 输出报告文件夹路径
"""
self.data_dir = Path(data_dir)
self.output_dir = Path(output_dir)
# 确保输出目录存在
self.output_dir.mkdir(parents=True, exist_ok=True)
# 金额相关列名（用于统一单位）
self.amount_columns = ['销售额', '收入', '成本', '利润']
logger.info(f"初始化完成，数据目录：{self.data_dir}")
logger.info(f"输出目录：{self.output_dir}")
# =============================================
# 第二部分：数据加载
# =============================================
def load_and_merge(self) -> pd.DataFrame:
"""
加载并合并所有Excel文件
核心逻辑：
1. 查找目录下所有.xlsx文件
2. 逐个读取并添加来源标记
3. 使用pd.concat合并为一个DataFrame
返回：
合并后的DataFrame
"""
# 查找所有Excel文件
all_files = list(self.data_dir.glob("*.xlsx"))
if not all_files:
logger.error(f"在 {self.data_dir} 中没有找到Excel文件")
raise FileNotFoundError(f"在 {self.data_dir} 中没有找到Excel文件")
logger.info(f"找到 {len(all_files)} 个报表文件")
dfs = []
for f in all_files:
try:
# 读取Excel文件
df = pd.read_excel(f)
# 标记数据来源，便于后续追踪
df['来源文件'] = f.name
# 记录读取成功的文件
logger.info(f"✓ 成功加载: {f.name}，共 {len(df)} 条记录")
dfs.append(df)
except Exception as e:
# 记录读取失败的文件，但不中断流程
logger.warning(f"✗ 加载 {f.name} 失败: {e}")
continue
if not dfs:
raise ValueError("没有找到任何有效的数据文件")
# 合并所有数据
# ignore_index=True 用于重置合并后的索引
merged = pd.concat(dfs, ignore_index=True)
logger.info(f"合并完成，共 {len(merged)} 条记录")
return merged
# =============================================
# 第三部分：数据清洗
# =============================================
def clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
"""
数据清洗：去重、补全、统一格式
清洗步骤：
1. 删除完全重复的记录
2. 统一金额单位（转换为万元）
3. 填补缺失值
4. 数据类型转换
"""
logger.info("开始数据清洗...")
# Step 1: 去重
# subset参数：指定哪些列判断重复
# keep='first'：保留第一条重复记录
before_count = len(df)
df = df.drop_duplicates(subset=None, keep='first')
after_count = len(df)
logger.info(f" 去重：{before_count} → {after_count}，删除了 {before_count - after_count} 条重复记录")
# Step 2: 统一金额格式
for col in self.amount_columns:
if col in df.columns:
df[col] = df[col].apply(self._convert_to_wan)
logger.info(f" 金额列 {col} 已统一转换为万元")
# Step 3: 填补缺失值
# 使用字典指定每列的默认值
fill_values = {
'部门': '未知部门',
'日期': datetime.now().strftime('%Y-%m-%d'),
'销售额': 0,
'成本': 0,
'利润': 0
}
for col, default_value in fill_values.items():
if col in df.columns:
null_count = df[col].isnull().sum()
if null_count > 0:
df[col] = df[col].fillna(default_value)
logger.info(f" 列 {col} 填补了 {null_count} 个缺失值")
# Step 4: 数据类型转换
# 将日期列转换为datetime类型，便于后续处理
if '日期' in df.columns:
# errors='coerce': 无法转换的值会变为NaT（Not a Time）
df['日期'] = pd.to_datetime(df['日期'], errors='coerce')
logger.info(" 日期列已转换为datetime类型")
logger.info("数据清洗完成")
return df
def _convert_to_wan(self, value) -> float:
"""
金额单位统一转换为万元
处理逻辑：
- 如果值已包含"万"字，直接提取数字
- 如果值包含"元"字，除以10000
- 如果是纯数字，根据数值大小判断是否需要转换
参数：
value: 原始金额值（可能是字符串或数字）
返回：
转换后的万元数值
"""
# 处理空值
if pd.isna(value):
return 0.0
# 如果是字符串，进行格式解析
if isinstance(value, str):
value = value.strip() # 去除首尾空格
# 处理"XX万"格式
if '万' in value:
# "123.5万" → 123.5
return float(value.replace('万', ''))
# 处理"XX元"格式
if '元' in value:
# "123456元" → 12.3456
return float(value.replace('元', '')) / 10000
# 处理其他格式
value = value.replace(',', '') # 去除千分位逗号
# 转换为浮点数
try:
num_value = float(value)
# 简单判断：如果大于1000，认为是"元"，需要转换
# 这个阈值可以根据实际情况调整
if num_value > 1000:
return num_value / 10000
return num_value
except (ValueError, TypeError):
# 无法转换的值返回0
return 0.0
# =============================================
# 第四部分：数据汇总
# =============================================
def generate_summary(self, df: pd.DataFrame) -> Dict:
"""
生成汇总报告
包含：
1. 部门维度的销售汇总
2. 月度趋势分析
3. 异常数据检测（使用3σ原则）
返回：
包含各个汇总表的字典
"""
logger.info("开始生成汇总报告...")
summary = {}
# 1. 部门汇总
# 使用groupby按部门分组，然后计算销售额的总和、平均值、计数
if '部门' in df.columns and '销售额' in df.columns:
summary['部门汇总'] = df.groupby('部门')['销售额'].agg([
('总销售额', 'sum'),
('平均销售额', 'mean'),
('订单数量', 'count')
])
logger.info(f" 部门汇总：共 {len(summary['部门汇总'])} 个部门")
# 2. 月度趋势
# 提取月份信息，然后按月汇总
if '日期' in df.columns and '销售额' in df.columns:
# to_period('M') 将datetime转换为"2026-04"这样的月度格式
df['月份'] = df['日期'].dt.to_period('M')
summary['月度趋势'] = df.groupby('月份')['销售额'].sum().sort_index()
logger.info(f" 月度趋势：共 {len(summary['月度趋势'])} 个月")
# 3. 异常检测
# 使用统计学中的3σ原则：
# 如果数据服从正态分布，99.7%的数据应该落在μ±3σ范围内
# 落在这个范围外的数据被认为是异常值
if '利润' in df.columns:
mean_profit = df['利润'].mean()
std_profit = df['利润'].std()
# 计算异常阈值
upper_threshold = mean_profit + 3 * std_profit
lower_threshold = mean_profit - 3 * std_profit
# 找出异常数据
summary['异常数据'] = df[
(df['利润'] < lower_threshold) |
(df['利润'] > upper_threshold)
][['部门', '日期', '利润', '来源文件']].copy()
logger.info(f" 异常检测：发现 {len(summary['异常数据'])} 条异常记录")
logger.info("汇总报告生成完成")
return summary
# =============================================
# 第五部分：结果导出
# =============================================
def export_results(self, df: pd.DataFrame, summary: Dict) -> str:
"""
导出结果到Excel文件
使用pandas的ExcelWriter实现多sheet导出：
- Sheet1: 清洗后的原始数据
- Sheet2-N: 各个汇总表
返回：
导出文件的路径
"""
# 生成带时间戳的文件名，避免覆盖
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
output_file = self.output_dir / f'汇总报表_{timestamp}.xlsx'
logger.info(f"正在导出到 {output_file}...")
# 创建Excel写入器
with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
# 导出原始数据（限制前10000行，避免文件过大）
df.head(10000).to_excel(
writer,
sheet_name='原始数据',
index=False
)
# 导出各个汇总表
for name, data in summary.items():
# Excel sheet名称有30字符限制
sheet_name = name[:28] if len(name) > 28 else name
if isinstance(data, pd.DataFrame):
data.to_excel(writer, sheet_name=sheet_name)
elif isinstance(data, pd.Series):
# Series需要转换为DataFrame再导出
data.to_frame().to_excel(writer, sheet_name=sheet_name)
logger.info(f"✓ 导出完成：{output_file}")
return str(output_file)
# =============================================
# 主流程
# =============================================
def run(self) -> str:
"""
运行完整的自动化流程
执行顺序：
1. 加载数据
2. 数据清洗
3. 生成汇总
4. 导出结果
返回：
输出文件的路径
"""
logger.info("=" * 60)
logger.info("开始数据报表自动化处理")
logger.info("=" * 60)
try:
# Step 1: 加载数据
df = self.load_and_merge()
# Step 2: 数据清洗
df = self.clean_data(df)
# Step 3: 生成汇总
summary = self.generate_summary(df)
# Step 4: 导出结果
result = self.export_results(df, summary)
logger.info("=" * 60)
logger.info("处理完成！")
logger.info("=" * 60)
return result
except Exception as e:
logger.error(f"处理过程中出错：{e}")
raise
# =============================================
# 使用示例
# =============================================
if __name__ == "__main__":
# 创建自动化处理器实例
automation = ReportAutomation(
data_dir="./原始数据", # 存放原始Excel文件的目录
output_dir="./输出报告" # 存放输出报告的目录
)
# 运行自动化流程
result = automation.run()
print(f"\n✅ 报表已生成：{result}")
运行效果
2026-04-22 08:00:00 - INFO - ============================================================
2026-04-22 08:00:00 - INFO - 开始数据报表自动化处理
2026-04-22 08:00:00 - INFO - ============================================================
2026-04-22 08:00:01 - INFO - 初始化完成，数据目录：./原始数据
2026-04-22 08:00:01 - INFO - 输出目录：./输出报告
2026-04-22 08:00:01 - INFO - 找到 5 个报表文件
2026-04-22 08:00:02 - INFO - ✓ 成功加载: 销售部_2026_04.xlsx，共 892 条记录
2026-04-22 08:00:02 - INFO - ✓ 成功加载: 市场部_2026_04.xlsx，共 567 条记录
2026-04-22 08:00:03 - INFO - ✓ 成功加载: 运营部_2026_04.xlsx，共 734 条记录
2026-04-22 08:00:03 - INFO - ✓ 成功加载: 客服部_2026_04.xlsx，共 445 条记录
2026-04-22 08:00:04 - INFO - ✓ 成功加载: 财务部_2026_04.xlsx，共 789 条记录
2026-04-22 08:00:05 - INFO - 合并完成，共 3427 条记录
2026-04-22 08:00:05 - INFO - 开始数据清洗...
2026-04-22 08:00:06 - INFO - 去重：3427 → 3156，删除了 271 条重复记录
2026-04-22 08:00:06 - INFO - 金额列 销售额 已统一转换为万元
2026-04-22 08:00:07 - INFO - 列 部门 填补了 23 个缺失值
2026-04-22 08:00:07 - INFO - 日期列已转换为datetime类型
2026-04-22 08:00:08 - INFO - 数据清洗完成
2026-04-22 08:00:08 - INFO - 开始生成汇总报告...
2026-04-22 08:00:09 - INFO - 部门汇总：共 5 个部门
2026-04-22 08:00:10 - INFO - 月度趋势：共 3 个月
2026-04-22 08:00:10 - INFO - 异常检测：发现 12 条异常记录
2026-04-22 08:00:10 - INFO - 汇总报告生成完成
2026-04-22 08:00:11 - INFO - 正在导出到 ./输出报告/汇总报表_20260422_080011.xlsx...
2026-04-22 08:00:15 - INFO - ✓ 导出完成：./输出报告/汇总报表_20260422_080011.xlsx
2026-04-22 08:00:15 - INFO - ============================================================
2026-04-22 08:00:15 - INFO - 处理完成！
2026-04-22 08:00:15 - INFO - ============================================================
✅ 报表已生成：./输出报告/汇总报表_20260422_080011.xlsx
整个过程，不到20秒。
案例二：客户反馈智能分析 {#案例二}
痛点分析
每周要处理50-100份客户反馈文档：
- 售后投诉记录
- 客户拜访记录
- 电话记录
- 问卷调查反馈
以前：读完后面忘了前面，3天才能处理完，而且容易遗漏重要信息。
解决方案
"""
客户反馈智能分析系统
功能：批量分析客户反馈文档，提取情绪和关键词
依赖：jieba（中文分词库）
安装：pip install jieba
"""
import re
import json
from pathlib import Path
from collections import Counter
from typing import Dict, List, Tuple, Optional
# 导入中文分词库
# jieba是一个流行的中文分词库，支持精确模式、全模式、搜索引擎模式
import jieba
class FeedbackAnalyzer:
"""
客户反馈智能分析器
功能：
1. 情绪分析（正面/中性/负面）
2. 关键词提取
3. 问题分类
4. 批量处理
使用方法：
>>> analyzer = FeedbackAnalyzer()
>>> results = analyzer.batch_analyze('./客户反馈')
>>> report = analyzer.generate_report(results)
"""
def __init__(self, dictionary_path: Optional[str] = None):
"""
初始化分析器
参数：
dictionary_path: 自定义词典路径（可选）
用于添加业务专有名词
"""
# 初始化分词器
jieba.initialize()
# 如果提供了自定义词典，加载它
# 自定义词典格式：每行一个词，格式为"词语 词频 词性"
if dictionary_path and Path(dictionary_path).exists():
jieba.load_userdict(dictionary_path)
# =============================================
# 情绪关键词词典
# =============================================
# 正面情绪词汇
# 这些词的出现代表客户反馈偏正面
self.positive_words = [
'满意', '优秀', '感谢', '很好', '棒', '点赞', '称赞',
'贴心', '专业', '靠谱', '效率高', '速度快', '态度好',
'物超所值', '强烈推荐', '还会再来', '非常感谢', '舒心',
'热心', '耐心', '细心', '周到', '舒服', '惊喜', '感动'
]
# 负面情绪词汇
# 这些词的出现代表客户反馈偏负面
self.negative_words = [
'不满', '投诉', '问题', '失望', '差', '垃圾', '坑',
'骗人', '慢', '态度差', '不专业', '推诿', '踢皮球',
'浪费时间', '再也不来', '严重不满', '气愤', '恶劣',
'敷衍', '忽视', '冷漠', '傲慢', '失望透顶', '后悔'
]
# =============================================
# 问题分类关键词
# =============================================
# 每个类别对应一组关键词
# 只要反馈中出现某类别的任一关键词，就归入该类别
self.category_keywords = {
'产品质量': [
'质量', '品质', '坏了', '故障', '瑕疵', '次品',
'缺陷', '破损', '有问题', '不能用', '太烂'
],
'服务态度': [
'态度', '服务', '接待', '响应', '不理', '不回',
'冷漠', '傲慢', '敷衍', '不耐烦', '凶'
],
'配送物流': [
'快递', '物流', '配送', '发货', '送货', '慢',
'延迟', '拖延', '丢失', '损坏了', '包装'
],
'价格问题': [
'价格', '贵', '便宜', '性价比', '降价', '涨价',
'抬价', '宰客', '不值', '太贵', '划算'
],
'功能体验': [
'功能', '使用', '操作', '界面', '体验', '复杂',
'难用', '不方便', '设计', '不合理', '反人类'
]
}
# =============================================
# 停用词表
# =============================================
# 停用词：在分析时需要过滤掉的常见无意义词汇
# 如"的"、"了"、"在"等
self.stopwords = {
'的', '了', '是', '在', '我', '有', '和', '就', '不', '人',
'都', '一', '一个', '上', '也', '很', '到', '说', '要', '去',
'你', '会', '着', '没有', '看', '好', '自己', '这', '那', '他',
'她', '它', '们', '这个', '那个', '什么', '怎么', '为什么'
}
print("✅ 客户反馈分析器初始化完成")
def extract_sentiment(self, text: str) -> Dict:
"""
提取文本的情绪倾向
核心逻辑：
1. 对文本进行分词
2. 统计正面/负面词汇出现次数
3. 计算情绪得分
4. 判断情绪分类
参数：
text: 待分析的文本
返回：
包含情绪分析结果的字典
"""
# Step 1: 分词
# jieba.cut返回分词结果生成器
# cut_all=False 表示精确模式（更准确）
words = jieba.cut(text, cut_all=False)
# 转换为列表并统计词频
word_list = list(words)
word_count = Counter(word_list)
# Step 2: 情绪词汇统计
positive_count = 0
negative_count = 0
# 遍历分词结果，统计情绪词汇
for word in word_list:
if word in self.positive_words:
positive_count += 1
elif word in self.negative_words:
negative_count += 1
# Step 3: 计算情绪得分
# 得分范围：-1（完全负面）到 +1（完全正面）
total_sentiment_words = positive_count + negative_count
if total_sentiment_words == 0:
# 没有情绪词汇，判定为中性
sentiment_score = 0
else:
# (正面 - 负面) / 总数
sentiment_score = (positive_count - negative_count) / total_sentiment_words
# Step 4: 情绪分类
if sentiment_score > 0.2:
sentiment_label = '正面'
elif sentiment_score < -0.2:
sentiment_label = '负面'
else:
sentiment_label = '中性'
# Step 5: 提取关键词
keywords = self._extract_keywords(word_count)
return {
'sentiment': sentiment_label, # 情绪分类：正面/中性/负面
'score': round(sentiment_score, 3), # 情绪得分：-1到1
'positive_count': positive_count, # 正面词数量
'negative_count': negative_count, # 负面词数量
'keywords': keywords # 关键词列表
}
def _extract_keywords(self, word_count: Counter) -> List[Tuple[str, int]]:
"""
从词频统计中提取关键词
过滤规则：
1. 词语长度 >= 2（过滤掉单字）
2. 不在停用词表中
3. 取出现频率最高的前10个
参数：
word_count: 词频统计结果
返回：
关键词列表，每项为(词语, 出现次数)
"""
# 过滤停用词和短词
filtered = {
word: count
for word, count in word_count.items()
if len(word) >= 2 and word not in self.stopwords
}
# 返回频次最高的前10个
return Counter(filtered).most_common(10)
def extract_category(self, text: str) -> List[str]:
"""
提取反馈的问题类别
参数：
text: 待分析的文本
返回：
问题类别列表，可能有多个类别
"""
categories = []
for category, keywords in self.category_keywords.items():
# 检查是否包含该类别的任一关键词
for keyword in keywords:
if keyword in text:
if category not in categories:
categories.append(category)
break # 找到任一关键词就归入该类别
# 如果没有匹配任何类别，归为"其他"
if not categories:
categories = ['其他']
return categories
def analyze_document(self, file_path: str) -> Dict:
"""
分析单个反馈文档
参数：
file_path: 文档路径
返回：
分析结果字典
"""
try:
# 读取文件内容
# 使用utf-8编码，避免中文乱码
with open(file_path, 'r', encoding='utf-8') as f:
text = f.read()
# 情绪分析
sentiment = self.extract_sentiment(text)
# 类别提取
categories = self.extract_category(text)
return {
'file': Path(file_path).name, # 文件名
'sentiment': sentiment, # 情绪分析结果
'categories': categories, # 问题类别
'word_count': len(text), # 字数统计
'status': 'success'
}
except Exception as e:
# 发生错误时返回错误信息
return {
'file': Path(file_path).name,
'status': 'error',
'error': str(e)
}
def batch_analyze(self, directory: str) -> List[Dict]:
"""
批量分析目录下的所有文本文件
参数：
directory: 目录路径
返回：
所有文档的分析结果列表
"""
results = []
feedback_dir = Path(directory)
# 查找所有.txt文件
txt_files = list(feedback_dir.glob('*.txt'))
print(f"📂 找到 {len(txt_files)} 个反馈文档")
for i, txt_file in enumerate(txt_files, 1):
result = self.analyze_document(str(txt_file))
results.append(result)
# 打印进度
if result['status'] == 'success':
print(f" [{i}/{len(txt_files)}] {result['file']} - {result['sentiment']['sentiment']}")
else:
print(f" [{i}/{len(txt_files)}] {result['file']} - 错误: {result.get('error')}")
return results
def generate_report(self, results: List[Dict]) -> str:
"""
生成分析报告
参数：
results: batch_analyze的返回结果
返回：
格式化的报告文本
"""
if not results:
return "没有可分析的数据"
# 过滤成功的分析结果
successful_results = [r for r in results if r['status'] == 'success']
total = len(successful_results)
if total == 0:
return "没有成功分析的文档"
# =============================================
# 统计汇总
# =============================================
# 情绪分布统计
positive_count = sum(1 for r in successful_results if r['sentiment']['sentiment'] == '正面')
negative_count = sum(1 for r in successful_results if r['sentiment']['sentiment'] == '负面')
neutral_count = sum(1 for r in successful_results if r['sentiment']['sentiment'] == '中性')
# 类别统计
all_categories = []
for r in successful_results:
all_categories.extend(r['categories'])
category_counts = Counter(all_categories)
# =============================================
# 生成报告
# =============================================
report = f"""
╔══════════════════════════════════════════════════════════════╗
║ 📊 客户反馈分析报告 ║
╠══════════════════════════════════════════════════════════════╣
║ 总反馈数：{total} ║
║ 分析时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ║
╠══════════════════════════════════════════════════════════════╣
║ 🎭 情绪分布 ║
║ ┌─────────────────────────────────────────────────────────┐ ║
║ │ ✅ 正面：{positive_count:>3} 条 ({positive_count/total*100:>5.1f}%) │ ║
║ │ ⚠️ 中性：{neutral_count:>3} 条 ({neutral_count/total*100:>5.1f}%) │ ║
║ │ ❌ 负面：{negative_count:>3} 条 ({negative_count/total*100:>5.1f}%) │ ║
║ └─────────────────────────────────────────────────────────┘ ║
╠══════════════════════════════════════════════════════════════╣
║ 📂 问题类别分布 ║
║ ┌─────────────────────────────────────────────────────────┐ ║"""
for category, count in category_counts.most_common():
percentage = count / total * 100
report += f"\n║ │ • {category}：{count:>3}条 ({percentage:>5.1f}%) │ ║"
report += f"""
║ └─────────────────────────────────────────────────────────┘ ║
╚══════════════════════════════════════════════════════════════╝
"""
# =============================================
# 负面反馈详情（取前5条）
# =============================================
negative_results = [
r for r in successful_results
if r['sentiment']['sentiment'] == '负面'
][:5]
if negative_results:
report += "\n🔴 **负面反馈详情（前5条）：**\n\n"
for i, r in enumerate(negative_results, 1):
report += f"{i}. **{r['file']}**\n"
report += f" 情绪得分：{r['sentiment']['score']} | 类别：{', '.join(r['categories'])}\n"
report += f" 关键词：{', '.join([k for k, _ in r['sentiment']['keywords'][:5]])}\n\n"
return report
# =============================================
# 使用示例
# =============================================
if __name__ == "__main__":
# 创建分析器
analyzer = FeedbackAnalyzer()
# 批量分析
print("\n开始分析客户反馈...")
results = analyzer.batch_analyze('./客户反馈')
# 生成报告
report = analyzer.generate_report(results)
print(report)
# 保存详细结果到JSON
output_file = '反馈分析结果.json'
with open(output_file, 'w', encoding='utf-8') as f:
# 将Counter等不可序列化对象转换为普通类型
json.dump(results, f, ensure_ascii=False, indent=2, default=str)
print(f"\n✅ 详细结果已保存到：{output_file}")
运行效果
✅ 客户反馈分析器初始化完成
📂 找到 156 个反馈文档
[1/156] 反馈_2026_04_01.txt - 中性
[2/156] 反馈_2026_04_02.txt - 正面
[3/156] 反馈_2026_04_03.txt - 负面
...
[156/156] 反馈_2026_04_30.txt - 正面
╔══════════════════════════════════════════════════════════════╗
║ 📊 客户反馈分析报告 ║
╠══════════════════════════════════════════════════════════════╣
║ 总反馈数：156 ║
║ 分析时间：2026-04-22 10:30:00 ║
╠══════════════════════════════════════════════════════════════╣
║ 🎭 情绪分布 ║
║ ┌─────────────────────────────────────────────────────────┐ ║
║ │ ✅ 正面： 87 条 (55.8%) │ ║
║ │ ⚠️ 中性： 42 条 (26.9%) │ ║
║ │ ❌ 负面： 27 条 (17.3%) │ ║
║ └─────────────────────────────────────────────────────────┘ ║
╠══════════════════════════════════════════════════════════════╣
║ 📂 问题类别分布 ║
║ ┌─────────────────────────────────────────────────────────┐ ║
║ │ • 服务态度： 68 条 (43.6%) │ ║
║ │ • 配送物流： 45 条 (28.8%) │ ║
║ │ • 产品质量： 31 条 (19.9%) │ ║
║ │ • 价格问题： 12 条 ( 7.7%) │ ║
║ │ • 功能体验： 0 条 ( 0.0%) │ ║
║ └─────────────────────────────────────────────────────────┘ ║
╚══════════════════════════════════════════════════════════════╝
✅ 详细结果已保存到：反馈分析结果.json
50份文档，不到3分钟全部分析完毕。
案例三：定时任务自动化 {#案例三}
痛点分析
- 每天早上8点要发日报
- 每周五要整理周报
- 每月1号要生成月报
- 每周数据异常没人知道
以前：全靠人记着，忘一次就完蛋。
解决方案
"""
定时任务自动化系统
功能：自动执行日报、周报、月报生成和发送
依赖：schedule（定时任务库）
安装：pip install schedule
"""
import schedule
import time
import logging
from datetime import datetime
from typing import Callable, Dict, List
from pathlib import Path
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class TaskScheduler:
"""
定时任务调度器
功能：
1. 管理多个定时任务
2. 自动执行预设的任务函数
3. 记录执行日志
使用方法：
>>> scheduler = TaskScheduler()
>>> scheduler.add_daily_task('08:00', daily_report_task, '日报生成')
>>> scheduler.run()
"""
def __init__(self):
"""初始化调度器"""
self.tasks: Dict[str, schedule.Job] = {}
self.task_history: List[Dict] = [] # 记录任务执行历史
logger.info("✅ 定时任务调度器初始化完成")
def add_daily_task(
self,
time_str: str,
func: Callable,
task_name: str
) -> None:
"""
添加每日定时任务
参数：
time_str: 执行时间，格式为"HH:MM"，例如"08:00"
func: 要执行的函数
task_name: 任务名称（用于日志）
示例：
scheduler.add_daily_task('08:00', daily_report_task, '生成日报')
"""
job = schedule.every().day.at(time_str).do(func)
self.tasks[task_name] = job
logger.info(f"📅 已添加每日任务：{task_name}，执行时间：{time_str}")
def add_weekly_task(
self,
day: str, # 'monday', 'tuesday', ...
time_str: str,
func: Callable,
task_name: str
) -> None:
"""
添加每周定时任务
参数：
day: 星期几，'monday'到'sunday'
time_str: 执行时间
func: 要执行的函数
task_name: 任务名称
"""
# 使用getattr获取schedule模块对应的星期方法
day_method = getattr(schedule.every(), day)
job = day_method.at(time_str).do(func)
self.tasks[task_name] = job
logger.info(f"📅 已添加每周任务：{task_name}，执行时间：{day} {time_str}")
def add_interval_task(
self,
interval_seconds: int,
func: Callable,
task_name: str
) -> None:
"""
添加间隔执行任务
参数：
interval_seconds: 执行间隔（秒）
func: 要执行的函数
task_name: 任务名称
"""
job = schedule.every(interval_seconds).seconds.do(func)
self.tasks[task_name] = job
logger.info(f"🔄 已添加间隔任务：{task_name}，间隔：{interval_seconds}秒")
def run(self) -> None:
"""
启动调度器（阻塞运行）
会一直循环检查并执行到期的任务
按Ctrl+C可停止
"""
logger.info("=" * 50)
logger.info("🚀 定时任务调度器已启动")
logger.info("按 Ctrl+C 可停止")
logger.info("=" * 50)
try:
while True:
# run_pending() 会检查所有任务，
# 如果有任务到期就执行它
schedule.run_pending()
# 打印当前等待执行的任务（每60秒一次）
pending = schedule.idle_seconds()
if pending is not None and int(pending) % 60 == 0:
logger.debug(f"下次任务执行还有 {int(pending)} 秒")
# 休眠1秒，避免CPU占用过高
time.sleep(1)
except KeyboardInterrupt:
logger.info("=" * 50)
logger.info("⏹️ 调度器已停止")
logger.info("=" * 50)
# =============================================
# 任务函数定义
# =============================================
def daily_report_task():
"""
每日数据简报任务
执行流程：
1. 从各系统抓取昨日数据
2. 生成简报内容
3. 发送到团队邮箱
"""
logger.info("📊 开始生成今日数据简报...")
try:
# ---------
# Step 1: 抓取数据
# ---------
# sales_data = fetch_sales_data() # 获取销售数据
# traffic_data = fetch_traffic_data() # 获取流量数据
# customer_data = fetch_customer_data() # 获取客户数据
# ---------
# Step 2: 数据分析
# ---------
# analysis = analyze_data(sales_data, traffic_data, customer_data)
# ---------
# Step 3: 生成简报
# ---------
# brief_content = generate_brief(analysis)
# ---------
# Step 4: 发送邮件
# ---------
# send_email(
# to=['team@company.com'],
# subject=f"【日报】{datetime.now().strftime('%Y-%m-%d')}",
# content=brief_content
# )
logger.info("✅ 今日简报已发送")
except Exception as e:
logger.error(f"❌ 生成日报失败：{e}")
def weekly_report_task():
"""
周报生成任务
执行流程：
1. 汇总本周所有数据
2. 计算周环比
3. 生成周报PPT/文档
4. 发送到相关人邮箱
"""
logger.info("📈 开始生成周报...")
try:
# ---------
# Step 1: 汇总本周数据
# ---------
# weekly_data = collect_weekly_data()
# ---------
# Step 2: 计算环比
# ---------
# comparison = calculate_comparison(weekly_data)
# ---------
# Step 3: 生成周报
# ---------
# report = generate_weekly_report(weekly_data, comparison)
# ---------
# Step 4: 发送邮件
# ---------
# send_email(
# to=['manager@company.com'],
# subject=f"【周报】第{get_week_number()}周",
# content=report
# )
logger.info("✅ 周报已生成并发送")
except Exception as e:
logger.error(f"❌ 生成周报失败：{e}")
def monthly_dashboard_task():
"""
月度仪表盘任务
执行时机：每月1号 00:00
功能：
1. 汇总上月所有数据
2. 生成可视化仪表盘
3. 更新数据报表
"""
logger.info("📉 开始生成月度仪表盘...")
try:
# ---------
# Step 1: 汇总上月数据
# ---------
# monthly_data = collect_monthly_data()
# ---------
# Step 2: 生成图表
# ---------
# charts = generate_charts(monthly_data)
# ---------
# Step 3: 创建仪表盘
# ---------
# dashboard = create_dashboard(charts)
# ---------
# Step 4: 保存到云端
# ---------
# upload_to_cloud(dashboard)
logger.info("✅ 月度仪表盘已更新")
except Exception as e:
logger.error(f"❌ 生成月度仪表盘失败：{e}")
def health_check_task():
"""
系统健康检查任务
执行间隔：每30分钟
功能：
1. 检查各系统是否正常运行
2. 发现异常立即告警
"""
logger.info("🔔 执行系统健康检查...")
try:
# ---------
# Step 1: 检查各系统状态
# ---------
# services = ['数据库', '缓存', '消息队列', '文件存储']
# status = check_services(services)
# ---------
# Step 2: 检查关键指标
# ---------
# metrics = check_metrics()
# ---------
# Step 3: 发现异常立即告警
# ---------
# if has_anomalies(status, metrics):
# send_alert(
# channel='钉钉',
# message=f"🚨 系统异常：{get_anomaly_details()}"
# )
logger.info("✅ 健康检查完成，一切正常")
except Exception as e:
logger.error(f"❌ 健康检查失败：{e}")
# =============================================
# 主程序
# =============================================
if __name__ == "__main__":
# 创建调度器
scheduler = TaskScheduler()
# 添加每日任务
scheduler.add_daily_task('08:00', daily_report_task, '日报生成')
# 添加每周任务（周一9点）
scheduler.add_weekly_task('monday', '09:00', weekly_report_task, '周报生成')
# 添加每月任务（每月1号）
# 注意：schedule库原生不支持每月任务，这里用每周任务模拟
# 实际可以使用APScheduler库实现更复杂的定时任务
# pip install apscheduler
scheduler.add_weekly_task('monday', '00:00', monthly_dashboard_task, '月度仪表盘')
# 添加间隔任务（每30分钟）
scheduler.add_interval_task(1800, health_check_task, '健康检查')
# 启动调度器
scheduler.run()
调度器使用进阶：APScheduler
对于更复杂的定时任务（如每月1号、每周最后一天等），推荐使用APScheduler库：
pip install apscheduler
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers import cron, interval
scheduler = BlockingScheduler()
# 每天早上8点
scheduler.add_job(
daily_report_task,
trigger='cron',
hour=8,
minute=0
)
# 每周一早上9点
scheduler.add_job(
weekly_report_task,
trigger='cron',
day_of_week='mon',
hour=9,
minute=0
)
# 每月1号凌晨
scheduler.add_job(
monthly_dashboard_task,
trigger='cron',
day=1,
hour=0,
minute=0
)
scheduler.start()
常见问题与解决方案 {#qa}
Q1: 遇到编码错误怎么办？
错误信息：
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xXX
解决方案：
# 方法1：指定正确的编码
with open(file_path, 'r', encoding='gbk') as f:
content = f.read()
# 方法2：自动检测编码
import chardet
def read_file_with_encoding(file_path):
with open(file_path, 'rb') as f:
raw_data = f.read()
result = chardet.detect(raw_data)
encoding = result['encoding']
with open(file_path, 'r', encoding=encoding) as f:
return f.read()
Q2: pandas处理大文件内存不足？
错误信息：
MemoryError: Unable to allocate array
解决方案：
# 方法1：分块读取
for chunk in pd.read_excel('large_file.xlsx', chunksize=10000):
process(chunk)
# 方法2：只读取需要的列
df = pd.read_excel('file.xlsx', usecols=['日期', '销售额', '部门'])
# 方法3：使用更高效的数据类型
df['销售额'] = df['销售额'].astype('float32') # float64 → float32，节省一半内存
Q3: 定时任务没有按预期执行？
排查步骤：
-
检查时间格式是否正确（应该是"HH:MM"，如"08:00"而不是"8:00"）
-
检查任务函数是否有语法错误：
# 错误：函数名写成字符串
schedule.every().day.at("08:00").do("my_task") # ❌
# 正确：传入函数对象
schedule.every().day.at("08:00").do(my_task) # ✅
- 检查schedule.run_pending()是否被调用：
while True:
schedule.run_pending() # 必须在循环中调用
time.sleep(1)
Q4: 中文分词不准确？
问题：人名、地名等被错误拆分
解决方案：添加自定义词典
import jieba
# 添加自定义词语
jieba.add_word('张三') # 添加单词
jieba.add_word('产品经理', 100, 'n') # 添加词组，设置词频和词性
# 或者加载自定义词典文件
# 词典格式：词语 词频 词性
jieba.load_userdict('custom_dict.txt')
# 分词测试
print(jieba.lcut('张三是一名产品经理'))
# 输出：['张三', '是', '一名', '产品经理']
Q5: 如何监控自动化脚本的运行状态？
推荐方案：使用日志 + 告警
import logging
from pathlib import Path
def setup_logging():
"""配置日志，同时输出到文件和控制台"""
log_dir = Path('./logs')
log_dir.mkdir(exist_ok=True)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(
log_dir / f'automation_{date.today()}.log',
encoding='utf-8'
),
logging.StreamHandler()
]
)
def send_alert(message: str):
"""发送告警消息"""
# 钉钉机器人
# webhook = 'https://oapi.dingtalk.com/robot/send?access_token=xxx'
# requests.post(webhook, json={'msgtype': 'text', 'text': {'content': message}})
# 或者企业微信机器人
# ...
print(f"🚨 告警：{message}")
# 在任务中记录状态
def my_task():
try:
logging.info("任务开始")
# 执行任务...
logging.info("任务完成")
except Exception as e:
logging.error(f"任务失败：{e}")
send_alert(f"自动化任务失败：{e}")
总结与延伸 {#总结}
核心要点回顾
-
自动化思维：遇到重复性工作，第一反应应该是"能不能自动化"，而不是"忍一忍就过去了"
-
最小化起步：从一个最简单的场景开始，不要一开始就想着完美
-
代码即文档：写清楚注释，让别人（和未来的自己）能看懂
-
日志很重要：记录运行状态，出问题能快速定位
延伸学习路径
| 阶段 | 学习内容 | 推荐资源 |
|---|---|---|
| 入门 | Python基础、pandas入门 | 《Python编程：从入门到实践》 |
| 进阶 | 正则表达式、文件处理 | Python官方文档 |
| 实用 | 邮件处理、定时任务 | schedule/APScheduler文档 |
| 高级 | 异常检测、机器学习 | scikit-learn官方教程 |
相关工具推荐
- 数据处理：pandas、openpyxl、xlrd/xlwt
- 中文处理：jieba、pkuseg、HanLP
- 邮件处理：yagmail、zmail
- 定时任务：schedule、APScheduler
- 日志管理：logging、loguru
- 配置文件：python-dotenv、pyyaml
完整源码获取
关注我的CSDN博客，回复"自动化"获取本文完整源码。
互动环节
你在工作中有没有遇到特别烦人的重复性工作？有没有想过用代码来解决？
欢迎在评论区分享，我们一起探讨更好的解决方案！
相关技术标签：
#Python #办公自动化 #pandas #数据分析 #定时任务 #效率提升 #办公效率 #自动化办公 #Python办公 #CSDN
本文首发于CSDN博客，原创不易，帮忙点个赞再走～
如需转载，请联系作者获得授权，并在文章头部注明原文链接。
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
12
0- 0
已为社区贡献1条内容
所有评论(0)