init project

.gitignore

@@ -0,0 +1,148 @@
+### Example user template template
+### Example user template
+
+# IntelliJ project files
+.idea
+*.iml
+out
+gen
+### Python template
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+

handle/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+# @Time : 2022/11/29 15:11
+# @Author : XuJiakai
+# @File : __init__.py
+# @Software: PyCharm
+
+if __name__ == '__main__':
+    pass

handle/search_winhc_data.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+# @Time : 2022/11/29 15:11
+# @Author : XuJiakai
+# @File : search_winhc_data
+# @Software: PyCharm
+
+if __name__ == '__main__':
+    pass

log.py

@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+# @Time : 2021/6/7 11:02
+# @Author : XuJiakai
+# @File : log
+# @Software: PyCharm
+
+import logging
+import os
+import time
+
+
+def get_log(log_name="my_log", write_file=False):
+    logging.basicConfig()
+    logger = logging.getLogger(log_name)
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+    # 日志输出格式
+    formatter = logging.Formatter('[%(asctime)s - ' + log_name + ':%(lineno)d] - %(levelname)s: %(message)s')
+
+    if write_file:
+        # 创建logs文件夹
+        cur_path = os.path.dirname(os.path.realpath(__file__))
+        log_path = os.path.join(cur_path, 'logs')
+        # 如果不存在这个logs文件夹，就自动创建一个
+        if not os.path.exists(log_path):
+            os.mkdir(log_path)
+        log_name = os.path.join(log_path, '%s.log' % time.strftime('%Y_%m_%d'))
+        fh = logging.FileHandler(log_name, 'a', encoding='utf-8')  # 这个是python3的
+        # 创建一个FileHandler，用于写到本地
+        fh.setLevel(logging.INFO)
+        fh.setFormatter(formatter)
+        logger.addHandler(fh)
+
+    # 创建一个StreamHandler,用于输出到控制台
+    ch = logging.StreamHandler()
+    ch.setLevel(logging.INFO)
+    ch.setFormatter(formatter)
+    logger.addHandler(ch)
+
+    return logger
+
+
+if __name__ == '__main__':
+    print(''.encode('utf-8'))

main.py

@@ -0,0 +1,16 @@
+# This is a sample Python script.
+
+# Press Shift+F10 to execute it or replace it with your code.
+# Press Double Shift to search everywhere for classes, files, tool windows, actions, and settings.
+
+
+def print_hi(name):
+    # Use a breakpoint in the code line below to debug your script.
+    print(f'Hi, {name}')  # Press Ctrl+F8 to toggle the breakpoint.
+
+
+# Press the green button in the gutter to run the script.
+if __name__ == '__main__':
+    print_hi('PyCharm')
+
+# See PyCharm help at https://www.jetbrains.com/help/pycharm/

sdk/KafkaSenderSDK.py

@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+# @Time : 2021/6/18 19:58
+# @Author : XuJiakai
+# @File : KafkaSenderSDK
+# @Software: PyCharm
+from kafka import KafkaProducer
+
+from log import get_log
+from sdk.WinhcAllClient import WinhcAllClient
+
+log = get_log('KafkaSenderSDK')
+
+
+class KafkaSenderSDK:
+    def __init__(self, topic_name, consumer_id):
+        hosts = WinhcAllClient().get_kafka_sender()
+
+
+if __name__ == '__main__':
+    pass

sdk/MongoDbFastScan.py

