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)