poyee-data-warehouse/dw_base/database/mongodb_utils.py

#!/usr/bin/env /usr/bin/python3
# -*- coding:utf-8 -*-

from typing import List

from pymongo import MongoClient


class MongoDBHandler:
    def __init__(self, url: str, port: int = None, username: str = None, password: str = None, check_db: str = None,
                 extra_config: List[str] = None):
        if not port or not username or not password or not check_db:
            uri = url
        else:
            uri = f'mongodb://{username}:{password}@{url}:{port}/{check_db}'
            if extra_config:
                uri = f'{uri}?{"&".join(extra_config)}'
        self.mongo_client = MongoClient(uri)


# ES_CLIENT = Elasticsearch('es-cn-nif1oiv5w0009di0f.public.elasticsearch.aliyuncs.com:9200',
#                               http_auth=('datafix', 'Aa123456'))
#
# INCLUDES = ['id', 'appInfo', 'appName', 'b2bClient', 'b2bInfo', 'b2bOpScope', 'b2bProduct', 'baikeInfo', 'blogName',
#                 'buyingInfo', 'certL2Name', 'contactAddress', 'entName', 'historyName', 'jobName', 'licenseContent',
#                 'licenseFileName', 'licenseOffice', 'miniAppName', 'opScope', 'patentName', 'semKeyword', 'semTitle',
#                 'seoInfo', 'seoKeyword', 'seoTitle', 'siteName', 'softwareProductName', 'tenderInfo', 'tenderName',
#                 'trademarkName', 'wechatName']
#
# def get_pid_by_name(company_name):
#     ent_col = MONGO_CLIENT.get_database('enterprise').get_collection('EnterpriseBaseInfo')
#     query = {'ENTNAME': company_name}
#     projection = {'PID': 1}
#     res_doc = ent_col.find_one(query, projection)
#     if res_doc and 'PID' in res_doc:
#         return res_doc['PID']
#
#     return None
#
#
# def search_by_name(company_name):
#     pid = get_pid_by_name(company_name)
#     print('pid: %s' % pid)
#     if pid:
#         company_details = search_by_pid(pid)
#         return company_details
#     return None
#
#
# def get_detail_from_es(index, pid, includes=['id']):
#     try:
#         return ES_CLIENT.get(index=index, id=pid, _source_includes=includes)
#     except NotFoundError as not_found_e:
#         pass
#     except Exception as e:
#         print('Unknown error when get detail from es. error:%s' % str(e))
#
#     return None
#
#
# def search_by_pid(pid):
#     return get_detail_from_es('company_info_prod', pid, INCLUDES)
#
#
# def main(local_save_path):
#     names = fp.readlines()
#     result_docs = []
#     for name in names:
#         print('company name: %s' % name)
#         detail = search_by_name(name.replace('\n', '').strip())
#         if detail is None:
#             print('can not found detail for name:%s' % name)
#             continue
#         source = detail['_source']
#         current_doc = []
#         # 为了保证顺序，需按INCLUDES遍历获取字段值
#         for field in INCLUDES:
#             if field in source:
#                 if isinstance(source[field], str) and source[field] != '':
#                     current_doc.append(source[field].replace('\t', ';').replace('\n', ';').replace('\r', ';'))
#                 elif isinstance(source[field], list) and source[field] != []:
#                     current_doc.append(
#                         ','.join(source[field]).replace('\t', ';').replace('\n', ';').replace('\r', ';'))
#                 else:
#                     current_doc.append('')
#             else:
#                 current_doc.append('')
#         result_docs.append(current_doc)
#     res_df = pd.DataFrame(result_docs, mysql_column_list=INCLUDES)
#     res_df.to_csv(local_save_path, sep='\t', index=False, encoding='utf-8', header=True)
#     # os.system("source /etc/profile;hadoop fs -put %s %s" % (local_save_path, hdfs_save_path))
#
#
# FLAGS = set('n ng nrfg nrt nt vn un'.split())
#
#
# def lcut(text):
#     if isinstance(text, list):
#         print(text)
#     return [
#         w.word for sentence in split('[^a-zA-Z0-9\u4e00-\u9fa5]+', text.strip())
#         for w in dt.cut(sentence) if len(w.word) > 2 and w.flag in FLAGS]
#
#
# class TFIDF:
#     def __init__(self, dictionary, model):
#         self.model = model
#         self.doc2bow = dictionary.doc2bow
#         self.id2word = {i: w for w, i in dictionary.token2id.items()}
#
#     @classmethod
#     def train(cls, texts):
#         texts1 = [lcut(text) for text in texts]
#         dictionary = Dictionary(texts1)
#         corpus = [dictionary.doc2bow(text) for text in texts1]
#         model = TfidfModel(corpus)
#         return cls(dictionary, model)
#
#     def extract(self, text, top_n=10):
#         vector = self.doc2bow(lcut(text))
#         key_words = sorted(self.model[vector], key=lambda x: x[1], reverse=True)
#         return [self.id2word[i] for i, j in key_words][:top_n]
#
#
# def extract_keywords(mysql_column_list):
#     columns_list = raw_data_df[mysql_column_list].values.tolist()
#     combined = [';'.join(row) for row in columns_list]
#     keywords = tf_idf_model.extract(';'.join(combined), top_n=50)
#     print(','.join(keywords))
#     return keywords
#
#
# if __name__ == '__main__':
#     fp = open(get_abs_path('data/seed_company_name_yidong.csv'))
#     local_save_path = '/root/wwj-hive-warehouse/data/zhong_qi_customer_info_yidong.csv'
#     hdfs_save_path = ''
#
#     main(local_save_path)
#     raw_data_df = pd.read_csv(local_save_path, sep='\t', encoding='utf-8')
#     raw_data_df = raw_data_df.fillna('')
#     raw_data_df['combined'] = raw_data_df[INCLUDES].apply(lambda row: ';'.join(row.values.astype(str)), axis=1)
#     all_text_info_list = raw_data_df['combined'].values.tolist()
#     tf_idf_model = TFIDF.train(all_text_info_list)
#
#     op_scope_keywords = extract_keywords(['opScope', 'b2bOpScope'])
#     product_keywords = extract_keywords(['appInfo', 'appName', 'b2bProduct'])
#     jobName_keywords = extract_keywords(['jobName'])
#     semKeyword_keywords = extract_keywords(['semKeyword'])
#     baike_keywords = extract_keywords(['baikeInfo'])
#     b2bInfo_keywords = extract_keywords(['b2bInfo'])
#     # opScope = raw_data_df[['opScope']].values.tolist()
#     # opScope_keywords = tf_idf_model.extract(opScope)
#     #
#     # opScope = raw_data_df[['opScope']].values.tolist()
#     # opScope_keywords = tf_idf_model.extract(opScope)
#     #
#     # opScope = raw_data_df[['opScope']].values.tolist()
#     # opScope_keywords = tf_idf_model.extract(opScope)
#     # pid = get_pid_by_name('杭州德玛瑞户外用品有限公司')
#     # print(pid)

