import hashlib import logging import os import json import requests import xml.etree.ElementTree as ET from domain.imdb.episode_domain import Episode from domain.lsp.generated.shanxi_xml import XMLBuilder from domain.lsp.value_object import CodeFactory from injector.models import MediaInfoSyncModel, TaskModel from utils.defines import ResponseCode from .base_injector import LspInjectorC2 from domain.injector.value_object import TaskStatusValue, DefinitionValue, ContentTypeValue from utils.date_utils import datetime_now logger = logging.getLogger("django") # 定义 山西移动code class CodeFactory: 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" 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 __init__(self, csp_code=""): self.csp_code = csp_code def gen_series_asset_id(self, series_id): flag_0 = (32 - len(str(series_id)) - len(self.Series) - len(self.csp_code) - 2) * "0" return "{}_{}{}_{}".format(self.Series, self.csp_code, flag_0, series_id) def gen_series_img_asset_id(self, series_id, flag="ver"): flag_0 = (32 - len(str(series_id)) - len(self.Series) - len(self.Image) - len(flag) - len( self.csp_code) - 5) * "0" return "{}_{}_{}_{}_{}_{}".format(self.Series, self.csp_code, self.Image, flag, flag_0, series_id) def gen_sub_asset_id(self, series_id, movie_id): flag_0 = (32 - len(str(series_id)) - len(str(movie_id)) - len(self.Program) - len(self.csp_code) - 3) * "0" return "{}_{}{}_{}_{}".format(self.Program, self.csp_code, flag_0, series_id, movie_id) def gen_sub_img_asset_id(self, series_id, movie_id, flag="ver"): flag_0 = (32 - len(str(series_id)) - len(self.Program) - len(self.Image) - len(flag) - len(self.csp_code) - len( str(movie_id)) - 6) * "0" return "{}_{}_{}_{}_{}_{}_{}".format(self.Program, self.csp_code, self.Image, flag, flag_0, series_id, movie_id) def gen_sub_video_asset_id(self, series_id, movie_id): flag_0 = (32 - len(str(series_id)) - len(self.Program) - len(self.Video) - len(self.csp_code) - len( str(movie_id)) - 5) * "0" return "{}_{}_{}_{}_{}_{}".format(self.Program, self.csp_code, self.Video, flag_0, series_id, movie_id) def gen_movie_asset_id(self, series_id, movie_id): flag_0 = (32 - len(str(series_id)) - len(str(movie_id)) - len(self.Movie) - len(self.csp_code) - 3) * "0" return "{}_{}{}_{}_{}".format(self.Movie, self.csp_code, flag_0, series_id, movie_id) def gen_movie_video_asset_id(self, series_id, movie_id): flag_0 = (32 - len(str(series_id)) - len(self.Movie) - len(self.Video) - len(self.csp_code) - len( str(movie_id)) - 5) * "0" return "{}_{}_{}_{}_{}_{}".format(self.Movie, self.csp_code, self.Video, flag_0, series_id, movie_id) def gen_movie_img_asset_id(self, series_id, movie_id, flag="ver"): flag_0 = (32 - len(str(series_id)) - len(self.Movie) - len(self.Image) - len(self.csp_code) - len( str(movie_id)) - len(flag) - 6) * "0" return "{}_{}_{}_{}_{}_{}_{}".format(self.Movie, self.csp_code, self.Image, flag, flag_0, series_id, movie_id) def gen_serialNo(self, task_id, flag, op, date): return "{}_{}_{}_{}".format(task_id, flag, op, date) def gen_serial_c2_no(self, task_id, flag, op, date): # 06 删除更新 return "{}_{}_{}_{}{}".format(task_id, flag, op, self.C2, date) # 定义山西移动的类 class ShanXiYiDong(LspInjectorC2): def __init__(self, alias="SHANXIYIDONG"): 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.csp_name = self.conf.csp_name self.cast_sep = "|" self.region_sep = "|" self.genre_sep = "," self.tag_sep = "," self.cps_code = "senyu2" # 这个找LSP确认 self.cp_id = "699413" self.code_factory = CodeFactory(self.cps_code) self.key = "4c2af42001f911ebb053691f1adea374" self.strategyId = "65" self.content_type_mapping = {"正片": 1, "宣传片": 3, "片花": 4, "花絮": 5, "其他": 7} self.action_mapping = {"0": "register", "1": "update", "2": "delete"} self.action_mapping1 = {"register": "0", "update": "1", "delete": "2"} def get_content_type(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_bit_rate_type(self, definition): if definition == DefinitionValue.P720: return "591" elif definition == DefinitionValue.P1080: return "111302" elif definition in (DefinitionValue.P3840, DefinitionValue.P4096, DefinitionValue.K4): return "121401" def _get_video_type(self, type): type_mapper = {"h264": "H.264", "hevc": "H.265", "mpeg4": "MPEG4", "mpeg2": "MPEG2", "avs": "AVS", "avs2": "AVS2"} return self.type_num_mapper[type_mapper[type]] def _get_resolution(self, definition, width): def_mapper = {DefinitionValue.P320: "2/3 D1", DefinitionValue.P480: "D1", DefinitionValue.P720: "720P", DefinitionValue.P1080: "1080P", DefinitionValue.K4: "4K(3840*2160)" if width == 3840 else "4K(4096*2160)"} return self.resolution_num_mapper[def_mapper[definition]] # 目前处理是取第一个国家 def get_regions(self, media_obj): regions = media_obj.regions if regions and len(regions): names = [region["name"] for region in regions] # 目前个别平台 需要改成内地，先分开处理,平台统一后在 统一处理 if names[0] == "中国大陆": return "内地" return names[0] return "" 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) if flag == "V34": h_pixs = 300 v_pixs = 400 _type = 4 elif flag == "V23": h_pixs = 400 v_pixs = 600 _type = 2 elif flag == "H169": h_pixs = 1920 v_pixs = 1080 _type = 1 elif flag == "H32": h_pixs = 600 v_pixs = 400 _type = 3 return {"filename": filename, "profile": profile, "h_pixs": str(h_pixs), "v_pixs": str(v_pixs), "len_size": "", "url": url, "_type": str(_type), "assetID": asset_id, "is_change": str(is_change)} 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(storage.definition), "usage": str(self.get_content_type(storage.content_type)), "is_change": str(is_change)} 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 "", "" return start_date, end_date 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: status = MediaInfoSyncModel.FINISH if int(op) == 0: _code, _msg = self.distribute_task(sync_id, asset_id, op, flag) remark = _msg elif int(op) == 1: # 先删除 在注入 op = self.code_factory.STATUS_DELETE _flag = self.code_factory.FLAG_SERIES_PROGRAM logger.info("更新，发起删除操作 op:{} flag:{}".format(op, _flag)) _code, _msg = self.distribute_task(sync_id, asset_id, op, _flag) if int(_code) in (4, 15): # 增添分发 op = self.code_factory.STATUS_REGISTER logger.info("删除失败，直接新增.op:{} flag:{}".format(op, flag)) self.distribute_task(sync_id, asset_id, op, flag) 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 series_callback_adi_other(self, sync_id, code, ottmsp, op, asset_id, flag): if int(code) == 0: if int(ottmsp) == 2: if int(op) == 0 or int(op) == 1: self.distribute_task(sync_id, asset_id, op, flag) 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: # 无视频更新 或者 删除 status = TaskStatusValue.INFO_COMPLETED if int(op) == 0: _code, _msg = self.distribute_task(task_id, asset_id, op, flag) remark = _msg elif int(op) == 1: # 先删除 在注入 op = self.code_factory.STATUS_DELETE flag = self.code_factory.FLAG_SERIES_PROGRAM logger.info("更新，发起删除操作 op:{} flag:{}".format(op, flag)) self.distribute_task(task_id, asset_id, op, flag) 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 program_callback_cdi(self, task_id, code): if int(code) == 0: remark = "cdi" status = TaskStatusValue.INFO_COMPLETED else: status = TaskStatusValue.INFO_FAILED remark = "cdi失败" logger.error( "{} 子集|电影 回调失败: task_id: {}".format(self.partner_alias, task_id)) self.update_task_info(task_id, status, remark) code = ResponseCode.SUCCESS.value desc = ResponseCode.SUCCESS.desc return code, desc def program_callback_cdi_other(self): code = ResponseCode.SUCCESS.value desc = ResponseCode.SUCCESS.desc return code, desc def callback_adi(self, body): logger.warning("callback_adi : {}".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.info( "adi serialNo:{} code:{} msg:{} ottmspFlag：{} op:{} type:{} ID:{}".format(ser_no, code, msg, ottmsp, op, type, id)) task_id, flag, op, _ = ser_no.split("_") if flag == "00": status, desc = self.series_callback_adi(task_id, code, ottmsp, op, id, flag, body) elif flag == "01": status, desc = self.program_callback_adi(task_id, code, ottmsp, op, id, flag, body) elif flag == "02": status, desc = self.series_callback_adi_other(task_id, code, ottmsp, op, id, flag, body) except Exception as e: logger.error("adi 执行分发error:{}".format(e)) status, desc = -1, "未携带xml报文体,xml解析失败." result = "{}|{}".format(status, desc) return result 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.info( "code:{} msg:{} sequence：{} componentid:{} action:{} serialNo:{}".format(code, msg, sequence, componentid, action, serialNo)) task_id, flag, op, _ = serialNo.split("_") if flag == "01": self.program_callback_cdi(task_id, code) elif flag == "03": self.program_callback_cdi_other() if int(code) == 0: # 准备分发 if op != "2": _code, _msg = self.distribute_task(task_id, asset_id, op, flag) self.update_task_info(task_id, None, _msg) status = 1 desc = "OK" except Exception as e: logger.info("cdi 执行分发error:{}".format(e)) status, desc = 0, "fail" return status, desc def parse_requestNo(self, requestNo): args = requestNo.split("_") if len(args) == 4: return args[0], args[1], args[2], args[3] elif len(args) == 5: return args[0], args[1], args[2], args[3] def callback(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") distrId = json_obj.get("distrId", "") resultMsg = resultMsg if resultMsg else "分发失败" logger.info("distribute_callback:{}".format(json_obj)) task_id, flag, op, _ = self.parse_requestNo(requestNo) remark = distrId + resultMsg if flag == "00": MediaInfoSyncModel.objects.filter(id=task_id).update(remark=remark) if code == "SUCC" and int(op) == 2: sync = MediaInfoSyncModel.objects.filter(id=task_id).first() self.remove_header_injector(sync) elif flag == "01": self.update_task_info(task_id, None, remark) if code == "SUCC" and int(op) == 2: task = TaskModel.objects.filter(id=task_id).first() self.remove_injector(task) elif flag == self.code_factory.FLAG_SERIES_PROGRAM: # 删除的添加 if code == "SUCC" and int(op) == 2: op = self.code_factory.STATUS_REGISTER _flag = self.code_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_factory.FLAG_SERIES: MediaInfoSyncModel.objects.filter(id=task_id).update(remark=resultMsg) elif _flag == self.code_factory.FLAG_PROGRAM: self.update_task_info(task_id, None, resultMsg) return {"code": 0, "msg": "success"} def movie_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_mapping1[action] if op == "0" or op == "2": v_is_change = "0" else: v_is_change = "1" img_is_change = "0" if is_other: serial_no = self.code_factory.gen_serialNo(task.id, "03", op, datetime_now()) asset_id = kwargs.get("asset_id") parent_id = kwargs.get("parent_id") video_id = kwargs.get("video_id") v34_id = kwargs.get("v34") v23_id = kwargs.get("v23") v169_id = kwargs.get("v169") v32_id = kwargs.get("v32") else: serial_no = self.code_factory.gen_serialNo(task.id, "01", op, datetime_now()) asset_id = self.code_factory.gen_movie_asset_id(media.id, storage.id) parent_id = self.code_factory.gen_series_asset_id(media.id) video_id = self.code_factory.gen_movie_video_asset_id(media.id, storage.id) v34_id = self.code_factory.gen_movie_img_asset_id(media.id, storage.id, "V34") v23_id = self.code_factory.gen_movie_img_asset_id(media.id, storage.id, "V23") v169_id = self.code_factory.gen_movie_img_asset_id(media.id, storage.id, "H169") v32_id = self.code_factory.gen_movie_img_asset_id(media.id, storage.id, "H32") 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.cp_id, media, action, **data) return xml def movie_xml_del(self, media, op, **kwargs): serial_no = self.code_factory.gen_serialNo("123456", "03", op, datetime_now()) asset_id = kwargs.get("asset_id") parent_id = kwargs.get("parent_id") video_id = kwargs.get("video_id") v34_id = kwargs.get("v34") v23_id = kwargs.get("v23") v169_id = kwargs.get("v169") v32_id = kwargs.get("v32") data = { "asset_id": asset_id, "serial_no": serial_no, "parent_id": parent_id, "country_origin": "", "actors": "", "directors": "", "type": "", "type2": "", "show_type": "Movie", "vst": "", "vet": "", "focus": "", "publish_no": "", "episode": {"title": "", "serial_num": ""}, "video": {"assetID": video_id}, "img_list": [{"assetID": v34_id}, {"assetID": v23_id}, {"assetID": v169_id}, {"assetID": v32_id}] } xml = XMLBuilder.build_movie(self.cp_id, media, "delete", **data) return xml def sub_series_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_mapping1[action] if op == "0" or op == "2": v_is_change = "0" else: v_is_change = "1" img_is_change = "0" if is_other: serial_no = self.code_factory.gen_serialNo(task.id, "03", op, datetime_now()) asset_id = kwargs.get("asset_id") parent_id = kwargs.get("parent_id") video_id = kwargs.get("video_id") v34_id = kwargs.get("v34") v32_id = kwargs.get("v32") else: serial_no = self.code_factory.gen_serialNo(task.id, "01", op, datetime_now()) asset_id = self.code_factory.gen_sub_asset_id(media.id, storage.id) parent_id = self.code_factory.gen_series_asset_id(media.id) video_id = self.code_factory.gen_sub_video_asset_id(media.id, storage.id) v32_id = self.code_factory.gen_sub_img_asset_id(media.id, storage.id, "H32") v34_id = self.code_factory.gen_sub_img_asset_id(media.id, storage.id, "V34") 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": [] "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.cp_id, media, action, **data) return xml def series_xml(self, sync, action, is_other=False, **kwargs): op = self.action_mapping1[action] # # if op == "0" or op == "2": # img_is_change = "0" # else: # img_is_change = "1" img_is_change = "0" if is_other: media = sync serial_no = self.code_factory.gen_serialNo(sync.id, "02", op, datetime_now()) else: media = sync.media serial_no = self.code_factory.gen_serialNo(sync.id, "00", op, datetime_now()) 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) if is_other: v34_id = kwargs.get("v34") v23_id = kwargs.get("v23") v169_id = kwargs.get("v169") v32_id = kwargs.get("v32") else: v34_id = self.code_factory.gen_series_img_asset_id(media.id, "V34") v23_id = self.code_factory.gen_series_img_asset_id(media.id, "V23") v169_id = self.code_factory.gen_series_img_asset_id(media.id, "H169") v32_id = self.code_factory.gen_series_img_asset_id(media.id, "H32") data = { "asset_id": self.code_factory.gen_series_asset_id(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, "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 series data:{}".format(data)) xml = XMLBuilder.build_series(self.cp_id, media, action, **data) return xml def save_album_xml_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")) # 剧头注入 向LSP发起xml内容 def exec_sync_header_cmd(self, sync, action): xml = self.series_xml(sync, action) headers = {"Content-Type": "text/xml; charset=utf-8"} self.save_album_xml_file(xml, "{}_{}".format(sync.id, "00")) logger.warning("exec_sync_header_cmd HEADER_ID: {}, Request: {}".format(sync.id, xml.decode())) response = requests.post(self.cms_api, data=xml, headers=headers).content result_str = str(response, encoding="utf-8") logger.warning( "shanxiyidong 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 the_sync = MediaInfoSyncModel.objects.filter(pk=sync.id) if the_sync.first().status == MediaInfoSyncModel.PENDING: the_sync.update(status=status, remark=msg) return # 剧头注入处理 def series_header_injector(self, sync, action="register"): if sync.status == MediaInfoSyncModel.PENDING: self.exec_sync_header_cmd(sync, action) return # 介质注入 向LSP发起xml内容 def exec_cmd(self, task, action): # 判断 是进行子集注入 还是 单片注入 if task.media.episodes == 1: xml = self.movie_xml(task, action) else: xml = self.sub_series_xml(task, action) self.save_album_xml_file(xml, "{}_{}".format(task.id, "01")) logger.warning("shanxiyidong exec task task_id: {}, Request:".format(task.id)) headers = {"Content-Type": "text/xml; charset=utf-8"} response = requests.post(self.cms_api, data=xml, headers=headers).content result_str = str(response, encoding="utf-8") logger.warning("shanxiyidong exec task task_id: {}, Response: {}".format(task.id, result_str)) code, msg = result_str.split("|") if code == "0": status = TaskStatusValue.FILE_TRANSFERING remark = "" else: status = TaskStatusValue.FILE_FAILED remark = msg self.update_task_info(task.id, status, remark) return status # 介质注入 def injector(self, task): status, flag = 0, 0 # 如果工单状态 在 pending 在就绪状态，我们就继续 if task.status in (TaskStatusValue.FILE_PENDING, TaskStatusValue.INFO_PENDING): status = self.exec_cmd(task, action="register") flag = self.get_continue_to_injector(status) return status, flag # 介质注入更新 def update(self, task): if task.status in (TaskStatusValue.FILE_PENDING, TaskStatusValue.INFO_PENDING): self.exec_cmd(task, action="update") # 介质注入删除 def remove(self, task): # 先删分发 在删注入 if task.media.episodes == 1: assert_id = self.code_factory.gen_movie_asset_id(task.media_id, task.storage_id) else: assert_id = self.code_factory.gen_sub_asset_id(task.media_id, task.storage_id) op = "2" flag = "01" _code, _msg = self.distribute_task(task.id, assert_id, op, flag) self.update_task_info(task.id, None, _msg) if int(_code) in (4, 15): # 没有分发，那就直接删除注入 self.remove_injector(task) def remove_injector(self, task): if task.status in (TaskStatusValue.FILE_PENDING, TaskStatusValue.INFO_PENDING): self.exec_cmd(task, action="delete") 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_update(self, sync): self.series_header_injector(sync, "update") def remove_header_injector(self, sync): self.series_header_injector(sync, "delete") # 删除剧头 def series_header_remove(self, sync): # 先删分发 在删注入 assert_id = self.code_factory.gen_series_asset_id(sync.media_id) op = "2" flag = "00" _code, _msg = self.distribute_task(sync.id, assert_id, op, flag) if int(_code) in (4, 15): # 没有分发，那就直接删除注入 self.remove_header_injector(sync) MediaInfoSyncModel.objects.filter(pk=sync.id).update(remark=_msg) 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 exec_distribute(self, task, action): media = task.media if media.episodes == 1: assert_id = self.code_factory.gen_movie_asset_id(media.id, task.storage_id) else: assert_id = self.code_factory.gen_sub_asset_id(media.id, task.storage_id) op = self.action_to_number(action) flag = "01" code, msg = self.distribute_task(task.id, assert_id, str(op), flag) if int(code) == 0: remark = "distribute" status = TaskStatusValue.INFO_TRANSFERING else: remark = msg status = TaskStatusValue.INFO_FAILED self.update_task_info(task.id, None, remark) return status 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 distribute_task(self, task_id, asset_id, op, flag): logger.info( "start distribute_task task_id:{} asset_id:{} op:{} flag:{}".format(task_id, asset_id, op, flag)) # r = TaskModel.objects.filter(pk=task_id).values("partner").first() partner = self.lsp_pk copy_right = 2 if partner and partner == 41: self.strategyId = "393" if partner and partner == 46: self.strategyId = "1080" copy_right = 1 try: request_no = self.code_factory.gen_serial_c2_no(task_id, flag, op, datetime_now()) sign = self.sign_md5((asset_id, self.cp_id, request_no, self.key)) op = self.action_to_number(self.action_mapping[op]) params = { "requestNo": request_no, "CPID": self.cp_id, "assetId": asset_id, "strategyId": self.strategyId, "typeOfOp": 1, "copyRight": copy_right, "priority": 1, "sign": sign, "op": op } logger.info("params:{}".format(params)) r = requests.post(self.conf.scatter_api, params=params) logger.info("distribute response:{}".format(r.content)) json_obj = r.json() code = json_obj.get("code") msg = json_obj.get("msg") except Exception as e: logger.info("distribute 分发出错:{}".format(e)) code, msg = -1, "执行分发失败" return code, msg def exec_header_other(self, media, action): xml = self.series_xml_other(media, action) headers = {"Content-Type": "text/xml; charset=utf-8"} logger.warning("other exec_sync_header_cmd Request: {}".format(xml)) response = requests.post(self.cms_api, data=xml, headers=headers).content logger.warning("other exec_sync_header_cmd Response: {}".format(response)) return