update

examples/tutorial_pyltp/ner.py

@@ -0,0 +1,58 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+"""
+https://huggingface.co/LTP
+"""
+import argparse
+import os
+
+from pyltp import Postagger, Segmentor, NamedEntityRecognizer
+
+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")
+    ner_model_path = os.path.join(args.ltp_data_dir, "ner.model")
+
+    segmentor = Segmentor(cws_model_path)
+    postagger = Postagger(pos_model_path)
+    recognizer = NamedEntityRecognizer(ner_model_path)
+
+    words = segmentor.segment(args.text)
+    postags = postagger.postag(words)
+    ner_tags = recognizer.recognize(words, postags)
+
+    print(words)
+    print(postags)
+    print(ner_tags)
+
+    segmentor.release()
+    postagger.release()
+    recognizer.release()
+
+    return
+
+
+if __name__ == "__main__":
+    main()

main.py

@@ -21,6 +21,11 @@ log.setup(log_directory=log_directory)
 import gradio as gr
 
 from toolbox.os.command import Command
+from toolbox.named_entity_recognization.named_entity_recognization import (
+    language_to_engines as ner_language_to_engines,
+    engine_to_tagger as ner_engine_to_tagger,
+    ner
+)
 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,
@@ -38,6 +43,11 @@ main_logger = logging.getLogger("main")
 def get_args():
     parser = argparse.ArgumentParser()
 
+    parser.add_argument(
+        "--ner_example_json_file",
+        default=(project_path / "ner_examples.json").as_posix(),
+        type=str
+    )
     parser.add_argument(
         "--pos_example_json_file",
         default=(project_path / "pos_examples.json").as_posix(),
@@ -52,6 +62,26 @@ def get_args():
     return args
 
 
+def run_ner(text: str, language: str, engine: str) -> str:
+    try:
+        main_logger.info(f"ner started. text: {text}, language: {language}, engine: {engine}")
+
+        begin = time.time()
+
+        words, postags, ner_tags = ner(text, language, engine)
+        result = ""
+        for word, ner_tag in zip(words, ner_tags):
+            row = f"{word}/{ner_tag}"
+            result += f"{row}\t"
+
+        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 run_pos_tag(text: str, language: str, engine: str) -> str:
     try:
         main_logger.info(f"pos tag started. text: {text}, language: {language}, engine: {engine}")
@@ -106,6 +136,8 @@ def shell(cmd: str):
 def main():
     args = get_args()
 
+    with open(args.ner_example_json_file, "r", encoding="utf-8") as f:
+        ner_examples: list = json.load(f)
     with open(args.pos_example_json_file, "r", encoding="utf-8") as f:
         pos_examples: list = json.load(f)
     with open(args.srl_example_json_file, "r", encoding="utf-8") as f:
@@ -200,6 +232,47 @@ def main():
                     fn=run_srl,
                 )
 
+            with gr.TabItem("ner"):
+                def ner_get_languages_by_engine(engine: str):
+                    language_list = list()
+                    for k, v in ner_language_to_engines.items():
+                        if engine in v:
+                            language_list.append(k)
+                    return gr.Dropdown(choices=language_list, value=language_list[0], label="language")
+
+                ner_language_choices = list(ner_language_to_engines.keys())
+                ner_engine_choices = list(ner_engine_to_tagger.keys())
+
+                ner_text = gr.Textbox(value="学而时习之，不亦悦乎。", lines=4, max_lines=50, label="text")
+
+                with gr.Row():
+                    ner_language = gr.Dropdown(
+                        choices=ner_language_choices, value=ner_language_choices[0],
+                        label="language"
+                    )
+                    ner_engine = gr.Dropdown(
+                        choices=ner_engine_choices, value=ner_engine_choices[0],
+                        label="engine"
+                    )
+                ner_engine.change(
+                    ner_get_languages_by_engine,
+                    inputs=[ner_engine],
+                    outputs=[ner_language],
+                )
+                ner_output = gr.Textbox(lines=4, max_lines=50, label="output")
+                ner_button = gr.Button(value="pos_tag", variant="primary")
+                ner_button.click(
+                    run_ner,
+                    inputs=[ner_text, ner_language, ner_engine],
+                    outputs=[ner_output],
+                )
+                gr.Examples(
+                    examples=ner_examples,
+                    inputs=[ner_text, ner_language, ner_engine],
+                    outputs=[ner_output],
+                    fn=run_ner,
+                )
+
             with gr.TabItem("shell"):
                 shell_text = gr.Textbox(label="cmd")
                 shell_button = gr.Button("run")

ner_examples.json

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

toolbox/named_entity_recognization/__init__.py

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

toolbox/named_entity_recognization/named_entity_recognization.py

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

toolbox/named_entity_recognization/pyltp_ner.py

@@ -0,0 +1,37 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+from functools import lru_cache
+import os
+
+ltp_data_dir = os.environ.get("LTP_DATA_DIR")
+
+from pyltp import Postagger, Segmentor, NamedEntityRecognizer
+
+
+@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")
+    ner_model_path = os.path.join(ltp_data_dir, "ner.model")
+
+    segmentor = Segmentor(cws_model_path)
+    pos_tagger = Postagger(pos_model_path)
+    recognizer = NamedEntityRecognizer(ner_model_path)
+
+    return segmentor, pos_tagger, recognizer
+
+
+def pyltp_ner(text: str, language: str) -> list:
+    segmentor, pos_tagger, recognizer = get_pyltp_srl_tagger()
+
+    words = segmentor.segment(text)
+    postags = pos_tagger.postag(words)
+    ner_tags = recognizer.recognize(words, postags)
+
+    return words, postags, ner_tags
+
+
+if __name__ == "__main__":
+    pass

toolbox/sementic_role_labeling/pyltp_srl.py

@@ -14,14 +14,14 @@ from pyltp import Parser, Postagger, Segmentor, SementicRoleLabeller
 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')
+    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")
 
-    if platform.system() == 'Windows':
-        srl_model_path = os.path.join(ltp_data_dir, 'pisrl_win.model')
+    if platform.system() == "Windows":
+        srl_model_path = os.path.join(ltp_data_dir, "pisrl_win.model")
     else:
-        srl_model_path = os.path.join(ltp_data_dir, 'pisrl.model')
+        srl_model_path = os.path.join(ltp_data_dir, "pisrl.model")
 
     segmentor = Segmentor(cws_model_path)
     pos_tagger = Postagger(pos_model_path)