# coding:utf-8 import pickle import os import requests import json import traceback import pandas as pd from odps import ODPS from odps.df import DataFrame from my_config import set_config from db_helper import HologresHelper, MysqlHelper, RedisHelper from log import Log from collections import defaultdict config_, env = set_config() log_ = Log() def get_odps_instance(project): odps = ODPS( access_id=config_.ODPS_CONFIG['ACCESSID'], secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'], project=project, endpoint=config_.ODPS_CONFIG['ENDPOINT'], ) return odps def execute_sql_from_odps(project, sql, connect_timeout=3000, read_timeout=500000, pool_maxsize=1000, pool_connections=1000): odps = ODPS( access_id=config_.ODPS_CONFIG['ACCESSID'], secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'], project=project, endpoint=config_.ODPS_CONFIG['ENDPOINT'], connect_timeout=connect_timeout, read_timeout=read_timeout, pool_maxsize=pool_maxsize, pool_connections=pool_connections ) records = odps.execute_sql(sql=sql) return records def exe_sql(project, sql, connect_timeout=3000, read_timeout=500000, pool_maxsize=1000, pool_connections=1000): odps = ODPS( access_id=config_.ODPS_CONFIG['ACCESSID'], secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'], project=project, endpoint=config_.ODPS_CONFIG['ENDPOINT'], connect_timeout=connect_timeout, read_timeout=read_timeout, pool_maxsize=pool_maxsize, pool_connections=pool_connections ) records = odps.execute_sql(sql) return records def get_data_from_odps(date, project, table, connect_timeout=3000, read_timeout=500000, pool_maxsize=1000, pool_connections=1000): """ 从odps获取数据 :param date: 日期 type-string '%Y%m%d' :param project: type-string :param table: 表名 type-string :param connect_timeout: 连接超时设置 :param read_timeout: 读取超时设置 :param pool_maxsize: :param pool_connections: :return: records """ odps = ODPS( access_id=config_.ODPS_CONFIG['ACCESSID'], secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'], project=project, endpoint=config_.ODPS_CONFIG['ENDPOINT'], connect_timeout=connect_timeout, read_timeout=read_timeout, pool_maxsize=pool_maxsize, pool_connections=pool_connections ) records = odps.read_table(name=table, partition='dt=%s' % date) return records def get_dataframe_from_odps(project, table, partition_spec_dict=None): """ 从odps获取数据 :param partition_spec_dict: 分区spec type-dict :param project: type-string :param table: 表名 type-string :return: odps.DataFrame """ odps = get_odps_instance(project) if partition_spec_dict: spec = ','.join(['{}={}'.format(k, partition_spec_dict[k]) for k in partition_spec_dict.keys()]) return DataFrame(odps.get_table(name=table)).filter_parts(spec) else: return DataFrame(odps.get_table(name=table)) def get_odps_df_of_max_partition(project, table, rb_spec=None): """ rb_spec: spec for right bound of partition names. type-dict return odps.DataFrame """ odps = get_odps_instance(project) t = odps.get_table(table) df = DataFrame(odps.get_table(table)) if rb_spec is None: return df.filter_parts(t.get_max_partition().partition_spec) else: spec = ','.join(['{}<{}'.format(k, rb_spec[k]) for k in rb_spec.keys()]) part_iter = t.iterate_partitions(spec=spec, reverse=True) try: partition = next(part_iter) return df.filter_parts(partition) except StopIteration: return None def get_odps_df_of_recent_partitions(project, table, n=1, rb_spec=None): """ rb_spec: spec for right bound of partition names. type-dict return odps.DataFrame """ odps = get_odps_instance(project) t = odps.get_table(table) df = DataFrame(odps.get_table(table)) spec = None if rb_spec: spec = ','.join(['{}<{}'.format(k, rb_spec[k]) for k in rb_spec.keys()]) part_iter = t.iterate_partitions(spec=spec, reverse=True) selected_parts = [] try: for i in range(0, n): partition = next(part_iter) selected_parts.append(partition) log_.info(f"table: {table}, selected part: {partition.name}") except StopIteration: log_.info(f"table: {table}, no more parts to iterate") return df.filter_parts(selected_parts) def check_table_partition_exits(date, project, table, connect_timeout=3000, read_timeout=500000, pool_maxsize=1000, pool_connections=1000): """ 判断表中是否存在这个分区 :param date: 日期 type-string '%Y%m%d' :param project: type-string :param table: 表名 type-string :param connect_timeout: 连接超时设置 :param read_timeout: 读取超时设置 :param pool_maxsize: :param pool_connections: :return: records """ odps = ODPS( access_id=config_.ODPS_CONFIG['ACCESSID'], secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'], project=project, endpoint=config_.ODPS_CONFIG['ENDPOINT'], connect_timeout=connect_timeout, read_timeout=read_timeout, pool_maxsize=pool_maxsize, pool_connections=pool_connections ) t = odps.get_table(name=table) return t.exist_partition(partition_spec=f'dt={date}') def check_table_partition_exits_v2(project, table, partition_spec_dict): """ 判断表中是否存在指定分区，并返回分区纪录数量 注：ODPS新版本移除了timeout等参数 :param project: 库名 type-string :param table: 表名 type-string :param partition_spec_dict: 分区spec type-dict :return: if_exist, num_records """ odps = get_odps_instance(project) t = odps.get_table(name=table) spec = ','.join(['{}={}'.format(k, partition_spec_dict[k]) for k in partition_spec_dict.keys()]) if t.exist_partition(partition_spec=spec): with t.open_reader(partition=spec) as reader: count = reader.count return True, count else: return False, 0 def write_to_pickle(data, filename, filepath=config_.DATA_DIR_PATH): """ 将数据写入pickle文件中 :param data: 数据 :param filename: 写入的文件名 :param filepath: 文件存放路径，默认为config_.DATA_DIR_PATH :return: None """ if not os.path.exists(filepath): os.makedirs(filepath) file = os.path.join(filepath, filename) with open(file, 'wb') as wf: pickle.dump(data, wf) def read_from_pickle(filename, filepath=config_.DATA_DIR_PATH): """ 从pickle文件读取数据 :param filename: 文件名 :param filepath: 文件存放路径，默认为config_.DATA_DIR_PATH :return: data """ file = os.path.join(filepath, filename) if not os.path.exists(file): return None with open(file, 'rb') as rf: data = pickle.load(rf) return data def send_msg_to_feishu(webhook, key_word, msg_text): """发送消息到飞书""" headers = {'Content-Type': 'application/json'} payload_message = { "msg_type": "text", "content": { "text": '{}: {}'.format(key_word, msg_text) } } response = requests.request('POST', url=webhook, headers=headers, data=json.dumps(payload_message)) print(response.text) def send_msg_to_feishu_new(webhook, key_word, title, msg_list): """发送消息到飞书""" headers = {'Content-Type': 'application/json'} content_list = [ [ { "tag": "text", "text": msg } ] for msg in msg_list ] payload_message = { "msg_type": "post", "content": { "post": { "zh_cn": { "title": f"{key_word}: {title}", "content": content_list, } } } } response = requests.request('POST', url=webhook, headers=headers, data=json.dumps(payload_message)) print(response.text) def request_post(request_url, request_data=None, **kwargs): """ post 请求 HTTP接口 :param request_url: 接口URL :param request_data: 请求参数 :return: res_data json格式 """ try: response = requests.post(url=request_url, json=request_data, **kwargs) if response.status_code == 200: res_data = json.loads(response.text) return res_data else: log_.info(f"response.status_code: {response.status_code}") return None except Exception as e: log_.error('url: {}, exception: {}, traceback: {}'.format(request_url, e, traceback.format_exc())) send_msg_to_feishu( webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'), key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'), msg_text='rov-offline{} - 接口请求失败：{}, exception: {}'.format(config_.ENV_TEXT, request_url, e) ) return None def request_get(request_url): """ get 请求 HTTP接口 :param request_url: 接口URL :return: res_data json格式 """ try: response = requests.get(url=request_url) if response.status_code == 200: res_data = json.loads(response.text) return res_data else: log_.info(f"response.status_code: {response.status_code}") return None except Exception as e: log_.error('url: {}, exception: {}, traceback: {}'.format(request_url, e, traceback.format_exc())) send_msg_to_feishu( webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'), key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'), msg_text='rov-offline{} - 接口请求失败：{}, exception: {}'.format(config_.ENV_TEXT, request_url, e) ) return None def data_normalization(data): """ 对结果做归一化处理(Min-Max Normalization)，将分数控制在[0, 100] :param data: type-list :return: normal_data, type-list 归一化后的数据 """ x_max = max(data) x_min = min(data) normal_data = [(x-x_min)/(x_max-x_min)*100 for x in data] return normal_data def filter_video_status(video_ids): """ 对视频状态进行过滤 :param video_ids: 视频id列表 type-list :return: filtered_videos """ i = 0 while i < 3: try: mysql_helper = MysqlHelper(mysql_info=config_.FILTER_MYSQL_INFO) video_status_sql = "SELECT t1.id AS 'video_id', " \ "t1.transcode_status AS 'transcoding_status', " \ "t2.audit_status AS 'audit_status', " \ "t2.video_status AS 'open_status', " \ "t2.recommend_status AS 'applet_rec_status', " \ "t2.app_recommend_status AS 'app_rec_status', " \ "t3.charge AS 'payment_status', " \ "case when t4.max_validate_count is null then 0 else t4.max_validate_count end AS 'encryption_status' " \ "FROM longvideo.wx_video t1 " \ "LEFT JOIN longvideo.wx_video_status t2 ON t1.id= t2.video_id " \ "LEFT JOIN longvideo.wx_video_detail t3 ON t1.id= t3.video_id " \ "LEFT JOIN longvideo.wx_video_pwd t4 ON t1.id= t4.video_id" if len(video_ids) == 1: sql = "SELECT video_id " \ "FROM ({}) " \ "WHERE audit_status = 5 " \ "AND applet_rec_status IN (1, -6) " \ "AND open_status = 1 " \ "AND payment_status = 0 " \ "AND encryption_status != 5 " \ "AND transcoding_status = 3 " \ "AND video_id IN ({});".format(video_status_sql, video_ids[0]) data = mysql_helper.get_data(sql=sql) else: data = [] for i in range(len(video_ids) // 200 + 1): sql = "SELECT video_id " \ "FROM ({}) " \ "WHERE audit_status = 5 " \ "AND applet_rec_status IN (1, -6) " \ "AND open_status = 1 " \ "AND payment_status = 0 " \ "AND encryption_status != 5 " \ "AND transcoding_status = 3 " \ "AND video_id IN {};".format(video_status_sql, tuple(video_ids[i*200:(i+1)*200])) select_res = mysql_helper.get_data(sql=sql) if select_res is not None: data += select_res filtered_videos = [int(temp[0]) for temp in data] return filtered_videos except Exception as e: log_.error(f"过滤失败, exception: {e}, traceback: {traceback.format_exc()}") send_msg_to_feishu( webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'), key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'), msg_text=f"rov-offline{config_.ENV_TEXT} - 过滤失败\n" f"retry count: {i}\n" f"exception: {e}\n" f"traceback: {traceback.format_exc()}" ) i += 1 if i == 1: return video_ids def filter_video_status_with_applet_rec(video_ids, applet_rec_status): """ 对视频状态进行过滤 :param video_ids: 视频id列表 type-list :param applet_rec_status: 小程序推荐状态 -6:待推荐 1:普通推荐 :return: filtered_videos """ i = 0 while i < 3: try: mysql_helper = MysqlHelper(mysql_info=config_.FILTER_MYSQL_INFO) video_status_sql = "SELECT t1.id AS 'video_id', " \ "t1.transcode_status AS 'transcoding_status', " \ "t2.audit_status AS 'audit_status', " \ "t2.video_status AS 'open_status', " \ "t2.recommend_status AS 'applet_rec_status', " \ "t2.app_recommend_status AS 'app_rec_status', " \ "t3.charge AS 'payment_status', " \ "case when t4.max_validate_count is null then 0 else t4.max_validate_count end AS 'encryption_status' " \ "FROM longvideo.wx_video t1 " \ "LEFT JOIN longvideo.wx_video_status t2 ON t1.id= t2.video_id " \ "LEFT JOIN longvideo.wx_video_detail t3 ON t1.id= t3.video_id " \ "LEFT JOIN longvideo.wx_video_pwd t4 ON t1.id= t4.video_id" if len(video_ids) == 1: sql = "SELECT video_id " \ "FROM ({}) " \ "WHERE audit_status = 5 " \ "AND applet_rec_status = {} " \ "AND open_status = 1 " \ "AND payment_status = 0 " \ "AND encryption_status != 5 " \ "AND transcoding_status = 3 " \ "AND video_id IN ({});".format(video_status_sql, applet_rec_status, video_ids[0]) data = mysql_helper.get_data(sql=sql) else: data = [] for i in range(len(video_ids) // 200 + 1): sql = "SELECT video_id " \ "FROM ({}) " \ "WHERE audit_status = 5 " \ "AND applet_rec_status = {} " \ "AND open_status = 1 " \ "AND payment_status = 0 " \ "AND encryption_status != 5 " \ "AND transcoding_status = 3 " \ "AND video_id IN {};".format(video_status_sql, applet_rec_status, tuple(video_ids[i*200:(i+1)*200])) select_res = mysql_helper.get_data(sql=sql) if select_res is not None: data += select_res filtered_videos = [int(temp[0]) for temp in data] return filtered_videos except Exception as e: log_.error(f"过滤失败, exception: {e}, traceback: {traceback.format_exc()}") send_msg_to_feishu( webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'), key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'), msg_text=f"rov-offline{config_.ENV_TEXT} - 过滤失败\n" f"retry count: {i}\n" f"exception: {e}\n" f"traceback: {traceback.format_exc()}" ) i += 1 if i == 1: return video_ids def filter_video_status_app(video_ids): """ 对视频状态进行过滤 - app :param video_ids: 视频id列表 type-list :return: filtered_videos """ i = 0 while i < 3: try: mysql_helper = MysqlHelper(mysql_info=config_.FILTER_MYSQL_INFO) video_status_sql = "SELECT t1.id AS 'video_id', " \ "t1.transcode_status AS 'transcoding_status', " \ "t2.app_audit_status AS 'app_audit_status', " \ "t2.original_status AS 'open_status', " \ "t2.recommend_status AS 'applet_rec_status', " \ "t2.app_recommend_status AS 'app_rec_status', " \ "t3.charge AS 'payment_status', " \ "case when t4.max_validate_count is null then 0 else t4.max_validate_count end AS 'encryption_status' " \ "FROM longvideo.wx_video t1 " \ "LEFT JOIN longvideo.wx_video_status t2 ON t1.id= t2.video_id " \ "LEFT JOIN longvideo.wx_video_detail t3 ON t1.id= t3.video_id " \ "LEFT JOIN longvideo.wx_video_pwd t4 ON t1.id= t4.video_id" if len(video_ids) == 1: sql = "SELECT video_id " \ "FROM ({}) " \ "WHERE app_audit_status = 5 " \ "AND app_rec_status IN (1, -6, 10) " \ "AND open_status = 1 " \ "AND payment_status = 0 " \ "AND encryption_status != 5 " \ "AND transcoding_status = 3 " \ "AND video_id IN ({});".format(video_status_sql, video_ids[0]) data = mysql_helper.get_data(sql=sql) else: data = [] for i in range(len(video_ids) // 200 + 1): sql = "SELECT video_id " \ "FROM ({}) " \ "WHERE app_audit_status = 5 " \ "AND app_rec_status IN (1, -6, 10) " \ "AND open_status = 1 " \ "AND payment_status = 0 " \ "AND encryption_status != 5 " \ "AND transcoding_status = 3 " \ "AND video_id IN {};".format(video_status_sql, tuple(video_ids[i*200:(i+1)*200])) select_res = mysql_helper.get_data(sql=sql) if select_res is not None: data += select_res filtered_videos = [int(temp[0]) for temp in data] return filtered_videos except Exception as e: log_.error(f"过滤失败, exception: {e}, traceback: {traceback.format_exc()}") send_msg_to_feishu( webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'), key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'), msg_text=f"rov-offline{config_.ENV_TEXT} - 过滤失败\n" f"retry count: {i}\n" f"exception: {e}\n" f"traceback: {traceback.format_exc()}" ) i += 1 if i == 1: return video_ids def filter_shield_video(video_ids, shield_key_name_list): """ 过滤屏蔽视频视频 :param video_ids: 需过滤的视频列表 type-list :param shield_key_name_list: 过滤视频 redis-key :return: filtered_videos 过滤后的列表 type-list """ if len(video_ids) == 0: return video_ids # 根据Redis缓存中的数据过滤 redis_helper = RedisHelper() for shield_key_name in shield_key_name_list: shield_videos_list = redis_helper.get_data_from_set(key_name=shield_key_name) if not shield_videos_list: continue shield_videos = [int(video) for video in shield_videos_list] video_ids = [int(video_id) for video_id in video_ids if int(video_id) not in shield_videos] return video_ids def filter_political_videos(video_ids): """ 过滤涉政视频 :param video_ids: 需过滤的视频列表 type-list :return: filtered_video_ids 过滤后的列表 type-list """ if len(video_ids) == 0: return video_ids # 根据Redis缓存中的数据过滤 redis_helper = RedisHelper() political_key_name = config_.POLITICAL_VIDEOS_KEY_NAME political_videos_list = redis_helper.get_data_from_set(key_name=political_key_name) if not political_videos_list: return video_ids political_videos = [int(video) for video in political_videos_list] filtered_video_ids = [int(video_id) for video_id in video_ids if int(video_id) not in political_videos] return filtered_video_ids def update_video_w_h_rate(video_ids, key_name): """ 获取横屏视频的宽高比，并存入redis中 （width/height>1) :param video_ids: videoId列表 type-list :param key_name: redis key :return: None """ # 获取数据 if len(video_ids) == 1: sql = "SELECT id, width, height, rotate FROM longvideo.wx_video WHERE id = {};".format(video_ids[0]) else: sql = "SELECT id, width, height, rotate FROM longvideo.wx_video WHERE id IN {};".format(tuple(video_ids)) mysql_helper = MysqlHelper(mysql_info=config_.MYSQL_INFO) data = mysql_helper.get_data(sql=sql) # 更新到redis info_data = {} for video_id, width, height, rotate in data: if int(width) == 0 or int(height) == 0: continue # rotate 字段值为 90或270时，width和height的值相反 if int(rotate) in (90, 270): w_h_rate = int(height) / int(width) else: w_h_rate = int(width) / int(height) if w_h_rate > 1: info_data[int(video_id)] = w_h_rate redis_helper = RedisHelper() # 删除旧数据 redis_helper.del_keys(key_name=key_name) # 写入新数据 if len(info_data) > 0: redis_helper.add_data_with_zset(key_name=key_name, data=info_data) def data_check(project, table, dt): """检查数据是否准备好""" odps = ODPS( access_id=config_.ODPS_CONFIG['ACCESSID'], secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'], project=project, endpoint=config_.ODPS_CONFIG['ENDPOINT'], connect_timeout=3000, read_timeout=500000, pool_maxsize=1000, pool_connections=1000 ) try: check_res = check_table_partition_exits(date=dt, project=project, table=table) if check_res: sql = f'select * from {project}.{table} where dt = {dt}' with odps.execute_sql(sql=sql).open_reader() as reader: data_count = reader.count else: data_count = 0 except Exception as e: data_count = 0 return data_count def get_feature_data(project, table, features, dt): """获取特征数据""" records = get_data_from_odps(date=dt, project=project, table=table) feature_data = [] for record in records: item = {} for feature_name in features: item[feature_name] = record[feature_name] feature_data.append(item) feature_df = pd.DataFrame(feature_data) return feature_df if __name__ == '__main__': # data_test = [9.20273281e+03, 7.00795065e+03, 5.54813112e+03, 9.97402494e-01, 9.96402495e-01, 9.96402494e-01] # data_normalization(data_test) # request_post(request_url=config_.NOTIFY_BACKEND_UPDATE_ROV_SCORE_URL, request_data={'videos': []}) # video_ids = [110, 112, 113, 115, 116, 117, 8289883] # update_video_w_h_rate(video_ids=video_ids, key_name='') project = config_.PROJECT_24H_APP_TYPE table = config_.TABLE_24H_APP_TYPE dt = '2022080115' check_res = check_table_partition_exits(date=dt, project=project, table=table) print(check_res)