import hashlib import re import os import random import logging import requests import xml.etree.ElementTree as ET from django.core.cache import cache from domain.injector.task_queue import TaskQueue from injector.models import TaskModel from injector.models import MediaInfoSyncModel from domain.imdb.episode_domain import Episode from domain.lsp.generated.common_xsd import Adi from domain.lsp.base_injector import LspInjectorC2 from domain.lsp.value_object import C2ApiUtil, CodeFactory from domain.injector.value_object import TaskStatusValue, DefinitionValue, ContentTypeValue, PlatformAliasValue from utils.date_utils import datetime_now from domain.lsp.generated.shanxi_xml import XMLBuilder from utils.defines import ResponseCode import json import traceback logger = logging.getLogger(__file__) class CodeMGFactory: Series = "ps" # 基础媒资信息id类别，供剧头使用 Program = "p" Movie = "mv" # 基础媒资分集id类别 Image = "img" # 基础媒资信息id类别 Video = "video" C2 = "c2" STATUS_DELETE = "2" STATUS_REGISTER = "0" FLAG_SERIES = "00" FLAG_PROGRAM = "01" FLAG_SERIES_PROGRAM = "06" class InjectorFlag: SERIES_FLAG = "00" PROGRAM_FLAG = "01" class DistAction: DELETE = "2" REGISTER = "0" class ImageFlag: VER = "ver" HOR = "hor" VER_34 = "V34" VER_23 = "V23" HOR_169 = "H169" HOR_32 = "H32" class AdiCallback: CODE_SUCCESS = 0 class ActionCode: REGISTER = "0" UPDATE = "1" DELETE = "2" def __init__(self, cp_id): self.cp_id = cp_id def __generate_code__(self, category, some_id): fmt = "{0}_{1}_{2}" code = fmt.format(category, self.cp_id, some_id) return code def __generate_no__(self, task_id, flag, op, platform): code = "{}_{}_{}_{}_{}".format(task_id, flag, op, platform, datetime_now()) return code def __generate_name__(self, task_id, some_id): fmt = "{0}_{1}" name = fmt.format(task_id, some_id) return name def get_flag_by_type(self, _type): if _type == self.Series: return self.FLAG_SERIES if _type in (self.Program, self.Movie): return self.FLAG_PROGRAM else: return self.FLAG_SERIES_PROGRAM def gen_series_code(self, series_id): return self.__generate_code__(self.Series, series_id) def gen_program_code(self, program_id): return self.__generate_code__(self.Program, program_id) def gen_movie_code(self, movie_id): return self.__generate_code__(self.Movie, movie_id) def gen_video_code(self, movie_id): return self.__generate_code__(self.Video, movie_id) def gen_series_img_code(self, series_id, flag="ver"): return self.__generate_code__(self.Image, "{}_{}".format(series_id, flag)) def gen_sub_img_code(self, program_id, flag="ver"): return self.__generate_code__(self.Image, "{}_{}".format(program_id, flag)) def gen_movie_img_code(self, movie, flag="ver"): return self.__generate_code__(self.Image, "{}_{}".format(movie, flag)) def gen_request_no(self, task_id, flag, op, platform): return self.__generate_no__(task_id, flag, op, platform) def gen_request_c_no(self, task_id, flag, op, platform): return self.__generate_no__(task_id, flag, op, platform + "c") def gen_xml_name(self, task_id, flag): return self.__generate_name__(task_id, flag) class NXCallBack: def callback_adi(self, body): logger.warning("NXCallBack : {}".format(body)) try: root = ET.fromstring(body) ser_no = root.get("serialNo") code = root.get("code") msg = root.get("msg") ottmsp = root.get("ottmspFlag") op = root.find(".//Asset").get("op") type = root.find(".//Asset").get("type") id = root.find(".//Asset").get("ID") logger.warning( "adi serialNo:{} code:{} msg:{} ottmspFlag：{} op:{} type:{} ID:{}".format(ser_no, code, msg, ottmsp, op, type, id)) task_id, flag, op, platform, _ = ser_no.split("_") if flag == CodeMGFactory.InjectorFlag.SERIES_FLAG: status, desc = self.series_callback_adi(task_id, code, ottmsp, op, id, flag, body) elif flag == CodeMGFactory.InjectorFlag.PROGRAM_FLAG: status, desc = self.program_callback_adi(task_id, code, ottmsp, op, id, flag, body) except Exception as e: logger.error("adi 执行分发error:{}".format(e)) traceback.print_exc() status, desc = -1, "未携带xml报文体,xml解析失败." result = "{}|{}".format(status, desc) return result def series_callback_adi(self, sync_id, code, ottmsp, op, asset_id, flag, body): if int(code) == 0: remark = "adi" status = MediaInfoSyncModel.SYNCING if int(ottmsp) == 2: sync = MediaInfoSyncModel.objects.filter(id=sync_id).first() status = MediaInfoSyncModel.PENDING2 sync.status = status sync.save() logger.warning( "{} 剧头 回调成功: header_id: {}, body: {}".format(self.partner_alias, sync_id, body)) self.series_header_injector(sync) elif int(code) in (1104, 1105) and int(op) == 2: sync = MediaInfoSyncModel.objects.filter(id=sync_id).first() status = MediaInfoSyncModel.PENDING2 sync.status = status sync.save() logger.warning( "{} 剧头 回调成功: header_id: {}, body: {}".format(self.partner_alias, sync_id, body)) self.series_header_injector(sync) else: status = MediaInfoSyncModel.FAILED remark = "失败" logger.error( "{} 剧头 回调失败: header_id: {}, body: {}".format(self.partner_alias, sync_id, body)) MediaInfoSyncModel.objects.filter(pk=sync_id).update(status=status, remark=remark) code = ResponseCode.SUCCESS.value desc = ResponseCode.SUCCESS.desc return code, desc def program_callback_adi(self, task_id, code, ottmsp, op, asset_id, flag, body): if int(code) == 0: remark = "adi" status = TaskStatusValue.FILE_TRANSFERING if int(ottmsp) == 2: # 无视频更新 或者 删除 if int(op) == 2: # 删除 task = TaskModel.objects.filter(id=task_id).first() status = TaskStatusValue.INFO_PENDING self.update_task_info_o(task, status, "del adi") self.remove_injector(task) elif int(code) in (1104, 1105) and int(op) == 2: # 不存在，已经下线 兼容旧数据，没有上传到咪咕的情况 task = TaskModel.objects.filter(id=task_id).first() status = TaskStatusValue.INFO_PENDING remark = "del adi no" task.status = status self.remove_injector(task) else: status = TaskStatusValue.FILE_FAILED remark = "adi失败" logger.error( "{} 剧头|子集|电影 回调失败: task_id: {}, body: {}".format(self.partner_alias, task_id, body)) self.update_task_info(task_id, status, remark) code = ResponseCode.SUCCESS.value desc = ResponseCode.SUCCESS.desc return code, desc def callback_cdi(self, body): logger.warning("callback_cdi: {}".format(body)) try: tree = ET.fromstring(body) attrib = tree.find(".//header").attrib code = attrib.get("code") msg = attrib.get("msg") sequence = attrib.get("sequence") componentid = attrib.get("component-id") serialNo = attrib.get("serialNo") action = attrib.get("action") asset_id = tree.find(".//task").attrib.get("original-content-id") logger.warning( "code:{} msg:{} sequence：{} componentid:{} action:{} serialNo:{}".format(code, msg, sequence, componentid, action, serialNo)) task_id, flag, op, platform, _ = serialNo.split("_") if int(code) == 0: logging.warning("cdi成功") remark = "cdi成功" status = TaskStatusValue.FILE_TRANSFERING # 准备分发 if op != "2": _code, _msg = self.distribute_task(task_id, asset_id, op, flag) remark = _msg else: status = TaskStatusValue.FILE_FAILED remark = "cdi失败" logger.error( "{} 子集|电影 回调失败: task_id: {}".format(self.partner_alias, task_id)) task = TaskModel.objects.filter(pk=task_id).first() self.update_task_info_o(task, status, remark) status = 1 desc = "OK" except Exception as e: logger.info("cdi 执行分发error:{}".format(e)) status, desc = 0, "fail" resp = XMLBuilder.build_cdi_response(sequence, status, desc) logger.info("cdi response status:{} desc:{} response:{}".format(status, desc, resp)) return resp def callback_strategy(self, body): # 这里要区分 是分发的回执 json_obj = json.loads(body) requestNo = json_obj.get("requestNo") code = json_obj.get("code") resultMsg = json_obj.get("resultMsg") assetId = json_obj.get("assetId") playUrl = json_obj.get("playUrl") resultMsg = resultMsg if resultMsg else "分发失败" logger.info("distribute callback:{}".format(json_obj)) task_id, flag, op, _, _ = requestNo.split("_") # 缓存地址 task_cache_key = self.PLAY_URL_KEY.format(task_id) cache.set(task_cache_key, playUrl, self.EXPIRE_TIME) if flag == "01": if int(op) == 0: if code == "SUCC": status = TaskStatusValue.INFO_PENDING else: status = TaskStatusValue.FILE_FAILED # 注入分发完成 task = TaskModel.objects.filter(pk=task_id).first() self.update_task_info_o(task, status, resultMsg) TaskQueue.update(task.level, task_id, status) # 准备未来注入 else: self.update_task_info(task_id, None, resultMsg) if int(op) == 2: if code == "SUCC": status = TaskStatusValue.FILE_PENDING else: status = TaskStatusValue.FILE_FAILED task = TaskModel.objects.filter(id=task_id).first() task.status = status task.save() self.remove_injector_mg(task) elif flag == self.code_mg_factory.FLAG_SERIES_PROGRAM: # pass # if code == "SUCC" and int(op) == 2: # op = self.code_mg_factory.STATUS_REGISTER # _flag = self.code_mg_factory.get_flag_by_type(assetId.split("_")[0]) # logger.info("删除成功，然后新增.op:{} flag:{}".format(op, _flag)) # self.distribute_task(task_id, assetId, op, flag) # if _flag == self.code_mg_factory.FLAG_SERIES: # MediaInfoSyncModel.objects.filter(id=task_id).update(remark=resultMsg) # elif _flag == self.code_mg_factory.FLAG_PROGRAM: # self.update_task_info(task_id, None, resultMsg) return {"code": 0, "msg": "success"} class NingXiaYiDong(LspInjectorC2, NXCallBack): def __init__(self, alias="NINGXIAYIDONG"): super().__init__(alias) self.lsp_id = self.conf.lsp_id self.csp_id = self.conf.csp_id self.cms_api = self.conf.cms_api self.cast_sep = "|" self.genre_sep = "|" self.mg_cps_code = "senyu2" # 这个找LSP确认 self.mg_cp_id = "699413" self.code_factory = CodeFactory(self.csp_id) self.code_mg_factory = CodeMGFactory(self.mg_cps_code) self.key = "4c2af42001f911ebb053691f1adea374" self.strategyId = "306" self.int_to_action = {"1": "register", "2": "update", "0": "delete"} self.action_to_op_map = {"register": "0", "update": "1", "delete": "2"} self.action_if6_mapping = {"0": "1", "2": "4"} self.mg_plat_no = PlatformAliasValue.NINGXIAYIDONG self.PLAY_URL_KEY = "mg_url_{}" self.EXPIRE_TIME = 3600 * 72 def get_media_effective_time(self, media_id): effective_time = Episode.get_media_effective_time(media_id, self.lsp_pk) start_date = effective_time["start_date"] end_date = effective_time["end_date"] if not (start_date and end_date): return "", "" start_time = "{} 00:00:00".format(start_date) end_time = "{} 23:59:59".format(end_date) return start_time, end_time def _get_bit_rate_type(self, definition): if definition == DefinitionValue.P320: return "LD" elif definition == DefinitionValue.P480: return "SD" elif definition == DefinitionValue.P720: return "HD" elif definition == DefinitionValue.P1080: return "HD" elif definition == DefinitionValue.K4: return "HD" return "SD" def _get_bit_rate_type_mg(self, definition): if definition == DefinitionValue.P720: return "591" elif definition == DefinitionValue.P1080: return "111302" def get_content_type(self, content_type): return { ContentTypeValue.POSITIVE: "正片", ContentTypeValue.CLIP: "花絮", ContentTypeValue.TRAILER: "预告片" }.get(content_type, "正片") def get_content_type_mg(self, content_type): if content_type == ContentTypeValue.POSITIVE: return 1 elif content_type == ContentTypeValue.CLIP: return 4 elif content_type == ContentTypeValue.TRAILER: return 3 else: return 7 def get_stbld_nums(self): the_num = TaskModel.objects.filter(partner_id=self.lsp_pk).count() if len(str(the_num)) <= 5: return str(the_num).rjust(5, "0") elif len(str(the_num)) <= 7: return str(the_num).rjust(7, "0") else: return "{}{}".format(random.randint(10, 99), str(the_num)[-5:]) def _have_episode_title(self, media, episode): fmt = "{0}{1}{2}" return fmt.format(episode.title, "_", episode.serial_num) def _nope_episode_title(self, media, episode): fmt = "{0}{1}{2}" media_name = self.transform_media_name(self.get_media_name_by_alias(media)) return fmt.format(media_name, "_", episode.serial_num) @staticmethod def transform_media_name(name): rule = re.compile("(第[一二三四五六七八九十]+?[集部系季篇])") digit_map = {"一": 1, "二": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9, "十": 10} result = name.replace(" ", "") find_list = re.findall(rule, name) for find in find_list: map_num = 0 join_list = list() for char in find: value = digit_map.get(char, None) if value: map_num += value else: join_list.append(char) join_str = "".join([join_list[0], str(map_num), join_list[-1]]) result = result.replace(find, join_str) return result def get_action_to_op(self, action): return self.action_to_op_map[str(action)] def get_int_to_action(self, int_action): return self.int_to_action[str(int_action)] def property_series(self, _id, volumn_count, media, action="REGIST"): the_property = [ {"Code": _id}, {"Name": self.transform_media_name(self.get_media_name_by_alias(media))}, {"Pinyin": ""}, {"Language": media.languages}, {"Description": media.storyline}, {"ProgramType2": self.get_genre(media)}, {"EnName": ""}, {"ReleaseYear": media.year}, {"VolumnCount": volumn_count}, {"OrgAirDate": ""}, {"Tag": ""}, {"OriginalCountry": self.get_regions(media)}, {"Score": media.douban_score}, {"Duration": int(media.total_seconds / 60) + 1}, {"ContentType": "Video"}, {"CopyRightPlatform": ""}, {"CopyRightArea": ""}, {"CopyRightCode": ""}, {"CopyRightIntegrity": ""}, {"CopyRightWay": ""}, {"CopyRightChain": ""}, {"CopyRightFile": ""}, {"CopyRightType": ""}, {"CopyRightCpName": ""}, {"IfPreview": 0}, ] return { "element_type": "Series", "id": _id, "code": _id, "action": action, "property": the_property } def property_program(self, _id, name, media, action="REGIST"): return { "element_type": "Program", "id": _id, "code": _id, "action": action, "property": [ {"Code": _id}, {"Name": name}, {"Description": ""}, {"ProgramType2": self.get_genre(media)}, {"IfPreview": 0} ] } def property_movie(self, _id, media_storage, _type=1, action="REGIST"): return { "element_type": "Movie", "id": _id, "code": _id, "action": action, "property": [ {"Code": _id}, {"BitRateId": int(media_storage.bit_rate)}, {"Definition": self._get_bit_rate_type(media_storage.definition)}, {"CdnPlatform": 18}, {"Dispatch": 0}, {"IsTranscode": 0} ] } def get_url_from_cache(self, task_id): task_cache_key = self.PLAY_URL_KEY.format(task_id) _url = cache.get(task_cache_key) url = "" play_url = "" if _url: play_url = _url.split("&stbId=")[0] url = _url.replace("&", "&") play_url = play_url.replace("&", "&") return url, play_url def _album_movie(self, task_id, media_storage, media, episode, media_ext, action): actors, directors, writers = self.get_media_person(media, actor=4, director=4, writer=2) vst, vet = self.get_media_effective_time(media.id) media_verpic = self.image_resize.get_oss_resize_params(self.partner_alias, media_ext.media_verpic, size=media_ext.verpic_size, to_ip=True) media_horpic = self.image_resize.get_oss_resize_params(self.partner_alias, media_ext.media_horpic, size=media_ext.horpic_size, to_ip=True) definition = self._get_bit_rate_type(media_storage.definition) param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) psid = self.code_factory.gen_series_code(media.id) p_series = self.property_series(psid, media.episodes, media, action=action) p_series["property"].extend([ {"ProgramType": self._get_category_desc(change_fields["category"])}, {"ActorDisplay": directors}, {"WriterDisplay": actors}, {"scriptWriter": writers}, {"Definition": definition}, {"IsCharge": str(change_fields["is_charge"])}, {"PictureURL1": self.to_ip_url(media_verpic)}, {"HorizontalPoster": self.to_ip_url(media_horpic)}, {"CopyRightBeginDate": vst}, {"CopyRightEndDate": vet} ]) ppid = self.code_factory.gen_program_code(media_storage.id) media_name = self.transform_media_name(self.get_media_name_by_alias(media)) p_program = self.property_program(ppid, media_name, media, action) p_program["property"].extend([ {"Definition": definition}, {"Duration": int(media_storage.seconds / 60) + 1}, {"VolumnCount": episode.serial_num}, {"isCharge": str(change_fields["is_charge"])}, {"ProgramType": self._get_category_desc(change_fields["category"])}, {"OriginalCountry": self.get_regions(media)}, {"videoType": self.get_content_type(media_storage.content_type)} ]) mapping_program = self.property_mapping("Series", "Program", psid, ppid, sequence=episode.serial_num, action=action) pmid = self.code_factory.gen_movie_code(media_storage.id) p_movie = self.property_movie(pmid, media_storage, action=action) url, play_url = self.get_url_from_cache(task_id) p_movie["property"].extend([{"PlayURL": play_url}]) p_movie["property"].extend([{"FileURL": url}]) mapping_movie = self.property_mapping("Program", "Movie", ppid, pmid, sequence=episode.serial_num, action=action) return { "object_list": [p_series, p_program, p_movie], "mapping_list": [mapping_program, mapping_movie] } def _album_series_header(self, media, media_ext, info_sync): action = info_sync.action_description() actors, directors, writers = self.get_media_person(media, actor=4, director=4, writer=2) vst, vet = self.get_media_effective_time(media.id) media_verpic = self.image_resize.get_oss_resize_params(self.partner_alias, media_ext.media_verpic, size=media_ext.verpic_size, to_ip=True) media_horpic = self.image_resize.get_oss_resize_params(self.partner_alias, media_ext.media_horpic, size=media_ext.horpic_size, to_ip=True) # 宁夏方要求剧头注入时 要填写清晰度，默认为高清 definition = "HD" param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) psid = self.code_factory.gen_series_code(media.id) p_series = self.property_series(psid, media.episodes, media, action=action) p_series["property"].extend([ {"ProgramType": self._get_category_desc(change_fields["category"])}, {"ActorDisplay": directors}, {"WriterDisplay": actors}, {"scriptWriter": writers}, {"Definition": definition}, {"IsCharge": str(change_fields["is_charge"])}, {"PictureURL1": self.to_ip_url(media_verpic)}, {"HorizontalPoster": self.to_ip_url(media_horpic)}, {"CopyRightBeginDate": vst}, {"CopyRightEndDate": vet} ]) return { "object_list": [p_series], "mapping_list": [] } def _album_sub_series(self, task_id, media_storage, media, episode, _episodes, action="REGIST"): psid = self.code_factory.gen_series_code(media.id) definition = self._get_bit_rate_type(media_storage.definition) episode_title = self.get_episode_title(media, episode) param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) ppid = self.code_factory.gen_program_code(media_storage.id) p_program = self.property_program(ppid, episode_title, media, action) p_program["property"].extend([ {"Definition": definition}, {"Duration": int(media_storage.seconds / 60) + 1}, {"VolumnCount": episode.serial_num}, {"ProgramType": self._get_category_desc(change_fields["category"])}, {"isCharge": str(change_fields["is_charge"])}, {"OriginalCountry": self.get_regions(media)}, {"videoType": self.get_content_type(media_storage.content_type)} ]) mapping_program = self.property_mapping("Series", "Program", psid, ppid, sequence=episode.serial_num, action=action) pmid = self.code_factory.gen_movie_code(media_storage.id) p_movie = self.property_movie(pmid, media_storage, action=action) url, play_url = self.get_url_from_cache(task_id) p_movie["property"].extend([{"PlayURL": play_url}]) p_movie["property"].extend([{"FileURL": url}]) mapping_movie = self.property_mapping("Program", "Movie", ppid, pmid, sequence=episode.serial_num, action=action) return { "object_list": [p_program, p_movie], "mapping_list": [mapping_program, mapping_movie], } def album(self, _header_id, action="REGIST"): header_id, _ = _header_id.split("_", 1) info_sync = MediaInfoSyncModel.objects.filter(pk=header_id).select_related("media").first() media = info_sync.media media_ext = media.mediaextfieldmodel data = self._album_series_header(media, media_ext, info_sync) xml = Adi.to_xml(data, flag="Nope") return xml def program(self, _task_id, action="REGIST"): task_id, _ = _task_id.split("_", 1) task = TaskModel.objects.filter(pk=task_id).select_related("storage").first() media_storage = task.storage # 媒介 media = media_storage.media # 媒体文件主体 episode = media_storage.episode # (分集)扩展信息 media_ext = media.mediaextfieldmodel # 媒体文件补充信息 _episodes = int(media.episodes) if media.episodes else 0 if _episodes > 1: data = self._album_sub_series(task_id, media_storage, media, episode, _episodes, action) else: # 单剧集 data = self._album_movie(task_id, media_storage, media, episode, media_ext, action) xml = Adi.to_xml(data, flag="Nope") return xml def exec_cmd(self, task_id): _task_id = self.generate_correlate_id(task_id) cmdfileUrl = self.get_cmd_ftp_url(self.csp_id, _task_id) # ftp协议 xml = C2ApiUtil.REQUESTXML.format(self.csp_id, self.lsp_id, "{}_{}".format(_task_id, datetime_now()), cmdfileUrl) logger.warning("EXEC_CMD Request: {}".format(xml)) headers = {"content-type": "text/xml", "SOAPAction": "application/soap+xml;charset=utf8"} response = requests.post(self.cms_api, data=xml, headers=headers).content logger.warning("EXEC_CMD Response: {}".format(response)) tree = ET.fromstring(response) code = tree.find(".//Result").text desc = tree.find(".//ErrorDescription").text if int(code) == 0: status = TaskStatusValue.INFO_TRANSFERING remark = "未来注入" else: status = TaskStatusValue.INFO_FAILED remark = desc TaskModel.objects.filter(pk=task_id).update(status=status, remark=remark) return status def save_album_xml_mg_file(self, xml, cmd_id): dir_path = self.get_xml_file_dir_path() if not os.path.exists(dir_path): os.makedirs(dir_path, exist_ok=True) file_path = self.get_xml_file_path(self.csp_id, cmd_id) with open(file_path, "w", encoding="utf-8") as f: f.write(str(xml, encoding="utf-8")) def exec_sync_header_cmd(self, sync): _sync_id = self.generate_correlate_id(sync.id, stage="series") cmd_file_url = self.get_cmd_ftp_url(self.csp_id, _sync_id) sync_id = "{}_{}".format(_sync_id, datetime_now()) xml = C2ApiUtil.REQUESTXML.format(self.csp_id, self.lsp_id, sync_id, cmd_file_url) logger.warning("{} exec_sync_header_cmd HEADER_ID: {}, Request: {}".format(self.lsp_id, sync.id, xml)) headers = {"content-type": "text/xml", "SOAPAction": "application/soap+xml;charset=utf8"} response = requests.post(self.cms_api, data=xml, headers=headers).content logger.warning("{} exec_sync_header_cmd HEADER_ID: {}, Response: {}".format(self.lsp_id, sync.id, response)) tree = ET.fromstring(response) code = tree.find(".//Result").text if int(code) == 0: status = MediaInfoSyncModel.SYNCING2 else: status = MediaInfoSyncModel.FAILED sync.status = status sync.save() return def exec_sync_header_mg_cmd(self, sync): action = self.get_int_to_action(sync.action) xml = self.series_xml(sync, action) headers = {"Content-Type": "text/xml; charset=utf-8"} xml_code = self.code_mg_factory.gen_xml_name(sync.id, CodeMGFactory.InjectorFlag.SERIES_FLAG) self.save_album_xml_mg_file(xml, xml_code) logger.warning("ningxiayidong_migu exec_sync_header_cmd HEADER_ID: {} Request".format(sync.id)) response = requests.post(self.conf.transcode_api, data=xml, headers=headers).content result_str = str(response, encoding="utf-8") logger.warning( "ningxiayidong_migu exec_sync_header_cmd HEADER_ID: {}, Response: {}".format(sync.id, result_str)) code, msg = result_str.split("|") if code == "0": status = MediaInfoSyncModel.SYNCING else: status = MediaInfoSyncModel.FAILED sync.status = status sync.remark = msg sync.save() return sync # the_sync = MediaInfoSyncModel.objects.filter(pk=sync.id).first() # if the_sync.status == MediaInfoSyncModel.PENDING: # 这里在判断 是为了防止同时有多人操作 # the_sync.status = status # the_sync.remark = msg # the_sync.save() def series_xml(self, sync, action, **kwargs): img_is_change = "0" op = self.get_action_to_op(action) media = sync.media serial_no = self.code_mg_factory.gen_request_no(sync.id, CodeMGFactory.InjectorFlag.SERIES_FLAG, op, self.mg_plat_no) media_ext = media.mediaextfieldmodel actors, directors, writers = self.get_media_person(media, actor=4, director=4, writer=2) vst, vet = self.get_media_effective_time(media.id) param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) v34_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.VER_34) v23_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.VER_23) v169_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.HOR_169) v32_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.HOR_32) data = { "asset_id": self.code_mg_factory.gen_series_code(media.id), "serial_no": serial_no, "country_origin": self.get_regions(media), "actors": actors, "directors": directors, "type": self._get_category_desc(change_fields["category"]), "type2": self.get_genre(media), "show_type": "Series", "vst": vst, "vet": vet, "focus": media_ext.focus, "publish_no": media_ext.approval_num, "img_list": [ self.parse_image(media_ext.media_verpic, CodeMGFactory.ImageFlag.VER_34, v34_id, img_is_change), self.parse_image(media_ext.media_verpic, CodeMGFactory.ImageFlag.VER_23, v23_id, img_is_change), self.parse_image(media_ext.media_horpic, CodeMGFactory.ImageFlag.HOR_169, v169_id, img_is_change), self.parse_image(media_ext.media_horpic, CodeMGFactory.ImageFlag.HOR_32, v32_id, img_is_change)], } logger.info("xml series data:{}".format(data)) xml = XMLBuilder.build_series(self.mg_cp_id, media, action, **data) return xml def get_image_pix_tuple(self, flag): pix_tuple = None if flag == "V34": pix_tuple = 300, 400, 4 elif flag == "V23": pix_tuple = 400, 600, 2 elif flag == "H169": pix_tuple = 1920, 1080, 1 elif flag == "H32": pix_tuple = 600, 400, 3 return pix_tuple def parse_image(self, url, flag, asset_id, is_change): filename = url.split("/")[-1] name, profile = filename.split(".") filename = "{}_{}_sc.{}".format(name, flag, profile) url = self.to_ip_url(url) url = "{}{}".format(url.split(name)[0], filename) h_pix, v_pix, _type = self.get_image_pix_tuple(flag) return {"filename": filename, "profile": profile, "h_pixs": str(h_pix), "v_pixs": str(v_pix), "len_size": "", "url": url, "_type": str(_type), "assetID": asset_id, "is_change": str(is_change)} def sign_md5(self, key_list): m = hashlib.md5() for item in key_list: m.update(item.encode()) return m.hexdigest() def action_to_number(self, action): if action.lower() == "update": return 2 elif action.lower() == "delete": return 4 else: return 1 def distribute_task(self, task_id, asset_id, op, flag): logger.warning( "start distribute_task task_id:{} asset_id:{} op:{} flag:{}".format(task_id, asset_id, op, flag)) try: request_no = self.code_mg_factory.gen_request_c_no(task_id, flag, op, self.mg_plat_no) sign = self.sign_md5((asset_id, self.mg_cp_id, request_no, self.key)) op = self.action_to_number(self.action_if6_mapping[op]) params = { "requestNo": request_no, "CPID": self.mg_cp_id, "assetId": asset_id, "IF6ID": self.strategyId, "typeOfOp": op, "medias": "", "sign": sign } logger.warning("params:{}".format(params)) r = requests.post(self.conf.storage_api, json=params) logger.warning("distribute response:{}".format(r.content)) json_obj = r.json() code = json_obj.get("code") msg = json_obj.get("msg") except Exception as e: logger.warning("distribute 分发出错:{}".format(e)) code, msg = -1, "执行分发失败" return code, msg def injector(self, task): status, flag = 0, 0 if task.status == TaskStatusValue.FILE_PENDING: status = self.exec_cdp_cmd(task) elif task.status == TaskStatusValue.INFO_PENDING: _task_id = self.generate_correlate_id(task.id) self.save_album_xml_file(self.csp_id, _task_id, param="program") status = self.exec_cmd(task.id) flag = self.get_continue_to_injector(status) return status, flag def get_continue_to_injector(self, status): return status not in TaskStatusValue.stop_injector() def save_album_xml_cdp_file(self, xml, cmd_id): dir_path = self.get_xml_file_dir_path() if not os.path.exists(dir_path): os.makedirs(dir_path, exist_ok=True) file_path = self.get_xml_file_path(self.csp_id, cmd_id) with open(file_path, "w", encoding="utf-8") as f: f.write(str(xml, encoding="utf-8")) def parser_video_info(self, storage, video_id, is_change): url = storage.url url = self.change_storage_url(url) filename = url.split("/")[-1] return {"fileName": filename, "fileSize": str(storage.size), "assetID": video_id, "url": url, "duration": str(storage.seconds), "bit_rate": self._get_bit_rate_type_mg(storage.definition), "usage": str(self.get_content_type_mg(storage.content_type)), "is_change": str(is_change)} def movie_cdp_xml(self, task, action, is_other=False, **kwargs): media = task.media media_ext = media.mediaextfieldmodel episode = task.episode storage = task.storage actors, directors, writers = self.get_media_person(media, actor=4, director=4, writer=2) vst, vet = self.get_media_effective_time(media.id) param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) episode_title = self.get_episode_title(media, episode) op = self.action_to_op_map[str(action).lower()] if op == "0" or op == "2": v_is_change = "0" else: v_is_change = "1" img_is_change = "0" serial_no = self.code_mg_factory.gen_request_no(task.id, CodeMGFactory.InjectorFlag.PROGRAM_FLAG, op, self.mg_plat_no) asset_id = self.code_mg_factory.gen_movie_code(storage.id) parent_id = self.code_mg_factory.gen_series_code(media.id) video_id = self.code_mg_factory.gen_video_code(storage.id) v34_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.VER_34) v23_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.VER_23) v169_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.HOR_169) v32_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.HOR_32) data = { "asset_id": asset_id, "serial_no": serial_no, "parent_id": parent_id, "country_origin": self.get_regions(media), "actors": actors, "directors": directors, "type": self._get_category_desc(change_fields["category"]), "type2": self.get_genre(media), "show_type": "Movie", "vst": vst, "vet": vet, "focus": media_ext.focus, "publish_no": media_ext.approval_num, "episode": {"title": episode_title, "serial_num": str(episode.serial_num)}, "video": self.parser_video_info(storage, video_id, v_is_change), "img_list": [self.parse_image(media_ext.media_verpic, "V34", v34_id, img_is_change), self.parse_image(media_ext.media_verpic, "V23", v23_id, img_is_change), self.parse_image(media_ext.media_horpic, "H169", v169_id, img_is_change), self.parse_image(media_ext.media_horpic, "H32", v32_id, img_is_change)] } logger.info("xml movie data:{}".format(data)) xml = XMLBuilder.build_movie(self.mg_cp_id, media, action, **data) return xml def get_episode_title_mg(self, media, episode): if episode.title: fmt = "第{0}集{1}{2}" return fmt.format(episode.serial_num, " ", episode.title) else: fmt = "{0}{1}第{2}集" return fmt.format(self.get_media_name_by_alias(media), " ", episode.serial_num) def sub_series_cdp_xml(self, task, action, is_other=False, **kwargs): media = task.media media_ext = media.mediaextfieldmodel episode = task.episode storage = task.storage actors, directors, writers = self.get_media_person(media, actor=4, director=4, writer=2) vst, vet = self.get_media_effective_time(media.id) param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) episode_title = self.get_episode_title_mg(media, episode) op = self.action_to_op_map[action.lower()] if op == "0" or op == "2": v_is_change = "0" else: v_is_change = "1" img_is_change = "0" serial_no = self.code_mg_factory.gen_request_no(task.id, CodeMGFactory.InjectorFlag.PROGRAM_FLAG, op, self.mg_plat_no) asset_id = self.code_mg_factory.gen_program_code(storage.id) parent_id = self.code_mg_factory.gen_series_code(media.id) video_id = self.code_mg_factory.gen_video_code(storage.id) v32_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.HOR_32) v34_id = self.code_mg_factory.gen_series_img_code(media.id, CodeMGFactory.ImageFlag.VER_34) data = { "asset_id": asset_id, "serial_no": serial_no, "parent_id": parent_id, "country_origin": self.get_regions(media), "actors": actors, "directors": directors, "type": self._get_category_desc(change_fields["category"]), "type2": self.get_genre(media), "show_type": "SubSeries", "vst": vst, "vet": vet, "focus": media_ext.focus, "publish_no": media_ext.approval_num, "episode": {"title": episode_title, "serial_num": str(episode.serial_num)}, "video": self.parser_video_info(storage, video_id, v_is_change), "img_list": [ self.parse_image(media_ext.media_horpic, "H32", v32_id, img_is_change) , self.parse_image(media_ext.media_verpic, "V34", v34_id, img_is_change)], } logger.info("xml sub_series data:{}".format(data)) xml = XMLBuilder.build_sub_series(self.mg_cp_id, media, action, **data) return xml def update_task_info(self, task_id, status=None, remark=None): task = TaskModel.objects.filter(pk=task_id).first() if status: task.status = status if remark: task.remark = remark task.save() def update_task_info_o(self, task, status=None, remark=None): if status: task.status = status if remark: task.remark = remark task.save() def exec_cdp_cmd(self, task, action="register"): # 判断 是进行子集注入 还是 单片注入 if task.media.episodes == 1: xml = self.movie_cdp_xml(task, action) else: xml = self.sub_series_cdp_xml(task, action) self.save_album_xml_cdp_file(xml, "mg_{}_{}".format(task.id, "01")) logger.warning("ningxiaydong exec task task_id: {}, Request:".format(task.id)) headers = {"Content-Type": "text/xml; charset=utf-8"} response = requests.post(self.conf.transcode_api, data=xml, headers=headers).content result_str = str(response, encoding="utf-8") logger.warning("ningxiaydong exec task task_id: {}, Response: {}".format(task.id, result_str)) code, msg = result_str.split("|") if code == "0": status = TaskStatusValue.FILE_TRANSFERING remark = "ok" else: status = TaskStatusValue.FILE_FAILED remark = msg self.update_task_info(task.id, status, remark) return status def update(self, task): if task.status in (TaskStatusValue.FILE_PENDING, TaskStatusValue.INFO_PENDING): _task_id = self.generate_correlate_id(task.id) self.save_album_xml_file(self.csp_id, _task_id, param="program", action="UPDATE") self.exec_cmd(task.id) def remove(self, task): # 1、先从分发删除 2、在删除咪咕注入 最后是 未来 # 先删分发 在删注入 if task.media.episodes == 1: asset_id = self.code_mg_factory.gen_movie_code(task.storage_id) else: asset_id = self.code_mg_factory.gen_program_code(task.storage_id) op = "2" flag = "01" _code, _msg = self.distribute_task(task.id, asset_id, op, flag) if int(_code) in (4, 15): # 没有分发，那就直接删除注入 logging.warning("没有分发直接删除咪咕注入") _status = TaskStatusValue.FILE_PENDING self.update_task_info_o(task, _status, _msg) self.remove_injector_mg(task) else: self.update_task_info_o(task, None, _msg) def remove_injector_mg(self, task): if task.status in (TaskStatusValue.FILE_PENDING,): self.exec_cdp_cmd(task, action="delete") def remove_injector(self, task): if task.status in (TaskStatusValue.INFO_PENDING,): _task_id = self.generate_correlate_id(task.id) self.save_album_xml_file(self.csp_id, _task_id, param="program", action="DELETE") self.exec_cmd(task.id) self.off_line_storage(task.partner_id, task.storage_id) logger.warning("{} 下线媒体: {}, task_id: {}".format(self.partner_alias, task.storage_id, task.id)) def series_header_injector(self, sync): # 先发送咪咕剧头工单 if sync.status == MediaInfoSyncModel.PENDING: logging.warning("剧头咪咕注入") sync = self.exec_sync_header_mg_cmd(sync) # 在发送未来工单 if sync.status == MediaInfoSyncModel.PENDING2: logging.warning("剧头未来注入") header_id = self.generate_correlate_id(sync.id, stage="series") self.save_album_xml_file(self.csp_id, header_id, param="series") self.exec_sync_header_cmd(sync) return def series_header_update(self, sync): self.series_header_injector(sync) def series_header_remove(self, sync): # 先删除咪咕注入剧头 self.series_header_injector(sync)