ai_qback/tools/sql_format.py

# -*- coding: utf-8 -*-
from tools.new_mysql import MySQLUploader
from tools.loglog import logger

from core.respone_format import *


class CRUD:
    def __init__(self):
        self.m = MySQLUploader()
        self.people_place_name = [] 
        self.get_people_place_name()

    def get_word_by_wordid(self, wordid):
        s = "select WordSpelling from dictionary_word where Id = %s"
        r = self.m.query_data(s, (wordid,))
        if r:
           
            word = r[0][0]
            return word
        return None

    def get_wordid_by_wordspelling(self, wordspelling,auto_insert=False):
        """加一个功能。大字典内没有这个单词就自动插入，返回id。auto_insert为真，自动插入大字典，获取其id"""
        s = "select Id from dictionary_word where wordspelling = %s"
        r = self.m.query_data(s, (wordspelling,))
        if r:
           
            wordid = r[0][0]
            return wordid

        if auto_insert:
            s = "insert into dictionary_word (WordSpelling) VALUES (%s);"
            self.m.execute_(s,(wordspelling,))
            s = "select Id from dictionary_word where wordspelling = %s"
            r = self.m.query_data(s, (wordspelling,))
            wordid = r[0][0]
            return wordid

   
    def get_exchange_prototype(self,wordspelling):
        s = "select Word from dictionary_exchange where Word = %s"
        r = self.m.query_data(s, (wordspelling,))
        if r:
            return list({i[0] for i in r})
        s2 = "select Word from dictionary_exchange where InflectedWordSpelling = %s"
        r2 = self.m.query_data(s2, (wordspelling,))
        if r2:
            return list({i[0] for i in r2})

    def get_word_meaning_by_wordid(self, wordid):
        s = "select Id,WordSpelling,WordMeaning from dictionary_meaningitem where WordId = %s"
        r = self.m.query_data(s, (wordid,))
        return r

   
    def get_people_place_name(self):
        s2 = "select word from people_place_name"
        r = self.m.query_data(s2)
        for i in r:
            self.people_place_name.append(i[0])


   
    def get_word_meaning_by_wordspelling(self, wordspelling, frequency):
        """根据单词获取其全部词义"""
       
        wordid = self.get_wordid_by_wordspelling(wordspelling)

       
        return_data = {"word_id": wordid, "frequency": frequency, "word": wordspelling,
                       "meanings": {"default": [], "sun_english": {"name": "春笋英语", "items": []}, "oxford": {"name": "牛津", "items": []}}}

       
        s = "select Id,WordMeaning from dictionary_meaningitem where WordSpelling = %s"
        r = self.m.query_data(s, (wordspelling,))
        for row_data in r:
            return_data["meanings"]["default"].append({"id": row_data[0], "text": row_data[1]})

       
        s2 = "select Id,WordMeaning from dictionary_meaningitem_spring_bamboo where WordSpelling = %s"
        r2 = self.m.query_data(s2, (wordspelling,))
        for row_data in r2:
            return_data["meanings"]["sun_english"]["items"].append({"id": row_data[0], "text": row_data[1]})

       
        s2 = "select Id,WordMeaning from dictionary_meaningitem_oxford where WordSpelling = %s"
        r2 = self.m.query_data(s2, (wordspelling,))
        for row_data in r2:
            return_data["meanings"]["oxford"]["items"].append({"id": row_data[0], "text": row_data[1]})

        return return_data

    def delete_word_meaning_by_wordmeaningid(self, wordmeaningid):
        s = "DELETE FROM dictionary_meaningitem where Id = %s"
        r = self.m.execute_(s, (wordmeaningid,))
        logger.info(f"根据词义id删除，{wordmeaningid}。结果{r}")
        return True if r is True else False


    def get_word_all_info(self,word_id, spell,frequency):
        def get_associational_words_info(word_meaning_id) -> list:
            return_data = []
            s = "select Id,BaseWordMeaningId,BaseWord,BaseWordMeaning,AssociationalWord,AssociationalWordMeaningId,AssociationalWordMeaning," \
                "AssociationReason,ReverseAssociationReason,CreatedTime,UpdatedTime " \
                "from dictionary_associational_word where BaseWordMeaningId = %s"
            r = self.m.query_data(s, (word_meaning_id,))
            if not r:
                return return_data

            for single_meaning in r:
                associational_id, base_word_meaning_id, base_word, base_word_meaning, associational_word, \
                associational_word_meaning_id, associational_word_meaning, association_reason,\
                reverse_association_reason, created_time, updated_time = single_meaning
                r_data = {"id":associational_id,"base_word":{"word":base_word,"meaning_id":base_word_meaning_id,"meaning":base_word_meaning},
                          "associational_word":{"word":associational_word,"meaning_id":associational_word_meaning_id,"meaning":associational_word_meaning},
                          "association_reason":association_reason,"reverse_association_reason":reverse_association_reason,
                          "create_time":created_time.strftime('%Y-%m-%d %H:%M:%S'),"update_time":updated_time.strftime('%Y-%m-%d %H:%M:%S')}
                return_data.append(r_data)

            return return_data

        def get_phrases_info(word_meaning_id) -> list:
            return_data = []
            s = "select Id,PhraseSpellingText,PhraseChineseTranslation,FromType,CreatedTime,UpdatedTime " \
                "from dictionary_phrase where WordMeaningId = %s"
            r = self.m.query_data(s, (word_meaning_id,))
            if not r:
                return return_data
            for single_phrase in r:
                phrase_id, phrase_spelling_text, phrase_chinese_translation, from_type, created_time, updated_time = single_phrase
                r_data = {"id":phrase_id,"english":phrase_spelling_text,"chinese":phrase_chinese_translation,"from":from_type,
                          "create_time":created_time.strftime('%Y-%m-%d %H:%M:%S'),"update_time":updated_time.strftime('%Y-%m-%d %H:%M:%S')}
                return_data.append(r_data)

            return return_data

        def get_exchanges_info(word_id) -> list:
            return_data = []
            s = "select Id,InflectedWordSpelling,Properties,WordTransformationDescription " \
                "from dictionary_exchange where WordId = %s"
            r = self.m.query_data(s, (word_id,))
            if not r:
                return return_data
            for single_exchange in r:
                exchange_id,spell,properties,description = single_exchange
                r_data = {"id": exchange_id, "spell": spell, "properties": properties, "description": description}
                return_data.append(r_data)

            return return_data

        return_data_all = {"word_id":word_id,"spell":spell,"frequency":frequency,"meanings":[],"exchanges":[]}
        if spell in self.people_place_name:
            return_data_all["type"] = "人名地名"
        return_data_all["type"] = "一般词汇"

        s = "select Id,WordMeaning,OperateAccount from dictionary_meaningitem where WordId = %s"
        r = self.m.query_data(s, (word_id,))
        if not r: 
            return resp_200(data=return_data_all)
        for single_meaning in r:
            meaning_id, word_meaning, operate_account = single_meaning
            single_meaning_item = {"id":meaning_id,"text":word_meaning,"editor":operate_account}
            associational_words_list = get_associational_words_info(meaning_id)
            single_meaning_item["associational_words"] = associational_words_list
            phrases_list = get_phrases_info(meaning_id)
            single_meaning_item["phrases"] = phrases_list
            return_data_all["meanings"].append(single_meaning_item)

        exchanges_info_list = get_exchanges_info(word_id)
        return_data_all["exchanges"] = exchanges_info_list

        return resp_200(data=return_data_all)

    def delete_associational_word(self,word_id,associational_id):
        s = "select Id from dictionary_meaningitem where WordId = %s"
        r = self.m.query_data(s, (word_id,))
        if not r:
            return resp_400(message="词义表内没有这个单词的词义")
        meaning_id = r[0][0]

        s = "select BaseWordMeaningId from dictionary_associational_word where Id = %s"
        r = self.m.query_data(s, (associational_id,))
       
        if r and r[0][0]==meaning_id:
            s = "DELETE FROM dictionary_associational_word where Id = %s"
            r = self.m.execute_(s, (associational_id,))
            logger.info(f"根据联想词id删除，{associational_id}。结果{r}")
            data = True if r is True else False
            return resp_200(data=[data]) if data else resp_500(message="数据库内部错误")
        else:
            logger.info(f"删除联想词时，单词id与联想词id校验失败。{r} {meaning_id}")
            return resp_400(message="单词id与联想词id校验失败")

    def delete_phrese_word(self,word_id,phrase_id):
        s = "select Id from dictionary_meaningitem where WordId = %s"
        r = self.m.query_data(s, (word_id,))
        if not r:
            return resp_400(message="词义表内没有这个单词的词义")
        meaning_id = r[0][0]

        s = "select WordMeaningId from dictionary_phrase where Id = %s"
        r = self.m.query_data(s, (phrase_id,))
       
        if r and r[0][0] == meaning_id:
            s = "DELETE FROM dictionary_phrase where Id = %s"
            r = self.m.execute_(s, (phrase_id,))
            logger.info(f"根据联想词id删除，{phrase_id}。结果{r}")
            data = True if r is True else False
            return resp_200(data=[data]) if data else resp_500(message="数据库内部错误")
        else:
            logger.info(f"删除语块时，单词id与语块id校验失败。{r} {meaning_id}")
            return resp_400(message="单词id与联想词id校验失败")

    def close_connection(self):
        """关闭数据库连接"""
        self.m.close_connection()


class UserCRUD:
    def __init__(self):
        self.m = MySQLUploader()

    def get_userinfo_by_account(self, account):
        s = "select id,account,password,uname,create_time from user where account = %s"
        r = self.m.query_data(s, (account,))
        if r:
           
            user_info = (r[0][0], r[0][1], r[0][2], r[0][3], r[0][4].strftime('%Y-%m-%d %H:%M:%S'))
            return user_info
        return None

    def close_connection(self):
        """关闭数据库连接"""
        self.m.close_connection()


if __name__ == '__main__':
    crud = CRUD()
   
   
   
   
   

    r = crud.get_wordid_by_wordspelling("abcdefg")
    print(type(r))
    print(r)

    crud.close_connection()