update

examples/tutorial_pyltp/README.md

@@ -2,9 +2,14 @@
 
 ```text
 工程路径: 
+https://github.com/HIT-SCIR/pyltp
 https://github.com/HuangFJ/pyltp
 
 模型文件: 
 https://ltp.ai/download.html
 
-```
+参考信息:
+https://ltp.readthedocs.io/zh-cn/v3.3.0/appendix.html 
+https://blog.csdn.net/weixin_43758551/article/details/104266953
+
+```

examples/tutorial_pyltp/srl.py

@@ -0,0 +1,69 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+import argparse
+import os
+
+from pyltp import Parser, Postagger, Segmentor, SementicRoleLabeller
+
+from project_settings import project_path
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument(
+        "--text",
+        default="元芳你怎么看？我就趴窗口上看呗！",
+        type=str
+    )
+    parser.add_argument(
+        "--ltp_data_dir",
+        default=(project_path / "data/pyltp_models/ltp_data_v3.4.0").as_posix(),
+        type=str
+    )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    args = get_args()
+
+    cws_model_path = os.path.join(args.ltp_data_dir, 'cws.model')
+    pos_model_path = os.path.join(args.ltp_data_dir, 'pos.model')
+    parser_model_path = os.path.join(args.ltp_data_dir, 'parser.model')
+    srl_model_path = os.path.join(args.ltp_data_dir, 'pisrl_win.model')
+
+    segmentor = Segmentor(cws_model_path)
+    pos_tagger = Postagger(pos_model_path)
+    parser = Parser(parser_model_path)
+    srl_labeler = SementicRoleLabeller(srl_model_path)
+
+    words = segmentor.segment(args.text)
+    postags = pos_tagger.postag(words)
+    arcs = parser.parse(words, postags)
+    roles = srl_labeler.label(words, postags, arcs)
+
+    for role in roles:
+        index = role[0]
+        role_ = [("INDEX", (index, index))] + role[1]
+        role_ = list(sorted(role_, key=lambda x: x[1][1]))
+
+        row = ""
+        for r in role_:
+            name = r[0]
+            start = r[1][0]
+            end = r[1][1]
+            arg_text = "".join(words[start:end+1])
+            row += f"{arg_text}/{name}\t"
+        print(row)
+
+    segmentor.release()
+    pos_tagger.release()
+    parser.release()
+    srl_labeler.release()
+
+    return
+
+
+if __name__ == "__main__":
+    main()

main.py

@@ -21,7 +21,16 @@ log.setup(log_directory=log_directory)
 import gradio as gr
 
 from toolbox.os.command import Command
-from toolbox.part_of_speech.part_of_speech import language_to_engines, engine_to_tagger, pos_tag
+from toolbox.part_of_speech.part_of_speech import (
+    language_to_engines as pos_language_to_engines,
+    engine_to_tagger as pos_engine_to_tagger,
+    pos_tag
+)
+from toolbox.sementic_role_labeling.sementic_role_labeling import (
+    language_to_engines as srl_language_to_engines,
+    engine_to_tagger as srl_engine_to_tagger,
+    srl
+)
 
 main_logger = logging.getLogger("main")
 
@@ -34,6 +43,11 @@ def get_args():
         default=(project_path / "pos_examples.json").as_posix(),
         type=str
     )
+    parser.add_argument(
+        "--srl_example_json_file",
+        default=(project_path / "srl_examples.json").as_posix(),
+        type=str
+    )
     args = parser.parse_args()
     return args
 
@@ -44,7 +58,7 @@ def run_pos_tag(text: str, language: str, engine: str) -> str:
 
         begin = time.time()
 
-        words, postags = pos_tag(text, engine)
+        words, postags = pos_tag(text, language, engine)
         result = ""
         for word, postag in zip(words, postags):
             row = f"{word}/{postag}"
@@ -58,6 +72,33 @@ def run_pos_tag(text: str, language: str, engine: str) -> str:
         return result
 
 
+def run_srl(text: str, language: str, engine: str) -> str:
+    try:
+        main_logger.info(f"srl started. text: {text}, language: {language}, engine: {engine}")
+
+        begin = time.time()
+
+        words, postags, arcs, roles = srl(text, language, engine)
+
+        result = ""
+        for role in roles:
+            row = ""
+            for r in role:
+                name = r[0]
+                start = r[1][0]
+                end = r[1][1]
+                arg_text = "".join(words[start:end+1])
+                row += f"{arg_text}/{name}\t"
+            result += f"{row}\n"
+
+        time_cost = time.time() - begin
+        result += f"\n\ntime_cost: {round(time_cost, 4)}"
+        return result
+    except Exception as e:
+        result = f"{type(e)}\n{str(e)}"
+        return result
+
+
 def shell(cmd: str):
     return Command.popen(cmd)
 
@@ -67,16 +108,8 @@ def main():
 
     with open(args.pos_example_json_file, "r", encoding="utf-8") as f:
         pos_examples: list = json.load(f)
-
-    def get_languages_by_engine(engine: str):
-        language_list = list()
-        for k, v in language_to_engines.items():
-            if engine in v:
-                language_list.append(k)
-        return gr.Dropdown(choices=language_list, value=language_list[0], label="language")
-
-    pos_language_choices = list(language_to_engines.keys())
-    pos_engine_choices = list(engine_to_tagger.keys())
+    with open(args.srl_example_json_file, "r", encoding="utf-8") as f:
+        srl_examples: list = json.load(f)
 
     # blocks
     with gr.Blocks() as blocks:
@@ -84,6 +117,16 @@ def main():
 
         with gr.Tabs():
             with gr.TabItem("part of speech"):
+                def pos_get_languages_by_engine(engine: str):
+                    language_list = list()
+                    for k, v in pos_language_to_engines.items():
+                        if engine in v:
+                            language_list.append(k)
+                    return gr.Dropdown(choices=language_list, value=language_list[0], label="language")
+
+                pos_language_choices = list(pos_language_to_engines.keys())
+                pos_engine_choices = list(pos_engine_to_tagger.keys())
+
                 pos_text = gr.Textbox(value="学而时习之，不亦悦乎。", lines=4, max_lines=50, label="text")
 
                 with gr.Row():
@@ -97,7 +140,7 @@ def main():
                     )
 
                 pos_engine.change(
-                    get_languages_by_engine,
+                    pos_get_languages_by_engine,
                     inputs=[pos_engine],
                     outputs=[pos_language],
                 )
@@ -116,6 +159,47 @@ def main():
                     fn=run_pos_tag,
                 )
 
+            with gr.TabItem("srl"):
+                def srl_get_languages_by_engine(engine: str):
+                    language_list = list()
+                    for k, v in pos_language_to_engines.items():
+                        if engine in v:
+                            language_list.append(k)
+                    return gr.Dropdown(choices=language_list, value=language_list[0], label="language")
+
+                srl_language_choices = list(srl_language_to_engines.keys())
+                srl_engine_choices = list(srl_engine_to_tagger.keys())
+
+                srl_text = gr.Textbox(value="学而时习之，不亦悦乎。", lines=4, max_lines=50, label="text")
+
+                with gr.Row():
+                    srl_language = gr.Dropdown(
+                        choices=srl_language_choices, value=srl_language_choices[0],
+                        label="language"
+                    )
+                    srl_engine = gr.Dropdown(
+                        choices=srl_engine_choices, value=srl_engine_choices[0],
+                        label="engine"
+                    )
+                srl_engine.change(
+                    srl_get_languages_by_engine,
+                    inputs=[srl_engine],
+                    outputs=[srl_language],
+                )
+                srl_output = gr.Textbox(lines=4, max_lines=50, label="output")
+                srl_button = gr.Button(value="pos_tag", variant="primary")
+                srl_button.click(
+                    run_srl,
+                    inputs=[srl_text, srl_language, srl_engine],
+                    outputs=[srl_output],
+                )
+                gr.Examples(
+                    examples=srl_examples,
+                    inputs=[srl_text, srl_language, srl_engine],
+                    outputs=[srl_output],
+                    fn=run_srl,
+                )
+
             with gr.TabItem("shell"):
                 shell_text = gr.Textbox(label="cmd")
                 shell_button = gr.Button("run")

srl_examples.json

@@ -0,0 +1,3 @@
+[
+  ["元芳你怎么看？我就趴窗口上看呗！", "chinese", "pyltp"]
+]

toolbox/part_of_speech/pyltp_pos_tagger.py

@@ -9,6 +9,12 @@ ltp_data_dir = os.environ.get("LTP_DATA_DIR")
 from pyltp import Postagger, Segmentor
 
 
+pos_name_amp = {
+    "nh": "人名",
+
+}
+
+
 @lru_cache(maxsize=5)
 def get_pyltp_pos_tagger():
     global ltp_data_dir

toolbox/sementic_role_labeling/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+
+
+if __name__ == "__main__":
+    pass

toolbox/sementic_role_labeling/pyltp_srl.py

@@ -0,0 +1,48 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+from functools import lru_cache
+import os
+from typing import List, Union
+
+ltp_data_dir = os.environ.get("LTP_DATA_DIR")
+
+from pyltp import Parser, Postagger, Segmentor, SementicRoleLabeller
+
+
+@lru_cache(maxsize=5)
+def get_pyltp_srl_tagger():
+    global ltp_data_dir
+
+    cws_model_path = os.path.join(ltp_data_dir, 'cws.model')
+    pos_model_path = os.path.join(ltp_data_dir, 'pos.model')
+    parser_model_path = os.path.join(ltp_data_dir, 'parser.model')
+    srl_model_path = os.path.join(ltp_data_dir, 'pisrl_win.model')
+
+    segmentor = Segmentor(cws_model_path)
+    pos_tagger = Postagger(pos_model_path)
+    parser = Parser(parser_model_path)
+    srl_labeler = SementicRoleLabeller(srl_model_path)
+
+    return segmentor, pos_tagger, parser, srl_labeler
+
+
+def pyltp_srl(text: str, language: str) -> list:
+    segmentor, pos_tagger, parser, srl_labeler = get_pyltp_srl_tagger()
+
+    words = segmentor.segment(text)
+    postags = pos_tagger.postag(words)
+    arcs = parser.parse(words, postags)
+    roles = srl_labeler.label(words, postags, arcs)
+
+    roles_ = list()
+    for role in roles:
+        index = role[0]
+        role_ = [("INDEX", (index, index))] + role[1]
+        role_ = list(sorted(role_, key=lambda x: x[1][1]))
+        roles_.append(role_)
+
+    return words, postags, arcs, roles_
+
+
+if __name__ == "__main__":
+    pass

toolbox/sementic_role_labeling/sementic_role_labeling.py

@@ -0,0 +1,28 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+from typing import Callable, Dict, List, Tuple, Union
+
+from toolbox.sementic_role_labeling.pyltp_srl import pyltp_srl
+
+
+language_to_engines = {
+    "chinese": ["pyltp"]
+}
+
+
+engine_to_tagger: Dict[str, Callable] = {
+    "pyltp": pyltp_srl
+}
+
+
+def srl(text: str, language: str, engine: str):
+    srl_tagger = engine_to_tagger.get(engine)
+    if srl_tagger is None:
+        raise AssertionError(f"engine {engine} not supported.")
+
+    words, postags, arcs, roles = srl_tagger(text, language)
+    return words, postags, arcs, roles
+
+
+if __name__ == "__main__":
+    pass