# mongo_url_a42 = 'mongodb://dw_all_ro:Dt#R30ES@' \
#                 'dds-m5e44df0967967a41.mongodb.rds.aliyuncs.com:3717,' \
#                 'dds-m5e44df0967967a42.mongodb.rds.aliyuncs.com:3717,' \
#                 'dds-m5e44df0967967a43.mongodb.rds.aliyuncs.com:3717' \
#                 '/admin?replicaSet=mgset-12596773&readReference=secondaryPreferred'
# a42_handler = MongoDBHandler(mongo_url_a42)
# mongo_url_b41 = 'mongodb://dw_all_rw:W#ioQseT@' \
#                 'dds-m5ed9ea9d9a653b41.mongodb.rds.aliyuncs.com:3717,' \
#                 'dds-m5ed9ea9d9a653b42.mongodb.rds.aliyuncs.com:3717,' \
#                 'dds-m5ed9ea9d9a653b43.mongodb.rds.aliyuncs.com:3717' \
#                 '/admin?replicaSet=mgset-45687639&readReference=secondaryPreferred'
# b41_handler = MongoDBHandler(mongo_url_b41)
# mongo_url_dev = 'mongodb://dev_dw_ro:Dt#R30ES@dds-m5e686962c7b71641431-pub.mongodb.rds.aliyuncs.com:3717/admin'
# dev_handler = MongoDBHandler(mongo_url_dev)
#
# if __name__ == '__main__':
#     mongo_handler = MongoDBHandler(url='dds-m5e44df0967967a42.mongodb.rds.aliyuncs.com',
#                                    port=3717,
#                                    username='dw_all_ro',
#                                    password='Dt#R30ES',
#                                    check_db='admin')
#     for db in mongo_handler.mongo_client.list_database_names():
#         print(db)