[关闭]
@heavysheep 2020-11-03T03:15:29.000000Z 字数 12014 阅读 789

代码质量-2020年10月

代码质量

BAD CODE-1

代码信息

项目: 北京网信办数据中心
归属人: 王子
业务逻辑: 审核文本的多个方法类，包含对不同关键词、规则的匹配、同音、象形等匹配方法。
代码链接: https://git.datagrand.com/wbyy_group/bjwxb/bjwxb_alarm/tree/master/app/controller
位置: 各方法类

代码片段

  1. class WordController(object):
  2.     data_path = "data"
  4.     def __init__(self):
  5.         self._element = "match"
  6.         self._word_dict = self._load_dict()
  7.         self._ac = self._compile_ac()
  9.     def match(self, text: str):
  10.         ...
  12.     def parser(self, text):
  13.         match_result = self.match(text)
  14.         for k in match_result:
  15.             match_result[k] = 0.99 if len(match_result[k]) > 1 else 0.9
  16.         return match_result
  18.     def _load_dict(self):
  19.         ...
  21.     def _compile_ac(self):
  22.         ac = ahocorasick.Automaton()
  23.         for w in self._word_dict.keys():
  24.             if w.strip():
  25.                 ac.add_word(w, w)
  26.         ac.make_automaton()
  27.         return ac
  1. class PinYinController(object):
  2.     data_path = "data"
  4.     def __init__(self):
  5.         self._element = "pinyin"
  6.         self._pinyin_dict = self._load_dict()
  7.         self._ac = self._compile_ac()
  9.     def match(self, text: str):
  10.         ...
  12.     def parser(self, text):
  13.         match_result = self.match(text)
  14.         for k in match_result:
  15.             match_result[k] = pow(2.2, len(match_result)) / 100
  16.         return match_result
  18.     def _load_dict(self):
  19.         ...
  20.         return pinyin_dict
  22.     def _compile_ac(self):
  23.         ac = ahocorasick.Automaton()
  24.         for w in self._pinyin_dict.keys():
  25.             if w.strip():
  26.                 ac.add_word(w, w)
  27.         ac.make_automaton()
  28.         return ac
  30.     @staticmethod
  31.     def _join(word: str, join_str: str = "_"):
  32.         return join_str.join(pypinyin.lazy_pinyin(word))

问题

  1. 不同方法类的部分方法重复，没有较好的抽象出基类；
  2. 缺乏注释、说明和示例，不能较好的说明业务和方法。

BAD CODE-2

代码信息

项目: 北京网信办数据中心
归属人: 韩伟
业务逻辑: 从ES中读取批量数据并聚类相似事件
代码链接: https://git.datagrand.com/wbyy_group/bjwxb/bjwxb_alarm/tree/master/app/controller
位置: 全文件

代码片段

  1. class ThreadWithReturn(threading.Thread):
  2.     """
  3.     定义带返回值的线程类
  4.     Args:
  5.         threading ([type]): [description]
  6.     """
  8.     def __init__(self, func, args=()):
  9.         super(ThreadWithReturn, self).__init__()
  10.         self.func = func
  11.         self.args = args
  13.     def run(self):
  14.         self.result = self.func(*self.args)
  16.     def get_result(self):
  17.         try:
  18.             return self.result
  19.         except Exception:
  20.             return None
  23. class SimiFinder:
  25.     def __init__(self):
  26.         self.es_recall = ESRecall()
  27.         self.text_simi_ = TextSimi()
  29.     @time_cost
  30.     def simi_finder(self, text_list):
  31.         try:
  32.             info_dict = self.process_data(text_list)
  34.             t_es = ThreadWithReturn(self._get_es_simi, args=(info_dict,))
  35.             t_es.start()
  36.             t_self_simi = ThreadWithReturn(
  37.                 self._get_self_simi, args=(info_dict,))
  38.             t_self_simi.start()
  40.             t_es.join()
  41.             es_simi_dict = t_es.get_result()
  42.             t_self_simi.join()
  43.             self_simi_dict = t_self_simi.get_result()
  45.             for i in range(len(text_list)):
  46.                 term_dict = text_list[i]
  47.                 item_id = term_dict['fields']['itemid']
  48.                 item_es_simi_list = es_simi_dict.get(item_id, [])
  49.                 item_self_simi_list = self_simi_dict.get(item_id, [])
  50.                 simi_result = item_es_simi_list + item_self_simi_list
  51.                 if simi_result:
  52.                     term_dict['fields'].setdefault('has_repeated', True)
  53.                 else:
  54.                     term_dict['fields'].setdefault('has_repeated', False)
  55.                 term_dict['fields'].setdefault('repeatedItems', simi_result)
  56.                 text_list[i] = term_dict
  57.         except:
  58.             import traceback
  59.             logger.error('Get simi error: {}'.format(traceback.format_exc()))
  60.             text_list=[]
  62.         return text_list

