common_func.py 2.7 KB

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