文件内容
scripts/communication_analyzer.py
#!/usr/bin/env python3
"""
亲子沟通分析器
Parent-Child Communication Analyzer
分析亲子对话模式，识别沟通问题，提供改进建议。
"""
import json
import os
import re
from datetime import datetime
from typing import Dict, List, Optional, Tuple
from collections import defaultdict
class CommunicationPattern:
"""沟通模式识别"""
# 负面沟通模式
NEGATIVE_PATTERNS = {
"criticism": {
"name": "批评",
"indicators": ["你总是", "你从来不", "你真", "你怎么又", "你怎么这么"],
"severity": 3
},
"contempt": {
"name": "轻蔑",
"indicators": ["真笨", "没出息", "像谁", "无语了", "服了"],
"severity": 4
},
"defensive": {
"name": "防御",
"indicators": ["那是因为", "我又没", "我不是说了", "那不是我的错"],
"severity": 2
},
"stonewalling": {
"name": "冷战",
"indicators": ["随便", "嗯", "哦", "行吧", "知道了", "不说话"],
"severity": 3
}
}
# 正面沟通模式
POSITIVE_PATTERNS = {
"empathy": {
"name": "共情",
"indicators": ["我理解你的感受", "我知道你很难过", "我理解你为什么"],
"weight": 2
},
"validation": {
"name": "肯定",
"indicators": ["你说得有道理", "我听到了", "我理解你的意思"],
"weight": 1
},
"solution": {
"name": "解决问题",
"indicators": ["我们一起想想", "有什么办法", "你觉得怎么做好"],
"weight": 2
},
"soft_start": {
"name": "柔和开场",
"indicators": ["我想和你聊聊", "我有个想法", "我们可以谈谈吗"],
"weight": 1
}
}
class CommunicationAnalyzer:
"""沟通分析器"""
def __init__(self, data_dir: str = None):
if data_dir is None:
data_dir = os.path.join(os.path.dirname(__file__), '..', 'data')
self.data_dir = os.path.abspath(data_dir)
os.makedirs(self.data_dir, exist_ok=True)
self.records_file = os.path.join(self.data_dir, 'communication_records.json')
def analyze_text(self, text: str) -> Dict:
"""分析单条文本"""
text_lower = text.lower()
results = {
"negative_patterns": [],
"positive_patterns": [],
"total_negative_score": 0,
"total_positive_score": 0,
"analysis": {}
}
# 检测负面模式
for pattern_id, pattern in CommunicationPattern.NEGATIVE_PATTERNS.items():
matches = []
for indicator in pattern["indicators"]:
if indicator in text_lower:
matches.append(indicator)
if matches:
results["negative_patterns"].append({
"pattern_id": pattern_id,
"name": pattern["name"],
"matches": matches,
"severity": pattern["severity"]
})
results["total_negative_score"] += pattern["severity"]
# 检测正面模式
for pattern_id, pattern in CommunicationPattern.POSITIVE_PATTERNS.items():
matches = []
for indicator in pattern["indicators"]:
if indicator in text:
matches.append(indicator)
if matches:
results["positive_patterns"].append({
"pattern_id": pattern_id,
"name": pattern["name"],
"matches": matches,
"weight": pattern["weight"]
})
results["total_positive_score"] += pattern["weight"]
# 综合评分
if results["total_negative_score"] > 0 or results["total_positive_score"] > 0:
ratio = results["total_positive_score"] / (results["total_positive_score"] + results["total_negative_score"])
results["analysis"]["health_ratio"] = round(ratio * 100, 1)
results["analysis"]["recommendation"] = self._get_recommendation(ratio)
else:
results["analysis"]["health_ratio"] = 50
results["analysis"]["recommendation"] = "未检测到明显的沟通模式，继续保持"
return results
def _get_recommendation(self, ratio: float) -> str:
"""根据健康比例给出建议"""
if ratio >= 0.8:
return "优秀的沟通模式！继续保持"
elif ratio >= 0.6:
return "良好的沟通，可以继续改进"
elif ratio >= 0.4:
return "需要关注和改进沟通方式"
else:
return "建议学习非暴力沟通技巧"
def analyze_conversation(self, conversation: List[Dict]) -> Dict:
"""分析一段对话"""
if not conversation:
return {"message": "对话记录为空"}
results = {
"total_messages": len(conversation),
"speaker_stats": defaultdict(lambda: {"messages": 0, "negative_score": 0, "positive_score": 0}),
"pattern_counts": defaultdict(int),
"overall_ratio": 0,
"messages": []
}
total_positive = 0
total_negative = 0
for msg in conversation:
speaker = msg.get("speaker", "unknown")
text = msg.get("text", "")
results["speaker_stats"][speaker]["messages"] += 1
analysis = self.analyze_text(text)
msg["analysis"] = analysis
results["messages"].append(msg)
results["speaker_stats"][speaker]["negative_score"] += analysis["total_negative_score"]
results["speaker_stats"][speaker]["positive_score"] += analysis["total_positive_score"]
total_positive += analysis["total_positive_score"]
total_negative += analysis["total_negative_score"]
for pattern in analysis["negative_patterns"]:
results["pattern_counts"][pattern["name"]] += 1
for pattern in analysis["positive_patterns"]:
results["pattern_counts"][pattern["name"]] += 1
# 计算总体健康比例
if total_positive + total_negative > 0:
results["overall_ratio"] = round(total_positive / (total_positive + total_negative) * 100, 1)
return results
def save_record(self, record: Dict, filename: str = None) -> str:
"""保存分析记录"""
if filename is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"comm_analysis_{timestamp}.json"
filepath = os.path.join(self.data_dir, filename)
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(record, f, ensure_ascii=False, indent=2)
return filepath
def interactive_analyze(self):
"""交互式分析"""
print("\n" + "="*50)
print("亲子沟通分析器")
print("="*50)
print("\n请输入对话内容（输入 'done' 结束输入）：")
print("-"*40)
conversation = []
speaker = "父母"
while True:
speaker = "父母" if speaker == "孩子" else "孩子"
print(f"\n[{speaker}]")
text = input().strip()
if text.lower() == "done":
break
if text:
conversation.append({
"speaker": speaker,
"text": text,
"timestamp": datetime.now().isoformat()
})
if not conversation:
print("没有输入对话内容")
return None
# 分析对话
print("\n" + "="*50)
print("分析结果")
print("="*50)
results = self.analyze_conversation(conversation)
self.print_analysis(results)
# 保存
filepath = self.save_record(results)
print(f"\n分析结果已保存至: {filepath}")
return results
def print_analysis(self, results: Dict):
"""打印分析结果"""
print(f"\n总消息数: {results['total_messages']}")
print(f"整体沟通健康度: {results['overall_ratio']}%")
print("\n【各发言人统计】:")
for speaker, stats in results["speaker_stats"].items():
pos = stats["positive_score"]
neg = stats["negative_score"]
ratio = pos / (pos + neg) * 100 if pos + neg > 0 else 50
print(f" {speaker}: {stats['messages']}条消息, 健康度{ratio:.0f}%")
if results["pattern_counts"]:
print("\n【沟通模式统计】:")
for pattern, count in sorted(results["pattern_counts"].items(), key=lambda x: x[1], reverse=True):
bar = "█" * min(count, 20)
print(f" {pattern}: {count}次 {bar}")
# 打印消息分析
print("\n【逐条分析】:")
for i, msg in enumerate(results["messages"], 1):
speaker = msg["speaker"]
text = msg["text"][:50] + "..." if len(msg["text"]) > 50 else msg["text"]
analysis = msg["analysis"]
print(f"\n {i}. [{speaker}] {text}")
if analysis["negative_patterns"]:
neg_names = [p["name"] for p in analysis["negative_patterns"]]
print(f" ⚠️ 负面: {', '.join(neg_names)}")
if analysis["positive_patterns"]:
pos_names = [p["name"] for p in analysis["positive_patterns"]]
print(f" ✅ 正面: {', '.join(pos_names)}")
print(f"\n【建议】: {results.get('analysis', {}).get('recommendation', 'N/A')}")
def main():
"""主函数"""
analyzer = CommunicationAnalyzer()
return analyzer.interactive_analyze()
if __name__ == "__main__":
main()