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