@@ -0,0 +1,80 @@
+# -*- coding: utf-8 -*-
+# @Time : 2021/12/29 13:48
+# @Author : XuJiakai
+# @File : MongoDbFastScan
+# @Software: PyCharm
+import threading
+
+import math
+from log import get_log
+
+log = get_log('MongoDbFastScan')
+
+
+class MongoDbFastScan:
+    def __init__(self, db):
+        self.collection_name = None
+        self.batch_size = 500
+        self.thread_num = 1
+        self.data_num = None
+        self.db = db
+        self.func = None
+
+    def __start_thread(self, skip, limit):
+        db = self.db
+        skip = int(skip)
+        limit = int(limit)
+        li = []
+        for i in db.get_collection(self.collection_name).find(no_cursor_timeout=True).skip(skip).limit(
+                limit).batch_size(self.batch_size):
+            li.append(i)
+            if len(li) >= self.batch_size:
+                log.info("start:{} end:{} handle ...".format(skip, skip + limit))
+                self.func(li)
+                li = []
+            pass
+        if len(li) != 0:
+            log.info("start:{} end:{} handle ...".format(skip, skip + limit))
+            self.func(li)
+            pass
+        log.info("start:{} end:{} success".format(skip, skip + limit))
+        pass
+
+    def scan(self, collection_name, func, thread_num=1, batch_size=500, skip=0):
+        self.collection_name = collection_name
+        self.func = func
+        db = self.db
+        count = db.get_collection(collection_name).count_documents(filter={})
+        if count == 0:
+            return
+        self.data_num = math.ceil(count / thread_num) * thread_num
+        log.info("data number: {} thread num: {}".format(self.data_num, self.thread_num))
+        self.thread_num = thread_num
+        self.batch_size = batch_size
+        skip_value = skip % (self.data_num / thread_num)
+        skip_continue = skip / (self.data_num / thread_num)
+
+        for i in range(thread_num):
+            if i < skip_continue:
+                continue
+            else:
+                ttt = threading.Thread(target=self.__start_thread, args=(
+                    i * (self.data_num / thread_num) + skip_value, self.data_num / thread_num - skip_value))
+                ttt.start()
+                pass
+            pass
+
+        pass
+
+
+if __name__ == '__main__':
+    m = MongoDbFastScan(None)
+
+
+    def func(li):
+        print(len(li))
+        pass
+
+
+    m.scan("a_xjk_recleanup_hk_20211221", func, thread_num=5)
+    pass

sdk/WinhcAllClient.py

@@ -0,0 +1,103 @@
+# -*- coding: utf-8 -*-
+# @Time : 2021/5/27 14:31
+# @Author : XuJiakai
+# @File : WinhcAllClient
+# @Software: PyCharm
+import psycopg2
+import pymysql
+from elasticsearch import Elasticsearch
+from kafka import KafkaConsumer, TopicPartition
+from kafka import KafkaProducer
+from odps import ODPS
+from pymongo import MongoClient
+
+from env.environment_switch import environment_switch
+from sdk.WinhcHbaseApi import WinhcHbaseApi
+
+
+class WinhcAllClient:
+    def __init__(self):
+        self.env = environment_switch()
+        pass
+
+    def get_hbase_client(self):
+        host = self.env.get_val('winhc_open_api.eci_data.host')
+        return WinhcHbaseApi(host)
+
+    def get_es_client(self, es_name='new'):
+        hosts = self.env.get_val('es.' + es_name + '.hosts')
+        username = self.env.get_val('es.' + es_name + '.username')
+        password = self.env.get_val('es.' + es_name + '.pwd')
+        use_ssl = False
+        return Elasticsearch(hosts.split(','), http_auth=(username, password), use_ssl=use_ssl)
+
+    def get_mongo_db(self, db='itslaw', mongo_name='itslaw'):
+        url = self.env.get_val('mongo.' + mongo_name + '.url')
+        c = MongoClient(url)
+        return c[db]
+
+    def get_mongo_collection(self, coll_name, db='itslaw', mongo_name='itslaw'):
+        db = self.get_mongo_db(db=db, mongo_name=mongo_name)
+        return db[coll_name]
+
+    def get_kafka_sender(self, name='base'):
+        hosts = self.env.get_val('kafka.' + name + '.hosts')
+        return KafkaProducer(bootstrap_servers=hosts,
+                             api_version=(0, 10),
+                             retries=3)
+
+    def get_kafka_consumer(self, group_id, name='base'):
+        hosts = self.env.get_val('kafka.' + name + '.hosts')
+        return KafkaConsumer(bootstrap_servers=hosts,
+                             group_id=group_id,
+                             api_version=(0, 10, 2),
+                             session_timeout_ms=25000,
+                             max_poll_records=100,
+                             fetch_max_bytes=1 * 1024 * 1024)
+
+    def get_kafka_consumer_by_reset_offset(self, topic, group_id, name='base', partitions=None):
+        consumer = self.get_kafka_consumer(group_id=group_id, name=name)
+
+        if not partitions:
+            p = consumer.partitions_for_topic(topic=topic)
+            partitions = [TopicPartition(topic=topic, partition=i) for i in p]
+
+        consumer.assign(partitions)
+        for i in partitions:
+            consumer.seek(i, 0)
+        return consumer
+
+    def get_mysql_client(self, db, name='test_firefly'):
+        host = self.env.get_val('mysql.' + name + '.host')
+        port = self.env.get_val('mysql.' + name + '.port')
+        user = self.env.get_val('mysql.' + name + '.user')
+        pwd = self.env.get_val('mysql.' + name + '.pwd')
+        return pymysql.connect(host=host, port=int(port), user=user, passwd=pwd, db=db, charset='utf8')
+
+    def get_holo_client(self, name='base', db=None):
+        host = self.env.get_val('holo.' + name + '.host')
+        port = self.env.get_val('holo.' + name + '.port')
+        if not db:
+            db = self.env.get_val('holo.' + name + '.dbname')
+        user = self.env.get_val('holo.' + name + '.username')
+        pwd = self.env.get_val('holo.' + name + '.pwd')
+        return psycopg2.connect(host=host, port=int(port), dbname=db, user=user, password=pwd)
+        pass
+
+
+
+
+
+
+def get_all_client():
+    return WinhcAllClient()
+    pass
+
+
+def get_odps_sdk(project='winhc_ng'):
+    return ODPS('LTAI4FynxS5nNuKyZ3LHhMX5', 'r6gWoySXC8kSK4qnfKRxEuWJ5uHIiE', project,
+                endpoint='http://service.odps.aliyun.com/api')
+
+
+if __name__ == '__main__':
+    pass

