860 lines
30 KiB
C#
860 lines
30 KiB
C#
using CloudBuilder.Core.DatabaseAccessor.Entity;
|
||
using Python.Runtime;
|
||
using System;
|
||
using System.Collections.Generic;
|
||
using System.Linq;
|
||
using System.Text;
|
||
using System.Threading.Tasks;
|
||
|
||
namespace CloudBuilder.Topshelf.Python
|
||
{
|
||
public class PythonCommand
|
||
{
|
||
private bool _disposed = false; // 释放标记
|
||
public dynamic PythonCreateScope { get; set; }
|
||
public IApplicationService ApplicationService;
|
||
public PythonCommand(IApplicationService applicationService)
|
||
{
|
||
ApplicationService = applicationService;
|
||
|
||
Runtime.PythonDLL = applicationService.Configuration["Python:PythonCommand"];
|
||
PythonEngine.PythonHome = Path.GetDirectoryName(Runtime.PythonDLL);
|
||
PythonEngine.Initialize();
|
||
|
||
PythonCreateScope = Py.CreateScope();
|
||
}
|
||
|
||
public void Exec(string arg)
|
||
{
|
||
using (Py.GIL())
|
||
{
|
||
PythonCreateScope.Exec(arg);//"from pyhanlp import HanLP"
|
||
}
|
||
}
|
||
|
||
public string GetValueExec(string arg)
|
||
{
|
||
using (Py.GIL())
|
||
{
|
||
PythonCreateScope.Exec(string.Format("s_value={0}", arg));
|
||
|
||
var value = PythonCreateScope.Eval("s_value");
|
||
|
||
return value.ToString();
|
||
}
|
||
}
|
||
|
||
public string[] GetValuesExec(string arg)
|
||
{
|
||
using (Py.GIL())
|
||
{
|
||
PythonCreateScope.Exec(string.Format("s_list={0}", arg));
|
||
|
||
var value = PythonCreateScope.Eval("s_list");
|
||
|
||
string[] st = value.As<string[]>();
|
||
|
||
return st;
|
||
}
|
||
}
|
||
|
||
// 核心：释放 Python.NET 资源
|
||
public void Dispose()
|
||
{
|
||
Dispose(true);
|
||
GC.SuppressFinalize(this); // 阻止析构函数重复执行
|
||
}
|
||
|
||
protected virtual void Dispose(bool disposing)
|
||
{
|
||
if (_disposed) return;
|
||
|
||
// 1. 释放 Python 作用域
|
||
if (PythonCreateScope != null)
|
||
{
|
||
PythonCreateScope.Dispose();
|
||
PythonCreateScope = null;
|
||
}
|
||
|
||
// 2. 关闭 Python 运行时（关键！）
|
||
if (PythonEngine.IsInitialized)
|
||
{
|
||
AppContext.SetSwitch("System.Runtime.Serialization.EnableUnsafeBinaryFormatterSerialization", true);
|
||
|
||
PythonEngine.Shutdown(); // 销毁 Python 解释器
|
||
}
|
||
|
||
_disposed = true;
|
||
}
|
||
}
|
||
}
|
||
/*
|
||
import os
|
||
import json
|
||
import logging
|
||
import sys
|
||
# web api通過cmd調用，如果沒有加入下面三行，會出錯
|
||
sys.stdout = open(os.devnull, 'w')
|
||
sys.stderr = open(os.devnull, 'w')
|
||
os.environ['TQDM_DISABLE'] = '1'
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
|
||
try:
|
||
import hanlp
|
||
from hanlp.components.mtl.multi_task_learning import MultiTaskLearning
|
||
from hanlp.components.mtl.tasks.tok.tag_tok import TaggingTokenization
|
||
from hanlp.components.mtl.tasks.ner.tag_ner import TaggingNamedEntityRecognition
|
||
|
||
# 尝试导入tqdm，如果不存在则使用替代
|
||
try:
|
||
from tqdm import tqdm
|
||
HAS_TQDM = True
|
||
except ImportError:
|
||
logger.warning("tqdm未安装，将不显示进度条")
|
||
HAS_TQDM = False
|
||
|
||
# 创建一个简单的tqdm替代
|
||
class tqdm:
|
||
def __init__(self, iterable, desc=None):
|
||
self.iterable = iterable
|
||
self.desc = desc
|
||
def __iter__(self):
|
||
return iter(self.iterable)
|
||
except ImportError as e:
|
||
logger.error(f"导入HanLP失败: {e}")
|
||
|
||
class HanLPProcessor:
|
||
"""HanLP自然语言处理工具类"""
|
||
|
||
def __init__(self,
|
||
hanlp_home='D:/Python38/hanlp/',
|
||
hf_home='D:/Python38/models/',
|
||
hf_endpoint='https://hf-mirror.com',
|
||
hanlp_url='https://ftp.hankcs.com/hanlp/',
|
||
surnames_path='D:/Python38/hanlp/dictionary/surnames.txt',
|
||
titles_path='D:/Python38/hanlp/dictionary/titles.txt',
|
||
words_path='D:/Python38/hanlp/dictionary/words.txt',
|
||
salutations_path='D:/Python38/hanlp/dictionary/salutations.txt'):
|
||
# 初始化ID计数器和父ID跟踪
|
||
self.item_id_counter = 1
|
||
self.current_father_id = 0
|
||
"""
|
||
初始化HanLP处理器
|
||
|
||
参数:
|
||
hanlp_home: HanLP资源缓存目录
|
||
hf_home: HuggingFace模型缓存目录
|
||
hf_endpoint: HuggingFace下载端点
|
||
hanlp_url: HanLP下载URL
|
||
surnames_path: 姓氏词典路径
|
||
titles_path: 头衔词典路径
|
||
words_path: 专有名词词典路径
|
||
salutations_path: 称呼词典路径
|
||
"""
|
||
# 存储配置信息
|
||
self.hanlp_home = hanlp_home
|
||
self.hf_home = hf_home
|
||
self.hf_endpoint = hf_endpoint
|
||
self.hanlp_url = hanlp_url
|
||
self.surnames_path = surnames_path
|
||
self.titles_path = titles_path
|
||
self.words_path = words_path
|
||
self.salutations_path = salutations_path
|
||
|
||
# 初始化组件
|
||
self.mtl = None
|
||
self.tok = None
|
||
self.ner = None
|
||
|
||
# 设置环境变量
|
||
self._set_environment_variables()
|
||
|
||
# 加载模型和词典
|
||
self._load_model()
|
||
self._load_dictionaries()
|
||
|
||
def _set_environment_variables(self):
|
||
"""设置必要的环境变量"""
|
||
try:
|
||
os.environ['HANLP_HOME'] = self.hanlp_home
|
||
os.environ['HF_HOME'] = self.hf_home
|
||
os.environ['HF_ENDPOINT'] = self.hf_endpoint
|
||
os.environ['HANLP_URL'] = self.hanlp_url
|
||
logger.info("环境变量设置成功")
|
||
except Exception as e:
|
||
logger.error(f"设置环境变量失败: {e}")
|
||
|
||
def _load_dict_from_file(self, file_path, default_tag='S-PERSON'):
|
||
"""从文件加载词典"""
|
||
dict_tags = {}
|
||
try:
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
for word in f:
|
||
word = word.strip()
|
||
if word: # 跳过空行
|
||
dict_tags[(word,)] = (default_tag,) # 保持元组形式
|
||
except Exception as e:
|
||
print(f"加载词典文件{file_path}失败: {e}")
|
||
return dict_tags
|
||
|
||
def _load_names(self, surnames_path, titles_path):
|
||
"""加载姓氏和头衔，生成姓名组合"""
|
||
try:
|
||
with open(surnames_path, 'r', encoding='utf-8') as f:
|
||
surnames = [line.strip() for line in f if line.strip()]
|
||
|
||
with open(titles_path, 'r', encoding='utf-8') as f:
|
||
titles = [line.strip() for line in f if line.strip()]
|
||
|
||
# 使用生成器表达式（节省内存）
|
||
full_names = (f'{surname}{title}' for surname in surnames for title in titles)
|
||
return full_names
|
||
except Exception as e:
|
||
print(f"加载姓名文件失败: {e}")
|
||
return []
|
||
|
||
def _load_model(self):
|
||
"""加载HanLP模型"""
|
||
try:
|
||
# CLOSE是自然语义标注的闭源语料库，BASE是中号模型，ZH中文
|
||
logger.info("开始加载HanLP模型...")
|
||
self.mtl = hanlp.load(hanlp.pretrained.mtl.CLOSE_TOK_POS_NER_SRL_DEP_SDP_CON_ELECTRA_SMALL_ZH)
|
||
self.tok = self.mtl['tok/coarse']
|
||
self.tok.dict_force = self.tok.dict_combine = None
|
||
self.ner = self.mtl['ner/msra']
|
||
logger.info("模型加载成功")
|
||
except Exception as e:
|
||
logger.error(f"加载模型失败: {e}")
|
||
self.mtl = None
|
||
self.tok = None
|
||
self.ner = None
|
||
|
||
def _load_dictionaries(self):
|
||
"""加载各种词典"""
|
||
try:
|
||
# 加载称呼词典
|
||
self.ner.dict_tags = self._load_dict_from_file(self.salutations_path)
|
||
|
||
# 读取专有名词词典
|
||
with open(self.words_path, 'r', encoding='utf-8') as f:
|
||
names = [line.strip() for line in f if line.strip()]
|
||
|
||
# 生成所有姓名组合
|
||
full_names = self._load_names(self.surnames_path, self.titles_path)
|
||
|
||
# 构建强制分词词典（使用进度条）
|
||
dict_force = {}
|
||
for name in tqdm(full_names, desc='读取人名词典'):
|
||
dict_force[name] = [name]
|
||
|
||
for name in names:
|
||
dict_force[name] = [name]
|
||
|
||
self.tok.dict_force = dict_force
|
||
except Exception as e:
|
||
print(f"加载词典失败: {e}")
|
||
|
||
def get_ner_json(self, text):
|
||
"""识别文本中的实体并返回JSON格式"""
|
||
if not text or not isinstance(text, str):
|
||
logger.warning("无效的输入文本")
|
||
return json.dumps([])
|
||
|
||
try:
|
||
if not self.mtl:
|
||
raise RuntimeError("模型未加载")
|
||
|
||
result = self.mtl(text, tasks=['tok/coarse', 'ner/msra'], skip_tasks=['tok/fine'])
|
||
|
||
# 检查结果格式
|
||
if 'ner/msra' not in result or not isinstance(result['ner/msra'], (list, tuple)):
|
||
logger.warning("模型返回结果格式异常")
|
||
return json.dumps([])
|
||
|
||
json_data = []
|
||
for item in result['ner/msra']:
|
||
# 确保item是正确的格式
|
||
if isinstance(item, (list, tuple)) and len(item) >= 4:
|
||
json_data.append({
|
||
'Text': str(item[0]),
|
||
'Type': str(item[1]),
|
||
'Start': int(item[2]),
|
||
'End': int(item[3])
|
||
})
|
||
|
||
return json.dumps(json_data, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"实体识别失败: {e}")
|
||
return json.dumps([])
|
||
|
||
def get_srl_json(self, text):
|
||
"""识别文本中的人物角色并返回JSON格式"""
|
||
if not text or not isinstance(text, str):
|
||
logger.warning("无效的输入文本")
|
||
return json.dumps([])
|
||
|
||
try:
|
||
if not self.mtl:
|
||
raise RuntimeError("模型未加载")
|
||
|
||
results = self.mtl(text, tasks=['tok/coarse', 'srl'], skip_tasks=['tok/fine'])
|
||
result = []
|
||
|
||
# 检查结果格式
|
||
if 'srl' not in results:
|
||
logger.warning("模型返回结果中未找到SRL数据")
|
||
return json.dumps([])
|
||
|
||
data = results['srl']
|
||
if not data:
|
||
return json.dumps(result)
|
||
|
||
for sentence in data:
|
||
if not isinstance(sentence, (list, tuple)):
|
||
continue
|
||
|
||
sentence_data = []
|
||
for item in sentence:
|
||
if not isinstance(item, (list, tuple)):
|
||
continue
|
||
|
||
# 确保至少有2个元素
|
||
if len(item) < 2:
|
||
continue
|
||
|
||
entry = {
|
||
'Text': str(item[0]),
|
||
'Type': str(item[1]),
|
||
'Start': int(item[2]) if len(item) > 2 else -1,
|
||
'End': int(item[3]) if len(item) > 3 else -1
|
||
}
|
||
sentence_data.append(entry)
|
||
|
||
if sentence_data: # 只添加非空的句子数据
|
||
result.append(sentence_data)
|
||
|
||
return json.dumps(result, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
logger.error(f"人物角色识别失败: {e}")
|
||
return json.dumps([])
|
||
|
||
def get_con(self, text):
|
||
"""解析文本的句法结构并返回结果"""
|
||
if not text or not isinstance(text, str):
|
||
logger.warning("无效的输入文本")
|
||
return None
|
||
|
||
try:
|
||
if not self.mtl:
|
||
raise RuntimeError("模型未加载")
|
||
|
||
# logger.info("开始句法分析...")
|
||
result = self.mtl(text, tasks=['con'])
|
||
# logger.info("句法分析完成")
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f"句法分析失败: {e}")
|
||
return None
|
||
|
||
def get_all(self,text):
|
||
result = self.mtl(text)
|
||
return result;
|
||
|
||
def get_con_json(self, text):
|
||
"""解析文本的句法结构并返回JSON格式字符串，便于C#对象处理"""
|
||
# 重置ID计数器和父ID，确保每次调用都从1开始
|
||
self.item_id_counter = 0
|
||
self.current_father_id = 0
|
||
|
||
# 初始化JSON数组用于存储所有解析结果元素
|
||
result_array = []
|
||
# 保存结果数组的引用，供packLabel方法使用
|
||
self.result_array = result_array
|
||
|
||
if not text or not isinstance(text, str):
|
||
logger.warning("无效的输入文本")
|
||
return json.dumps([])
|
||
|
||
if not self.mtl:
|
||
raise RuntimeError("模型未加载")
|
||
|
||
# logger.info("开始句法分析...")
|
||
results = self.mtl(text, tasks=['con'])
|
||
# logger.info("句法分析完成")
|
||
|
||
# 检查结果格式
|
||
if 'con' not in results:
|
||
logger.warning("模型返回结果中未找到CON数据")
|
||
return json.dumps([])
|
||
|
||
#从results中截取"con": [截取字符串]}
|
||
# 将results转为字符串并提取con字段值
|
||
results_str = str(results)
|
||
# logger.info(results_str)
|
||
# 只取出top的内容
|
||
new_results = self.get_con_top('"con": [',results_str)
|
||
level=0
|
||
|
||
self.item_id_counter += 1
|
||
|
||
# 调用pack_childen处理所有节点，所有生成的元素将通过packLabel方法添加到result_array
|
||
try:
|
||
self.pack_childen(new_results, level)
|
||
except Exception as e:
|
||
logger.warning(f"pack_childen处理时出错: {e}")
|
||
|
||
# 清理引用，避免内存泄漏
|
||
delattr(self, 'result_array')
|
||
|
||
# 将结果数组转换为JSON字符串返回，便于C#交互
|
||
return json.dumps(result_array, ensure_ascii=False)
|
||
|
||
def pack_childen(self, text, level, father_id=None):
|
||
# 如果没有提供father_id，则使用当前类的current_father_id
|
||
if father_id is None:
|
||
father_id = self.current_father_id
|
||
|
||
childen_json = []
|
||
json_temp = self.split(text, ',')
|
||
# 去除前后的[]
|
||
childen = self.substring_before_after_one(json_temp[1])
|
||
# 分隔元素
|
||
childens = self.split(childen, ',')
|
||
label = json_temp[0]
|
||
|
||
if json_temp[0] == '_':
|
||
# "_", ["二愣子"] - 提取列表中的值
|
||
# 当标签为"_"时，提取childen中的值并直接返回值字符串
|
||
label = self.substring_before_after_one(childen)
|
||
# 保存当前的current_father_id
|
||
current_father = self.current_father_id
|
||
# 设置当前元素的父ID
|
||
self.current_father_id = father_id
|
||
# 创建节点
|
||
top_json = self.packLabel(label, level, 0)
|
||
# 恢复原来的current_father_id
|
||
self.current_father_id = current_father
|
||
# logger.info(top_json)
|
||
return
|
||
|
||
# 保存当前的current_father_id
|
||
current_father = self.current_father_id
|
||
# 设置当前元素的父ID
|
||
self.current_father_id = father_id
|
||
# 创建当前节点
|
||
top_json = self.packLabel(label, level, len(childens))
|
||
# 获取当前节点的ID作为子节点的父ID
|
||
current_node_id = top_json['ItemId']
|
||
# 恢复原来的current_father_id
|
||
self.current_father_id = current_father
|
||
|
||
# logger.info(top_json)
|
||
|
||
# 递归处理子节点，传递当前节点的ID作为父ID
|
||
for i, item in enumerate(childens):
|
||
childen_temp = self.substring_before_after_one(item)
|
||
childen_json.append(self.pack_childen(childen_temp, level + 1, current_node_id))
|
||
|
||
def get_root(self, text):
|
||
json_str = self.get_task(text,'dep')
|
||
|
||
for index, dep in enumerate(json_str["dep"]):
|
||
if('root'==dep[1]):
|
||
return json_str["tok/fine"][index];
|
||
return ''
|
||
|
||
def get_task(self, text, task):
|
||
'''
|
||
执行指定的任务并跳过fine分词
|
||
|
||
参数:
|
||
text: 输入文本
|
||
task: 要执行的任务名称
|
||
|
||
返回:
|
||
任务执行结果
|
||
'''
|
||
if not text or not isinstance(text, str):
|
||
logger.warning('无效的输入文本')
|
||
return None
|
||
|
||
try:
|
||
if not self.mtl:
|
||
raise RuntimeError('模型未加载')
|
||
|
||
# 跳过tok/fine任务
|
||
result = self.mtl(text, tasks=[task])
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f'执行任务{task}失败: {e}')
|
||
return None
|
||
|
||
def get_tok_fine(self, text):
|
||
"""细粒度分词，返回C#可JsonSerializer的数组JSON"""
|
||
if not text or not isinstance(text, str):
|
||
logger.warning("无效的输入文本")
|
||
return json.dumps([], ensure_ascii=False)
|
||
|
||
try:
|
||
if not self.mtl:
|
||
raise RuntimeError("模型未加载")
|
||
|
||
results = self.mtl(text, tasks=['tok/fine'], skip_tasks=['tok/coarse'])
|
||
|
||
if 'tok/fine' not in results or not isinstance(results['tok/fine'], (list, tuple)):
|
||
logger.warning("模型返回结果格式异常")
|
||
return json.dumps([], ensure_ascii=False)
|
||
|
||
return json.dumps(results['tok/fine'], ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"细粒度分词失败: {e}")
|
||
return json.dumps([], ensure_ascii=False)
|
||
|
||
def get_tok_coarse(self, text):
|
||
"""粗粒度分词，返回C#可JsonSerializer的数组JSON"""
|
||
if not text or not isinstance(text, str):
|
||
logger.warning("无效的输入文本")
|
||
return json.dumps([], ensure_ascii=False)
|
||
|
||
try:
|
||
if not self.mtl:
|
||
raise RuntimeError("模型未加载")
|
||
|
||
results = self.mtl(text, tasks=['tok/coarse'], skip_tasks=['tok/fine'])
|
||
|
||
if 'tok/coarse' not in results or not isinstance(results['tok/coarse'], (list, tuple)):
|
||
logger.warning("模型返回结果格式异常")
|
||
return json.dumps([], ensure_ascii=False)
|
||
|
||
return json.dumps(results['tok/coarse'], ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"粗粒度分词失败: {e}")
|
||
return json.dumps([], ensure_ascii=False)
|
||
|
||
def get_con_top(self,from_text,text):
|
||
if '"con": [' in text:
|
||
start = text.find(from_text) + len(from_text)
|
||
# 寻找对应的结束括号
|
||
end = start
|
||
bracket_count = 1
|
||
while end < len(text) and bracket_count > 0:
|
||
if text[end] == '[':
|
||
bracket_count += 1
|
||
elif text[end] == ']':
|
||
bracket_count -= 1
|
||
end += 1
|
||
new_results = text[start:end-1] # -1 是为了去掉最后的']'
|
||
else:
|
||
new_results = ''
|
||
return new_results
|
||
|
||
def substring_before_after_one(self, text):
|
||
if not text or not isinstance(text, str):
|
||
return text
|
||
if len(text) < 2:
|
||
return text
|
||
return text[1:-1]
|
||
|
||
def split(self, text, char):
|
||
if not text:
|
||
return []
|
||
|
||
result = []
|
||
current = []
|
||
in_quotes = False # 是否在引号内
|
||
bracket_level = 0 # 括号嵌套级别
|
||
|
||
i = 0
|
||
while i < len(text):
|
||
c = text[i]
|
||
|
||
# 处理引号
|
||
if c == '"':
|
||
in_quotes = not in_quotes
|
||
current.append(c)
|
||
|
||
# 处理括号
|
||
elif c == '[' and not in_quotes:
|
||
bracket_level += 1
|
||
current.append(c)
|
||
elif c == ']' and not in_quotes:
|
||
if bracket_level > 0:
|
||
bracket_level -= 1
|
||
current.append(c)
|
||
|
||
# 遇到分隔符且不在括号内和引号内
|
||
elif c == char and bracket_level == 0 and not in_quotes:
|
||
# 添加当前部分到结果
|
||
part = ''.join(current).strip()
|
||
# 清理引号（如果有）
|
||
if part.startswith('"') and part.endswith('"'):
|
||
part = part[1:-1]
|
||
result.append(part)
|
||
current = []
|
||
|
||
# 处理转义字符
|
||
elif c == '\\' and i + 1 < len(text):
|
||
current.append(c)
|
||
current.append(text[i + 1])
|
||
i += 1
|
||
|
||
else:
|
||
current.append(c)
|
||
|
||
i += 1
|
||
|
||
# 处理最后一个元素
|
||
if current:
|
||
part = ''.join(current).strip()
|
||
# 清理引号（如果有）
|
||
if part.startswith('"') and part.endswith('"'):
|
||
part = part[1:-1]
|
||
result.append(part)
|
||
|
||
return result
|
||
|
||
def packLabel(self, label, level, children):
|
||
# 保存当前ID作为返回值
|
||
current_id = self.item_id_counter
|
||
|
||
# 创建包含ID和父ID的字典
|
||
result = {
|
||
"ItemId": current_id,
|
||
"FatherId": self.current_father_id,
|
||
"Label": label,
|
||
"Level": level,
|
||
"Children": children
|
||
}
|
||
|
||
# 将生成的元素添加到result_array（如果存在）
|
||
if hasattr(self, 'result_array'):
|
||
self.result_array.append(result)
|
||
|
||
# 更新计数器
|
||
self.item_id_counter += 1
|
||
|
||
return result
|
||
|
||
def extract_underscore_value(self, text):
|
||
"""
|
||
从"_", ["值"]格式的文本中提取值，并直接返回字符串值
|
||
|
||
Args:
|
||
text: 格式为"_", ["值"]的字符串
|
||
|
||
Returns:
|
||
提取的字符串值
|
||
"""
|
||
# 分割文本以获取值部分
|
||
parts = self.split(text, ',')
|
||
if len(parts) < 2:
|
||
return ""
|
||
|
||
# 获取第二个部分（包含值的列表）
|
||
value_part = parts[1]
|
||
# 去除前后的[]和可能的引号
|
||
value = self.substring_before_after_one(value_part).strip('"')
|
||
|
||
# 直接返回提取的值
|
||
return value
|
||
|
||
def get_surnames_json(self):
|
||
"""返回姓氏词典数组的JSON格式字符串，供C#使用"""
|
||
try:
|
||
with open(self.surnames_path, 'r', encoding='utf-8') as f:
|
||
surnames = [line.strip() for line in f if line.strip()]
|
||
|
||
return json.dumps(surnames, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"获取姓氏词典失败: {e}")
|
||
return json.dumps([])
|
||
|
||
processor=HanLPProcessor()
|
||
|
||
# 测试split函数
|
||
|
||
# 测试split函数
|
||
def test_split_function():
|
||
# 测试用例1: 简单的逗号分隔
|
||
text1 = ' "TOP",[["IP", [["NP", [["_", ["二愣子"]]]], ["VP", [["VP", [["_", ["姓"]], ["NP", [["_", ["韩"]]]]]], ["VP", [["_", ["名"]], ["NP", [["_", ["立"]]]]]]]], ["_", ["。"]]]]]'
|
||
result1 = processor.split(text1, ',')
|
||
print("测试用例1结果:")
|
||
for i, item in enumerate(result1):
|
||
print(f" {i}: {item}")
|
||
|
||
text1='二愣子姓韩名立。'
|
||
print(processor.get_con_json(text1))
|
||
# 测试用例2: 括号内包含逗号
|
||
# text2 = '"NP",[["NN",["_", "张三"]],["NN",["_", "李四"]]]'
|
||
# result2 = processor.split(text2, ',')
|
||
# print("\n测试用例2结果:")
|
||
|
||
def test_extract_underscore_value():
|
||
"""
|
||
测试新添加的extract_underscore_value函数
|
||
"""
|
||
print("\n测试extract_underscore_value函数:")
|
||
|
||
# 测试用例1: 基本格式
|
||
test_text1 = '"_", ["二愣子"]'
|
||
result1 = processor.extract_underscore_value(test_text1)
|
||
print(f"输入: {test_text1}")
|
||
print(f"输出: {result1}")
|
||
|
||
# 测试用例2: 其他值
|
||
test_text2 = '"_", ["张三"]'
|
||
result2 = processor.extract_underscore_value(test_text2)
|
||
print(f"\n输入: {test_text2}")
|
||
print(f"输出: {result2}")
|
||
|
||
# 测试用例3: 空值情况
|
||
test_text3 = '"_", []'
|
||
result3 = processor.extract_underscore_value(test_text3)
|
||
print(f"\n输入: {test_text3}")
|
||
print(f"输出: {result3}")
|
||
|
||
# 测试修复后的pack_childen方法对"_", ["二愣子"]格式的处理
|
||
def test_pack_childen_with_underscore():
|
||
"""
|
||
测试pack_childen方法对"_", ["值"]格式的处理
|
||
"""
|
||
print("\n测试pack_childen方法对'_', ['二愣子']格式的处理:")
|
||
# 测试用例：直接调用pack_childen处理"_", ["二愣子"]格式
|
||
test_input = '"_", ["二愣子"]'
|
||
print(f"输入: {test_input}")
|
||
try:
|
||
# 调用pack_childen方法，使用level=0
|
||
result = processor.pack_childen(test_input, 0)
|
||
# 由于pack_childen在处理"_"标签时会直接返回，我们需要查看logger输出
|
||
print("测试完成，请检查日志输出")
|
||
except Exception as e:
|
||
print(f"测试出错: {e}")
|
||
# for i, item in enumerate(result2):
|
||
# print(f" {i}: {item}")
|
||
|
||
# 如果直接运行此文件，则执行测试
|
||
def test_id_generation():
|
||
"""
|
||
测试ItemId和FatherId的生成逻辑
|
||
"""
|
||
print("开始测试ID生成...")
|
||
# 创建处理器实例
|
||
processor = HanLPProcessor()
|
||
|
||
# 测试基本的packLabel方法
|
||
print("\n测试基本的packLabel方法:")
|
||
result1 = processor.packLabel("TOP", 0, 1)
|
||
print(result1)
|
||
|
||
# 测试生成第二个节点，应该有正确的父子关系
|
||
print("\n测试生成第二个节点:")
|
||
# 对于第二个节点，我们手动设置它的父ID为第一个节点的ID
|
||
processor.current_father_id = result1['ItemId']
|
||
result2 = processor.packLabel("TOP", 0, 1)
|
||
print(result2)
|
||
|
||
# 测试pack_childen方法
|
||
print("\n测试pack_childen方法:")
|
||
# 重置处理器的ID计数器
|
||
processor.item_id_counter = 1
|
||
processor.current_father_id = 0
|
||
|
||
# 使用一个简单的测试字符串来模拟解析结果，确保格式正确
|
||
test_input = "TOP, [IP, [NP, [NN, 你]], [VP, [VV, 好]]]"
|
||
try:
|
||
processor.pack_childen(test_input, 0)
|
||
print("pack_childen测试成功")
|
||
except Exception as e:
|
||
print(f"pack_childen测试失败: {e}")
|
||
# 为了演示，我们可以直接测试packLabel的组合
|
||
print("\n直接测试ID和父子关系:")
|
||
processor.item_id_counter = 1
|
||
processor.current_father_id = 0
|
||
|
||
# 模拟用户要求的输出格式
|
||
result1 = processor.packLabel("TOP", 0, 1)
|
||
print(f"{{'ItemId':{result1['ItemId']},'FatherId':{result1['FatherId']},'Label': '{result1['Label']}', 'Level': {result1['Level']}, 'Children': {result1['Children']}}}")
|
||
|
||
# 第二个节点的父ID是第一个节点的ID
|
||
processor.current_father_id = result1['ItemId']
|
||
result2 = processor.packLabel("TOP", 0, 1)
|
||
print(f"{{'ItemId':{result2['ItemId']},'FatherId':{result2['FatherId']},'Label': '{result2['Label']}', 'Level': {result2['Level']}, 'Children': {result2['Children']}}}")
|
||
|
||
print("\nID生成测试完成")
|
||
|
||
def test_con_json():
|
||
"""
|
||
测试get_con_json方法返回的JSON格式字符串
|
||
"""
|
||
print("\n开始测试get_con_json方法...")
|
||
processor = HanLPProcessor()
|
||
|
||
# 测试空输入
|
||
empty_result = processor.get_con_json("")
|
||
print(f"空输入测试: {empty_result}")
|
||
print(f"空输入结果类型: {type(empty_result)}")
|
||
# 验证可以解析为空数组
|
||
try:
|
||
parsed_empty = json.loads(empty_result)
|
||
print(f"空输入解析结果: {parsed_empty}, 类型: {type(parsed_empty)}")
|
||
except json.JSONDecodeError as e:
|
||
print(f"空输入JSON解析失败: {e}")
|
||
|
||
# 测试None输入
|
||
none_result = processor.get_con_json(None)
|
||
print(f"None输入测试: {none_result}")
|
||
print(f"None输入结果类型: {type(none_result)}")
|
||
# 验证可以解析为空数组
|
||
try:
|
||
parsed_none = json.loads(none_result)
|
||
print(f"None输入解析结果: {parsed_none}, 类型: {type(parsed_none)}")
|
||
except json.JSONDecodeError as e:
|
||
print(f"None输入JSON解析失败: {e}")
|
||
|
||
# 尝试测试有效输入（注意：由于可能没有加载模型，这里可能会抛出异常）
|
||
try:
|
||
# 尝试简单的中文文本
|
||
text = "二愣子姓韩名立。"
|
||
print(f"测试有效输入: {text}")
|
||
result = processor.get_con_json(text)
|
||
print(f"返回结果类型: {type(result)}")
|
||
print(f"返回结果字符串长度: {len(result)}")
|
||
|
||
# 验证返回的是字符串
|
||
assert isinstance(result, str), "返回结果应该是字符串类型"
|
||
print("✓ 验证通过：返回结果是字符串类型")
|
||
|
||
# 尝试解析JSON字符串
|
||
try:
|
||
parsed_result = json.loads(result)
|
||
print(f"✓ JSON解析成功，解析后类型: {type(parsed_result)}")
|
||
print(f"解析后数组长度: {len(parsed_result)}")
|
||
|
||
# 如果解析成功且有元素，显示部分内容
|
||
if parsed_result:
|
||
print(f"解析后第一个元素: {parsed_result[0]}")
|
||
if len(parsed_result) > 1:
|
||
print(f"解析后第二个元素: {parsed_result[1]}")
|
||
except json.JSONDecodeError as e:
|
||
print(f"✗ JSON解析失败: {e}")
|
||
|
||
print(f"完整返回结果: {result}")
|
||
except Exception as e:
|
||
print(f"有效输入测试出错（可能是模型未加载）: {e}")
|
||
|
||
print("get_con_json方法测试完成")
|
||
|
||
if __name__ == "__main__":
|
||
# test_split_function()
|
||
# test_id_generation()
|
||
# test_extract_underscore_value()
|
||
# test_pack_childen_with_underscore()
|
||
# test_con_json()
|
||
# processor=HanLPProcessor()
|
||
print("")
|
||
|
||
|
||
*/ |