from sqlalchemy import create_engine from urllib import parse import json import numpy as np import logging from concurrent_log import ConcurrentTimedRotatingFileHandler # 创建数据库连接引擎 def get_db_engine(db_info): db_con_str = 'mysql+pymysql://%s:%s@%s:%d/%s' % \ (db_info['username'], parse.quote_plus(db_info['password']), db_info['host'], db_info['port'], db_info['database']) engine = create_engine(db_con_str, connect_args={'charset': 'utf8'}) return engine # insert into 替换为 replace into 存在重复唯一键时,先删除此行数据,然后插入新的数据 def mysql_replace_into(table, conn, keys, data_iter): from sqlalchemy.dialects.mysql import insert from sqlalchemy.ext.compiler import compiles from sqlalchemy.sql.expression import Insert @compiles(Insert) def replace_string(insert, compiler, **kw): s = compiler.visit_insert(insert, **kw) s = s.replace("INSERT INTO", "REPLACE INTO") return s data = [dict(zip(keys, row)) for row in data_iter] conn.execute(table.table.insert(replace_string=""), data) class NpEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, np.integer): return int(obj) elif isinstance(obj, np.floating): return float(obj) elif isinstance(obj, np.ndarray): return obj.tolist() else: return super(NpEncoder, self).default(obj) # 生成logger对象 def get_logger(log_file_name, log_name): log_formatter = logging.Formatter('%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s', '%Y/%m/%d %I:%M:%S %p') log_handler = ConcurrentTimedRotatingFileHandler(filename=log_file_name, when="midnight", backupCount=100) log_handler.setFormatter(log_formatter) logger = logging.getLogger(log_name) logger.addHandler(log_handler) logger.setLevel(logging.DEBUG) # 日志打印级别 return logger