sdk/WinhcElasticSearchSDK.py

@@ -0,0 +1,103 @@
+# -*- coding: utf-8 -*-
+# @Time : 2021/5/24 17:15
+# @Author : XuJiakai
+# @File : WinhcElasticSearchSDK
+# @Software: PyCharm
+
+from log import get_log
+from elasticsearch.exceptions import NotFoundError
+
+from sdk.WinhcAllClient import WinhcAllClient
+
+log = get_log('WinhcElasticSearchSDK')
+
+
+class WinhcElasticSearchSDK:
+    def __init__(self, es_name):
+        self.es = WinhcAllClient().get_es_client(es_name)
+
+    def exists(self, index, doc_type, _id):
+        return self.es.transport.perform_request("HEAD", "/" + index + "/" + doc_type + "/" + _id)
+
+    def get(self, index, doc_type, _id):
+        try:
+            res = self.es.get(index=index, doc_type=doc_type, id=_id)['_source']
+            return res
+        except NotFoundError:
+            return None
+
+    def query_record_num(self, index, doc_type, dsl):
+        dsl['size'] = 0
+        return self.es.search(body=dsl, index=index, doc_type=doc_type)['hits']['total']
+        pass
+
+    def query(self, index, doc_type, dsl):
+        res = self.es.search(body=dsl, index=index, doc_type=doc_type)['hits']['hits']
+        if len(res) < 1:
+            return []
+        docs = []
+        for i in range(len(res)):
+            _id = res[i]['_id']
+            doc = res[i]['_source']
+            doc['_id'] = _id
+            docs.append(doc)
+        return docs
+
+    def perform_get(self, url):
+        return self.es.transport.perform_request(
+            "GET",
+            url,
+        )
+        pass
+
+    def perform_post(self, url, body):
+        return self.es.transport.perform_request(
+            "POST",
+            url,
+            body=body
+        )
+        pass
+
+    def scan(self, index, doc_type, query_dsl, func, scroll='5m', timeout='5m', size=500):
+        queryData = self.es.search(
+            index=index,
+            doc_type=doc_type,
+            scroll=scroll,
+            timeout=timeout,
+            size=size,
+            body=query_dsl
+        )
+        hits = queryData.get("hits").get("hits")
+        total = queryData["hits"]["total"]
+        log.info(
+            "total record : %s", total
+        )
+
+        log.info("Processing a batch of data: %s - %s", 0, size)
+        func(hits)
+        scroll_id = queryData["_scroll_id"]
+        for i in range(int(total / size)):
+            res = self.es.scroll(scroll_id=scroll_id, scroll=scroll)
+            log.info("Processing a batch of data: %s - %s", (i + 1) * size, (i + 2) * size)
+            func(res["hits"]["hits"])
+        log.info("scan successful ! ")
+
+
+def get_es_sdk(env="new"):
+    return WinhcElasticSearchSDK(es_name=env)
+    pass
+
+
+def get_old_es():
+    return WinhcElasticSearchSDK(es_name='old')
+
+
+def get_new_es():
+    return WinhcElasticSearchSDK(es_name='new')
+
+
+if __name__ == '__main__':
+    old_es = get_old_es()
+    res = old_es.get("wenshu_detail3", "wenshu_detail_type", "1faa94669633b078966da8905c6aa9b5")
+    print(res)
+    pass

