import re from domain.imdb.media_domain import Media from domain.imdb.value_object import CategoryValue from domain.injector.media_storage_domain import MediaStorage from domain.injector.task_queue import TaskQueue from domain.lsp.base_injector import LspInjectorG from domain.injector.value_object import TaskStatusValue, TaskModelActionValue import requests import logging import json from django.core.cache import cache from injector.models import TaskModel, PartnerMediaStorageModel logger = logging.getLogger("django") class NmgYdGt(LspInjectorG): 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="NMGYDGT"): super().__init__(alias) self.key = "v0j#efkgFdFCn" self.salt = "gitv@!_tv_party" self.level = 10 self.genre_map = {"文艺": "文学艺术", "相声": "相声小品"} 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 = "（来源于德生）" 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 injector(self, task): status = 0 if task.status == TaskStatusValue.FILE_PENDING: # 执行介质注入 status = self.exec_storage_injector(task) self.just_sub_count(status, task) elif task.status == TaskStatusValue.INFO_PENDING: # 执行媒资注入 status = self.exec_media_injector(task) flag = self.get_continue_to_injector(status) return status, flag def just_sub_count(self, status, task): if status in [0, TaskStatusValue.FILE_FAILED, TaskStatusValue.FILE_COMPLETED]: count = TaskQueue.get_count(task.level) logger.warning("nm_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 exec_storage_injector(self, task): if task.level == self.level: count = TaskQueue.get_count(task.level) # logger.warning("trans_count code task_id:{}, count:{}".format(task.id, count)) if count: count = int(count) else: count = 0 if count < 19: TaskQueue.add_1(self.level) # logger.warning("trans_count<19") else: # logger.warning("trans_count>=19") return task.status # 把介质uri 发送给银河 银河拉取 logger.info("request injector storage") storage = task.storage data = {"cp_id": self.conf.csp_id, "album_id": task.media_id, "video_id": task.storage_id, "bit_id": self.get_storage_definition(storage.definition, storage.v_codec), "uri": storage.hls_url, "file_type": "m3u8" } logger.info(data) resp = requests.post(self.conf.storage_api, json=data) logger.info("response injector storage:{}".format(resp.json())) code = resp.json().get("code") message = resp.json().get("message") if code == "A000000": # 表示成功 status = TaskStatusValue.FILE_TRANSFERING else: status = TaskStatusValue.FILE_FAILED task.status = status task.remark = message task.save() return status def exec_media_injector(self, task): # 分为两步 1、通知gitv增量 2、提供单专辑接口供调用 request_body = {"data": [{"album_id": str(task.media_id), "is_show": str(1)}]} request_body_str = json.dumps(request_body, separators=(",", ":")) print(request_body_str) logger.info("request injector info") url = "{}?cpId={}&t={}".format(self.conf.cms_api, self.conf.csp_id, self.make_sign(request_body_str)) print(url) print(request_body) resp = requests.post(url, json=request_body) logger.info("response injector info:{}".format(resp.json())) code = resp.json()["code"] message = resp.json()["message"] if code != "A00000": status = TaskStatusValue.INFO_FAILED else: status = TaskStatusValue.INFO_TRANSFERING TaskModel.objects.filter(pk=task.id).update(status=status, remark=message) return status def gen_m3u8_uri(self, task): # 创建一个m3u8文件，然后在文件中描述地址url return None def storage_callback(self, request): logger.info("nmgydgt_storage_callback storage :{}".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() # 结束了介质注入，继续激发媒资信息注入通知 if task.status == TaskStatusValue.FILE_COMPLETED: task.status = TaskStatusValue.INFO_PENDING task.save() TaskQueue.update_cron_task(task.level, task.id, task.status) # 缓存 当前操作的storage_id # key = "{}_{}".format(task.partner.alias, task.media_id) # resp = cache.get(key) # cache.set(storgae_id) self.sub_count(task) return {"code": "A000000", "message": "success", "timestamp": "20180905120000" } def album(self, media_id, action): if not media_id: # "epg 信息未按标准提供" return "E00002", {} media_domain = Media(media_id) videos, common = self._assemble_videos(media_domain) print(videos) print(common) album = self._assemble_album(media_domain, videos, common) ret = { "album": album, "code": "A00000" } logger.warning("nmgydgt_pull_media_id:{} album: {}".format(media_id, album)) return ret def _assemble_videos(self, media_domain, limit=50): 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] 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": self.get_storage_is_show_action(storage.id, self.lsp_pk, task.action), "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 _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) is_show = 1 # 判断is_show 是1还是0 如果是删除操作，且删除的是最后一个子集 则需要联通媒资信息也删除 # 判断是否有video有 is_show # is_delete = False # for video in videos: # if video["is_show"]: # is_delete = True # if self.is_movie(media): # is_show = 0 # else: # storage_id_list = TaskModel.objects.values("storage_id").filter(media_id=media.id, partner_id=self.lsp_pk) # count = PartnerMediaStorageModel.objects.filter(storage_id__in=storage_id_list).distinct().count() # if len(storage_id_list) > count: # is_show = 1 # else: # is_show = 0 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": self._get_genre(media), "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": is_show, "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 callback(self, request_body): logger.info("nmgydgt media_info callback {}".format(request_body)) body = json.loads(request_body) storage_ids = body["videos"] code = body["code"] message = body["message"] 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] 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}) response_data = {"code": "A00000"} return response_data 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) def _get_area(self, media): names = [] for region in media.regions: if region["name"] in ("中国大陆", "内地"): names.append("中国") else: names.append(region["name"]) return ";".join(name for name in names) def _get_genre(self, media): genres = [] for genre in media.genres: if genre["name"] in self.genre_map: genres.append(self.genre_map[genre["name"]]) else: genres.append(genre["name"]) return ";".join(genres) def update(self, task): self.exec_media_injector(task) self.update_action(task.id) def remove(self, task): self.off_line_storage(task.partner_id, task.storage_id) self.exec_media_injector(task) def active_task(self, task, *args): if task and task.is_offline() and task.is_status_completed(): # 修改删除状态 PartnerMediaStorageModel.objects.filter(storage_id=task.storage_id, partner_id=task.partner_id).update( is_show=1) # 重新注入 task.status = TaskStatusValue.INFO_PENDING task.action = TaskModelActionValue.ACTIVE task.save() self.exec_media_injector(task) return True def un_active_task(self, task, *args): if task and task.is_online() and task.is_status_completed(): # 重新注入 task.status = TaskStatusValue.INFO_PENDING task.action = TaskModelActionValue.UN_ACTIVE task.save() self.exec_media_injector(task) return True