import datetime 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 Gitv(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 self.change_video = False self.start_date = datetime.datetime(year=2025, month=4, day=2) self.only_media = [14643, 14664] self.storage_epsoide_maping = {} self.change_csp_media = [14743, 14742, 14741, 14740, 14739, 14738, 14737, 14736, 14735, 14734, 14733, 14732, 14717, 14718, 14719, 14720, 14721, 14722, 14723, 14724, 14725, 14727, 14728, 14729] self.new_csp = "5690" self.add_prefix_media = [3857, 3855, 15223, 15224, 15225, 15226, 15227, 15228, 15229, 9908, 3085, 15230, 15195, 15231, 15232, 15233, 13448, 15234, 15235, 15199, 1638, 13447, 15236, 15237, 15222, 15238, 13370, 13368, 15239, 13357, 13363, 13457, 15240, 15241, 15242, 13362, 13367, 13364, 15481] self.prefix_text = '（节目来源于“德生文化”）' self.temp_media_episode_title = [17384] 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 gen_desc(self, media): if media.id in self.add_prefix_media: return self.prefix_text + media.storyline return media.storyline # 有分集标题时 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 get_csp_by_media(self, media_id): if media_id in self.change_csp_media: return self.new_csp return self.lsp.csp_id # 向转码服务提交本地文件 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.get_csp_by_media(task.media_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 just_change(self, task): if task.create_time > self.start_date and task.media_id not in self.only_media: self.change_video = True return True return False def gen_video_id(self, task): if self.change_video: if task.replace: return task.replace return "gitv{}".format(task.episode_id) return task.storage_id # 向全局存储提交转码后的文件地址 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.get_csp_by_media(task.media_id), "album_id": storage.media_id, "video_id": self.gen_video_id(task), "bit_id": self.get_storage_definition(storage.definition, storage.v_codec), "file_type": "m3u8", "uri": url } logger.warning("--- 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("gitv 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.get_csp_by_media(media_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("gitv 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) self.storage_epsoide_maping[task.storage_id] = task.episode_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) self.storage_epsoide_maping[task.storage_id] = task.episode_id 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 self.just_change(task) 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("pull_media_id:{} album:{}".format(media_id, album)) return ret def get_episode_title(self, media, episode): if self. media.id in self.temp_media_episode_title: title = "{}第{}集".format(media.name, episode.serial_num) return title # 获取注入时变更的媒体类型 param_dict = {"category": media.category} category = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict)["category"] if category in CategoryValue.movie_category(): # 电影系列 title = self.get_media_name_by_alias(media) elif category in (CategoryValue.TVSERIES, CategoryValue.CHILDREN, CategoryValue.ANIMATION, CategoryValue.EDUCATION, CategoryValue.GAME): # 电视剧、少儿(儿童)、动漫、教育、游戏 if episode.title: title = self._have_episode_title(media, episode) else: title = self._nope_episode_title(media, episode) elif category in (CategoryValue.VARIETY, CategoryValue.SHORT_FILM, CategoryValue.DOCUMENTARY): title = "{} {}期".format(self.get_media_name_by_alias(media), episode.serial_num) else: title = "{} {}".format(self.get_media_name_by_alias(media), episode.title) return title 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", "episode") video_list = [] common_video_info = {} old_task_list = [] now_task_list = [] for task in task_list: if self.just_change(task): now_task_list.append(task) else: old_task_list.append(task) if old_task_list: for task in old_task_list: episode = task.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) _id = self.gen_video_id(task) video = { "video_id": str(task.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, "begin_time": episode.begin, "end_time": episode.end, "m3u8s": [{"m3u8": [{"ver_id": self.get_storage_definition(storage.definition, storage.v_codec)}]}] } video_list.append(video) return video_list, common_video_info if now_task_list: return self.handler_now_task(now_task_list, media) def handler_now_task(self, task_list, media): video_list = [] common_video_info = {} self.change_video = True episode_ids = [] repeat_task = {} # 查询已经注入的本节目，完成的，非删除的节目子集 for task in task_list: if task.episode_id in episode_ids: if task.episode_id not in repeat_task: repeat_task[task.episode_id] = [task] else: repeat_task[task.episode_id].append(task) else: episode_ids.append(task.episode_id) for task in task_list: if task.episode_id not in episode_ids: continue episode = task.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) _id = self.gen_video_id(task) re_task_list = repeat_task.get(task.episode_id, []) re_task_list.append(task) m3u8s_list = [{"ver_id": self.get_storage_definition(task.storage.definition, task.storage.v_codec)} for task in re_task_list] video = { "video_id": str(_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, "begin_time": episode.begin, "end_time": episode.end, "m3u8s": [{"m3u8": m3u8s_list}] } 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 episodes = media.episodes 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), "duration": int(media.total_seconds), "genre": ";".join(genre["name"] for genre in media.genres), "publish_time": media.date_published.replace('-', ''), "episode_total": media.episodes, "album_desc": self.gen_desc(media), "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.get('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.get('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.get('content_type') } return album # 转码成功回调 def transcode_callback(self, file_url, gitv_task_id: str, message): logger.warning("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("gitv video callback :{}".format(request.body)) code = request.POST['code'] storgae_id = request.POST['video_id'] message = request.POST['message'] album_id = request.POST['album_id'] if code == "E000001": status = TaskStatusValue.FILE_FAILED remark = "" else: status = TaskStatusValue.FILE_COMPLETED remark = message task = TaskModel.objects.filter(media_id=album_id, partner_id=1).first() query_info = { "partner_id": self.lsp_pk, "status": TaskStatusValue.FILE_TRANSFERING } self.just_change(task) if self.change_video or "gitv" in storgae_id: query_info["episode_id"] = str(storgae_id).replace("gitv", "").replace("a", "") else: query_info["storage_id"] = storgae_id 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("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"] new_storage_id_list = [] # 更新任务状态 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() task = TaskModel.objects.filter(media_id=media_id, partner_id=self.lsp_pk, status=TaskStatusValue.INFO_TRANSFERING).first() self.just_change(task) if self.change_video: storage_id_list = [str(_id).replace("gitv", "").replace("a", "") for _id in storage_id_list] if storage_id_list: query_info = { "partner_id": self.lsp_pk, "status": TaskStatusValue.INFO_TRANSFERING } if self.change_video: query_info["episode_id__in"] = storage_id_list else: query_info["storage_id__in"] = storage_id_list 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 } if self.change_video: queryset = TaskModel.objects.filter(**query_info).values("storage_id") new_storage_id_list = [item["storage_id"] for item in queryset] TaskModel.objects.filter(**query_info).update(**{"status": status, "remark": remark}) if self.change_video: storage_id_list = new_storage_id_list if storage_id_list: logger.warning("info callback storage_id_list {}".format(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")