sdk/WinhcEsCompanyV8Sdk.py

@@ -0,0 +1,162 @@
+# -*- coding: utf-8 -*-
+# @Time : 2020/12/10 10:50
+# @Author : XuJiakai
+# @File : WinhcEsV8Sdk
+# @Software: PyCharm
+import json
+import logging
+import re
+
+from elasticsearch.exceptions import NotFoundError
+from zhconv import convert
+
+from sdk.WinhcElasticSearchSDK import WinhcElasticSearchSDK
+
+logging.info("Begin")
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+fmt = logging.Formatter('%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s')
+
+
+class winhc_es_company_v8_sdk(WinhcElasticSearchSDK):
+    def __init__(self, es_name='new'):
+        super().__init__(es_name)
+        self.pattern = re.compile('[^\\u4e00-\\u9fa50-9a-zA-Z]')
+        self.index = "winhc-company-v9"
+        self.doc_type = 'company'
+
+    def get_doc_by_company_id(self, company_id):
+        try:
+            res = self.es.get(index=self.index, doc_type=self.doc_type, id=company_id)['_source']
+            res['company_id'] = company_id
+            return res
+        except NotFoundError:
+            return None
+
+    def exists_company_id(self, company_id):
+        return super().exists(self.index, self.doc_type, company_id)
+
+    def query_by_company(self, company_name):
+        """
+        :param company_name:
+        :return: 最多反回20条数据
+        """
+        cname = self.clean_up(company_name)
+        dsl = """{"query":{"bool":{"must":[{"bool":{"should":[{"term":{"cname.value.keyword":{"value":"%s"}}},{"term":{"history_name.value.keyword":{"value":"%s"}}}]}}]}}}""" % (
+            cname, cname)
+        return self.__get_doc_by_dsl(dsl)
+
+    def query_by_credit_code_reg_number(self, credit_code, reg_number):
+        """
+        :param company_name:
+        :return: 最多反回20条数据
+        """
+        dsl_pre = """{"query":{"bool":{"must":[{"bool":{"filter":[{"bool":{"must_not":[{"term":{"deleted":{"value":9}}}]}}]}},{"bool":{"should":["""
+        dsl_post = """]}}]}}}"""
+        dsl = []
+        if credit_code is not None and credit_code.strip() != '':
+            dsl.append("""{"term":{"credit_code":{"value":"%s"}}}""" % credit_code)
+            pass
+        if reg_number is not None and reg_number.strip() != '':
+            dsl.append("""{"term":{"reg_number":{"value":"%s"}}}""" % reg_number)
+            pass
+
+        if len(dsl) == 0:
+            raise AttributeError("reg_number or credit_code is empty !")
+        dsl = dsl_pre + ','.join(dsl) + dsl_post
+        return self.__get_doc_by_dsl(dsl)
+
+    def clean_up(self, name):
+        return self.pattern.sub("", name)
+
+    def __get_doc_by_dsl(self, dsl):
+        res = super().query(index=self.index, doc_type=self.doc_type, dsl=dsl)
+        return res
+
+    def query(self, company_name):
+        """
+        :param company_name:
+        :return: 最多反回20条数据
+        """
+        logger.info("es查询:%s" % (company_name))
+
+        def get_term(n, is_clean=False):
+            if is_clean:
+                k = "value"
+            else:
+                k = 'show.keyword'
+            all_term = [{
+                "term": {
+                    "cname." + k: {
+                        "value": n
+                    }
+                }
+            }, {
+                "term": {
+                    "cname.simplified_chinese.keyword": {
+                        "value": n
+                    }
+                }
+            }, {
+                "term": {
+                    "history_name." + k: {
+                        "value": n
+                    }
+                }
+            }, {
+                "term": {
+                    "history_name.simplified_chinese.keyword": {
+                        "value": n
+                    }
+                }
+            }]
+            return all_term
+
+        all_term = get_term(company_name)
+
+        cname = self.clean_up(company_name)
+        if company_name != cname:
+            all_term.extend(get_term(cname))
+
+        simplified_chinese = convert(company_name, 'zh-cn')
+        if company_name != simplified_chinese:
+            all_term.extend(get_term(simplified_chinese))
+
+        simplified_cname = self.clean_up(simplified_chinese)
+        if company_name != simplified_cname:
+            all_term.extend(get_term(simplified_cname))
+
+        dsl = {
+            "size": 20,
+            "_source": [
+                "cname.show"
+            ],
+            "query": {
+                "bool": {
+                    "must": [
+                        {
+                            "bool": {
+                                "should": all_term
+                            }
+                        },
+                        {
+                            "term": {
+                                "deleted": {
+                                    "value": "0"
+                                }
+                            }
+                        }
+                    ]
+                }
+            }
+        }
+        # print(json.dumps(dsl))
+        return self.__get_doc_by_dsl(dsl)
+
+
+if __name__ == '__main__':
+    sdk = winhc_es_company_v8_sdk('new')
+    res = sdk.query('成安建筑营造有限公司')
+    print(res)
+    # print(sdk.query_by_company('香港澳德利集团有限公司'))
+    pass

