221 lines
8.4 KiB
Python
221 lines
8.4 KiB
Python
import re
|
||
|
||
|
||
class Preprocessor:
|
||
"""
|
||
预处理器：从原始对话中提取包含故事的段落
|
||
|
||
核心思路：
|
||
1. 过滤纯课堂管理内容（签到、纪律等）
|
||
2. 识别"个案对话段"——老师与某位学员的深度互动
|
||
3. 保留完整对话上下文（师生互动），不过度过滤
|
||
4. 按话题边界切分为合理的段落
|
||
"""
|
||
|
||
# 课堂管理关键词（出现这些词的整行直接过滤）
|
||
MANAGEMENT_KEYWORDS = ["签到", "打卡", "禁言", "扣三个一", "进组", "会议室链接"]
|
||
|
||
# 纯课堂管理发言模式（整条消息都是管理内容）
|
||
MANAGEMENT_PATTERNS = [
|
||
re.compile(r"大家.*签到"),
|
||
re.compile(r"请.*进组"),
|
||
re.compile(r"会议室"),
|
||
re.compile(r"纪律"),
|
||
re.compile(r"昵称.*改为"),
|
||
re.compile(r"欢迎.*进入"),
|
||
]
|
||
|
||
# 老师关键词
|
||
TEACHER_KEYWORDS = ["卢慧老师", "静静", "督导"]
|
||
|
||
def is_teacher(self, speaker: str) -> bool:
|
||
"""判断是否为老师/助教"""
|
||
return any(kw in speaker for kw in self.TEACHER_KEYWORDS)
|
||
|
||
def is_management_line(self, line: dict) -> bool:
|
||
"""判断是否为课堂管理行"""
|
||
content = line.get("content", "")
|
||
# 短内容 + 管理关键词 = 管理行
|
||
if len(content) < 30:
|
||
return any(kw in content for kw in self.MANAGEMENT_KEYWORDS)
|
||
return any(p.search(content) for p in self.MANAGEMENT_PATTERNS)
|
||
|
||
def preprocess(self, paragraphs: list[dict]) -> list[dict]:
|
||
"""
|
||
完整预处理流程：
|
||
1. 过滤课堂管理
|
||
2. 识别个案对话段（保留师生互动）
|
||
3. 按主要讲述者拆分（每个段落只包含一个主要讲述者）
|
||
"""
|
||
# 第一步：过滤课堂管理行
|
||
filtered = [p for p in paragraphs if not self.is_management_line(p)]
|
||
|
||
# 第二步：识别个案对话段
|
||
segments = self._extract_story_segments(filtered)
|
||
|
||
# 第三步：按主要讲述者拆分
|
||
segments = self._split_by_speaker(segments)
|
||
|
||
return segments
|
||
|
||
def _split_by_speaker(self, segments: list[list[dict]]) -> list[list[dict]]:
|
||
"""
|
||
将每个段落按讲述者拆分成子段落。
|
||
同一个讲述者的所有对话（即使中间隔了其他人的插话）收集在一起，
|
||
老师发言分配给前一个活跃的学员作为上下文。
|
||
"""
|
||
result = []
|
||
for seg in segments:
|
||
# 按讲述者收集对话，保留老师发言作为上下文
|
||
speaker_blocks = {} # speaker -> [lines]
|
||
last_student = None
|
||
|
||
for line in seg:
|
||
speaker = line.get("speaker", "")
|
||
if self.is_teacher(speaker):
|
||
# 老师发言归入前一个活跃的学员
|
||
if last_student and last_student in speaker_blocks:
|
||
speaker_blocks[last_student].append(line)
|
||
else:
|
||
if speaker not in speaker_blocks:
|
||
speaker_blocks[speaker] = []
|
||
speaker_blocks[speaker].append(line)
|
||
last_student = speaker
|
||
|
||
# 每个讲述者的对话作为一个子段落
|
||
for speaker, lines in speaker_blocks.items():
|
||
student_content = sum(
|
||
len(l.get("content", ""))
|
||
for l in lines
|
||
if not self.is_teacher(l.get("speaker", ""))
|
||
)
|
||
# 过滤掉内容太少的（可能是插话而非讲述者）
|
||
if student_content > 80 and len(lines) >= 3:
|
||
# 计算该讲述者在自己段落中的字数占比
|
||
# 如果老师发言占比超过50%，说明这主要是老师在讲课，不是学员故事
|
||
teacher_chars = sum(
|
||
len(l.get("content", ""))
|
||
for l in lines
|
||
if self.is_teacher(l.get("speaker", ""))
|
||
)
|
||
total_chars = student_content + teacher_chars
|
||
if total_chars > 0 and teacher_chars / total_chars > 0.5:
|
||
continue # 老师讲太多，跳过
|
||
result.append(lines)
|
||
|
||
return result
|
||
|
||
def _extract_story_segments(self, paragraphs: list[dict]) -> list[list[dict]]:
|
||
"""
|
||
从过滤后的对话中提取可能包含故事的段落。
|
||
|
||
策略：
|
||
- 找到学员的"长发言"（>50字）作为故事线索
|
||
- 向前向后扩展，包含老师的引导和互动
|
||
- 直到遇到另一位学员的长发言或长时间沉默
|
||
- 最小段落 10 行，最大段落 200 行
|
||
"""
|
||
segments = []
|
||
i = 0
|
||
used = set() # 已分配的行索引
|
||
|
||
while i < len(paragraphs):
|
||
p = paragraphs[i]
|
||
|
||
# 跳过老师单独发言（非互动）
|
||
if self.is_teacher(p.get("speaker", "")):
|
||
i += 1
|
||
continue
|
||
|
||
# 跳过短发言（< 30字不太可能开启故事）
|
||
if len(p.get("content", "")) < 30:
|
||
i += 1
|
||
continue
|
||
|
||
# 跳过已分配的行
|
||
if p.get("index", i) in used:
|
||
i += 1
|
||
continue
|
||
|
||
# 找到一个学员发言，开始扩展
|
||
segment_start = i
|
||
segment_end = i
|
||
|
||
# 向前扩展（找老师之前的引导）
|
||
expand_start = max(0, i - 5)
|
||
for j in range(i - 1, expand_start - 1, -1):
|
||
if j in used:
|
||
break
|
||
prev = paragraphs[j]
|
||
# 如果前一个是老师发言，包含
|
||
if self.is_teacher(prev.get("speaker", "")):
|
||
segment_start = j
|
||
# 如果前一个是同一学员的发言，包含
|
||
elif prev.get("speaker") == p.get("speaker"):
|
||
segment_start = j
|
||
else:
|
||
# 其他学员发言，停止
|
||
break
|
||
|
||
# 向后扩展（找老师的回应和后续互动）
|
||
current_speaker = p.get("speaker", "")
|
||
for j in range(i + 1, min(len(paragraphs), i + 200)):
|
||
if j in used:
|
||
segment_end = j - 1
|
||
break
|
||
|
||
curr = paragraphs[j]
|
||
curr_speaker = curr.get("speaker", "")
|
||
curr_content = curr.get("content", "")
|
||
|
||
# 老师发言：包含（互动引导）
|
||
if self.is_teacher(curr_speaker):
|
||
segment_end = j
|
||
continue
|
||
|
||
# 同一学员继续发言：包含
|
||
if curr_speaker == current_speaker:
|
||
segment_end = j
|
||
continue
|
||
|
||
# 其他学员短回应（< 20字）：可能是插话，跳过但不终止
|
||
if len(curr_content) < 20:
|
||
continue
|
||
|
||
# 其他学员长发言：新的话题开始，终止
|
||
break
|
||
|
||
# 提取段落
|
||
segment = paragraphs[segment_start:segment_end + 1]
|
||
|
||
# 过滤掉太短的段落（< 8行）
|
||
if len(segment) >= 8:
|
||
# 检查是否有足够的学员实质性内容（非纯疗愈练习）
|
||
student_content_len = sum(
|
||
len(s.get("content", ""))
|
||
for s in segment
|
||
if not self.is_teacher(s.get("speaker", ""))
|
||
and len(s.get("content", "")) > 15
|
||
)
|
||
|
||
# 学员实质内容 > 100字才认为是故事
|
||
if student_content_len > 100:
|
||
segments.append(segment)
|
||
for s in segment:
|
||
used.add(s.get("index", 0))
|
||
i = segment_end + 1
|
||
continue
|
||
|
||
i += 1
|
||
|
||
return segments
|
||
|
||
|
||
def is_teacher(speaker: str) -> bool:
|
||
"""判断是否为老师/助教发言"""
|
||
teacher_keywords = ["卢慧老师", "静静", "督导", "老师"]
|
||
return any(kw in speaker for kw in teacher_keywords)
|
||
|
||
|
||
preprocessor = Preprocessor()
|