| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091 | from sqlalchemy import create_enginefrom urllib import parseimport jsonimport numpy as npimport loggingfrom concurrent_log import ConcurrentTimedRotatingFileHandlerfrom config.url import get_none_water_mark_urlimport requestsimport pandas as pdimport base64# 创建数据库连接引擎def get_db_engine(db_info):    db_con_str = 'mysql+pymysql://%s:%s@%s:%d/%s' % \                 (db_info['user'],                  parse.quote_plus(db_info['password']),                  db_info['host'],                  db_info['port'],                  db_info['database'])    engine = create_engine(db_con_str, connect_args={'charset': 'utf8mb4'})    return engine# insert into 替换为 replace into 存在重复唯一键时,先删除此行数据,然后插入新的数据def mysql_replace_into(table, conn, keys, data_iter):    from sqlalchemy.dialects.mysql import insert    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 Noneif __name__ == '__main__':    r = get_toutiao_none_water_mark_and_time_efficient_url('v02033290000budu3753giguv9qli6bg')    print(r)
 |