问题

  1. 未发现使用在Flask内部使用线程的必要原因
  2. 函数间职责分离不够彻底
  3. 业务逻辑嵌套抽象不足，较难阅读

BAD CODE-3

代码信息

项目: 某情感分类项目
归属人: 房悦竹
业务逻辑: 获取评论内容的情感得分
代码链接: https://git.datagrand.com/wbyy_group/text_sentiment/blob/master/text_sentiment/src/sentiment_dict.py
位置: 88:207

代码片段

  1. ...
  3.     # @timing
  4.     def get_review_score(self, review):
  5.         """
  6.         返回基于通用词典规则的评论得分
  7.         :param review: 评论内容
  8.         :return: review_score: [x, y], x-正面分, y-负面分, result: {0:x, 1:y}, x-负面分, y-正面分, hit_word_list:
  9.         """
  10.         sentences = tp.split_sentence(review)
  11.         senti_score = []
  12.         hit_word_list = []
  14.         for sen in sentences:
  15.             zhuanzhe_flag = 0  # set zhuanzhe flag
  16.             # seg = self.wordseg.cut(sen, with_stop_word=False, model=None)
  17.             seg = self.wordseg.lcut(sen)
  18.             seg = process_contain_no(seg)  # 特殊处理分词结果的“不”字
  19.             cur_word_pos = 0  # current word position
  20.             senti_word_pos = 0  # sentiment word position
  21.             pos_score = 0
  22.             neg_score = 0
  23.             seg_len = len(seg)
  24.             for word in seg:
  25.                 current_neg = 0
  26.                 current_pos = 0
  27.                 if word in self.zhuanzhe_dict:
  28.                     zhuanzhe_flag = self.zhuanzhe_dict[word]
  29.                     hit_word = 'zhuanzhe:' + word
  30.                     hit_word_list.append(hit_word)
  32.                 if word in self.pos_dict:
  33.                     hit_word = 'pos:' + word
  34.                     hit_word_list.append(hit_word)
  35.                     doubt_flag = False  # 是否找到疑问词
  36.                     current_pos += 1
  37.                     start_pos = max(cur_word_pos - 5, senti_word_pos)
  39.                     for w in seg[start_pos:cur_word_pos]:
  40.                         current_pos = self.match_adv(w, current_pos, hit_word_list)
  42.                     if cur_word_pos != 0:
  43.                         s_idx = max([cur_word_pos - 2, 0])  # 往前找两个词
  44.                         for w in seg[s_idx:cur_word_pos]:
  45.                             current_pos, doubt_flag, hit_word_list = self.match_doubt_forward(w, current_pos,
  46.                                                                                               hit_word_list)
  47.                             if doubt_flag:
  48.                                 break
  50.                     if not doubt_flag:  # 找到疑问词后不再向后寻找
  51.                         if cur_word_pos < seg_len - 1:
  52.                             e_idx = min([cur_word_pos + 3, seg_len])  # 往后寻找两个词
  53.                             for w in seg[cur_word_pos + 1:e_idx]:
  54.                                 current_pos, doubt_flag, hit_word_list = self.match_doubt_backward(w, current_pos,
  55.                                                                                                    hit_word_list)
  56.                                 if doubt_flag:
  57.                                     break
  59.                     pos_score += current_pos
  60.                     senti_word_pos = cur_word_pos
  62.                 elif word in self.neg_dict:
  63.                     hit_word = 'neg:' + word
  64.                     hit_word_list.append(hit_word)
  65.                     doubt_flag = False  # 是否找到疑问词
  66.                     current_neg += 1
  67.                     start_pos = max(cur_word_pos - 5, senti_word_pos)
  68.                     for w in seg[start_pos:cur_word_pos]:
  69.                         current_neg = self.match_adv(w, current_neg, hit_word_list)
  70.                     if cur_word_pos != 0:
  71.                         s_idx = max([cur_word_pos - 2, 0])  # 往前找两个词
  72.                         for w in seg[s_idx:cur_word_pos]:
  73.                             current_neg, doubt_flag, hit_word_list = self.match_doubt_forward(w, current_neg,
  74.                                                                                               hit_word_list)
  75.                             if doubt_flag:
  76.                                 break
  78.                     if not doubt_flag:  # 找到疑问词后不再向后寻找
  79.                         if cur_word_pos < seg_len - 1:
  80.                             e_idx = min([cur_word_pos + 3, seg_len])  # 往后寻找两个词
  81.                             for w in seg[cur_word_pos + 1:e_idx]:
  82.                                 current_neg, doubt_flag, hit_word_list = self.match_doubt_backward(w, current_neg,
  83.                                                                                                    hit_word_list)
  84.                                 if doubt_flag:
  85.                                     break
  86.                     neg_score += current_neg
  87.                     senti_word_pos = cur_word_pos
  89.                 elif word in self.insult_dict:
  90.                     hit_word = 'insult:' + word
  91.                     hit_word_list.append(hit_word)
  92.                     current_neg += 1 * 2  # 对在insult_dict中的词赋予2.0的权重
  93.                     neg_score += current_neg
  95.                 cur_word_pos += 1
  96.             pos_score, neg_score = zhuanzhe_process(zhuanzhe_flag, pos_score, neg_score, senti_score)
  97.             senti_score.append(transform_to_positive_num(pos_score, neg_score))
  99.         # pay more attention to last sentence
  100.         if senti_score[-1][0] > senti_score[-1][1]:
  101.             senti_score[-1][0] *= 1.5
  102.         if senti_score[-1][1] > senti_score[-1][0]:
  103.             senti_score[-1][1] *= 1.5
  105.         review_score = sum_sentences_score(senti_score)  # [Pos, Neg]
  107.         result = {}
  108.         if review_score[0] == 0 and review_score[1] == 0:
  109.             result[0] = 0.5
  110.             result[1] = 0.5
  111.         elif review_score[0] == 0:  # 词典正面分为0
  112.             result[0] = 0.75  # 原值为1，改为0.8
  113.             result[1] = 0.25
  114.         elif review_score[1] == 0:  # 词典负面分为0
  115.             result[0] = 0.25
  116.             result[1] = 0.75
  117.         else:
  118.             temp_sum = review_score[0] + review_score[1]
  119.             result[0] = round(review_score[1] / temp_sum, 2)
  120.             result[1] = round(review_score[0] / temp_sum, 2)
  122.         return review_score, result, hit_word_list
  124. ...

