情绪记录分析程序 - 智能决策课程实践一、实际应用场景描述作为一名全栈开发工程师我在过去三年中经历了多个高强度项目周期。长期的技术攻坚和团队协作让我意识到情绪管理对工作效率和个人健康至关重要。典型场景- 周一晨会前感到焦虑因为担心任务分配不均- 代码评审通过时感到成就感但被同事指出bug时感到挫败- 项目上线成功后兴奋但遇到线上故障时沮丧- 每天晚上回顾时想不起今天情绪波动的具体原因基于智能决策课程中的决策树算法、权重计算和模式识别理论我设计了一个情绪记录分析系统帮助用户量化情绪变化识别关键影响因素并提供科学的调节建议。二、引入痛点现有解决方案的问题痛点 传统方案缺陷记录碎片化 纸质日记容易丢失手机备忘录难以统计分析缺乏量化分析 主观描述无法计算影响力权重无智能建议 需要自己查找调节方法效率低下数据孤立 情绪与具体事件、环境因素割裂无法找关联无趋势预测 不能根据历史数据预测情绪风险点我们的解决方案- ✅ 结构化数据存储支持多维度分析- ✅ 基于智能决策的权重计算模型- ✅ 自动生成个性化调节建议- ✅ 可视化趋势图表- ✅ 智能预警高风险时段三、核心逻辑讲解智能决策算法架构┌─────────────────────────────────────────────────────────────┐│ 情绪记录分析系统 │├─────────────────────────────────────────────────────────────┤│ 数据采集层 ││ ├── 情绪评分 (1-10分) ││ ├── 触发事件分类 (工作/人际/健康/环境/其他) ││ └── 环境因素 (睡眠/天气/咖啡因等) │├─────────────────────────────────────────────────────────────┤│ 智能决策层 ││ ├── 决策树分析: 构建情绪影响决策树 ││ ├── 权重计算: 使用信息增益计算各因素影响力 ││ ├── 关联规则: Apriori算法挖掘频繁模式 ││ └── 聚类分析: K-means识别相似情境的情绪模式 │├─────────────────────────────────────────────────────────────┤│ 输出层 ││ ├── 影响因素排行榜 ││ ├── 个性化调节建议 ││ └── 可视化分析报告 │└─────────────────────────────────────────────────────────────┘核心算法详解1. 权重计算基于信息增益信息增益 H(情绪分布) - Σ [P(因子值) × H(该因子值下的情绪分布)]2. 决策树分裂准则选择使信息增益最大的特征作为分裂节点递归构建决策树。3. 关联规则挖掘使用Apriori算法发现如睡眠不足 高压工作日 → 负面情绪的模式。四、代码模块化实现项目结构emotion_analyzer/│├── main.py # 程序入口├── config.py # 配置文件├── data_models.py # 数据模型定义├── emotion_recorder.py # 情绪记录模块├── analyzer_engine.py # 智能分析引擎├── suggestion_generator.py # 建议生成器├── visualization.py # 可视化模块├── utils.py # 工具函数├── storage.py # 数据存储模块├── requirements.txt # 依赖包└── README.md # 说明文档核心代码文件config.py - 配置文件配置文件 - 情绪记录分析系统包含系统常量、分类定义和阈值设置from dataclasses import dataclassfrom typing import List, Dictdataclassclass EmotionConfig:情绪配置类# 情绪评分范围MIN_EMOTION_SCORE: int 1MAX_EMOTION_SCORE: int 10# 情绪标签映射EMOTION_LABELS: Dict[int, str] Nonedef __post_init__(self):self.EMOTION_LABELS {1: 极度负面, 2: 很负面, 3: 负面, 4: 略负面,5: 中性,6: 略正面, 7: 正面, 8: 很正面, 9: 极正面, 10: 狂喜}dataclassclass EventCategories:触发事件分类WORK: List[str] NoneINTERPERSONAL: List[str] NoneHEALTH: List[str] NoneENVIRONMENT: List[str] NoneOTHER: List[str] Nonedef __post_init__(self):self.WORK [任务压力, 截止日期, 会议, 代码评审,项目进度, 职场冲突, 晋升/考核, 技术难题]self.INTERPERSONAL [家庭关系, 朋友交往, 恋爱关系, 社交活动,人际冲突, 被认可, 被批评, 孤独感]self.HEALTH [睡眠质量, 运动情况, 饮食状况, 身体疼痛,疲劳程度, 疾病, 精力水平, 休息充分度]self.ENVIRONMENT [天气, 噪音, 光线, 温度,通勤, 办公环境, 居住条件, 经济压力]self.OTHER [兴趣爱好, 娱乐活动, 学习成长, 意外事件, 回忆, 期待]dataclassclass AnalysisThresholds:分析阈值配置HIGH_IMPACT_THRESHOLD: float 0.15 # 高影响力阈值LOW_EMOTION_WARNING: int 4 # 负面情绪警告线POSITIVE_EMOTION_TARGET: int 7 # 正面情绪目标线RISK_PATTERN_THRESHOLD: int 3 # 风险模式出现次数阈值# 全局配置实例EMOTION_CONFIG EmotionConfig()EVENT_CATEGORIES EventCategories()THRESHOLDS AnalysisThresholds()# 调节建议数据库SUGGESTIONS_DB {work: {high_stress: [尝试番茄工作法25分钟专注5分钟休息,将大任务拆解为可执行的小步骤,主动向领导沟通任务优先级和资源需求,下班后进行15分钟冥想放松大脑],code_review: [提前进行自我代码审查减少被批评焦虑,将反馈视为学习机会而非个人攻击,建立代码质量标准清单降低不确定性],deadline: [制定详细的时间缓冲计划,提前识别潜在风险点并制定预案,学会合理拒绝不合理的截止日期要求]},interpersonal: {conflict: [使用我感受陈述法代替指责式沟通,暂停对话冷静后再处理争议,寻求双赢解决方案而非零和博弈],loneliness: [每周安排至少一次有意义的社交活动,加入兴趣社群寻找志同道合的伙伴,培养独处时的充实感减少对他人的过度依赖],recognition: [记录每日成就建立正向反馈循环,主动向信任的人分享成功经历,设定可实现的小目标积累成就感]},health: {poor_sleep: [建立固定睡眠时间表即使周末也保持一致,睡前1小时避免使用电子设备,创造舒适的睡眠环境黑暗、安静、凉爽],fatigue: [评估工作量是否合理必要时调整优先级,每90分钟工作中插入短暂休息,检查是否存在贫血、甲状腺等潜在健康问题],lack_exercise: [从每天10分钟轻度运动开始逐步增加,选择喜欢的运动方式提高坚持概率,利用碎片时间进行伸展运动]},environment: {bad_weather: [准备室内替代活动计划,使用明亮灯光改善室内光照,播放喜欢的音乐改善心情],noise: [使用降噪耳机或白噪音掩盖干扰,与相关人员协商安静工作时间,调整工作安排到相对安静的时段]}}data_models.py - 数据模型定义数据模型定义模块使用Python dataclass定义核心数据结构from dataclasses import dataclass, fieldfrom datetime import datetimefrom typing import Optional, List, Dictimport uuiddataclassclass TriggerEvent:触发事件数据模型属性:category: 事件类别 (work/interpersonal/health/environment/other)sub_category: 子类别名称description: 详细描述intensity: 强度评分 (1-10)is_positive: 是否为正面触发因素category: strsub_category: strdescription: strintensity: int 5 # 默认中等强度is_positive: bool False # 默认为负面触发def __post_init__(self):# 数据验证if not 1 self.intensity 10:raise ValueError(f强度必须在1-10之间当前值: {self.intensity})valid_categories [work, interpersonal, health, environment, other]if self.category not in valid_categories:raise ValueError(f无效类别: {self.category}有效值为: {valid_categories})dataclassclass EnvironmentalFactors:环境因素数据模型属性:sleep_hours: 睡眠时长 (小时)sleep_quality: 睡眠质量评分 (1-10)weather: 天气状况caffeine_intake: 咖啡因摄入量 (mg)exercise_minutes: 运动时长 (分钟)social_interaction: 社交互动时长 (分钟)sleep_hours: float 7.0sleep_quality: int 5weather: str 晴朗caffeine_intake: int 200 # mgexercise_minutes: int 30social_interaction: int 60 # minutesdef __post_init__(self):# 数据验证if not 0 self.sleep_hours 24:raise ValueError(f睡眠时长必须在0-24小时之间)if not 1 self.sleep_quality 10:raise ValueError(f睡眠质量必须在1-10之间)dataclassclass EmotionRecord:情绪记录数据模型 - 核心数据结构属性:record_id: 唯一标识符timestamp: 记录时间戳emotion_score: 情绪评分 (1-10)emotion_label: 情绪标签trigger_events: 触发事件列表environmental_factors: 环境因素notes: 备注信息tags: 自定义标签emotion_score: inttrigger_events: List[TriggerEvent]environmental_factors: EnvironmentalFactorstimestamp: datetime field(default_factorydatetime.now)record_id: str field(default_factorylambda: str(uuid.uuid4())[:8])emotion_label: str notes: str tags: List[str] field(default_factorylist)def __post_init__(self):# 数据验证from config import EMOTION_CONFIGif not EMOTION_CONFIG.MIN_EMOTION_SCORE self.emotion_score EMOTION_CONFIG.MAX_EMOTION_SCORE:raise ValueError(f情绪评分必须在{EMOTION_CONFIG.MIN_EMOTION_SCORE}-{EMOTION_CONFIG.MAX_EMOTION_SCORE}之间)# 自动设置情绪标签self.emotion_label EMOTION_CONFIG.EMOTION_LABELS.get(self.emotion_score, 未知)# 如果没有触发事件添加默认记录if not self.trigger_events:self.trigger_events [TriggerEvent(other, 未明确, 未记录具体触发因素)]def to_dict(self) - Dict:转换为字典格式便于存储和分析return {record_id: self.record_id,timestamp: self.timestamp.isoformat(),emotion_score: self.emotion_score,emotion_label: self.emotion_label,trigger_events: [{category: e.category,sub_category: e.sub_category,description: e.description,intensity: e.intensity,is_positive: e.is_positive}for e in self.trigger_events],environmental_factors: {sleep_hours: self.environmental_factors.sleep_hours,sleep_quality: self.environmental_factors.sleep_quality,weather: self.environmental_factors.weather,caffeine_intake: self.environmental_factors.caffeine_intake,exercise_minutes: self.environmental_factors.exercise_minutes,social_interaction: self.environmental_factors.social_interaction},notes: self.notes,tags: self.tags}classmethoddef from_dict(cls, data: Dict) - EmotionRecord:从字典创建记录对象env_factors EnvironmentalFactors(**data[environmental_factors])trigger_events [TriggerEvent(**e) for e in data[trigger_events]]return cls(record_iddata[record_id],timestampdatetime.fromisoformat(data[timestamp]),emotion_scoredata[emotion_score],emotion_labeldata[emotion_label],trigger_eventstrigger_events,environmental_factorsenv_factors,notesdata.get(notes, ),tagsdata.get(tags, []))dataclassclass AnalysisResult:分析结果数据模型属性:top_influencers: 影响力最高的因素列表correlation_matrix: 相关性矩阵risk_patterns: 识别出的风险模式suggestions: 个性化调节建议trend_analysis: 趋势分析结果confidence_score: 分析置信度top_influencers: List[Dict]correlation_matrix: Dict[str, Dict[str, float]]risk_patterns: List[Dict]suggestions: List[Dict]trend_analysis: Dictconfidence_score: float 0.0def to_summary(self) - str:生成结果摘要summary_lines [ * 50, 情绪分析结果摘要, * 50]# 影响力因素summary_lines.append(\n 主要影响因素 TOP 5:)for i, influencer in enumerate(self.top_influencers[:5], 1):summary_lines.append(f {i}. {influencer[factor]}: f影响力{influencer[impact_weight]:.3f}, f出现频次{influencer[frequency]})# 风险模式if self.risk_patterns:summary_lines.append(\n⚠️ 识别的风险模式:)for pattern in self.risk_patterns:summary_lines.append(f • {pattern[pattern]}: f风险等级{pattern[risk_level]})# 建议数量summary_lines.append(f\n 生成调节建议: {len(self.suggestions)} 条)return \n.join(summary_lines)storage.py - 数据存储模块数据存储模块支持JSON文件存储可扩展为SQLite/PostgreSQLimport jsonimport osfrom datetime import datetimefrom typing import List, Optionalfrom data_models import EmotionRecord, AnalysisResultimport configclass EmotionStorage:情绪数据存储类功能:- 保存单条记录- 批量加载历史记录- 删除记录- 数据备份与恢复def __init__(self, storage_path: str emotion_data.json):初始化存储Args:storage_path: 数据存储文件路径self.storage_path storage_pathself._ensure_storage_exists()def _ensure_storage_exists(self):确保存储文件存在if not os.path.exists(self.storage_path):self._save_data([])print(f✅ 创建新的数据存储文件: {self.storage_path})def _load_data(self) - List[dict]:加载所有数据try:with open(self.storage_path, r, encodingutf-8) as f:data json.load(f)return data if isinstance(data, list) else []except json.JSONDecodeError:print(f⚠️ 数据文件损坏创建新文件)return []except Exception as e:print(f❌ 读取数据失败: {e})return []def _save_data(self, data: List[dict]):保存数据到文件try:with open(self.storage_path, w, encodingutf-8) as f:json.dump(data, f, ensure_asciiFalse, indent2)except Exception as e:print(f❌ 保存数据失败: {e})raisedef save_record(self, record: EmotionRecord) - bool:保存单条情绪记录Args:record: 情绪记录对象Returns:是否保存成功try:data self._load_data()# 检查是否已存在相同ID的记录existing_ids [item.get(record_id) for item in data]if record.record_id in existing_ids:print(f⚠️ 记录 {record.record_id} 已存在跳过保存)return Falsedata.append(record.to_dict())self._save_data(data)print(f✅ 记录保存成功: ID{record.record_id}, 情绪{record.emotion_label})return Trueexcept Exception as e:print(f❌ 保存记录失败: {e})return Falsedef load_all_records(self) - List[EmotionRecord]:加载所有历史记录Returns:情绪记录对象列表raw_data self._load_data()records []for item in raw_data:try:record EmotionRecord.from_dict(item)records.append(record)except Exception as e:print(f⚠️ 解析记录失败 (ID{item.get(record_id, unknown)}): {e})continueprint(f 成功加载 {len(records)} 条历史记录)return recordsdef load_records_by_date_range(self,start_date: datetime,end_date: datetime) - List[EmotionRecord]:按日期范围加载记录Args:start_date: 开始日期end_date: 结束日期Returns:指定日期范围内的记录列表all_records self.load_all_records()filtered [r for r in all_recordsif start_date r.timestamp end_date]print(f 日期范围 {start_date.date()} ~ {end_date.date()}: 找到 {len(filtered)} 条记录)return filtereddef delete_record(self, record_id: str) - bool:删除指定记录Args:record_id: 记录IDReturns:是否删除成功data self._load_data()original_count len(data)data [item for item in data if item.get(record_id) ! record_id]if len(data) original_count:self._save_data(data)print(f️ 记录 {record_id} 已删除)return Trueelse:print(f⚠️ 未找到记录 {record_id})return Falsedef get_statistics(self) - dict:获取基本统计信息Returns:统计信息字典records self.load_all_records()if not records:return {total_records: 0}scores [r.emotion_score for r in records]return {total_records: len(records),date_range: {earliest: min(r.timestamp for r in records).strftime(%Y-%m-%d),latest: max(r.timestamp for r in records).strftime(%Y-%m-%d)},emotion_stats: {average: round(sum(scores) / len(scores), 2),min: min(scores),max: max(scores),positive_ratio: round(len([s for s in scores if s 6]) / len(scores) * 100, 1)}}def backup_data(self, backup_path: Optional[str] None) - str:备份数据Args:backup_path: 备份文件路径默认添加时间戳Returns:备份文件路径if backup_path is None:timestamp datetime.now().strftime(%Y%m%d_%H%M%S)backup_path femotion_backup_{timestamp}.jsondata self._load_data()with open(backup_path, w, encodingutf-8) as f:json.dump(data, f, ensure_asciiFalse, indent2)print(f 数据已备份至: {backup_path})return backup_pathemotion_recorder.py - 情绪记录模块情绪记录模块提供交互式界面收集用户输入的情绪数据from datetime import datetimefrom typing import Optionalfrom data_models import EmotionRecord, TriggerEvent, EnvironmentalFactorsfrom storage import EmotionStoragefrom config import EVENT_CATEGORIES, EMOTION_CONFIG, THRESHOLDSclass EmotionRecorder:情绪记录器类功能:- 交互式记录情绪数据- 验证输入有效性- 保存到存储系统def __init__(self, storage: Optional[EmotionStorage] None):初始化记录器Args:storage: 数据存储实例默认创建新实例self.storage storage or EmotionStorage()self.current_record: Optional[EmotionRecord] Nonedef start_interactive_session(self) - Optional[EmotionRecord]:启动交互式记录会话Returns:创建的记录对象如果取消则返回Noneprint(\n * 60)print( 情绪记录系统 - 开始新记录)print( * 60)try:# 1. 记录情绪评分emotion_score self._get_emotion_score()if emotion_score is None:return None# 2. 记录触发事件trigger_events self._get_trigger_events()# 3. 记录环境因素env_factors self._get_environmental_factors()# 4. 记录备注和标签notes self._get_notes()tags self._get_tags()# 5. 创建记录对象self.current_record EmotionRecord(emotion_scoreemotion_score,trigger_eventstrigger_events,environmental_factorsenv_factors,notesnotes,tagstags)# 6. 显示记录摘要self._display_record_summary()# 7. 确认保存if self._confirm_save():if self.storage.save_record(self.current_record):print(\n 记录已成功保存)return self.current_recordelse:print(\n❌ 保存失败请重试)return Noneelse:print(\n⏹️ 记录已取消未保存)return Noneexcept KeyboardInterrupt:print(\n\n⏹️ 用户中断记录取消)return Noneexcept Exception as e:print(f\n❌ 记录过程出错: {e})return Nonedef _get_emotion_score(self) - Optional[int]:获取情绪评分print(\n--- 第一步情绪评分 ---)print(请为当前情绪打分 (1-10分):)print(1极度负面, 5中性, 10狂喜)print(利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