Python文本分析实战：基于Pandas的团队发言频率统计与可视化

最近在整理团队训练数据时发现一个很有意思的需求如何系统化统计某个成员在特定时间段内的发言频率这种需求在团队管理、社群运营中都很常见。本文将以统计2026上半年训练中善禹的发言次数为实际案例完整讲解从数据收集到分析可视化的全流程解决方案。无论你是团队负责人需要了解成员参与度还是开发者需要处理类似的统计需求这套方案都能直接复用。我们将使用Python作为主要工具涵盖数据预处理、文本分析、结果统计和可视化展示四个核心环节。1. 需求分析与技术选型1.1 业务场景理解在实际的团队训练场景中统计特定成员的发言次数可以帮助管理者评估成员的参与活跃度分析训练内容的互动情况为后续训练计划提供数据支持发现潜在的沟通问题1.2 技术方案设计针对寻雨2026上半年成训喊过多少次善禹这个具体需求我们需要解决以下几个技术问题数据来源训练记录可能存在于聊天记录、会议纪要、语音转文字文本等不同格式中时间范围筛选需要准确识别2026年上半年的时间范围2026-01-01至2026-06-30关键词识别如何准确识别善禹这个关键词包括可能的别名、错别字等情况统计精度确保统计的准确性和去重逻辑1.3 工具栈选择数据处理Pandas NumPy文本分析正则表达式 jieba分词中文场景可视化Matplotlib Seaborn数据存储CSV/Excel文件适用于中小规模数据2. 环境准备与数据规范2.1 开发环境配置# 所需库安装命令 pip install pandas numpy matplotlib seaborn jieba openpyxl2.2 数据格式标准化在实际项目中训练记录可能来自多个渠道。我们需要先统一数据格式# 标准化的数据结构示例 import pandas as pd from datetime import datetime # 定义标准数据格式 sample_data { timestamp: [2026-01-15 09:30:00, 2026-02-20 14:15:00, 2026-03-10 10:00:00], speaker: [张三, 李四, 王五], content: [今天善禹提出了很好的建议, 我觉得善禹的方案可行, 善禹上次说的那个问题], session_id: [train_20260115, train_20260220, train_20260310] } df pd.DataFrame(sample_data) print(df.head())2.3 时间范围处理# 时间范围定义和验证函数 def validate_time_range(start_date, end_date): 验证时间范围的合理性 try: start datetime.strptime(start_date, %Y-%m-%d) end datetime.strptime(end_date, %Y-%m-%d) if start end: raise ValueError(开始日期不能晚于结束日期) if (end - start).days 365: print(警告时间范围超过一年请确认统计需求) return start, end except ValueError as e: print(f日期格式错误{e}) return None, None # 2026年上半年时间范围 start_2026h1 2026-01-01 end_2026h1 2026-06-30 start_dt, end_dt validate_time_range(start_2026h1, end_2026h1)3. 核心统计逻辑实现3.1 关键词匹配算法针对中文名称的匹配需要考虑多种情况import re import jieba class NameMatcher: def __init__(self, target_name): self.target_name target_name self.patterns self._generate_patterns() def _generate_patterns(self): 生成多种匹配模式 patterns [] # 精确匹配 patterns.append(re.compile(r\b re.escape(self.target_name) r\b)) # 忽略大小写 patterns.append(re.compile(re.escape(self.target_name), re.IGNORECASE)) # 分词匹配针对中文 words jieba.lcut(self.target_name) if len(words) 1: pattern_str |.join([re.escape(word) for word in words]) patterns.append(re.compile(pattern_str)) return patterns def count_mentions(self, text): 统计文本中目标名称的出现次数 if not isinstance(text, str): return 0 total_count 0 for pattern in self.patterns: matches pattern.findall(text) total_count len(matches) return total_count # 使用示例 matcher NameMatcher(善禹) test_text 善禹今天提出了很好的建议善禹的方案很优秀 print(f出现次数{matcher.count_mentions(test_text)})3.2 时间过滤与数据筛选def filter_data_by_time(df, start_date, end_date, time_columntimestamp): 按时间范围筛选数据 # 确保时间列是datetime类型 df[time_column] pd.to_datetime(df[time_column]) # 筛选时间范围 mask (df[time_column] start_date) (df[time_column] end_date) filtered_df df[mask].copy() print(f原始数据量{len(df)}) print(f筛选后数据量{len(filtered_df)}) print(f时间范围{start_date} 至 {end_date}) return filtered_df def analyze_mentions(df, target_name, content_columncontent): 分析数据框中目标名称的提及情况 matcher NameMatcher(target_name) # 统计每条的提及次数 df[mention_count] df[content_column].apply(matcher.count_mentions) # 汇总统计 total_mentions df[mention_count].sum() mentioned_records df[df[mention_count] 0] stats { total_mentions: total_mentions, mentioned_records_count: len(mentioned_records), mention_rate: len(mentioned_records) / len(df) if len(df) 0 else 0, average_per_record: total_mentions / len(df) if len(df) 0 else 0 } return df, stats4. 完整实战案例4.1 模拟数据生成为了演示完整流程我们先生成模拟的训练数据import random from datetime import datetime, timedelta def generate_training_data(num_records1000): 生成模拟训练数据 speakers [张三, 李四, 王五, 赵六, 钱七, 孙八, 周九, 吴十] sessions [ftrain_20260{i}{j} for i in range(1, 7) for j in range(1, 29) if i ! 2 or j 28] data [] start_date datetime(2026, 1, 1) for i in range(num_records): # 随机生成时间2026年上半年 days_offset random.randint(0, 181) # 1月1日到6月30日 hours_offset random.randint(0, 23) minutes_offset random.randint(0, 59) timestamp start_date timedelta(daysdays_offset, hourshours_offset, minutesminutes_offset) speaker random.choice(speakers) session random.choice(sessions) # 生成内容有一定概率包含善禹 content_templates [ 今天讨论的主题是{}, 我觉得{}的方案不错, 关于{}提出的问题, 按照{}的建议执行, {}上次说的那个方法, 常规工作汇报, 项目进度更新, 技术方案讨论 ] template random.choice(content_templates) if {} in template and random.random() 0.3: # 30%概率包含名称 content template.format(善禹) else: content template.replace({}, 这个) data.append({ timestamp: timestamp.strftime(%Y-%m-%d %H:%M:%S), speaker: speaker, content: content, session_id: session }) return pd.DataFrame(data) # 生成测试数据 training_df generate_training_data(1500) print(生成的模拟数据示例) print(training_df.head())4.2 执行统计分析def comprehensive_analysis(df, target_name, start_date, end_date): 执行完整的统计分析 print( * 50) print(f开始分析{target_name}在{start_date}至{end_date}的提及情况) print( * 50) # 1. 时间筛选 filtered_df filter_data_by_time(df, start_date, end_date) if len(filtered_df) 0: print(指定时间范围内无数据) return None, None # 2. 提及分析 analyzed_df, stats analyze_mentions(filtered_df, target_name) # 3. 详细统计 print(\n 基本统计 ) print(f总记录数{len(filtered_df)}) print(f总提及次数{stats[total_mentions]}) print(f涉及记录数{stats[mentioned_records_count]}) print(f提及率{stats[mention_rate]:.2%}) print(f平均每记录提及{stats[average_per_record]:.2f}次) # 4. 按发言人统计 speaker_stats analyzed_df.groupby(speaker)[mention_count].agg([sum, count]).reset_index() speaker_stats speaker_stats.rename(columns{sum: mention_times, count: total_records}) speaker_stats[mention_ratio] speaker_stats[mention_times] / speaker_stats[total_records] print(\n 按发言人统计 ) print(speaker_stats.sort_values(mention_times, ascendingFalse)) # 5. 按月统计 analyzed_df[month] analyzed_df[timestamp].dt.month monthly_stats analyzed_df.groupby(month)[mention_count].sum().reset_index() print(\n 按月统计 ) print(monthly_stats) return analyzed_df, stats # 执行分析 result_df, final_stats comprehensive_analysis(training_df, 善禹, 2026-01-01, 2026-06-30)4.3 结果可视化import matplotlib.pyplot as plt import seaborn as sns def visualize_results(df, target_name): 可视化分析结果 plt.style.use(seaborn-v0_8) fig, axes plt.subplots(2, 2, figsize(15, 12)) # 1. 月度提及趋势 monthly_data df.groupby(df[timestamp].dt.month)[mention_count].sum() axes[0, 0].bar(monthly_data.index, monthly_data.values, colorskyblue, alpha0.7) axes[0, 0].set_title(f{target_name}月度提及趋势) axes[0, 0].set_xlabel(月份) axes[0, 0].set_ylabel(提及次数) # 2. 发言人提及分布 speaker_data df.groupby(speaker)[mention_count].sum().sort_values(ascendingFalse) axes[0, 1].barh(speaker_data.index[:10], speaker_data.values[:10], colorlightcoral) axes[0, 1].set_title(发言人提及排名Top 10) axes[0, 1].set_xlabel(提及次数) # 3. 提及时间分布 df[hour] df[timestamp].dt.hour hourly_data df.groupby(hour)[mention_count].sum() axes[1, 0].plot(hourly_data.index, hourly_data.values, markero, colorgreen) axes[1, 0].set_title(提及时间分布) axes[1, 0].set_xlabel(小时) axes[1, 0].set_ylabel(提及次数) # 4. 会话提及密度 session_data df.groupby(session_id)[mention_count].sum().sort_values(ascendingFalse) axes[1, 1].hist(session_data.values, bins20, colororange, alpha0.7) axes[1, 1].set_title(会话提及密度分布) axes[1, 1].set_xlabel(每次会话提及次数) axes[1, 1].set_ylabel(会话数量) plt.tight_layout() plt.savefig(f{target_name}_mention_analysis.png, dpi300, bbox_inchestight) plt.show() # 执行可视化 if result_df is not None: visualize_results(result_df, 善禹)5. 数据源适配与实战技巧5.1 常见数据源处理在实际项目中数据可能来自多种格式def load_data_from_sources(file_path, source_type): 从不同数据源加载数据 if source_type csv: df pd.read_csv(file_path) elif source_type excel: df pd.read_excel(file_path, engineopenpyxl) elif source_type json: df pd.read_json(file_path) else: raise ValueError(f不支持的数据源类型{source_type}) # 数据清洗和标准化 df standardize_dataframe(df) return df def standardize_dataframe(df): 标准化数据框格式 # 列名标准化 column_mapping { 时间: timestamp, 发言时间: timestamp, 发言人: speaker, 姓名: speaker, 内容: content, 发言内容: content, 会话ID: session_id, 会议ID: session_id } df df.rename(columnscolumn_mapping) # 确保必要列存在 required_columns [timestamp, speaker, content] for col in required_columns: if col not in df.columns: raise ValueError(f缺少必要列{col}) return df5.2 高级匹配技巧对于复杂的匹配需求可以进一步优化class AdvancedNameMatcher(NameMatcher): def __init__(self, target_name, aliasesNone): super().__init__(target_name) self.aliases aliases or [] self._extend_patterns() def _extend_patterns(self): 扩展匹配模式包括别名 for alias in self.aliases: self.patterns.append(re.compile(r\b re.escape(alias) r\b)) self.patterns.append(re.compile(re.escape(alias), re.IGNORECASE)) def fuzzy_match(self, text, threshold0.8): 模糊匹配基于编辑距离 from difflib import SequenceMatcher words jieba.lcut(text) matches 0 for word in words: # 检查与目标名称的相似度 similarity SequenceMatcher(None, word, self.target_name).ratio() if similarity threshold: matches 1 return matches # 使用高级匹配器 advanced_matcher AdvancedNameMatcher(善禹, aliases[善宇, 小雨, 小禹])6. 常见问题与解决方案6.1 数据质量问题处理def handle_data_quality_issues(df): 处理常见的数据质量问题 issues_found [] # 1. 处理空值 initial_count len(df) df df.dropna(subset[timestamp, speaker, content]) if len(df) initial_count: issues_found.append(f移除空值记录{initial_count - len(df)}条) # 2. 时间格式统一 try: df[timestamp] pd.to_datetime(df[timestamp]) except Exception as e: issues_found.append(f时间格式转换错误{e}) # 3. 去重处理 duplicate_count df.duplicated().sum() if duplicate_count 0: df df.drop_duplicates() issues_found.append(f移除重复记录{duplicate_count}条) # 4. 内容清洗 df[content] df[content].astype(str).str.strip() print(数据质量问题处理完成) for issue in issues_found: print(f - {issue}) return df6.2 性能优化建议当处理大规模数据时需要考虑性能优化def optimize_performance(df): 数据处理的性能优化 # 1. 使用更高效的数据类型 df[speaker] df[speaker].astype(category) df[session_id] df[session_id].astype(category) # 2. 索引优化 if timestamp in df.columns: df df.sort_values(timestamp).reset_index(dropTrue) # 3. 内存使用优化 df df.infer_objects() return df def batch_processing(large_df, batch_size10000): 分批处理大规模数据 results [] total_batches (len(large_df) // batch_size) 1 for i in range(total_batches): start_idx i * batch_size end_idx min((i 1) * batch_size, len(large_df)) batch_df large_df.iloc[start_idx:end_idx].copy() batch_result analyze_mentions(batch_df, 善禹) results.append(batch_result) print(f处理进度{i1}/{total_batches}) # 合并结果 final_df pd.concat([r[0] for r in results]) final_stats combine_statistics([r[1] for r in results]) return final_df, final_stats7. 生产环境最佳实践7.1 配置化管理将关键参数配置化便于维护# config.py class AnalysisConfig: TARGET_NAME 善禹 TIME_RANGE { start: 2026-01-01, end: 2026-06-30 } DATA_SOURCE { type: csv, # csv, excel, json path: data/training_records.csv } MATCHING { aliases: [善宇, 小雨], fuzzy_threshold: 0.8 } OUTPUT { report_path: reports/, visualization: True } # 使用配置 config AnalysisConfig()7.2 错误处理与日志记录import logging from datetime import datetime def setup_logging(): 设置日志记录 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(fanalysis_{datetime.now().strftime(%Y%m%d_%H%M%S)}.log), logging.StreamHandler() ] ) def safe_analysis(config): 带错误处理的安全分析 logger logging.getLogger(__name__) try: logger.info(开始数据分析流程) # 数据加载 df load_data_from_sources(config.DATA_SOURCE[path], config.DATA_SOURCE[type]) logger.info(f成功加载数据记录数{len(df)}) # 数据处理 df handle_data_quality_issues(df) # 执行分析 result_df, stats comprehensive_analysis( df, config.TARGET_NAME, config.TIME_RANGE[start], config.TIME_RANGE[end] ) logger.info(分析完成) return result_df, stats except Exception as e: logger.error(f分析过程出错{e}) raise7.3 结果导出与报告生成def generate_report(result_df, stats, config): 生成分析报告 report_content f # {config.TARGET_NAME}提及分析报告 ## 时间范围{config.TIME_RANGE[start]} 至 {config.TIME_RANGE[end]} ### 关键统计指标 - 总记录数{len(result_df)} - 总提及次数{stats[total_mentions]} - 提及率{stats[mention_rate]:.2%} - 平均每记录提及{stats[average_per_record]:.2f}次 ### 详细分析 此处包含详细的数据分析结果 # 保存报告 report_file f{config.OUTPUT[report_path]}report_{datetime.now().strftime(%Y%m%d_%H%M%S)}.md with open(report_file, w, encodingutf-8) as f: f.write(report_content) # 保存处理后的数据 result_file f{config.OUTPUT[report_path]}processed_data.csv result_df.to_csv(result_file, indexFalse, encodingutf-8-sig) return report_file, result_file通过这套完整的解决方案我们不仅能够准确统计善禹在2026年上半年的提及次数还能获得丰富的分析维度和可视化结果。这种方法可以轻松适配到其他类似的统计需求中为团队管理和决策提供数据支持。在实际应用中建议根据具体的数据特点和分析需求适当调整匹配算法和统计逻辑。特别是对于大规模数据或实时分析场景可以考虑引入数据库存储和更高效的流式处理方案。