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