| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778 | 
							- import base64
 
- import json
 
- import logging
 
- import numpy as np
 
- import pandas as pd
 
- import requests
 
- from concurrent_log import ConcurrentTimedRotatingFileHandler
 
- from config.url_and_db import get_none_water_mark_url
 
- # insert into 替换为 replace into 存在重复唯一键时,先删除此行数据,然后插入新的数据
 
- def mysql_replace_into(table, conn, keys, data_iter):
 
-     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
 
- # 获取头条无水印有时效性的链接
 
- def get_toutiao_none_water_mark_and_time_efficient_url(vid):
 
-     """
 
-     依据 vid,得到无水印,有时效性的链接
 
-     :return:
 
-     """
 
-     request_path = get_none_water_mark_url + str(vid)
 
-     request = requests.get(request_path)
 
-     response_data = json.loads(request.text)
 
-     if response_data.get('message') == 'success' and response_data.get('data') and response_data.get('data').get('video_list'):
 
-         video_df = pd.DataFrame()
 
-         for k, v in response_data.get('data').get('video_list').items():
 
-             single_video_df = pd.DataFrame([v])
 
-             video_df = video_df.append(single_video_df)
 
-         video_df.sort_values(by='size', ascending=True, inplace=True)  # 获取分辨率最高的素材
 
-         main_url = video_df['main_url'].values[-1]
 
-         main_url_decode = base64.b64decode(main_url)
 
-         main_url_decode = str(main_url_decode, encoding='utf-8')  # bytes to str
 
-         return main_url_decode
 
-     else:
 
-         return None
 
- if __name__ == '__main__':
 
-     r = get_toutiao_none_water_mark_and_time_efficient_url('v02033290000budu3753giguv9qli6bg')
 
-     print(r)
 
 
  |