问题

  1. 函数代码过长，未实现良好抽象；
  2. 嵌套逻辑过多，导致可读性差和维护困难；
  3. 某些权重参数应做一定说明，描述定义依据。

GOODCODE-1

代码信息

项目: VOC核心
业务逻辑: 封装ES DSL
代码链接: https://git.datagrand.com/voc/voc_api/blob/master/app/common/es.py
位置: 25:368行

代码片段

  1. class ESQuery(object):
  2.     def __init__(self, index: str = None, doc_type: str = None, bool: dict = None, collapse: dict = None,
  3.                  dsl: dict = None):
  4.         """
  5.         ES 查询dsl封装
  6.         :param index: ES index
  7.         :param doc_type: ES doc_type(如有)
  8.         :param bool: bool dsl
  9.         :param collapse: collapse dsl.折叠dsl
  10.         :param dsl: source dsl.传入时直接生成dsl，和bool collapse等解析参数有所出入
  12.         example:
  13.             # [build condition]
  14.             filter_condition = Query.term(field="is_match", value=True)
  15.             agg_condition = [Aggs.terms(name="group", field="website.keyword", size=500)]
  17.             # [input query]
  18.             query = ESQuery(index="index_1", doc_type="type_1", bool=bools(filter=filter_condition))
  19.             query.aggs(condition=agg_condition)
  21.             # [modify attr]
  22.             query.size(0)
  23.             query.sort("datetime:desc")
  25.             # [get result]
  26.             query_result, _ = query.search()
  27.         """
  28.         if dsl and bool:
  29.             raise ValueError("不允许同时传入'dsl' 'bool'参数")
  30.         if dsl and collapse:
  31.             raise ValueError("不允许同时传入'dsl' 'collapse'参数")
  33.         self.index = index
  34.         self.doc_type = doc_type
  36.         self.params = {}
  38.         self.bool = bool
  40.         self.dsl = dsl if dsl is not None else {}
  41.         if bool:
  42.             self.dsl["query"] = bool
  43.         if collapse:
  44.             self.dsl["collapse"] = collapse
  46.     def bools(self, must: (dict, list) = None, filter: (dict, list) = None, should: (dict, list) = None,
  47.               must_not: (dict, list) = None):
  48.         """
  49.         bool 条件
  50.         :param must: must条件
  51.         :param filter: filter条件
  52.         :param should: should条件
  53.         :param must_not: must not条件
  54.         """
  55.         for name, condition_info in (("must", must), ("filter", filter), ("should", should), ("must_not", must_not)):
  56.             if self.dsl.get("query") is None:
  57.                 self.dsl["query"] = {"bool": {}}
  58.             if self.dsl["query"].get("bool") is None:
  59.                 self.dsl["query"]["bool"] = {}
  60.             if condition_info is not None:
  61.                 condition = _parser_condition(condition_info)
  62.                 if name not in self.dsl["query"]["bool"]:
  63.                     self.dsl["query"]["bool"][name] = condition
  64.                 else:
  65.                     self.dsl["query"]["bool"][name].extend(condition)
  67.     def aggs(self, condition):
  68.         if isinstance(condition, list):
  69.             condition = {k: v for c in condition for k, v in c.items()}
  70.         self.dsl["aggs"] = condition
  72.     def range(self, range_condition: dict):
  73.         """
  74.         区间条件配置方法 对应DSL _source
  75.         ES文档: [https://www.elastic.co/guide/en/elasticsearch/reference/5.5/query-dsl-range-query.html]
  76.         :param range_condition: 区间条件
  77.         """
  78.         if self.dsl.get("query") is None:
  79.             self.dsl["query"] = {"bool": {}}
  80.         if self.dsl["query"].get("bool") is None:
  81.             self.dsl["query"]["bool"] = {}
  82.         if self.dsl["query"]["bool"].get("must") is None:
  83.             self.dsl["query"]["bool"]["must"] = range_condition
  84.             return
  86.         for i, condition in enumerate(self.dsl["query"]["bool"]["must"]):
  87.             if "range" in condition.keys():
  88.                 self.dsl["query"]["bool"]["must"][i] = range_condition
  89.                 return
  90.         self.dsl["query"]["bool"]["must"].append(range_condition)
  92.     def source(self, include: (str, list) = None, exclude: (str, list) = None):
  93.         """
  94.         字段配置方法 对应DSL _source
  95.         ES文档: [https://www.elastic.co/guide/en/elasticsearch/reference/5.5/search-request-source-filtering.html]
  96.         :param include: 保留字段
  97.         :param exclude: 不保留字段
  98.         """
  99.         self.dsl["_source"] = {}
  100.         if include:
  101.             self.dsl["_source"]["includes"] = include
  102.         if exclude:
  103.             self.dsl["_source"]["excludes"] = exclude
  105.     def pagination(self, page: int = 1, per_page: int = 20):
  106.         """
  107.         分页方法 对应DSL from/size
  108.         ES文档: [https://www.elastic.co/guide/en/elasticsearch/reference/5.5/search-request-from-size.html]
  109.         :param page: 页数，从1开始，默认1
  110.         :param per_page: 页容量，默认20
  111.         """
  112.         if page is None:
  113.             page = 1
  114.         if per_page is None:
  115.             per_page = 20
  117.         from_ = (page - 1) * per_page
  118.         self.dsl["from"] = from_
  119.         self.dsl["size"] = per_page
  121.     def sort(self, condition: (str, dict)):
  122.         """
  123.         排序方法，对应DSL sort
  124.         ES文档: [https://www.elastic.co/guide/en/elasticsearch/reference/5.5/search-request-sort.html]
  125.         注意此方法是对query对象的排序方法
  126.         :param condition: 排序条件
  127.         """
  128.         if isinstance(condition, dict):
  129.             condition = ",".join(("{}:{}".format(c_k, c_v) for c_k, c_v in condition.items()))
  130.         self.params["sort"] = condition
  132.     def size(self, num: int):
  133.         """
  134.         单容量方法，对应DSL size
  135.         注意和pagination有冲突
  136.         :param num: 容量
  137.         """
  138.         self.dsl.update({"size": num})
  140.     def scroll(self, scroll: str):
  141.         """
  142.         滚动方法 对应DSL scroll
  143.         ES文档: [https://www.elastic.co/guide/en/elasticsearch/reference/5.5/search-request-scroll.html]
  144.         :param scroll: 滚动ID
  145.         """
  146.         self.params.update({"scroll": scroll})
  148.     def search(self, debug=False):
  149.         """
  150.         search方法 对应ES request body search方法
  151.         ES文档: [https://www.elastic.co/guide/en/elasticsearch/reference/5.5/search-uri-request.html]
  152.         :param debug: 是否打印(info日志)DSL request body
  153.         """
  154.         if debug:
  155.             logger.info(json.dumps(self.dsl, ensure_ascii=False, indent=4))
  156.         if self.dsl.get("size") is None:
  157.             self.size(50)
  158.         result = es.search(body=self.dsl, index=self.index, doc_type=self.doc_type, params=self.params)
  159.         result.update({"pager_info": self._get_page_info(result)})
  160.         return result
  162.     def _get_page_info(self, result: dict):
  163.         """
  164.         分页信息方法，用于返回分页相关情况说明
  165.         :param result: ES查询结果
  166.         :return: 分页相关信息
  167.         """
  168.         pager_info = {}
  170.         if self.dsl.get("from") is not None:
  171.             pager_info["page"] = self.dsl["from"]
  172.             pager_info["per_page"] = self.dsl["size"]
  173.             pager_info["total"] = result["hits"]["total"]
  174.             pager_info["pages"] = math.ceil(pager_info["total"] / pager_info["per_page"])
  176.         if self.params.get("sort") is not None:
  177.             pager_info["order_by"] = self.params["sort"]
  179.         return pager_info
  181.     def clone(self):
  182.         return ESQuery(index=self.index, doc_type=self.doc_type, dsl=deepcopy(self.dsl))

