self-constructing_graph/graph/router/nlp_models.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy.sql import select, or_, and_, func,distinct
from db.schemas  import *
from db.models import *
from db.database import  get_db
from utils.response import resp_200
from typing import List
from math import ceil
import json
from datetime import datetime
import jieba
import jieba.posseg as pseg

router = APIRouter()

################################### MOD ###############################################
@router.get("/api/model-list/{page}/{page_size}")
def model_list_page(page: int, page_size: int, db: Session = Depends(get_db)):
    count = db.query(DbKgModels).count()

    total_page = ceil(count / page_size)

    start = 1
    if page <= total_page:
        start = (page - 1) * page_size

    results = db.query(DbKgModels).limit(page_size).offset(start).all()

    codes = [KgModels.model_validate(node) for node in results]

    return resp_200(
        data={"total": count, "pages": page, "size": page_size, "records": [item.model_dump() for item in codes]})


@router.post("/api/model-run")
def model_run(req: NLPRequest, db: Session = Depends(get_db)):
    if req.category == "word_cut":
        if req.name == "jieba":
            result = []
            category_list = []
            words = pseg.cut(req.content)
            index = 0
            for w, flag in words:
                data = {'name': w, 'category': flag, 'category_name': flag, 'start': index, 'end': index + len(w)}
                node = db.query(DbKgNode).filter(DbKgNode.name == w).first()
                if node != None:
                    data['category'] = node.category
                    data['category_name'] = node.category
                    if not (node.category in category_list):
                        category_list.append(node.category)
                    result.append(data)
                else:
                    if data['category'] == 'n':
                        result.append(data)
                index = index + len(w)
           
            return resp_200(data={"records": result})

    medical_re.load_schema("/home/kg-server/nlp/medical_re/predicate.json")
    model4s, model4po = medical_re.load_model()
    text = req.content
    # text = '据报道称，新冠肺炎患者经常会发热、咳嗽，少部分患者会胸闷、乏力，其病因包括: 1.自身免疫系统缺陷\n2.人传人。'  # content是输入的一段文字
    # text = '''诊断: 烫伤程度，分三度。 一度伤、烫伤只损伤皮肤表层，局部轻度红肿、无水泡、疼痛明显，应立即脱去衣袜后，将创面放入冷水中浸洗半小时，再用麻油、菜油涂擦创面。 二度伤、烫伤是真皮损伤，局部红肿疼痛，有大小不等的水泡，大水泡可用消毒针刺破水泡边缘放水，涂上烫伤膏后包扎，松紧要适度。 三度伤、烫伤是皮下，脂肪、肌肉、骨骼都有损伤，并呈灰或红褐色，此时应用干净布包住创面及时送往医院。切不可在创面上涂紫药水或膏类药物，影响病情况观察与处理。 鉴别诊断：水火烫伤处理的原则是首先除去热源，迅速离开现场，用各种灭火方法，如水浸、水淋、就地卧倒翻滚、立即将湿衣服脱去或剪破、淋水，将肢体浸泡在冷水中，直到疼痛消失为止。还可用湿毛巾或床单盖在伤处，再往上喷洒冷水。不要弄破水泡。 水火烫伤可分为一度烫伤(红斑性，皮肤变红，并有火辣辣的刺痛感)，二度烫伤(水泡性，患处产生水泡)，三度烫伤(坏死性，皮肤剥落)。对局部较小面积轻度烫伤，可在家中施治，在清洁创面后，可外涂京万红、美宝润湿烧伤膏等。对大面积烫伤，宜尽早送医院治疗。'''
    res = medical_re.get_triples(text, model4s, model4po)
    # print(json.dumps(res, ensure_ascii=False, indent=True))
    return resp_200(data={"records": res})

################################### NLP ###############################################
from nlp.medical_ner import medical_ner

@router.get("/api/nlp-parse")
def nlp_test():
    # with open("../nlp/medical_re/predicate.json", 'r', encoding='utf-8', errors='replace') as f:
    #     data = json.load(f)
    #     print(data.keys())
    #medical_re.load_schema("/home/kg-server/nlp/medical_re/predicate.json")
    #model4s, model4po = medical_re.load_model()
    #text = req.content
    #text = '据报道称，新冠肺炎患者经常会发热、咳嗽，少部分患者会胸闷、乏力，其病因包括: 1.自身免疫系统缺陷\n2.人传人。'  # content是输入的一段文字
    #text = '''诊断: 烫伤程度，分三度。 一度伤、烫伤只损伤皮肤表层，局部轻度红肿、无水泡、疼痛明显，应立即脱去衣袜后，将创面放入冷水中浸洗半小时，再用麻油、菜油涂擦创面。 二度伤、烫伤是真皮损伤，局部红肿疼痛，有大小不等的水泡，大水泡可用消毒针刺破水泡边缘放水，涂上烫伤膏后包扎，松紧要适度。 三度伤、烫伤是皮下，脂肪、肌肉、骨骼都有损伤，并呈灰或红褐色，此时应用干净布包住创面及时送往医院。切不可在创面上涂紫药水或膏类药物，影响病情况观察与处理。 鉴别诊断：水火烫伤处理的原则是首先除去热源，迅速离开现场，用各种灭火方法，如水浸、水淋、就地卧倒翻滚、立即将湿衣服脱去或剪破、淋水，将肢体浸泡在冷水中，直到疼痛消失为止。还可用湿毛巾或床单盖在伤处，再往上喷洒冷水。不要弄破水泡。 水火烫伤可分为一度烫伤(红斑性，皮肤变红，并有火辣辣的刺痛感)，二度烫伤(水泡性，患处产生水泡)，三度烫伤(坏死性，皮肤剥落)。对局部较小面积轻度烫伤，可在家中施治，在清洁创面后，可外涂京万红、美宝润湿烧伤膏等。对大面积烫伤，宜尽早送医院治疗。'''
    sentence = '''诊断: 烫伤程度，分三度。 一度伤、烫伤只损伤皮肤表层，局部轻度红肿、无水泡、疼痛明显，应立即脱去衣袜后，将创面放入冷水中浸洗半小时，再用麻油、菜油涂擦创面。 二度伤、烫伤是真皮损伤，局部红肿疼痛，。'''
    my_pred = medical_ner()
    res = my_pred.predict_sentence(sentence)
    print("---")
    #res = medical_re.get_triples(text, model4s, model4po)
    #print(json.dumps(res, ensure_ascii=False, indent=True))
    return resp_200(data={"records":res})

################################### DATASET ############################################
@router.get("/api/dataset-list/{page}/{page_size}")
def model_list_page(page: int, page_size: int, db: Session = Depends(get_db)):
    count = db.query(DbKgDataset).count()

    total_page = ceil(count / page_size)
    start = 1
    if page <= total_page:
        start = (page - 1) * page_size
    results = db.query(DbKgDataset).limit(page_size).offset(start).all()
    print(results)
    codes = [KgDataset.model_validate(node) for node in results]

    return resp_200(
        data={"total": count, "pages": page, 'total_page':total_page, "size": page_size, "records": [item.model_dump() for item in codes]})


nlp_router = router