Server-py/sri-pysdk/xclient/xmongo.py

# update: 2022-3-30-15
"""
$eq 等于
$ne 不等
$in []
db.collection.find({ field: { $not: { $in: [value1, value2, value3] } } })
$regex 模糊查找
(>) 大于 - $gt
(<) 小于 - $lt
(>=) 大于等于 - $gte
(<= ) 小于等于 - $lte


pip install pymongo==3.11.2 -i https://pypi.tuna.tsinghua.edu.cn/simple
"""
from pymongo import MongoClient, TEXT
from bson.objectid import ObjectId


class Client(MongoClient):

    def __init__(self, host='sri-thirdparty-mongo', port=27017, database='vms', username='admin', password='admin'):
        """
        内部访问:
            host: sri-thirdparty-mongo
            port: 27017
        远程访问:
            host: 118.190.217.96、192.168.20.162
            port: 7030
        """
        super().__init__(f'mongodb://{username}:{password}@{host}:{port}')
        self.db = self[database]

    def run_command(self, command=None, command_type='dict', use_admin=False):
        """
        Command:
            print(self.db.command('serverStatus'))
            print(self.db.command('dbstats'))
            print(self['admin'].command('listDatabases', 1))  # 查看数据库信息
            print(self['admin'].command({'setParameter': 1, 'internalQueryExecMaxBlockingSortBytes': 52428800}))  # 设置排序内存限制
            print(self['admin'].command({'getParameter': 1, 'internalQueryExecMaxBlockingSortBytes': 1}))  # 查询排序内存限制
            print(self.db.command('collstats', 'WorkFlowPlugin'))  # 查看表信息
            print(self.db.command({'collStats': 'WorkFlowPlugin'}))  # 查看表信息
        Usage:
            self.run_command(
                command={'setParameter': 1, 'internalQueryExecMaxBlockingSortBytes': 52428800},
                use_admin=True
            )  # 设置排序内存限制
        """
        if use_admin and command_type == 'dict':
            return self['admin'].command(command)

    def get_indexes(self, table_name):
        """
        获取索引
        # return collection.indexes.find()
        # return collection.list_indexes()
        """
        collection = self.db[table_name]
        return collection.index_information()

    def add_index(self, table_name, index_list):
        """
        table_name: Xxx
        index_list: [("field", 1)]
        # return collection.create_index([('sec_id', TEXT)], name='search_index_zyq', default_language='english')
        """
        collection = self.db[table_name]
        return collection.create_index(index_list)

    def del_index(self, table_name, index_or_name):
        """
        删除索引 https://www.dotnetperls.com/create-index-mongodb
        table_name: Xxx
        index_list: [("field", 1)]
        """
        collection = self.db[table_name]
        return collection.drop_index(index_or_name)

    def add(self, table_name, data):
        """
        保存数据
        doc: https://www.cnblogs.com/aademeng/articles/9779271.html
        """
        table = self.db[table_name]
        if type(data) == dict:
            return str(table.insert_one(data).inserted_id)
        elif type(data) == list:
            return [str(i) for i in table.insert_many(data).inserted_ids]

    def filter(self, table_name, condition_dict, sort_field=None):
        """
        筛选数据
        condition_dict: 筛选条件
        sort_field: [('排序字段', -1)]
        """
        table = self.db[table_name]
        if sort_field:
            return table.find(condition_dict).sort(sort_field)
        else:
            return table.find(condition_dict)

    def filter_by_aggregate(self, table_name, condition=None, sort_dict=None, page=None, size=None):
        """
        聚合查询
        condition: 条件
        sort_dict: 排序 {'字段': -1}
        page: 页数
        size: 条数
        """
        table = self.db[table_name]
        parameters = list()
        if condition:
            parameters.append({'$match': condition})
        if sort_dict:
            parameters.append({'$sort': sort_dict})
        if type(page) == int and type(size) == int:
            skip = 0 if page == 1 else (page - 1) * size
            parameters.append({'$skip': skip})  # 跳过多少条
            parameters.append({'$limit': size})
        return table.aggregate(parameters, allowDiskUse=True)

    def get_count(self, table_name, condition_dict=None):
        """
        获取数据数量
        """
        table = self.db[table_name]
        if not condition_dict:
            condition_dict = {}
        return table.count_documents(condition_dict)

    def get_all(self, table_name, sort_field=None):
        """
        获取数据 / 排序
        table_name: '表名'
        sort_field: [('排序字段', -1)]
        """
        table = self.db[table_name]
        if sort_field:
            return table.find().sort(sort_field)
        else:
            return table.find()

    def get_one(self, table_name, unique_dict):
        """
        单条获取
        """
        table = self.db[table_name]
        data = table.find_one(unique_dict)
        if data:
            return data
        else:
            return dict()

    def get_last_one(self, table_name, sort_list=None):
        """
        获取最后一条
        table_name: '集合名称'
        sort_list: [('_id', -1)]
        """
        table = self.db[table_name]
        if not sort_list:
            sort_list = [('_id', -1)]
        return table.find_one(sort=sort_list)

    def get_one_by_id(self, table_name, object_id):
        """
        单条获取
        """
        table = self.db[table_name]
        return table.find_one({'_id': ObjectId(str(object_id))})

    def update_one_by_id(self, table_name, object_id, update_dict, need_back=False):
        """
        单条数据
        """
        table = self.db[table_name]
        modified_count = table.update_one({'_id': ObjectId(str(object_id))}, {'$set': update_dict}).modified_count
        if need_back:
            item = self.get_one_by_id(table_name, object_id)
            item.pop('_id')
            item['uuid'] = object_id
            return item
        else:
            return modified_count

    def update_all(self, table_name, condition_dict, update_dict):
        """
        批量更新
        """
        table = self.db[table_name]
        return table.update_many(condition_dict, {'$set': update_dict}).matched_count

    def update_one(self, table_name, unique_dict, update_dict, default_dict=None, need_back=False):
        """
        # if get:
        #     update update_one_by_id
        # else:
        #     insert default_dict
        # return: get
        """
        data = self.get_one(table_name, unique_dict)
        if data:
            object_id = str(data.get('_id'))
            update_dict.update(unique_dict)
            self.update_one_by_id(table_name, object_id, update_dict)
        else:
            if not default_dict:
                default_dict = dict()
            if '_id' in default_dict:
                del default_dict['_id']
            default_dict.update(update_dict)
            default_dict.update(unique_dict)
            object_id = self.add(table_name, default_dict)
        if need_back:
            return self.get_one_by_id(table_name, object_id)
        else:
            return object_id

    def remove(self, table_name, condition_dict):
        """
        批量删除
        """
        table = self.db[table_name]
        return table.delete_many(condition_dict).deleted_count

    def remove_one_by_id(self, table_name, object_id):
        """
        单条删除
        """
        table = self.db[table_name]
        return table.delete_many({'_id': ObjectId(str(object_id))}).deleted_count

    def distinct(self, table_name, field):
        """
        去重
        """
        table = self.db[table_name]
        return table.distinct(field)

    def get_all_vague(self, table_name, unique_dict):
        """
        模糊查询获取
        """
        table = self.db[table_name]
        return table.find(unique_dict)

    def get_aggregate(self, table_name, condition):
        """
        分组聚合查询（by wdj）
        """
        table = self.db[table_name]
        return table.aggregate(condition)

    def get_col_list(self, table_name, condition, _dic):
        """
        只返回某一列的数据（by wdj）
        """
        table = self.db[table_name]
        return table.find(condition, _dic)

    # def find_by_aggregate(self, table_name,ip):
    #     """
    #     分组查询获取   {"$match":{"ip":ip}   {"$group":{"_id":"","counter":{"$sum":1}}}
    #     """
    #     table = self.db[table_name]
    #     count= table.aggregate([{"$group":{"_id":"$find_plugin","count":{"$sum":"$risk_score"}}}])
    #     return count

    def deep_filter(self, collection, field_path, filter_type='equal', filter_value=None, return_count=False,
                    sort_field=None):
        """
        field_path: 递进字段名（适用字典类型）
        filter_type: 筛选类型 contain !contain equal !equal in !in
        sort_field: [('排序字段', -1)]
        """
        if filter_type == 'equal':
            filter_rule = {'$eq': filter_value}
        elif filter_type == 'contain':
            filter_rule = {'$regex': f'^.*{filter_value}.*$'}
        else:
            filter_rule = {}
        condition = {
            field_path: filter_rule
        }
        if return_count:
            return self.get_count(collection, condition)
        elif sort_field:
            return self.filter(collection, condition, sort_field)
        else:
            return self.filter(collection, condition)

    def del_collect(self):
        db_collections = self.db.collection_names()
        for collection in db_collections:
            self.db[collection].drop()

    @staticmethod
    def uuid2oid(uuid):
        return ObjectId(uuid)


