import re import json import logging import requests from django.core.cache import cache from utils import date_utils from domain.injector.task_queue import TaskQueue, TaskCacheHash from domain.lsp.base_injector import LspInjectorG from injector.models import TaskModel from domain.imdb.media_domain import Media from domain.injector.media_storage_domain import MediaStorage from domain.injector.value_object import TaskStatusValue from domain.imdb.value_object import CategoryValue from utils.decorators import injector_g_cache_timeout logger = logging.getLogger(__file__) class GitvBase(LspInjectorG): TRANS_CODE_TASK_CACHE = "gitv_trans_code_task:{gitv_task_id}" TRANS_CODE_FILE_URL_CACHE = "gitv_trans_code_file_url:{storage_id}" EXPIRE_TIME = 60 * 60 * 24 * 30 Media_Required_Args = ('year', 'name', 'category', 'genres', 'date_published', 'episodes', 'storyline', 'regions', 'languages') Episode_Required_Args = ('serial_num', 'episode_horpic') Video_Required_Args = ('seconds', 'content_type', 'definition', 'pro_type', 'dub') MediaExt_Required_Args = ('media_verpic', 'media_horpic', 'focus', 'rating') Cast_required_Args = ('actor', 'director') def __init__(self, alias="GITV"): super().__init__(alias) self.key = "v0j#efkgFdFCn" self.salt = "gitv@!_tv_party" self.level = 8 def check_args(self, storage_id): """校验参数是不是全""" video = MediaStorage.model.objects.select_related('media', 'episode').get(id=storage_id) self.check_module_args(video.media, self.Media_Required_Args) self.check_module_args(video.episode, self.Episode_Required_Args) self.check_module_args(video, self.Video_Required_Args) self.check_module_args(video.media.mediaextfieldmodel, self.MediaExt_Required_Args) self.check_media_cast(video.media.celebritymediarelationmodel_set.all(), self.Cast_required_Args) return # 有分集标题时 def _have_episode_title(self, media, episode): fmt = "第{0}集{1}" return fmt.format(episode.serial_num, episode.title) # 无分集标题时 def _nope_episode_title(self, media, episode): fmt = "{0}第{1}集" return fmt.format(self.get_media_name_by_alias(media), episode.serial_num) # 向转码服务提交本地文件 def inject_local_video(self, task): if task.level == self.level: count = TaskQueue.get_count(task.level) if count: count = int(count) else: count = 0 if count < 49: TaskQueue.add_1(self.level) # logger.warning("trans_count<50") else: logger.warning("trans_count>=50") return task.status task = TaskModel.objects.filter(pk=task.id).select_related("storage").first() file_name = task.storage.name.strip() suffix = file_name.split(".")[-1] # 文件后缀 # 转码参数 data = { "cp_id": self.lsp.csp_id, "uri": task.storage.url, "media_type": suffix, # 银河目前支持的文件类型mp4,ts "file_name": task.storage.name, } try: response = requests.post(self.lsp.transcode_api, data=data) logger.warning("trans code task_id:{}, response:{}".format(task.id, response)) response = response.json() code = response["code"] if code == "A000000": # 缓存银河转码task id task_cache_key = self.TRANS_CODE_TASK_CACHE.format(gitv_task_id=response["task_id"]) cache.set(task_cache_key, task.id, self.EXPIRE_TIME) # 更新任务状态 status = TaskStatusValue.TRANS_CODE_TRANSFERING remark = "" TaskModel.objects.filter(pk=task.id).update(status=status, remark=remark) else: # 失败之后才更改状态为失败 status = TaskStatusValue.TRANS_CODE_FAILED remark = response["message"] TaskModel.objects.filter(pk=task.id).update(status=status, remark=remark) except Exception as e: status = 0 logger.error("gitv trans code error:{}, storage_id:{} data:{} ".format(e, task.storage.id, data)) return status # 向全局存储提交转码后的文件地址 def inject_transcoded_video(self, task): storage = task.storage file_url_cache_key = self.TRANS_CODE_FILE_URL_CACHE.format(storage_id=storage.id) url = cache.get(file_url_cache_key) data = { "cp_id": self.lsp.csp_id, "album_id": storage.media_id, "video_id": storage.id, "bit_id": self.get_storage_definition(storage.definition, storage.v_codec), "file_type": "m3u8", "uri": url } logger.warning("--- gdyd_upload_storage storage_id:{}, url:{}, data:{}, key:{}".format(storage.id, url, data, file_url_cache_key)) try: response = requests.post(self.lsp.storage_api, data=data) response = response.json() logger.warning('transcode inject resp:', response) code = response["code"] if code == "A000000": status = TaskStatusValue.FILE_TRANSFERING remark = "" TaskModel.objects.filter(pk=task.id).update(status=status, remark=remark) else: status = TaskStatusValue.FILE_FAILED remark = response["message"] TaskModel.objects.filter(pk=task.id).update(status=status, remark=remark) except Exception as e: status = 0 logger.error("gdyd_upload_storage error:{}, url:{}, data:{}".format(e, url, data)) return status # 通知对方拉取增量数量 # @injector_g_cache_timeout def verdict_inject_media_info_call(self, task_id, media_id, status): data_list = [{ "album_id": str(media_id), "is_show": 1 # 目前不考虑下片 }] logger.warning("通知对方拉取 params:{}".format(data_list)) # python语言逗号冒号后面有空格；银河用的编程语言会将空格去掉，导致md5值不一致 body = json.dumps({"data": data_list}, separators=(",", ":")) url = "{}?cpId={}&t={}".format(self.lsp.cms_api, self.lsp.csp_id, self.make_sign(body)) try: response = requests.post(url, data=body, headers={"Content-Type": "application/json"}) logger.warning("通知对方拉取 response:{}".format(response.content)) response = response.json() code = response["code"] message = response["message"] if code != "A00000": status = TaskStatusValue.INFO_FAILED TaskModel.objects.filter(pk=task_id).update(status=status, remark=message) except Exception as e: status = TaskStatusValue.INFO_FAILED message = e TaskModel.objects.filter(pk=task_id).update(status=status, remark=message) logger.error("gdyd_upload_info error:{}, url:{}, body:{}".format(e, url, body)) return status # 执行信息注入流程 def call_for_inject_media_info(self, task, action="REGIST"): # 考虑批量拉取时task状态 status = TaskStatusValue.INFO_TRANSFERING TaskModel.objects.filter(pk=task.id).update(status=status, remark="") if not TaskCacheHash.has_key(task.partner_id, task.media_id): logging.warning("album:add:task:{}".format(task.id)) TaskCacheHash.add(self.lsp_pk, task.media_id, task.storage_id, task.id) status = self.verdict_inject_media_info_call(task.id, task.media_id, status) else: logging.warning("album:add:task:1:{}".format(task.id)) TaskCacheHash.add(self.lsp_pk, task.media_id, task.storage_id, task.id, 1) if status == TaskStatusValue.INFO_FAILED: logging.warning("album:status:failed:{}".format(task.id)) TaskCacheHash.delete(self.lsp_pk, task.media_id, [task.storage_id]) self.continue_push(task.media_id) return status def call_for_inject_media_info1(self, media_id): if not TaskCacheHash.has_key(self.lsp_pk, media_id): field_value_list = TaskCacheHash.get_all(self.lsp_pk, media_id, 1) logging.warning("field_value_list:{}".format(field_value_list)) for field, val in field_value_list: TaskCacheHash.add(self.lsp_pk, media_id, field, val) TaskCacheHash.delete(self.lsp_pk, media_id, 1) # 注入流程入口 def injector(self, task): status = 0 if task.status == TaskStatusValue.TRANS_CODE_PENDING: status = self.inject_local_video(task) if task.status == TaskStatusValue.FILE_PENDING: status = self.inject_transcoded_video(task) if task.status == TaskStatusValue.INFO_PENDING: # self.sub_count(task) status = self.call_for_inject_media_info(task) flag = self.get_continue_to_injector(status) self.just_sub_count(status, task) return status, flag def just_sub_count(self, status, task): if status in [0, TaskStatusValue.TRANS_CODE_FAILED, TaskStatusValue.FILE_FAILED, TaskStatusValue.FILE_TRANSFERING]: count = TaskQueue.get_count(task.level) logger.warning("task_sub_count:{} task_id:{} status:{}".format(count, task.id, status)) self.sub_count(task) def sub_count(self, task): if task.level == self.level: TaskQueue.sub_1(task.level) # 提供银河拉取分集信息 def album(self, media_id, action): logging.warning("album:{}".format(media_id)) if not media_id: # "epg 信息未按标准提供" return "E00002", {} media_domain = Media(media_id) videos, common = self._assemble_videos(media_domain) logging.warning("album videos:{}".format(videos)) logging.warning("album common:{}".format(common)) logging.warning("album:{}".format(media_id)) album = self._assemble_album(media_domain, videos, common) ret = { "album": album, "code": "A00000" } # pull_media_id:2231 logger.warning("gdyd_pull_media_id:{} album:{}".format(media_id, album)) return ret def _assemble_videos(self, media_domain, limit=5): media = media_domain.inst # query_task = { # "storage__media_id": media.id, # "partner_id": self.lsp_pk, # "status": TaskStatusValue.INFO_TRANSFERING # 上传信息成功后会将任务状态置为 INFO_TRANSFERING # } # task_list = TaskModel.objects.filter(**query_task).select_related("storage", "storage__episode")[0: limit] task_list = TaskCacheHash.get_values(self.lsp_pk, media_domain.media_id) task_list_1 = TaskCacheHash.get_values(self.lsp_pk, media_domain.media_id, 1) field_value_list = TaskCacheHash.get_all(self.lsp_pk, media_domain.media_id, 1) logging.warning("album task_list:{}".format(task_list)) logging.warning("album task_list_1:{}".format(task_list_1)) logging.warning("album field_value_list:{}".format(field_value_list)) for field, val in field_value_list.items(): TaskCacheHash.add(self.lsp_pk, media_domain.media_id, field, val) TaskCacheHash.del_key(self.lsp_pk, media_domain.media_id, 1) tasks = task_list + task_list_1 task_list = TaskModel.objects.filter(id__in=tasks).select_related("storage", "storage__episode") video_list = [] common_video_info = {} for task in task_list: episode = task.storage.episode episode_name = self.get_episode_title(media, episode) storage = task.storage common_video_info['pro_type'] = storage.pro_type common_video_info['dub'] = storage.dub common_video_info['content_type'] = storage.content_type param_dict = {"is_charge": 1} is_fee = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict)["is_charge"] _video_horpic = self.image_resize.get_oss_resize_params(self.partner_alias, episode.episode_horpic, size=episode.horpic_size, to_ip=True) video = { "video_id": str(storage.id), "video_name": episode_name, "duration": storage.seconds, "play_order": episode.serial_num, "video_horpic": self.to_ip_url(_video_horpic), "is_fee": is_fee, "is_show": int(self.get_storage_is_show(storage.id, self.lsp_pk)), "definition": self.get_storage_definition(storage.definition, storage.v_codec), "content_type": storage.content_type, 'video_desc': episode.title, 'sub_title': episode.title, "m3u8s": [{"m3u8": [{"ver_id": self.get_storage_definition(storage.definition, storage.v_codec)}]}] } video_list.append(video) return video_list, common_video_info def year_map(self, year): return year def _get_area(self, media): return ";".join(region["name"] for region in media.regions) def _assemble_album(self, media_domain, videos, common): media = media_domain.inst director_obj_list, actor_obj_list, _ = media_domain.get_media_celebrity(media) media_ext = media.mediaextfieldmodel param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) _album_verpic = self.image_resize.get_oss_resize_params(self.partner_alias, media_ext.media_verpic, size=media_ext.verpic_size, to_ip=True) _album_horpic = self.image_resize.get_oss_resize_params(self.partner_alias, media_ext.media_horpic, size=media_ext.horpic_size, to_ip=True) album = { "videos": [{"video": videos}], "album_id": str(media.id), "year": str(self.year_map(media.year)), "album_name": self.get_media_name_by_alias(media), "genre": ";".join(genre["name"] for genre in media.genres), "publish_time": media.date_published.replace('-', ''), "episode_total": media.episodes, "album_desc": media.storyline, "area": self._get_area(media), "language": media.languages, "album_verpic": self.to_ip_url(_album_verpic), "album_horpic": self.to_ip_url(_album_horpic), "is_fee": change_fields["is_charge"], "is_show": 1, "focus": media_ext.focus, "pro_type": common['pro_type'], "age_group": self.get_age_group(media_ext.rating), "actor": ";".join(actor.name for actor in actor_obj_list), "director": ";".join(director.name for director in director_obj_list), "dub": common['dub'], "is_series": 0 if change_fields["category"] in (CategoryValue.MOVIE, CategoryValue.ANIMATION_MOVIE) else 1, "episode_updated": len(videos), "chn_name": CategoryValue.category_map()[change_fields["category"]], "content_type": common['content_type'] } return album # 转码成功回调 def transcode_callback(self, file_url, gitv_task_id: str, message): logger.warning("gdyd_on_trancode_callback with url: {}, task_id: {}".format(file_url, gitv_task_id)) task_cache_key = self.TRANS_CODE_TASK_CACHE.format(gitv_task_id=gitv_task_id) task_id = cache.get(task_cache_key) if not task_id: logger.error("trans code callback error, task_cache_key:{}".format(task_cache_key)) return {"code": "A000001", "message": "失败"} query = TaskModel.objects.filter(pk=task_id) remark = "" if not file_url: res = requests.get("http://api.trans.gitv.tv/query", data={"task_id": gitv_task_id}).json() if res["code"] != "A000000": remark = message else: file_url = res["data"][0]["file_url"] remark = "" if file_url else res["data"][0]["msg"] task = query.first() if file_url: query.update(status=TaskStatusValue.TRANS_CODE_COMPLETED, remark="") file_url_cache_key = self.TRANS_CODE_FILE_URL_CACHE.format(storage_id=task.storage.id) cache.set(file_url_cache_key, file_url, self.EXPIRE_TIME) status = TaskStatusValue.FILE_PENDING query.update(status=status) else: status = TaskStatusValue.TRANS_CODE_FAILED query.update(status=status, remark=remark) TaskQueue.update_cron_task(task.level, task_id, status) return {"code": "A000000", "message": "成功"} # 介质拉取成功回调 def storage_callback(self, request): logger.warning("gdyd_video_callback :{}".format(request.body)) code = request.POST['code'] storgae_id = request.POST['video_id'] message = request.POST['message'] if code == "E000001": status = TaskStatusValue.FILE_FAILED remark = "" else: status = TaskStatusValue.FILE_COMPLETED remark = message query_info = { "storage_id": storgae_id, "partner_id": self.lsp_pk, "status": TaskStatusValue.FILE_TRANSFERING } task = TaskModel.objects.filter(**query_info).first() if task: task.status = status task.remark = remark task.save(update_fields=["status", "remark"]) TaskQueue.update_cron_task(task.level, task.id, status) # 添加注入信息任务 query_info["status"] = TaskStatusValue.FILE_COMPLETED task = TaskModel.objects.filter(**query_info).first() if task: task.status = TaskStatusValue.INFO_PENDING task.save(update_fields=["status"]) TaskQueue.update_cron_task(task.level, task.id, TaskStatusValue.INFO_PENDING) # 目前只处理了银河TV response response_data = { "code": "A000000", "message": "success", "timestamp": date_utils.datetime_now() } return response_data # 分集信息拉取成功回调 def callback(self, request_body): logger.warning("gdyd_info_callback {}".format(request_body)) body = json.loads(request_body) storage_ids = body["videos"] code = body["code"] message = body["message"] # /injector/media/4435?partner=GITV req_url = body["req_url"] # 更新任务状态 if code == "A00000": status = TaskStatusValue.INFO_COMPLETED remark = "" else: status = TaskStatusValue.INFO_FAILED remark = message _storage_id_list = storage_ids.strip(",").split(",") storage_id_list = [i for i in _storage_id_list if i != ""] media_id = re.search(r"\d+", req_url).group() if storage_id_list: query_info = { "storage_id__in": storage_id_list, "partner_id": self.lsp_pk, "status": TaskStatusValue.INFO_TRANSFERING } else: media_id = re.search(r"\d+", req_url).group() query_info = { "storage__media_id": int(media_id), "partner_id": self.lsp_pk, "status": TaskStatusValue.INFO_TRANSFERING } TaskModel.objects.filter(**query_info).update(**{"status": status, "remark": remark}) if storage_id_list: TaskCacheHash.delete(self.lsp_pk, media_id, storage_id_list) self.continue_push(media_id) response_data = {"code": "A00000"} return response_data def continue_push(self, media_id): if TaskCacheHash.get_size(self.lsp_pk, media_id) == 0 and TaskCacheHash.get_size(self.lsp_pk, media_id, 1) > 0: self.call_for_inject_media_info1(media_id) def update(self, task): self.call_for_inject_media_info(task, action="UPDATE") def remove(self, task): self.off_line_storage(task.partner_id, task.storage_id) self.call_for_inject_media_info(task, action="DELETE") class GuangDongMobile(GitvBase): def __init__(self, alias="GITV_GDYD"): super().__init__(alias) self.level = 9 def year_map(self, year): year = int(year) if year < 1980: return "更早" elif year < 1990: return "80年代" elif year < 2000: return "90年代" elif year <= 2010: return "2000-2010" else: return year # 有分集标题时 def _have_episode_title(self, media, episode): fmt = "第{0}集{1}{2}" return fmt.format(episode.serial_num, " ", episode.title) # 无分集标题时 def _nope_episode_title(self, media, episode): fmt = "{0}{1}第{2}集" return fmt.format(self.get_media_name_by_alias(media), " ", episode.serial_num) # 参数在JSON中 def storage_callback(self, request): logger.warning("gdyd_video callback :{}".format(request.body)) jobj = json.loads(request.body) code = jobj['code'] storgae_id = jobj['video_id'] message = jobj['message'] if code == "E000001": status = TaskStatusValue.FILE_FAILED remark = "" else: status = TaskStatusValue.FILE_COMPLETED remark = message query_info = { "storage_id": storgae_id, "partner_id": self.lsp_pk, "status": TaskStatusValue.FILE_TRANSFERING } task = TaskModel.objects.filter(**query_info).first() if task: task.status = status task.remark = remark task.save(update_fields=["status", "remark"]) TaskQueue.update_cron_task(task.level, task.id, status) # 添加注入信息任务 query_info["status"] = TaskStatusValue.FILE_COMPLETED task = TaskModel.objects.filter(**query_info).first() if task: task.status = TaskStatusValue.INFO_PENDING task.save(update_fields=["status"]) TaskQueue.update_cron_task(task.level, task.id, TaskStatusValue.INFO_PENDING) # 目前只处理了银河TV response response_data = { "code": "A000000", "message": "success", "timestamp": date_utils.datetime_now() } return response_data pass