rov-server/utils.py

import json
import time
import aiohttp

from datetime import datetime
import aiocache
from config import set_config
from log import Log

config_ = set_config()
log_ = Log()


async def request_post(url, data, timeout=1):
    """
    post 请求 HTTP接口
    :param request_url: 接口URL
    :param request_data: 请求参数
    :param timeout: 超时时间，默认为1秒，type-float
    :return: res_data json格式
    """

    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(url, data=data) as res:
                if res.status == 200:
                    res_data = json.loads(await res.text())
                    print('res_data', res_data)
                    return res_data
                else:
                    return None
    except:
        log_.error('url: {} timeout, exception: {}'.format(url, e))
        return None


async def get_videos_remain_view_count(app_type, videos):
    """
    获取视频在流量池中的剩余可分发数
    :param app_type: 产品标识 type-int
    :param videos: 视频信息 (视频id, 流量池标记) type-list，[{'videoId': video_id, 'flowPool': flow_pool}, ...]
    :return: data type-list，[(video_id, flow_pool, view_count), ...]
    """
    if not videos:
        return []

    request_data = {'appType': app_type, 'videos': videos}
    result = await request_post(url=config_.GET_REMAIN_VIEW_COUNT_URL, data=request_data, timeout=1)

    if result is None:
        return []

    if result['code'] != 0:
        log_.info('获取视频在流量池中的剩余可分发数失败')
        return []

    data = [(item['videoId'], item['flowPool'], item['distributeCount']) for item in result['data']]
    return data


async def get_videos_local_distribute_count(video_id, flow_pool):
    """
    获取流量池视频本地分发数
    :param video_id: video_id
    :param flow_pool: 流量池标记
    :return: current_count 本地记录的分发数
    """
    redis_h = datetime.now().hour
    if datetime.now().minute >= 30:
        redis_h += 0.5
    key_name = config_.LOCAL_DISTRIBUTE_COUNT_PREFIX + str(redis_h)
    video = '{}-{}'.format(video_id, flow_pool)
    current_count = await aiocache.get_score_with_value(key_name=key_name, value=video)
    if current_count is not None:
        return current_count
    else:
        return None


class FilterVideos(object):
    """视频过滤"""
    def __init__(self, app_type, video_ids, mid='', uid=''):
        """
        初始化
        :param app_type: 产品标识 type-int
        :param video_ids: 需过滤的视频列表 type-list
        :param mid: mid type-string
        :param uid: uid type-string
        """
        self.app_type = app_type
        self.mid = mid
        self.uid = uid
        self.video_ids = video_ids

    async def filter_videos(self):
        """视频过滤"""
        # 预曝光过滤
        st_pre = time.time()
        filtered_pre_result = await self.filter_video_previewed(self.video_ids)
        et_pre = time.time()
        log_.info('filter by previewed: app_type = {}, result = {}, execute time = {}ms'.format(
            self.app_type, filtered_pre_result, (et_pre - st_pre) * 1000))
        if not filtered_pre_result:
            return None

        # 视频状态过滤
        st_status = time.time()
        filtered_status_result = await self.filter_video_status(video_ids=filtered_pre_result)
        et_status = time.time()
        log_.info('filter by video status: result = {}, execute time = {}ms'.format(
            filtered_status_result, (et_status - st_status) * 1000))
        if not filtered_status_result:
            return None

        # 视频已曝光过滤
        st_viewed = time.time()
        filtered_viewed_result = await self.filter_video_viewed(video_ids=filtered_status_result)
        et_viewed = time.time()
        log_.info('filter by viewed: app_type = {}, mid = {}, uid = {}, result = {}, execute time = {}ms'.format(
            self.app_type, self.mid, self.uid, filtered_viewed_result, (et_viewed - st_viewed) * 1000))
        if not filtered_viewed_result:
            return None
        else:
            return filtered_viewed_result

    async def filter_video_previewed(self, video_ids):
        """
        预曝光过滤
        :param video_ids: 需过滤的视频列表 type-list
        :return: filtered_videos  过滤后的列表  type-list
        """
        # 根据Redis缓存中的数据过滤
        # key拼接
        key_name = config_.PREVIEW_KEY_PREFIX + '{}.{}'.format(self.app_type, self.mid)
        pe_videos_list = await aiocache.get_data_from_set(key_name)
        if not pe_videos_list:
            return video_ids
        pe_videos = [eval(video) for video in pe_videos_list]
        filtered_videos = [video_id for video_id in video_ids if video_id not in pe_videos]
        return filtered_videos

    async def filter_video_status(self, video_ids):
        """
        对视频状态进行过滤
        :param video_ids: 视频id列表 type-list
        :return: filtered_videos
        """
        b_time = time.time()
        if len(video_ids) == 1:
            sql = "set hg_experimental_enable_shard_pruning=off; " \
                  "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 IS NULL " \
                  "AND transcoding_status = 3 " \
                  "AND video_id IN ({});".format(config_.VIDEO_STATUS, video_ids[0])
        else:
            sql = "set hg_experimental_enable_shard_pruning=off; " \
                  "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 IS NULL " \
                  "AND transcoding_status = 3 " \
                  "AND video_id IN {};".format(config_.VIDEO_STATUS, tuple(video_ids))

        data = await aiocache.pg_getdata(sql=sql)
        filtered_videos = [temp[0] for temp in data]
        e_time = time.time()
        print('holo time',e_time-b_time)
        return filtered_videos

    async def filter_video_viewed(self, video_ids, types=(1,)):
        """
        调用后端接口过滤用户已观看视频
        :param video_ids: 视频id列表 type-list
        :param types: 过滤参数 type-tuple, 默认(1, )  1-已观看 2-视频状态 3-是否进入老年人社区 4-话题状态 5-推荐状态
        :return: filtered_videos
        """
        # 调用http接口
        request_data = {"appType": self.app_type,
                        "mid": self.mid,
                        "uid": self.uid,
                        "types": list(types),
                        "videoIds": video_ids}
        result = await request_post(url=config_.VIDEO_FILTER_URL, data=request_data, timeout=2.5)

        if result is None:
            log_.info('过滤失败，types: {}'.format(types))
            return video_ids

        if result['code'] != 0:
            log_.info('过滤失败，types: {}'.format(types))
            return video_ids

        filtered_videos = result['data']
        return filtered_videos


if __name__ == '__main__':
    filter_ = FilterVideos(app_type=1, mid='22', uid='www', video_ids=[1, 2, 3, 55])
    filter_.filter_videos()
    filter_.filter_video_status(video_ids=[1, 3, 5])