if __name__ == '__main__':
    """
    sudo python3 -m pip install pymongo
    sudo python3 /home/server/projects/taiwuict/cscec-8bur-vms/supplement-python/clients/db_mongo.py
    """
    # --- init ---

    # mdb = Client(database='vms', host='192.168.0.16', port=7030)
    # mdb = Client(database='vms', host='192.168.51.242', port=7030)
    # mdb = Client(database='vms', host='192.168.1.233', port=7030)
    # mdb = Client(database='vms', host='192.168.0.16', port=7030)
    # mdb = Client(database='vms', host='10.8.20.115', port=7030)  # 晨鸣大厦
    # mdb = Client(database='vms', host='192.168.15.195', port=7030)
    # mdb = Client(database='vms', host='192.168.0.15', port=7030)
    # mdb = Client(database='vms', host='192.168.101.199', port=7030)  # 章丘项目
    # mdb = Client(database='vms', host='192.168.15.195', port=7030)  # 第3现场
    # mdb = Client(database='vms', host='192.168.1.242', port=7030)  # 第4现场
    # mdb = Client(database='ar', host='58.34.94.176', port=7030)  # 第4现场
    mdb = Client(host='127.0.0.1', port=47017, database='ar', username='admin', password='admin')

    # --- test ---
    # data = {
    #     'username': 'aabbss',
    #     'password': 'aabbss',
    #     'role_id': 'aabbss',
    #     'create_at': 11223344,
    # }
    # uuid = mdb.add('User', data)
    # --- test ---
    # __condition = {
    #     'face_name': {'$eq': None},
    # }
    # total = mdb.get_count('Face', __condition)
    # print(total)
    # total = mdb.remove('Face', __condition)
    # print(total)
    # total = mdb.get_count('Face', __condition)
    # print(total)

    # --- test ---
    __condition = {
        # 'face_name': {'$ne': None},
        # 'face_name': {'$eq': None},
        'face_name': {'$regex': '陈忠福'},
        # 'face_features': {'$eq': b'\x80\x03N.'},
    }
    items = mdb.filter('Face', __condition)
    # for item in items:
    #     print(item)
    #     # print(item.keys())
    # total = mdb.get_count('Face', __condition)
    # print(total)

    # --- test ---
    # _id = '62cbbefba007e25f12acea9d'
    # out = mdb.get_one_by_id('Face', _id)
    # print(out)

    # --- test ---
    # condition = {
    #     # 'face_name': {'$eq': None},
    #     'face_name': {'$ne': None},
    # }
    # total = mdb.get_count('Face', condition)
    # print(total)

    # --- test ---
    # items = mdb.get_all('Face')
    # for item in items:
    #     if 'c7304b' in str(item.get('_id')):
    #         print(item)

    # --- test ---
    # __condition = {
    #     # 'face_feature_info_list': {'$ne': None},
    #     'face_feature_info_list': {'$eq': None},
    # }
    # # items = mdb.filter('Face', __condition)
    # # for item in items:
    # #     print(item)
    # #     break
    # total = mdb.get_count('Face', __condition)
    # print(total)
    # total = mdb.remove('Face', __condition)
    # print(total)
    # total = mdb.get_count('Face', __condition)
    # print(total)

    # --- test ---
    # unique_dict = {
    #     'prc_id': '371324199110201974',
    # }
    # item = mdb.get_one('Face', unique_dict)
    # print(item.get('cscec8_id'))

    # --- test ---
    # __condition = {
    #     'face_type_uuid_list': {'$in': ['62a190ca84b2c038bac3519d', 'aa', '62a1a438024e59b2e654467a']}
    # }
    # items = mdb.filter('Face', __condition)
    # for item in items:
    #     print(item.get('face_name'))

    # --- test ---
    # __condition = {
    #     # 'prc_id': {'$ne': None},
    #     # 'face_name': {'$ne': None},
    # }
    # items = mdb.filter('Face', __condition)
    # for item in items:
    #     print(item)
    #     mdb.update_one_by_id('Face', str(item.get('_id')), {'upload_is': None})

    # --- test ---
    # total = mdb.get_count('Face', __condition)
    # print(total)

    # --- test ---
    # # _id = '6295942092325efa38758bf1'
    # # _id = '629591b792325efa38758be2'
    # # _id = '6295925992325efa38758be5'
    # _id = '6295915592325efa38758bdf'
    # # mdb.update_one_by_id('Face', _id, {'prc_id': '370102198802249999'})
    # # mdb.update_one_by_id('Face', _id, {'prc_id': '370102198802241111'})
    # mdb.update_one_by_id('Face', _id, {'cscec8_id': 'f6bae517-0a28-429c-b65b-b16548ab2ca4'})

    # --- test ---
    # import time
    #
    # __condition = dict(
    #     modify_at={'$gte': time.time() - 86400 * 100},
    #     # modify_at={'$gte': methods.now_ts() - 3600},
    # )
    # items = mdb.filter('Face', __condition)
    # for item in items:
    #     print(item.keys())
    # total = mdb.get_count('Face', __condition)
    # print(total)

    # --- test ---
    # __uuid = '6244068e64481ae85ca38ef9'
    # item = mdb.get_one_by_id('Face', __uuid)
    # print(item)

    # --- test ---
    # _uuid = '6243fa35d169a1bf065f8f02'
    # item = mdb.get_one_by_id('Face', _uuid)
    # print(item.get('face_feature_info_list')[0])

    # --- test ---
    # """
    # Face: 陌生人脸表
    # Face.face_name: 姓名
    # """
    # __condition = {
    #     # 'face_name': {'$ne': None},
    #     # 'face_name': {'$eq': None},
    #     'face_name': {'$regex': '王'},
    # }
    # items = mdb.filter('Face', __condition)
    # for item in items:
    #     print(item)
    #     break
    # total = mdb.get_count('Face', __condition)
    # print(total)

    # --- test ---
    # __condition = {
    #     'name': {'$eq': '未知类型'}
    # }
    # count = mdb.get_count('FaceType', __condition)
    # print(count)

    # --- test ---
    # items = mdb.get_all('VisitorInfo')
    # for item in items:
    #     _list = item.get('aaabbbccc', [])
    #     print('_list', type(_list), _list)
    #     break

    # --- test ---
    # _condition = {'ipv4': '192.168.30.232'}
    # items = mdb.filter_by_aggregate('UserCamera', condition=_condition, sort_dict={'create_at': -1}, page=1,
    #                                 size=10)
    # count = 0
    # for item in items:
    # 	count += 1
    # print(count)

    # --- test ---
    # _uuid = '604c22d22043eb79cb425c9c'
    # out = mdb.remove('AgentDetectionRecord', {'camera_uuid': _uuid})
    # # out = mdb.get_one('AgentDetectionRecord', {'camera_uuid': _uuid})
    # print(out)

    # --- test ---
    # _condition = {
    #     '_id': {'$in': [ObjectId('605994eb7c29e9dc3887e639')]}
    # }
    # items = mdb.filter('AgentDetectionRecord', _condition)
    # for item in items:
    #     print(item)

    # --- test ---
    # out = mdb.get_count('VisitorInfo')
    # print(out)