utils.py 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707
  1. import traceback
  2. import requests
  3. import json
  4. import time
  5. import gevent
  6. import pandas as pd
  7. import random
  8. from datetime import datetime
  9. # from db_helper import HologresHelper, RedisHelper, MysqlHelper
  10. from db_helper import RedisHelper, MysqlHelper
  11. from config import set_config
  12. from log import Log
  13. config_ = set_config()
  14. log_ = Log()
  15. def send_msg_to_feishu(msg_text):
  16. """发送消息到飞书"""
  17. # webhook地址
  18. webhook = 'https://open.feishu.cn/open-apis/bot/v2/hook/8de4de35-30ed-4692-8854-7a154e89b2f2'
  19. # 自定义关键词key_word
  20. key_word = '服务报警'
  21. headers = {'Content-Type': 'application/json'}
  22. payload_message = {
  23. "msg_type": "text",
  24. "content": {
  25. "text": '{}: {}'.format(key_word, msg_text)
  26. }
  27. }
  28. response = requests.request('POST', url=webhook, headers=headers, data=json.dumps(payload_message))
  29. # print(response.text)
  30. def request_post(request_url, request_data, timeout):
  31. """
  32. post 请求 HTTP接口
  33. :param request_url: 接口URL
  34. :param request_data: 请求参数
  35. :param timeout: 超时时间,单位为秒,type-float or tuple(connect_timeout, read_timeout)
  36. :return: res_data json格式
  37. """
  38. try:
  39. headers = {"Connection": "close"}
  40. response = requests.post(url=request_url, json=request_data, timeout=timeout, headers=headers)
  41. if response.status_code == 200:
  42. res_data = json.loads(response.text)
  43. return res_data
  44. else:
  45. return None
  46. except Exception as e:
  47. log_.error('url: {}, exception: {}, traceback: {}'.format(request_url, e, traceback.format_exc()))
  48. return None
  49. def request_get(request_url, timeout):
  50. """
  51. get 请求 HTTP接口
  52. :param request_url: 接口URL
  53. :param timeout: 超时时间,单位为秒,type-float or tuple(connect_timeout, read_timeout)
  54. :return: res_data json格式
  55. """
  56. try:
  57. response = requests.get(url=request_url, timeout=timeout)
  58. if response.status_code == 200:
  59. res_data = json.loads(response.text)
  60. return res_data
  61. else:
  62. return None
  63. except Exception as e:
  64. log_.error('url: {}, exception: {}, traceback: {}'.format(request_url, e, traceback.format_exc()))
  65. return None
  66. def get_user_has30day_return(mid):
  67. """
  68. 获取用户近30天是否有回流
  69. :param mid: mid
  70. :return: data, type
  71. """
  72. if not mid:
  73. return None
  74. # 获取redis中存储的状态值
  75. user_key = f"{config_.KEY_NAME_PREFIX_USER_HAS30DAY_RETURN}{mid}"
  76. redis_helper = RedisHelper()
  77. data = redis_helper.get_data_from_redis(key_name=user_key)
  78. if data is not None:
  79. return int(data)
  80. else:
  81. request_url = f"{config_.GET_USER_30DayReturnCnt_URL}{mid}"
  82. result = request_get(request_url=request_url, timeout=0.1)
  83. if result is None:
  84. return None
  85. if result['code'] != 0:
  86. return None
  87. data = result['data']
  88. if data is True:
  89. redis_data = 1
  90. else:
  91. redis_data = 0
  92. redis_helper.set_data_to_redis(key_name=user_key, value=redis_data, expire_time=2 * 3600)
  93. return redis_data
  94. def get_videos_remain_view_count(app_type, videos):
  95. """
  96. 获取视频在流量池中的剩余可分发数
  97. :param app_type: 产品标识 type-int
  98. :param videos: 视频信息 (视频id, 流量池标记) type-list,[{'videoId': video_id, 'flowPool': flow_pool}, ...]
  99. :return: data type-list,[(video_id, flow_pool, view_count), ...]
  100. error_flag 错误标记,True为错误
  101. """
  102. error_flag = False
  103. if not videos:
  104. return [], error_flag
  105. request_data = {'appType': app_type, 'videos': videos}
  106. result = request_post(request_url=config_.GET_REMAIN_VIEW_COUNT_URL, request_data=request_data, timeout=(0.1, 1))
  107. if result is None:
  108. error_flag = True
  109. return [], error_flag
  110. if result['code'] != 0:
  111. log_.info('获取视频在流量池中的剩余可分发数失败')
  112. error_flag = True
  113. return [], error_flag
  114. data = [(item['videoId'], item['flowPool'], item['distributeCount']) for item in result['data']]
  115. return data, error_flag
  116. def get_videos_local_distribute_count(video_id, flow_pool):
  117. """
  118. 获取流量池视频本地分发数
  119. :param video_id: video_id
  120. :param flow_pool: 流量池标记
  121. :return: current_count 本地记录的分发数
  122. """
  123. # redis_h = datetime.now().hour
  124. # if datetime.now().minute >= 30:
  125. # redis_h += 0.5
  126. # key_name = config_.LOCAL_DISTRIBUTE_COUNT_PREFIX + str(redis_h)
  127. key_name = f'{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{video_id}:{flow_pool}'
  128. redis_helper = RedisHelper()
  129. # video = '{}-{}'.format(video_id, flow_pool)
  130. # current_count = redis_helper.get_score_with_value(key_name=key_name, value=video)
  131. current_count = redis_helper.get_data_from_redis(key_name=key_name)
  132. if current_count is not None:
  133. return int(current_count)
  134. else:
  135. return None
  136. def update_video_w_h_rate(video_id, key_name):
  137. """
  138. 获取横屏视频的宽高比,并存入redis中 (width/height>1)
  139. :param video_id: videoId type-int
  140. :param key_name: redis key
  141. :return: None
  142. """
  143. # 获取数据
  144. sql = "SELECT id, width, height, rotate FROM longvideo.wx_video WHERE id = {};".format(video_id)
  145. mysql_helper = MysqlHelper()
  146. data = mysql_helper.get_data(sql=sql)
  147. if len(data) == 0:
  148. return
  149. # 更新到redis
  150. width, height, rotate = int(data[0][1]), int(data[0][2]), int(data[0][3])
  151. if width == 0 or height == 0:
  152. return
  153. if rotate in (90, 270):
  154. w_h_rate = height / width
  155. else:
  156. w_h_rate = width / height
  157. if w_h_rate > 1:
  158. info_data = {int(video_id): w_h_rate}
  159. else:
  160. return
  161. redis_helper = RedisHelper()
  162. # 写入新数据
  163. if len(info_data) > 0:
  164. redis_helper.add_data_with_zset(key_name=key_name, data=info_data)
  165. class FilterVideos(object):
  166. """视频过滤"""
  167. def __init__(self, request_id, app_type, video_ids, mid='', uid=''):
  168. """
  169. 初始化
  170. :param request_id: request_id
  171. :param app_type: 产品标识 type-int
  172. :param video_ids: 需过滤的视频列表 type-list
  173. :param mid: mid type-string
  174. :param uid: uid type-string
  175. """
  176. self.request_id = request_id
  177. self.app_type = app_type
  178. self.mid = mid
  179. self.uid = uid
  180. self.video_ids = video_ids
  181. def filter_video_status_h(self, video_ids, rule_key, data_key, ab_code, province_code, key_flag=''):
  182. """召回小时级更新的视频状态过滤"""
  183. # 根据Redis缓存中的数据过滤
  184. redis_helper = RedisHelper()
  185. # 获取不符合推荐状态的视频
  186. if ab_code in [code for _, code in config_.AB_CODE['region_rank_by_h'].items()]:
  187. if key_flag == 'region_24h':
  188. key_prefix = f"{config_.REGION_H_VIDEO_FILER_24H}{province_code}."
  189. elif key_flag == 'day_24h':
  190. key_prefix = f"{config_.H_VIDEO_FILER_24H}{province_code}."
  191. else:
  192. key_prefix = f"{config_.REGION_H_VIDEO_FILER}{province_code}."
  193. elif ab_code in [code for _, code in config_.AB_CODE['rank_by_24h'].items()]:
  194. key_prefix = config_.H_VIDEO_FILER_24H
  195. elif key_flag == '24h':
  196. key_prefix = config_.H_VIDEO_FILER_24H
  197. else:
  198. key_prefix = config_.H_VIDEO_FILER
  199. filter_videos_list = redis_helper.get_data_from_set(
  200. key_name=f"{key_prefix}{self.app_type}.{data_key}.{rule_key}"
  201. )
  202. if not filter_videos_list:
  203. return video_ids
  204. filter_videos = [int(video) for video in filter_videos_list]
  205. filtered_videos = [video_id for video_id in video_ids if video_id not in filter_videos]
  206. return filtered_videos
  207. def filter_videos_h(self, rule_key, data_key, ab_code, province_code, key_flag='', pool_type='rov'):
  208. """召回小时级更新的视频过滤"""
  209. # 预曝光过滤
  210. # st_pre = time.time()
  211. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  212. # et_pre = time.time()
  213. # log_.info({
  214. # 'logTimestamp': int(time.time() * 1000),
  215. # 'request_id': self.request_id,
  216. # 'app_type': self.app_type,
  217. # 'mid': self.mid,
  218. # 'uid': self.uid,
  219. # 'operation': 'preview_filter',
  220. # 'request_videos': self.video_ids,
  221. # 'preview_filter_result': filtered_pre_result,
  222. # 'executeTime': (time.time() - st_pre) * 1000
  223. # })
  224. if not filtered_pre_result:
  225. return None
  226. # 视频状态过滤
  227. # st_status = time.time()
  228. filtered_status_result = self.filter_video_status_h(video_ids=filtered_pre_result, rule_key=rule_key,
  229. data_key=data_key, ab_code=ab_code,
  230. province_code=province_code, key_flag=key_flag)
  231. # et_status = time.time()
  232. # log_.info({
  233. # 'logTimestamp': int(time.time() * 1000),
  234. # 'request_id': self.request_id,
  235. # 'app_type': self.app_type,
  236. # 'mid': self.mid,
  237. # 'uid': self.uid,
  238. # 'operation': 'status_filter',
  239. # 'request_videos': filtered_pre_result,
  240. # 'status_filter_result': filtered_status_result,
  241. # 'executeTime': (time.time() - st_status) * 1000
  242. # })
  243. if not filtered_status_result:
  244. return None
  245. # 视频已曝光过滤
  246. st_viewed = time.time()
  247. filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_status_result)
  248. # et_viewed = time.time()
  249. log_.info({
  250. 'logTimestamp': int(time.time() * 1000),
  251. 'pool_type': pool_type,
  252. 'request_id': self.request_id,
  253. 'app_type': self.app_type,
  254. 'mid': self.mid,
  255. 'uid': self.uid,
  256. 'operation': 'view_filter',
  257. 'request_videos': filtered_status_result,
  258. 'view_filter_result': filtered_viewed_result,
  259. 'executeTime': (time.time() - st_viewed) * 1000
  260. })
  261. if not filtered_viewed_result:
  262. return None
  263. else:
  264. return [int(video_id) for video_id in filtered_viewed_result]
  265. def filter_videos(self, pool_type='rov', region_code=None, shield_config=None):
  266. """视频过滤"""
  267. # 预曝光过滤
  268. st_pre = time.time()
  269. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  270. # print("filtered_pre:", (time.time()-st_pre)*1000)
  271. # et_pre = time.time()
  272. # log_.info({
  273. # 'logTimestamp': int(time.time() * 1000),
  274. # 'request_id': self.request_id,
  275. # 'app_type': self.app_type,
  276. # 'mid': self.mid,
  277. # 'uid': self.uid,
  278. # 'operation': 'preview_filter',
  279. # 'request_videos': self.video_ids,
  280. # 'preview_filter_result': filtered_pre_result,
  281. # 'executeTime': (time.time() - st_pre) * 1000
  282. # })
  283. if not filtered_pre_result:
  284. return None
  285. # 视频状态过滤采用离线定时过滤方案
  286. # 视频状态过滤
  287. # st_status = time.time()
  288. # filtered_status_result = self.filter_video_status(video_ids=filtered_pre_result)
  289. # et_status = time.time()
  290. # log_.info('filter by video status: result = {}, execute time = {}ms'.format(
  291. # filtered_status_result, (et_status - st_status) * 1000))
  292. # if not filtered_status_result:
  293. # return None
  294. # 视频已曝光过滤
  295. st_viewed = time.time()
  296. filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_pre_result)
  297. # print("filtered_pre:", (time.time() - st_viewed) * 1000)
  298. # et_viewed = time.time()
  299. # log_.info({
  300. # 'logTimestamp': int(time.time() * 1000),
  301. # 'pool_type': pool_type,
  302. # 'request_id': self.request_id,
  303. # 'app_type': self.app_type,
  304. # 'mid': self.mid,
  305. # 'uid': self.uid,
  306. # 'operation': 'view_filter',
  307. # 'request_videos': filtered_pre_result,
  308. # 'view_filter_result': filtered_viewed_result,
  309. # 'executeTime': (time.time() - st_viewed) * 1000
  310. # })
  311. if not filtered_viewed_result:
  312. return None
  313. filtered_viewed_videos = [int(video_id) for video_id in filtered_viewed_result]
  314. if pool_type == 'flow':
  315. # 流量池视频需过滤屏蔽视频
  316. if region_code is None or shield_config is None:
  317. return filtered_viewed_videos
  318. else:
  319. shield_key_name_list = shield_config.get(region_code, None)
  320. if shield_key_name_list is not None:
  321. filtered_shield_video_ids = self.filter_shield_video(
  322. video_ids=filtered_viewed_videos, shield_key_name_list=shield_key_name_list
  323. )
  324. log_.info({
  325. 'logTimestamp': int(time.time() * 1000),
  326. 'pool_type': pool_type,
  327. 'request_id': self.request_id,
  328. 'app_type': self.app_type,
  329. 'mid': self.mid,
  330. 'uid': self.uid,
  331. 'operation': 'shield_filter',
  332. 'request_videos': filtered_viewed_videos,
  333. 'shield_filter_result': filtered_shield_video_ids,
  334. 'executeTime': (time.time() - st_viewed) * 1000
  335. })
  336. # print("filtered_pre flow:", (time.time() - st_viewed) * 1000)
  337. return filtered_shield_video_ids
  338. else:
  339. return filtered_viewed_videos
  340. else:
  341. return filtered_viewed_videos
  342. def filter_video_previewed(self, video_ids):
  343. """
  344. 预曝光过滤
  345. :param video_ids: 需过滤的视频列表 type-list
  346. :return: filtered_videos 过滤后的列表 type-list
  347. """
  348. pre_time = time.time()
  349. if not self.mid or self.mid == 'null':
  350. # mid为空时,不做预曝光过滤
  351. return video_ids
  352. # 根据Redis缓存中的数据过滤
  353. redis_helper = RedisHelper()
  354. # key拼接
  355. key_name = f"{config_.PREVIEW_KEY_PREFIX}{self.app_type}:{self.mid}"
  356. #print("key_name:", key_name)
  357. pe_videos_list = redis_helper.get_data_from_set(key_name)
  358. #print("pe_videos_list:", pe_videos_list)
  359. # log_.info('****app_type = {}, mid = {}, uid = {}, pe_videos_list = {}'.format(
  360. # self.app_type, self.mid, self.uid, pe_videos_list))
  361. # log_.info('****app_type = {}, mid = {}, uid = {}, video_ids = {}'.format(
  362. # self.app_type, self.mid, self.uid, video_ids))
  363. if not pe_videos_list:
  364. return video_ids
  365. pe_videos = [int(video) for video in pe_videos_list]
  366. #print("pe_videos:", len(pe_videos))
  367. filtered_videos = [video_id for video_id in video_ids if video_id not in pe_videos]
  368. #print(f"pre res: {filtered_videos}\nexecute_time: {(time.time() - pre_time) * 1000}")
  369. return filtered_videos
  370. # def filter_video_status(self, video_ids):
  371. # """
  372. # 对视频状态进行过滤
  373. # :param video_ids: 视频id列表 type-list
  374. # :return: filtered_videos
  375. # """
  376. # if len(video_ids) == 1:
  377. # sql = "set hg_experimental_enable_shard_pruning=off; " \
  378. # "SELECT video_id " \
  379. # "FROM {} " \
  380. # "WHERE audit_status = 5 " \
  381. # "AND applet_rec_status IN (1, -6) " \
  382. # "AND open_status = 1 " \
  383. # "AND payment_status = 0 " \
  384. # "AND encryption_status != 5 " \
  385. # "AND transcoding_status = 3 " \
  386. # "AND video_id IN ({});".format(config_.VIDEO_STATUS, video_ids[0])
  387. # else:
  388. # sql = "set hg_experimental_enable_shard_pruning=off; " \
  389. # "SELECT video_id " \
  390. # "FROM {} " \
  391. # "WHERE audit_status = 5 " \
  392. # "AND applet_rec_status IN (1, -6) " \
  393. # "AND open_status = 1 " \
  394. # "AND payment_status = 0 " \
  395. # "AND encryption_status != 5 " \
  396. # "AND transcoding_status = 3 " \
  397. # "AND video_id IN {};".format(config_.VIDEO_STATUS, tuple(video_ids))
  398. #
  399. # hologres_helper = HologresHelper()
  400. # data = hologres_helper.get_data(sql=sql)
  401. # filtered_videos = [int(temp[0]) for temp in data]
  402. # return filtered_videos
  403. def filter_video_viewed(self, video_ids, types=(1, 6,)):
  404. """
  405. 调用后端接口过滤用户已观看视频
  406. :param video_ids: 视频id列表 type-list
  407. :param types: 过滤参数 type-tuple, 默认(1, )
  408. 1-已观看 2-视频状态 3-是否进入老年人社区 4-话题状态 5-推荐状态 6-白名单过滤 7-涉政视频过滤
  409. :return: filtered_videos
  410. """
  411. # 获取对应端的过滤参数types
  412. types = config_.FILTER_VIEWED_TYPES_CONFIG.get(self.app_type, None)
  413. if types is None:
  414. types = config_.FILTER_VIEWED_TYPES_CONFIG.get('other')
  415. request_data = {"appType": self.app_type,
  416. "mid": self.mid,
  417. "uid": self.uid,
  418. "types": list(types),
  419. "videoIds": video_ids}
  420. # print(request_data)
  421. # 调用http接口
  422. result = request_post(request_url=config_.VIDEO_FILTER_URL, request_data=request_data, timeout=(0.1, 1))
  423. # print("result:", result)
  424. if result is None:
  425. # print("result is None")
  426. # log_.info('过滤失败,types: {}'.format(types))
  427. return []
  428. if result['code'] != 0:
  429. # log_.info('过滤失败,types: {}'.format(types))
  430. return []
  431. filtered_videos = result['data']
  432. return filtered_videos
  433. def filter_video_viewed_new(self, video_ids):
  434. """
  435. 调用后端接口过滤用户已观看视频
  436. :param video_ids: 视频id列表 type-list
  437. :param types: 过滤参数 type-tuple, 默认(1, )
  438. 1-已观看 2-视频状态 3-是否进入老年人社区 4-话题状态 5-推荐状态 6-白名单过滤 7-涉政视频过滤
  439. :return: filtered_videos
  440. """
  441. # 获取对应端的过滤参数types
  442. st_time = time.time()
  443. types = config_.FILTER_VIEWED_TYPES_CONFIG.get(self.app_type, None)
  444. #print(types)
  445. if types is None:
  446. types = config_.FILTER_VIEWED_TYPES_CONFIG.get('other')
  447. if 6 in types:
  448. types = list(types)
  449. types.remove(6)
  450. #print(types)
  451. request_data = {"appType": self.app_type,
  452. "mid": self.mid,
  453. "uid": self.uid,
  454. "types": list(types),
  455. "videoIds": video_ids}
  456. # 调用http接口
  457. result = request_post(request_url=config_.VIDEO_FILTER_URL, request_data=request_data, timeout=(0.1, 1))
  458. #print(f"view res: {result}\nexecute_time: {(time.time() - st_time) * 1000}")
  459. if result is None:
  460. # log_.info('过滤失败,types: {}'.format(types))
  461. return []
  462. if result['code'] != 0:
  463. # log_.info('过滤失败,types: {}'.format(types))
  464. return []
  465. filtered_videos = result['data']
  466. return filtered_videos
  467. def filter_shield_video(self, video_ids, shield_key_name_list):
  468. """
  469. 过滤屏蔽视频视频
  470. :param video_ids: 需过滤的视频列表 type-list
  471. :param shield_key_name_list: 过滤视频 redis-key
  472. :return: filtered_videos 过滤后的列表 type-list
  473. """
  474. # print("filter_shield_video:", len(filter_shield_video))
  475. if len(video_ids) == 0:
  476. return video_ids
  477. # 根据Redis缓存中的数据过滤
  478. redis_helper = RedisHelper()
  479. for shield_key_name in shield_key_name_list:
  480. video_ids = [
  481. int(video_id) for video_id in video_ids
  482. if not redis_helper.data_exists_with_set(key_name=shield_key_name, value=video_id)
  483. ]
  484. # shield_videos_list = redis_helper.get_data_from_set(key_name=shield_key_name)
  485. # if not shield_videos_list:
  486. # continue
  487. # shield_videos = [int(video) for video in shield_videos_list]
  488. # video_ids = [int(video_id) for video_id in video_ids if int(video_id) not in shield_videos]
  489. # print("video_ids:", len(video_ids))
  490. return video_ids
  491. def new_filter_video(self):
  492. """视频过滤"""
  493. # 1. 预曝光过滤
  494. st_pre = time.time()
  495. #print("new_filter video_ids:", self.video_ids)
  496. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  497. if not filtered_pre_result:
  498. return None
  499. # log_.info({
  500. # 'logTimestamp': int(time.time() * 1000),
  501. # 'request_id': self.request_id,
  502. # 'app_type': self.app_type,
  503. # 'mid': self.mid,
  504. # 'uid': self.uid,
  505. # 'operation': 'preview_filter',
  506. # 'request_videos': self.video_ids,
  507. # 'preview_filter_result': filtered_pre_result,
  508. # 'executeTime': (time.time() - st_pre) * 1000
  509. # })
  510. #2. 视频已曝光过滤
  511. st_viewed = time.time()
  512. #print("---filtered viewed---")
  513. #print("filtered_pre_result:",filtered_pre_result)
  514. filtered_viewed_result = self.filter_video_viewed(video_ids=filtered_pre_result)
  515. if not filtered_viewed_result:
  516. return None
  517. return filtered_viewed_result
  518. def new_flow_video(self, vid_list, flow_vids_set, region_code, shield_config):
  519. flow_video_list = []
  520. normal_video_list = []
  521. for v_id in vid_list:
  522. if v_id in flow_vids_set:
  523. flow_video_list.append(v_id)
  524. else:
  525. normal_video_list.append(v_id)
  526. shield_key_name_list = shield_config.get(region_code, None)
  527. if shield_key_name_list is not None:
  528. filtered_shield_video_ids = self.filter_shield_video(
  529. video_ids=flow_video_list, shield_key_name_list=shield_key_name_list
  530. )
  531. return normal_video_list, filtered_shield_video_ids
  532. else:
  533. return normal_video_list, flow_video_list
  534. def filter_movie_religion_video(self, video_ids):
  535. """过滤白名单视频(影视,宗教)"""
  536. # 影视 + 宗教: rov.filter.movie.{videoId}
  537. # 宗教: rov.filter.religion.{videoId}
  538. st_time = time.time()
  539. if self.app_type not in [config_.APP_TYPE['WAN_NENG_VIDEO'],
  540. config_.APP_TYPE['LAO_HAO_KAN_VIDEO'],
  541. config_.APP_TYPE['ZUI_JING_QI'],
  542. config_.APP_TYPE['H5']]:
  543. # 过滤 影视 + 宗教
  544. keys = [f"rov.filter.movie.{video_id}" for video_id in video_ids]
  545. elif self.app_type in [config_.APP_TYPE['WAN_NENG_VIDEO'],
  546. config_.APP_TYPE['ZUI_JING_QI'],
  547. config_.APP_TYPE['H5']]:
  548. # 过滤 影视 + 宗教
  549. keys = [f"rov.filter.religion.{video_id}" for video_id in video_ids]
  550. else:
  551. #print(f"m_r res: {video_ids}\nexecute_time: {(time.time() - st_time) * 1000}")
  552. return video_ids
  553. redis_helper = RedisHelper(redis_info=config_.REDIS_INFO_FILTER)
  554. filter_videos = []
  555. for i in range(len(keys)//1000 + 1):
  556. video_ids_temp = video_ids[i*1000:(i+1)*1000]
  557. if len(video_ids_temp) == 0:
  558. break
  559. mget_res = redis_helper.mget(keys=keys[i*1000:(i+1)*1000])
  560. filter_videos.extend([int(data) for data in mget_res if data is not None])
  561. if len(filter_videos) > 0:
  562. filtered_videos = set(video_ids) - set(filter_videos)
  563. #print(f"m_r res: {list(filtered_videos)}\nexecute_time: {(time.time() - st_time) * 1000}")
  564. return list(filtered_videos)
  565. else:
  566. #print(f"m_r res: {video_ids}\nexecute_time: {(time.time() - st_time) * 1000}")
  567. return video_ids
  568. def filter_videos_new(self, region_code=None, shield_config=None, flow_set=None):
  569. """视频过滤"""
  570. # 预曝光过滤
  571. st_pre = time.time()
  572. #print("self.video_ids:", len(self.video_ids))
  573. filtered_pre_result = self.filter_video_previewed(self.video_ids)
  574. if not filtered_pre_result:
  575. return None
  576. #print("filtered_pre_result:", len(filtered_pre_result))
  577. #print(filtered_pre_result)
  578. # 视频已曝光过滤/白名单过滤
  579. st_viewed = time.time()
  580. t = [
  581. gevent.spawn(self.filter_video_viewed_new, filtered_pre_result),
  582. gevent.spawn(self.filter_movie_religion_video, filtered_pre_result)]
  583. gevent.joinall(t)
  584. filtered_result_list = [i.get() for i in t]
  585. #print("filtered_result_list1:",filtered_result_list[0])
  586. #print("filtered_result_list2:",filtered_result_list[1])
  587. filtered_viewed_set = set('')
  588. for i in filtered_result_list[0]:
  589. filtered_viewed_set.add(int(i))
  590. filter_video_set =set('')
  591. for j in filtered_result_list[1]:
  592. filter_video_set.add(int(j))
  593. filtered_viewed_result = list(filtered_viewed_set & filter_video_set)
  594. #print(f"view&m_r res: {filtered_viewed_result}\nexecute_time: {(time.time() - st_viewed) * 1000}")
  595. #print("filtered:",len(filtered_viewed_result))
  596. if not filtered_viewed_result:
  597. return None
  598. filtered_viewed_videos = [int(video_id) for video_id in filtered_viewed_result]
  599. #print("result:", filtered_viewed_videos)
  600. if flow_set is None:
  601. return filtered_viewed_videos
  602. else:
  603. # 流量池视频需过滤屏蔽视频
  604. if region_code is None or shield_config is None:
  605. return filtered_viewed_videos
  606. else:
  607. normal_recall_ids = []
  608. left_flow_ids = []
  609. for vid in filtered_viewed_videos:
  610. if vid in flow_set:
  611. left_flow_ids.append(vid)
  612. else:
  613. normal_recall_ids.append(vid)
  614. shield_key_name_list = shield_config.get(region_code, None)
  615. if shield_key_name_list is not None:
  616. filtered_shield_video_ids = self.filter_shield_video(
  617. video_ids=left_flow_ids, shield_key_name_list=shield_key_name_list
  618. )
  619. return normal_recall_ids+filtered_shield_video_ids
  620. else:
  621. return filtered_viewed_videos
  622. if __name__ == '__main__':
  623. user = [
  624. ('weixin_openid_o0w175fDc8pNnywrYN49E341tKfI', ''),
  625. ('weixin_openid_o0w175YwC3hStzcR5DAQdbgzdMeI', ''),
  626. ('weixin_openid_o0w175ftZDl6VJVDx9la3WVPh7mU', '15900461'),
  627. ('weixin_openid_o0w175SPqpCVRcp7x1XvnX4qpIvI', '19659040'),
  628. ('weixin_openid_o0w175cOnguapyWIrDrHkOWl4oFQ', '31210128'),
  629. ('weixin_openid_o0w175UXYId-o71e1Q3SOheYNteQ', '33099722'),
  630. ('weixin_openid_o0w175QQ5b42AtOe50bchrFgcttA', ''),
  631. ('weixin_openid_o0w175bgaPlfLsp3YLDKWqLWtXX8', '35371534'),
  632. ('weixin_openid_o0w175eRpvbmV6nOhM1VTyyLICWA', '30488803'),
  633. ('weixin_openid_o0w175TZYvG47pQkOjyJFoxQuqsw', '')
  634. ]
  635. video_df = pd.read_csv('./data/videoids.csv')
  636. videoid_list = video_df['videoid'].tolist()
  637. for mid, uid in user:
  638. video_ids = random.sample(videoid_list, 1000)
  639. start_time = time.time()
  640. filter_ = FilterVideos(request_id=f'{mid} - {uid}', app_type=0, mid=mid, uid=uid, video_ids=video_ids)
  641. res = filter_.filter_videos_new()
  642. print(f"res: {res}\nexecute_time: {(time.time() - start_time) * 1000}")
  643. # filter_.filter_video_status(video_ids=[1, 3, 5])
  644. # videos = [{'videoId': 9034659, 'flowPool': '3#11#3#1637824188547'}, {'videoId': 9035052, 'flowPool': '3#11#3#1637824172827'}]
  645. # res = get_videos_remain_view_count(4, videos)
  646. # print(res)
  647. # text = '测试 @李倩'
  648. # send_msg_to_feishu(text)
  649. # update_video_w_h_rate(video_id=113, key_name='')
  650. # mid = "weixin_openid_obHDW5c4g3aULfCWh-68LcUSxCB"
  651. # request_url = f"{config_.GET_USER_30DayReturnCnt_URL}{mid}"
  652. # res = request_get(request_url=request_url, timeout=100)
  653. # res = get_user_has30day_return(mid=mid)
  654. # print(res, type(res))