亮点

统一封装了ES DSL，避免操作字典描述逻辑，更具可读性。

GOODCODE-2

代码信息

项目: BERT文本相似度
业务逻辑: 推断文本相似度业务方法
代码链接: https://git.datagrand.com/wbyy_group/text_simi_bert/blob/master/app/src/text_simi.py
位置: 79:120行

代码片段

  1.     def get_simi(self, query_list, candidate_list=None, simi='bert'):
  2.         """
  3.         返回相似度
  4.         :param self_simi: 是否自相似
  5.         :param query_list: query 文本list
  6.         :param candidate_list: 待相似文本list
  7.         :return: 两个list的相似度，[len(query_list), len(candidate_list)]
  8.         """
  9.         if candidate_list:
  10.             corpus_list = query_list + candidate_list
  11.         else:
  12.             corpus_list = query_list
  14.         if simi != 'bert':
  15.             # 分词
  16.             feature_list = [[word for word in jieba.cut(text) if word not in self.stop_word_dict] for text in corpus_list]
  18.             # 建立字典
  19.             dictionary = Dictionary(feature_list)
  20.             corpus_features = [dictionary.doc2bow(
  21.                 feature) for feature in feature_list]
  23.             # 构建tfidf
  24.             tf_idf_model = TfidfModel(corpus_features)
  25.             query_tfidf = tf_idf_model[corpus_features[:len(query_list)]]
  26.             candidate_tfidf = tf_idf_model[corpus_features[len(query_list):]]
  28.             # 计算相似度
  29.             sparse_matrix = SparseMatrixSimilarity(
  30.                 candidate_tfidf, len(dictionary.token2id))
  31.             simi_result = sparse_matrix.get_similarities(query_tfidf)
  32.         else:
  33.             query_vec = self._get_vecs(query_list)
  34.             if candidate_list:
  35.                 corpus_vec = self._get_vecs(candidate_list)
  36.             else:
  37.                 corpus_vec = query_vec
  38.             label_vec = query_vec / (query_vec ** 2).sum(axis=1, keepdims=True) ** 0.5
  39.             corpus_list = corpus_vec / (corpus_vec ** 2).sum(axis=1, keepdims=True) ** 0.5
  40.             simi_result = np.dot(label_vec, corpus_list.T)
  42.         return simi_result

亮点

  1. 逻辑清晰，分离美观，方法说明及注释比较到位
添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注