aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/conventionalrp/core/thulac_parser.py523
1 files changed, 523 insertions, 0 deletions
diff --git a/src/conventionalrp/core/thulac_parser.py b/src/conventionalrp/core/thulac_parser.py
new file mode 100644
index 0000000..075c3bd
--- /dev/null
+++ b/src/conventionalrp/core/thulac_parser.py
@@ -0,0 +1,523 @@
+"""
+基于清华 THULAC 的智能解析器
+
+使用 THULAC (THU Lexical Analyzer for Chinese) 进行中文词法分析，
+自动识别 TRPG 日志中的对话、动作、旁白等内容类型，
+大幅简化规则配置。
+
+THULAC 是清华大学自然语言处理与社会人文计算实验室研制推出的
+一套中文词法分析工具包，具有中文分词和词性标注功能。
+
+词性标注说明：
+- n/名词 np/人名 ns/地名 ni/机构名 nz/其它专名
+- m/数词 q/量词 mq/数量词 t/时间词 f/方位词 s/处所词
+- v/动词 a/形容词 d/副词
+- h/前接成分 k/后接成分
+- i/习语 j/简称 r/代词 c/连词 p/介词
+- u/助词 y/语气助词 e/叹词 o/拟声词
+- g/语素 w/标点 x/其它
+"""
+
+import re
+from typing import List, Dict, Optional, Tuple
+import logging
+
+logger = logging.getLogger(__name__)
+
+try:
+ import thulac
+ THULAC_AVAILABLE = True
+except ImportError:
+ THULAC_AVAILABLE = False
+ logger.warning("THULAC not installed. Please install with: pip install thulac")
+
+
+class THULACParser:
+ """
+ 基于 THULAC 的智能 TRPG 日志解析器
+
+ 特性：
+ 1. 使用 THULAC 进行中文分词和词性标注
+ 2. 基于词性和上下文自动识别内容类型
+ 3. 极简规则配置（只需配置分隔符）
+ 4. 支持自定义词典扩展
+ """
+
+ # 默认分隔符配置（可通过 load_rules 覆盖）
+ DEFAULT_DELIMITERS = {
+ "dialogue": [
+ ('"', '"'), # 英文双引号
+ ('"', '"'), # 中文双引号
+ ("'", "'"), # 单引号
+ ],
+ "thought": [
+ ("【", "】"), # 中文方括号
+ ("[", "]"), # 英文方括号
+ ],
+ "action": [
+ ("（", "）"), # 中文括号
+ ("(", ")"), # 英文括号
+ ("*", "*"), # 星号
+ ("**", "**"), # 双星号
+ ],
+ "ooc": [
+ ("//", "\n"), # 双斜杠到行尾
+ (">>", "\n"), # 双右尖括号到行尾
+ ]
+ }
+
+ # THULAC 词性到内容类型的映射
+ POS_TYPE_MAPPING = {
+ # 动词相关 -> 动作
+ "v": "action", # 动词
+
+ # 名词相关 -> 旁白
+ "n": "narration", # 名词
+ "np": "narration", # 人名
+ "ns": "narration", # 地名
+ "ni": "narration", # 机构名
+ "nz": "narration", # 其它专名
+
+ # 代词 -> 对话（第一人称/第二人称倾向于对话）
+ "r": "dialogue", # 代词
+
+ # 副词/形容词 -> 旁白
+ "d": "narration", # 副词
+ "a": "narration", # 形容词
+
+ # 量词/数词 -> 旁白
+ "m": "narration", # 数词
+ "q": "narration", # 量词
+ "mq": "narration", # 数量词
+
+ # 时间/方位/处所 -> 旁白
+ "t": "narration", # 时间词
+ "f": "narration", # 方位词
+ "s": "narration", # 处所词
+
+ # 语气词/叹词 -> 对话
+ "y": "dialogue", # 语气助词
+ "e": "dialogue", # 叹词
+ }
+
+ # 动作动词关键词（优先级更高）
+ ACTION_VERBS = {
+ "走", "跑", "看", "听", "摸", "拿", "放", "打开", "关闭",
+ "推", "拉", "举", "扔", "跳", "爬", "坐", "站", "躺",
+ "进入", "离开", "接近", "远离", "转身", "回头", "低头", "抬头",
+ "微笑", "大笑", "哭", "喊", "叫", "说", "讲", "念", "读",
+ "投掷", "检定", "攻击", "防御", "躲避"
+ }
+
+ # 对话相关关键词
+ DIALOGUE_INDICATORS = {
+ "我", "你", "他", "她", "我们", "你们", "他们",
+ "吗", "呢", "啊", "哦", "嗯", "哼", "咦", "哎",
+ }
+
+ def __init__(self, seg_only: bool = False, user_dict: str = None):
+ """
+ 初始化 THULAC 解析器
+
+ Args:
+ seg_only: 是否只进行分词（不标注词性）
+ user_dict: 用户自定义词典路径
+ """
+ if not THULAC_AVAILABLE:
+ raise ImportError(
+ "THULAC is not installed. Please install with:\n"
+ "pip install thulac\n"
+ "Note: First installation may take a few minutes to download models."
+ )
+
+ self.seg_only = seg_only
+ self.thulac = thulac.thulac(
+ seg_only=seg_only,
+ user_dict=user_dict if user_dict else None
+ )
+
+ self.delimiters = self.DEFAULT_DELIMITERS.copy()
+ self.custom_words = {}
+ self.statistics = {
+ "total_parsed": 0,
+ "dialogue_count": 0,
+ "action_count": 0,
+ "thought_count": 0,
+ "narration_count": 0,
+ "ooc_count": 0,
+ }
+
+ logger.info(f"THULACParser initialized with seg_only={seg_only}")
+
+ def load_rules(self, rules_path: str = None, rules_dict: Dict = None):
+ """
+ 加载简化的规则配置
+
+ Args:
+ rules_path: 规则文件路径（JSON5 格式）
+ rules_dict: 直接传入规则字典
+
+ 规则格式示例：
+ {
+ "delimiters": {
+ "dialogue": [["\"", "\""], [""", """]],
+ "action": [["(", ")"], ["*", "*"]],
+ "thought": [["【", "】"]]
+ },
+ "custom_words": {
+ "骰子": "n",
+ "检定": "v",
+ "守秘人": "np"
+ }
+ }
+ """
+ import json5
+ from pathlib import Path
+
+ if rules_path:
+ if not Path(rules_path).exists():
+ raise FileNotFoundError(f"Rules file not found: {rules_path}")
+ with open(rules_path, "r", encoding="utf-8") as f:
+ rules_dict = json5.load(f)
+
+ if not rules_dict:
+ logger.info("No rules provided, using default delimiters")
+ return
+
+ # 加载分隔符配置
+ if "delimiters" in rules_dict:
+ for content_type, delimiter_pairs in rules_dict["delimiters"].items():
+ self.delimiters[content_type] = [tuple(pair) for pair in delimiter_pairs]
+
+ # 加载自定义词汇（词 -> 词性）
+ if "custom_words" in rules_dict:
+ self.custom_words = rules_dict["custom_words"]
+ logger.info(f"Loaded {len(self.custom_words)} custom words")
+
+ logger.info("Rules loaded successfully")
+
+ def _extract_delimited_content(self, text: str) -> List[Dict]:
+ """
+ 提取分隔符标记的内容
+
+ Returns:
+ List of {type, content, start, end, delimiter}
+ """
+ results = []
+
+ for content_type, delimiter_pairs in self.delimiters.items():
+ for start_delim, end_delim in delimiter_pairs:
+ # 转义正则特殊字符
+ start_pattern = re.escape(start_delim)
+ end_pattern = re.escape(end_delim)
+
+ # 处理到行尾的情况
+ if end_delim == "\n":
+ pattern = f"{start_pattern}(.+?)(?:\n|$)"
+ else:
+ pattern = f"{start_pattern}(.+?){end_pattern}"
+
+ for match in re.finditer(pattern, text):
+ results.append({
+ "type": content_type,
+ "content": match.group(1),
+ "start": match.start(),
+ "end": match.end(),
+ "delimiter": (start_delim, end_delim),
+ "confidence": 1.0 # 分隔符匹配的置信度为 100%
+ })
+
+ # 按位置排序
+ results.sort(key=lambda x: x["start"])
+ return results
+
+ def _analyze_with_thulac(self, text: str) -> List[Dict]:
+ """
+ 使用 THULAC 分析文本
+
+ Returns:
+ List of {type, content, words, tags, confidence}
+ """
+ # 使用 THULAC 进行分词和词性标注
+ result = self.thulac.cut(text, text=False) # 返回 [(word, pos), ...]
+
+ if not result:
+ return [{
+ "type": "narration",
+ "content": text,
+ "words": [],
+ "tags": [],
+ "confidence": 0.5,
+ "method": "thulac"
+ }]
+
+ # 分离词和词性
+ words = [item[0] for item in result]
+ tags = [item[1] for item in result]
+
+ # 应用自定义词性（如果有）
+ for i, word in enumerate(words):
+ if word in self.custom_words:
+ tags[i] = self.custom_words[word]
+
+ # 基于词性和内容推断类型
+ content_type = self._infer_content_type(words, tags)
+
+ # 计算置信度
+ confidence = self._calculate_confidence(words, tags, content_type)
+
+ return [{
+ "type": content_type,
+ "content": text,
+ "words": words,
+ "tags": tags,
+ "confidence": confidence,
+ "method": "thulac"
+ }]
+
+ def _infer_content_type(self, words: List[str], tags: List[str]) -> str:
+ """
+ 基于词性和内容推断内容类型
+
+ 策略：
+ 1. 检查是否包含动作动词 -> action
+ 2. 检查是否包含对话指示词 -> dialogue
+ 3. 统计主导词性 -> 按映射表判断
+ """
+ # 策略1: 检查动作动词
+ for word in words:
+ if word in self.ACTION_VERBS:
+ return "action"
+
+ # 策略2: 检查对话指示词
+ dialogue_indicators = sum(1 for w in words if w in self.DIALOGUE_INDICATORS)
+ if dialogue_indicators >= 2: # 至少2个对话指示词
+ return "dialogue"
+
+ # 策略3: 统计词性
+ pos_count = {}
+ for tag in tags:
+ if tag == "w": # 忽略标点
+ continue
+ pos_count[tag] = pos_count.get(tag, 0) + 1
+
+ if not pos_count:
+ return "narration"
+
+ # 找出最常见的词性
+ dominant_pos = max(pos_count.items(), key=lambda x: x[1])[0]
+
+ # 特殊规则：如果有动词，倾向于判断为动作
+ if "v" in pos_count and pos_count["v"] >= len(words) * 0.3:
+ return "action"
+
+ # 根据主导词性映射
+ return self.POS_TYPE_MAPPING.get(dominant_pos, "narration")
+
+ def _calculate_confidence(self, words: List[str], tags: List[str],
+ content_type: str) -> float:
+ """
+ 计算分析置信度
+
+ 基于以下因素：
+ 1. 词性标注的一致性
+ 2. 关键词匹配度
+ 3. 文本长度
+ """
+ if not words or not tags:
+ return 0.5
+
+ base_confidence = 0.5
+
+ # 因素1: 关键词匹配
+ if content_type == "action":
+ action_word_count = sum(1 for w in words if w in self.ACTION_VERBS)
+ if action_word_count > 0:
+ base_confidence += 0.3
+ elif content_type == "dialogue":
+ dialogue_word_count = sum(1 for w in words if w in self.DIALOGUE_INDICATORS)
+ if dialogue_word_count >= 2:
+ base_confidence += 0.3
+
+ # 因素2: 词性一致性
+ unique_pos = len(set(tag for tag in tags if tag != "w"))
+ if unique_pos == 1:
+ base_confidence += 0.2
+ elif unique_pos <= 3:
+ base_confidence += 0.1
+
+ return min(1.0, base_confidence)
+
+ def _merge_results(self, delimited: List[Dict], thulac_results: List[Dict],
+ text: str) -> List[Dict]:
+ """
+ 合并分隔符提取和 THULAC 分析结果
+
+ 优先级：分隔符标记 > THULAC 分析
+ """
+ if not delimited:
+ return thulac_results
+
+ results = []
+ covered_ranges = set()
+
+ # 首先添加所有分隔符标记的内容
+ for item in delimited:
+ results.append(item)
+ for i in range(item["start"], item["end"]):
+ covered_ranges.add(i)
+
+ # 然后分析未被覆盖的文本片段
+ uncovered_segments = []
+ start = 0
+ for i in range(len(text)):
+ if i in covered_ranges:
+ if start < i:
+ uncovered_segments.append((start, i))
+ start = i + 1
+ if start < len(text):
+ uncovered_segments.append((start, len(text)))
+
+ # 对未覆盖的片段使用 THULAC 分析
+ for start, end in uncovered_segments:
+ segment = text[start:end].strip()
+ if segment:
+ thulac_result = self._analyze_with_thulac(segment)
+ for item in thulac_result:
+ item["start"] = start
+ item["end"] = end
+ results.append(item)
+
+ # 按位置重新排序
+ results.sort(key=lambda x: x.get("start", 0))
+ return results
+
+ def parse_line(self, line: str) -> Dict:
+ """
+ 解析单行日志
+
+ Args:
+ line: 日志行
+
+ Returns:
+ {
+ "metadata": {...},
+ "content": [...]
+ }
+ """
+ if not line or not line.strip():
+ return {"metadata": {}, "content": []}
+
+ # 提取元数据（时间戳、发言人等）
+ metadata = self._extract_metadata(line)
+
+ # 移除元数据后的内容
+ content_text = self._remove_metadata(line, metadata)
+
+ # 1. 提取分隔符标记的内容
+ delimited = self._extract_delimited_content(content_text)
+
+ # 2. 使用 THULAC 分析未标记的内容
+ thulac_results = []
+ if not delimited or len(delimited) == 0:
+ thulac_results = self._analyze_with_thulac(content_text)
+
+ # 3. 合并结果
+ content = self._merge_results(delimited, thulac_results, content_text)
+
+ # 更新统计
+ self.statistics["total_parsed"] += 1
+ for item in content:
+ type_key = f"{item['type']}_count"
+ if type_key in self.statistics:
+ self.statistics[type_key] += 1
+
+ return {
+ "metadata": metadata,
+ "content": content
+ }
+
+ def _extract_metadata(self, line: str) -> Dict:
+ """提取元数据（时间戳、发言人）"""
+ metadata = {}
+
+ # 常见的元数据格式
+ patterns = [
+ r"^\[(.+?)\]\s*<(.+?)>", # [时间] <发言人>
+ r"^(.+?)\s*\|\s*(.+?)\s*:", # 时间 | 发言人:
+ r"^<(.+?)>\s*@\s*(.+?)$", # <发言人> @ 时间
+ ]
+
+ for pattern in patterns:
+ match = re.search(pattern, line)
+ if match:
+ metadata["timestamp"] = match.group(1)
+ metadata["speaker"] = match.group(2)
+ break
+
+ return metadata
+
+ def _remove_metadata(self, line: str, metadata: Dict) -> str:
+ """移除元数据，返回纯内容"""
+ if not metadata:
+ return line
+
+ # 移除匹配到的元数据部分
+ patterns = [
+ r"^\[.+?\]\s*<.+?>\s*",
+ r"^.+?\s*\|\s*.+?\s*:\s*",
+ r"^<.+?>\s*@\s*.+?\s*",
+ ]
+
+ for pattern in patterns:
+ line = re.sub(pattern, "", line, count=1)
+
+ return line.strip()
+
+ def parse_log(self, log_path: str) -> List[Dict]:
+ """
+ 解析完整的 TRPG 日志文件
+
+ Args:
+ log_path: 日志文件路径
+
+ Returns:
+ 解析结果列表
+ """
+ from pathlib import Path
+
+ if not Path(log_path).exists():
+ raise FileNotFoundError(f"Log file not found: {log_path}")
+
+ with open(log_path, "r", encoding="utf-8") as f:
+ lines = f.readlines()
+
+ results = []
+ for i, line in enumerate(lines):
+ line = line.strip()
+ if not line:
+ continue
+
+ try:
+ result = self.parse_line(line)
+ result["line_number"] = i + 1
+ results.append(result)
+ except Exception as e:
+ logger.error(f"Error parsing line {i+1}: {e}")
+ results.append({
+ "line_number": i + 1,
+ "error": str(e),
+ "raw_line": line
+ })
+
+ logger.info(f"Parsed {len(results)} lines from {log_path}")
+ return results
+
+ def get_statistics(self) -> Dict:
+ """获取解析统计信息"""
+ return self.statistics.copy()
+
+ def reset_statistics(self):
+ """重置统计信息"""
+ for key in self.statistics:
+ self.statistics[key] = 0
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-20 20:32:06 +0000