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()