common_func.py 3.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091
  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. from config.url import get_none_water_mark_url
  8. import requests
  9. import pandas as pd
  10. import base64
  11. # 创建数据库连接引擎
  12. def get_db_engine(db_info):
  13. db_con_str = 'mysql+pymysql://%s:%s@%s:%d/%s' % \
  14. (db_info['user'],
  15. parse.quote_plus(db_info['password']),
  16. db_info['host'],
  17. db_info['port'],
  18. db_info['database'])
  19. engine = create_engine(db_con_str, connect_args={'charset': 'utf8mb4'})
  20. return engine
  21. # insert into 替换为 replace into 存在重复唯一键时,先删除此行数据,然后插入新的数据
  22. def mysql_replace_into(table, conn, keys, data_iter):
  23. from sqlalchemy.dialects.mysql import insert
  24. from sqlalchemy.ext.compiler import compiles
  25. from sqlalchemy.sql.expression import Insert
  26. @compiles(Insert)
  27. def replace_string(insert, compiler, **kw):
  28. s = compiler.visit_insert(insert, **kw)
  29. s = s.replace("INSERT INTO", "REPLACE INTO")
  30. return s
  31. data = [dict(zip(keys, row)) for row in data_iter]
  32. conn.execute(table.table.insert(replace_string=""), data)
  33. class NpEncoder(json.JSONEncoder):
  34. def default(self, obj):
  35. if isinstance(obj, np.integer):
  36. return int(obj)
  37. elif isinstance(obj, np.floating):
  38. return float(obj)
  39. elif isinstance(obj, np.ndarray):
  40. return obj.tolist()
  41. else:
  42. return super(NpEncoder, self).default(obj)
  43. # 生成logger对象
  44. def get_logger(log_file_name, log_name):
  45. log_formatter = logging.Formatter('%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s', '%Y/%m/%d %I:%M:%S %p')
  46. log_handler = ConcurrentTimedRotatingFileHandler(filename=log_file_name, when="midnight", backupCount=100)
  47. log_handler.setFormatter(log_formatter)
  48. logger = logging.getLogger(log_name)
  49. logger.addHandler(log_handler)
  50. logger.setLevel(logging.DEBUG) # 日志打印级别
  51. return logger
  52. # 获取头条无水印有时效性的链接
  53. def get_toutiao_none_water_mark_and_time_efficient_url(vid):
  54. """
  55. 依据 vid,得到无水印,有时效性的链接
  56. :return:
  57. """
  58. request_path = get_none_water_mark_url + str(vid)
  59. request = requests.get(request_path)
  60. response_data = json.loads(request.text)
  61. if response_data.get('message') == 'success' and response_data.get('data') and response_data.get('data').get('video_list'):
  62. video_df = pd.DataFrame()
  63. for k, v in response_data.get('data').get('video_list').items():
  64. single_video_df = pd.DataFrame([v])
  65. video_df = video_df.append(single_video_df)
  66. video_df.sort_values(by='size', ascending=True, inplace=True) # 获取分辨率最高的素材
  67. main_url = video_df['main_url'].values[-1]
  68. main_url_decode = base64.b64decode(main_url)
  69. main_url_decode = str(main_url_decode, encoding='utf-8') # bytes to str
  70. return main_url_decode
  71. else:
  72. return None
  73. if __name__ == '__main__':
  74. r = get_toutiao_none_water_mark_and_time_efficient_url('v02033290000budu3753giguv9qli6bg')
  75. print(r)