sdk/WinhcHbaseApi.py

@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+# @Time : 2021/6/17 10:36
+# @Author : XuJiakai
+# @File : WinhcHbaseApi
+# @Software: PyCharm
+import requests as r
+import json
+
+
+class WinhcHbaseApi:
+    def __init__(self, host):
+        self.host = "http://" + host+":8288"
+        pass
+
+    def bulk_get_human(self, rs):
+        res = r.post(self.host + "/hbase/bulk-get/human", json=rs)
+        res = json.loads(res.text)
+        return res
+
+    def get_record(self, table_name, rowkey):
+        res = r.get(self.host + "/hbase/get/%s/%s" % (table_name, rowkey))
+        return json.loads(res.text)['data']
+
+    def scan(self, table_name, prefix, size):
+        res = r.get(self.host + "/hbase/scan/%s/%s_?size=%s" % (table_name, prefix, size))
+        return json.loads(res.text)['data']
+
+
+if __name__ == '__main__':
+    pass

sdk/__init__.py

@@ -0,0 +1,15 @@
+# -*- coding: utf-8 -*-
+# @Time : 2021/5/27 14:44
+# @Author : XuJiakai
+# @File : __init__.py
+# @Software: PyCharm
+
+from sdk.WinhcAllClient import get_all_client
+from sdk.WinhcElasticSearchSDK import get_old_es, get_new_es, get_es_sdk
+
+__all__ = [
+    'get_all_client'
+    , 'get_old_es'
+    , 'get_new_es'
+    , 'get_es_sdk'
+]

spider/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+# @Time : 2022/11/24 16:38
+# @Author : XuJiakai
+# @File : __init__.py
+# @Software: PyCharm
+
+if __name__ == '__main__':
+    pass

utils/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+# @Time : 2022/11/24 16:38
+# @Author : XuJiakai
+# @File : __init__.py
+# @Software: PyCharm
+
+if __name__ == '__main__':
+    pass

utils/base_utils.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+# @Time : 2022/11/24 16:57
+# @Author : XuJiakai
+# @File : base_utils
+# @Software: PyCharm
+
+def to_list(string: str):
+    string = string.strip()
+    return [i.strip() for i in string.split("\n") if i.strip() != '']
+    pass
+
+
+def parse_env_and_name(val, env):
+    if '.' in val:
+        tmp = val.split('.')
+        env = tmp[0]
+        val = tmp[1]
+    return env, val
+
+
+if __name__ == '__main__':
+    pass

utils/clipboard_util.py

@@ -0,0 +1,36 @@
+# -*- coding: utf-8 -*-
+# @Time : 2020/6/13 10:21
+# @Author : XuJiakai
+# @File : clipboard_util
+# @Software: PyCharm
+
+
+def get_text():
+    import win32con
+    import win32clipboard as wincld
+    wincld.OpenClipboard()
+    text_result = wincld.GetClipboardData(win32con.CF_UNICODETEXT)
+    wincld.CloseClipboard()
+    return text_result
+
+
+def set_text(info):
+    import win32con
+    import win32clipboard as wincld
+    wincld.OpenClipboard()
+    wincld.EmptyClipboard()
+    wincld.SetClipboardData(win32con.CF_UNICODETEXT, info)
+    wincld.CloseClipboard()
+
+
+def pause_break(print_msg=""):
+    if print_msg != '':
+        print(print_msg)
+
+    input("按任意键继续...\n")
+    pass
+
+
+if __name__ == '__main__':
+    pause_break()
+    pass