common_func.py 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758
  1. from sqlalchemy import create_engine
  2. from urllib import parse
  3. import json
  4. import numpy as np
  5. import logging
  6. from concurrent_log import ConcurrentTimedRotatingFileHandler
  7. # 创建数据库连接引擎
  8. def get_db_engine(db_info):
  9. db_con_str = 'mysql+pymysql://%s:%s@%s:%d/%s' % \
  10. (db_info['username'],
  11. parse.quote_plus(db_info['password']),
  12. db_info['host'],
  13. db_info['port'],
  14. db_info['database'])
  15. engine = create_engine(db_con_str, connect_args={'charset': 'utf8'})
  16. return engine
  17. # insert into 替换为 replace into 存在重复唯一键时,先删除此行数据,然后插入新的数据
  18. def mysql_replace_into(table, conn, keys, data_iter):
  19. from sqlalchemy.dialects.mysql import insert
  20. from sqlalchemy.ext.compiler import compiles
  21. from sqlalchemy.sql.expression import Insert
  22. @compiles(Insert)
  23. def replace_string(insert, compiler, **kw):
  24. s = compiler.visit_insert(insert, **kw)
  25. s = s.replace("INSERT INTO", "REPLACE INTO")
  26. return s
  27. data = [dict(zip(keys, row)) for row in data_iter]
  28. conn.execute(table.table.insert(replace_string=""), data)
  29. class NpEncoder(json.JSONEncoder):
  30. def default(self, obj):
  31. if isinstance(obj, np.integer):
  32. return int(obj)
  33. elif isinstance(obj, np.floating):
  34. return float(obj)
  35. elif isinstance(obj, np.ndarray):
  36. return obj.tolist()
  37. else:
  38. return super(NpEncoder, self).default(obj)
  39. # 生成logger对象
  40. def get_logger(log_file_name, log_name):
  41. log_formatter = logging.Formatter('%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s', '%Y/%m/%d %I:%M:%S %p')
  42. log_handler = ConcurrentTimedRotatingFileHandler(filename=log_file_name, when="midnight", backupCount=100)
  43. log_handler.setFormatter(log_formatter)
  44. logger = logging.getLogger(log_name)
  45. logger.addHandler(log_handler)
  46. logger.setLevel(logging.DEBUG) # 日志打印级别
  47. return logger