import os import logging import requests import datetime import xml.etree.ElementTree as ET from injector.models import TaskModel from injector.models import MediaInfoSyncModel from domain.injector.value_object import TaskStatusValue, ContentTypeValue, DefinitionValue from domain.lsp.base_injector import LspInjectorC2 from domain.lsp.value_object import CodeFactory, C2ApiUtil from domain.lsp.generated.hebeiyidong_xsd import Adi from utils.defines import ResponseCode from utils.date_utils import datetime_now logger = logging.getLogger(__file__) class HeBeiYiDong(LspInjectorC2): def __init__(self, alias="HEBEIYIDONG"): 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.code_factory = CodeFactory(self.csp_id) self.code_sep = "-" self.genre_sep = '，' @staticmethod def bit_rate_transform(bit_rate): if bit_rate >= 2147483: return 2147482 if bit_rate <= 0: return 1 return bit_rate # program节目清晰度 @staticmethod def _get_definition(definition_value): if definition_value == DefinitionValue.P320: return 0 elif definition_value == DefinitionValue.P720: return 1 elif definition_value == DefinitionValue.P1080: return 2 elif definition_value == DefinitionValue.K4: return 3 return 0 # 分辨率类型 @staticmethod def _get_resolution(definition_value): if definition_value in (DefinitionValue.P320, DefinitionValue.P480): return 5 elif definition_value == DefinitionValue.P720: return 6 elif definition_value == DefinitionValue.P1080: return 8 elif definition_value == DefinitionValue.K4: return 10 return 5 # 河北转换code def transform_base_code(self, code): resolution = code.split(self.code_sep)[::-1] return "".join(resolution) def _get_tags(self, media): tmp1 = ";".join(i["name"] for i in media.genres) if media.genres else "剧情" param_dict = {"category": media.category} category = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict)["category"] tmp2 = self._get_category_desc(category) tmp3 = "国产" if len(media.regions) == 1 and media.regions[0]["name"] == "中国" else "" return ";".join((tmp1, tmp2, tmp3)).strip(";") def property_series(self, sid, media, media_ext, action): 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) param_dict = {"category": media.category} category = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict)["category"] return { "property": [ {"Name": self.get_media_name_by_alias(media)}, {"SortName": self.get_media_name_by_alias(media)}, {"OrgAirDate": media.date_published.replace("-", "")}, {"Price": 0}, {"VolumnCount": media.episodes}, {"Status": 1}, {"Description": media.storyline}, {"Type": self._get_category_desc(category)}, {"Keywords": self._get_tags(media)}, {"Tags": self._get_tags(media)}, {"Rating": media.douban_score}, {"ContentProvider": self.csp_name}, {"CustContentType": 1}, # 河北剧头内容形态对应未存储，默认都是正片 {"ReleaseYear": media.year}, {"OriginalCountry": self.get_regions(media)}, ], "extend_info_list": [ {"field_id": "AlbumVerpic", "field_value": self.to_ip_url(media_verpic)}, {"field_id": "AlbumHorpic", "field_value": self.to_ip_url(media_horpic)}, ], "element_type": "Series", "id": sid, "code": sid, "action": action } def property_program(self, pid, media, media_ext, media_storage, action="REGIST"): param_dict = {"category": media.category} category = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict)["category"] return { "property": [ {"OriginalCountry": self.get_regions(media)}, {"Language": media.languages}, {"OrgAirDate": media.date_published.replace("-", "")}, {"ReleaseYear": media.year}, {"Status": 1}, {"Type": self._get_category_desc(category)}, {"DefinitionFlag": self._get_definition(media_storage.definition)}, {"ViewPoint": media_ext.focus}, {"CustContentType": ContentTypeValue.trans_hebeiyidong(media_storage.content_type)}, {"Rating": media.douban_score}, {"ContentProvider": self.csp_name}, {"PgmCategory": self._get_category_desc(category)}, ], "element_type": "Program", "id": pid, "code": pid, "action": action } def property_movie(self, mid, media_storage, action="REGIST"): return { "property": [ {"Type": ContentTypeValue.trans(media_storage.content_type)}, {"FileURL": self.get_storage_ftp_url(media_storage.url)}, {"AudioType": self.get_audio_type(media_storage.a_codec)}, {"Duration": "{0:06d}".format( int(str(datetime.timedelta(seconds=media_storage.seconds)).replace(":", "")))}, {"BitRateType": self._get_bit_rate_type(media_storage.definition)}, {"BitRate": self.bit_rate_transform(int(media_storage.bit_rate / 1024))}, {"Resolution": self._get_resolution(media_storage.definition)}, ], "element_type": "Movie", "id": mid, "code": mid, "action": action } def property_picture(self, _id, file_url, _type=1, action="REGIST"): return { "element_type": "Picture", "id": _id, "code": _id, "action": action, "property": [ {"Type": _type}, {"FileURL": self.to_ip_url(file_url)}, ] } def _album_movie(self, media_storage, media, episode, media_ext, action="REGIST"): tags = self._get_tags(media) genres = self.get_genre(media) if media.genres else "剧情" search_name = self.transform_chinese_to_letter(self.get_media_name_by_alias(media)) vst, vet = self.get_episode_effective_time(media.id, episode.serial_num) actors, directors, writers = self.get_media_person(media, actor=5, director=5, writer=5) _sid = self.code_factory.gen_series_code(media.id) sid = self.transform_base_code(_sid) series = self.property_series(sid, media, media_ext, action) series["property"].extend([ {"LicensingWindowStart": vst}, {"LicensingWindowEnd": vet}, {"SearchName": search_name}, {"Director": directors}, {"ScriptWriter": writers}, {"Genre": genres} ]) series["extend_info_list"].append({"field_id": "ActorDisplay", "field_value": actors}) _pid = self.code_factory.gen_program_code(media_storage.id) pid = self.transform_base_code(_pid) program = self.property_program(pid, media, media_ext, media_storage, action=action) mapping_program = self.property_mapping("Series", "Program", sid, pid, sequence=episode.serial_num, action=action) program["property"].extend([ {"Name": self.get_media_name_by_alias(media)}, {"SortName": self.get_media_name_by_alias(media)}, {"OriginalName": self.get_media_name_by_alias(media)}, {"Tags": tags}, {"Keywords": tags}, {"Description": media.storyline}, {"SearchName": self.transform_chinese_to_letter(self.get_media_name_by_alias(media))}, {"ScriptWriter": writers}, {"Genre": genres}, {"ActorDisplay": actors}, {"WriterDisplay": writers}, {"LicensingWindowStart": vst}, {"LicensingWindowEnd": vet}, {"PgmSndClass": genres}, {"SeriesFlag": 1}, ]) 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) program["extend_info_list"] = [ {"field_id": "Duration", "field_value": media_storage.seconds}, {"field_id": "TvVerpic", "field_value": self.to_ip_url(media_verpic)}, {"field_id": "TvHorpic", "field_value": self.to_ip_url(media_horpic)} ] _mid = self.code_factory.gen_movie_code(media_storage.id) mid = self.transform_base_code(_mid) movie = self.property_movie(mid, media_storage, action=action) mapping_movie = self.property_mapping("Program", "Movie", pid, mid, action=action) return { "object_list": [series, program, movie], "mapping_list": [mapping_program, mapping_movie], } def _album_sub_series(self, media_storage, media, episode, media_ext, action="REGIST"): tags = self._get_tags(media) title = self.get_episode_title(media, episode) search_name = self.transform_chinese_to_letter(self.get_media_name_by_alias(media)) genres = self.get_genre(media) if media.genres else "剧情" vst, vet = self.get_episode_effective_time(media.id, episode.serial_num) actors, directors, writers = self.get_media_person(media, actor=5, director=5, writer=5) _pid = self.code_factory.gen_program_code(media_storage.id) pid = self.transform_base_code(_pid) _sid = self.code_factory.gen_series_code(media.id) sid = self.transform_base_code(_sid) series = self.property_series(sid, media, media_ext, action) series["property"].extend([ {"LicensingWindowStart": vst}, {"LicensingWindowEnd": vet}, {"SearchName": search_name}, {"Director": directors}, {"ScriptWriter": writers}, {"Genre": genres} ]) series["extend_info_list"].append({"field_id": "ActorDisplay", "field_value": actors}) program = self.property_program(pid, media, media_ext, media_storage, action=action) program["property"].extend([ {"Name": title}, {"SortName": title}, {"OriginalName": title}, {"Tags": tags}, {"Keywords": tags}, {"Description": episode.storyline}, {"SearchName": self.transform_chinese_to_letter(search_name)}, {"ScriptWriter": writers}, {"Genre": genres}, {"ActorDisplay": actors}, {"WriterDisplay": writers}, {"LicensingWindowStart": vst}, {"LicensingWindowEnd": vet}, {"SeriesFlag": 1}, {"PgmSndClass": genres}, ]) episode_horpic = self.image_resize.get_oss_resize_params(self.partner_alias, episode.episode_horpic, size=episode.horpic_size, to_ip=True) program["extend_info_list"] = [ {"field_id": "Duration", "field_value": media_storage.seconds}, {"field_id": "TvHorpic", "field_value": self.to_ip_url(episode_horpic)} ] _mid = self.code_factory.gen_movie_code(media_storage.id) mid = self.transform_base_code(_mid) movie = self.property_movie(mid, media_storage, action=action) mapping_movie = self.property_mapping("Program", "Movie", pid, mid, action=action) _sid = self.code_factory.gen_series_code(media.id) sid = self.transform_base_code(_sid) mapping_program = self.property_mapping("Series", "Program", sid, pid, sequence=episode.serial_num, action=action) return { "object_list": [series, program, movie], "mapping_list": [mapping_program, mapping_movie] } def _album_series_header(self, media, media_ext, action="REGIST"): vst, vet = self.get_media_effective_time(media.id) search_name = self.transform_chinese_to_letter(self.get_media_name_by_alias(media)) genre = self.get_genre(media) if media.genres else "剧情" actors, directors, writers = self.get_media_person(media, actor=5, director=5, writer=5) _sid = self.code_factory.gen_series_code(media.id) sid = self.transform_base_code(_sid) series = self.property_series(sid, media, media_ext, action) series["property"].extend([ {"Genre": genre}, {"LicensingWindowStart": vst}, {"LicensingWindowEnd": vet}, {"SearchName": search_name}, {"Director": directors}, {"ScriptWriter": writers} ]) series["extend_info_list"].append({"field_id": "ActorDisplay", "field_value": actors}) return { "object_list": [series, ], "mapping_list": [] } def delete_olddata_of_three_parties(self, sid=None, pid=None, mid=None, flag="series"): """ 河北媒资注入情况，有之前注入过的数据在之前的第三方平台上，需要对过期数据进行删除操作，然后在运营平台重新进行注入。 :return: """ if flag == "series": data = { "object_list": [{ "element_type": "Series", "id": sid, "code": sid, "action": "DELETE" }], "mapping_list": [] } else: data = { "object_list": [{ "element_type": "Program", "id": pid, "code": pid, "action": "DELETE" }, { "element_type": "Movie", "id": mid, "code": mid, "action": "DELETE" }], "mapping_list": [{ "parent_type": "Series", "parent_id": sid, "element_type": "Program", "element_id": pid, "parent_code": sid, "element_code": pid, "action": "DELETE" }, { "parent_type": "Program", "parent_id": pid, "element_type": "Movie", "element_id": mid, "parent_code": pid, "element_code": mid, "action": "DELETE" }] } return data def album(self, _header_id, series_code=None, is_delete=0): if is_delete: data = self.delete_olddata_of_three_parties(sid=series_code) else: 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 action = info_sync.action_description() data = self._album_series_header(media, media_ext, action=action) xml = Adi.to_xml(data, flag="Nope") return xml def program(self, _task_id, action="REGIST", series_code=None, program_code=None, movie_code=None, is_delete=0): if is_delete: data = self.delete_olddata_of_three_parties(sid=series_code, pid=program_code, mid=movie_code, flag="program") else: 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(media_storage, media, episode, media_ext, action) else: data = self._album_movie(media_storage, media, episode, media_ext, action) xml = Adi.to_xml(data, flag="Nope") return xml def save_album_xml_file(self, csp_id, cmd_id, action="REGIST", param="series", series_code=None, program_code=None, movie_code=None, is_delete=0): if param == "series": xml = self.album(cmd_id, series_code=series_code, is_delete=is_delete) else: xml = self.program( cmd_id, action=action, series_code=series_code, program_code=program_code, movie_code=movie_code, is_delete=is_delete ) 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(csp_id, cmd_id) with open(file_path, "w") as f: f.write(xml) def callback(self, request_body): logger.error("{} 回执请求体内容 {}".format(self.partner_alias, request_body)) params = C2ApiUtil.callback_params_config(self.partner_alias) re_correlate_id = params["correlate_id"] re_cmd_result = params["cmd_result"] re_result_file_url = params["result_fileURL"] re_response_xml = params["response_xml"] try: tree = ET.fromstring(request_body) _correlate_id = tree.find(re_correlate_id).text if "sywh" in _correlate_id: cor_id, flag, text_flag = _correlate_id, "-1", "删除其它平台工单" else: cor_id, flag, _ = _correlate_id.split("_") text_flag = "剧头" if flag == "00" else "子集|电影" code = tree.find(re_cmd_result).text result_file_url = tree.find(re_result_file_url).text if cor_id: if flag == "00": code, desc = self.callback_album(code, cor_id, result_file_url) elif flag == "01": code, desc = self.callback_program(code, cor_id, result_file_url) else: code, desc = 0, "delete {} Successful.".format(cor_id) else: logger.error( "{} callback 回调错误, code:{}, {}: id: {}".format(self.partner_alias, code, cor_id, text_flag)) code = -1 desc = ResponseCode.PARAM_ERROR.desc except Exception: code, desc = -1, "未携带xml报文体,xml解析失败." xml = re_response_xml.format(desc, code) return xml def exec_cmd(self, task_id, is_delete=0): cmd_file_url = self.get_cmd_ftp_url(self.csp_id, task_id) request_task_id = task_id if is_delete else "{}_{}".format(task_id, datetime_now()) xml = C2ApiUtil.REQUESTXML.format(self.csp_id, self.lsp_id, request_task_id, cmd_file_url) logger.warning("{} EXEC_CMD Request: {}".format(self.lsp_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_CMD Response: {}".format(self.lsp_id, response)) tree = ET.fromstring(response) code = tree.find(".//multiRef[@id='id1']") desc = tree.find(".//multiRef[@id='id0']/ErrorDescription") if code is not None and int(code.text) == 0: remark = "" status = TaskStatusValue.FILE_TRANSFERING else: status = TaskStatusValue.FILE_FAILED remark = "" if desc is None else desc.text if not is_delete: _task_id, _ = task_id.split("_", 1) the_task = TaskModel.objects.filter(pk=_task_id) if the_task.first().status in TaskStatusValue.pending(): the_task.update(status=status, remark=remark) return status def exec_sync_header_cmd(self, header_id, is_delete=0): cmd_file_url = self.get_cmd_ftp_url(self.csp_id, header_id) requests_header_id = header_id if is_delete else "{}_{}".format(header_id, datetime_now()) xml = C2ApiUtil.REQUESTXML.format(self.csp_id, self.lsp_id, requests_header_id, cmd_file_url) logger.warning("{} exec_sync_header_cmd HEADER_ID: {}, Request: {}".format(self.lsp_id, header_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, header_id, response)) tree = ET.fromstring(response) code = tree.find(".//multiRef[@id='id1']") # desc = tree.find(".//multiRef[@id='id0']/ErrorDescription") if code is not None and int(code.text) == 0: status = MediaInfoSyncModel.SYNCING else: status = MediaInfoSyncModel.FAILED if not is_delete: _header_id, _ = header_id.split("_", 1) the_sync = MediaInfoSyncModel.objects.filter(pk=_header_id) if the_sync.first().status == MediaInfoSyncModel.PENDING: the_sync.update(status=status) return def injector(self, task, series_code=None, program_code=None, movie_code=None): status, flag = 0, 0 if program_code: # 工单号增加当前日期天数，防止删除后工单重复 _task_id, is_delete = program_code, 1 elif task.status in (TaskStatusValue.FILE_PENDING, TaskStatusValue.INFO_PENDING): _task_id, is_delete = self.generate_correlate_id(task.id), 0 else: _task_id, is_delete = None, 0 if _task_id: self.save_album_xml_file(self.csp_id, _task_id, param="program", series_code=series_code, program_code=program_code, movie_code=movie_code, is_delete=is_delete) status = self.exec_cmd(_task_id, is_delete=is_delete) flag = self.get_continue_to_injector(status) return status, flag 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): 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="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, series_code=None): if series_code: # 工单号增加当前日期天数，防止删除后工单重复 _header_id, is_delete = series_code, 1 elif sync.status == MediaInfoSyncModel.PENDING: _header_id, is_delete = self.generate_correlate_id(sync.id, stage="series"), 0 else: _header_id, is_delete = None, 0 if _header_id: self.save_album_xml_file(self.csp_id, _header_id, series_code=series_code, is_delete=is_delete) self.exec_sync_header_cmd(_header_id, is_delete=is_delete) def series_header_update(self, sync, series_code=None): self.series_header_injector(sync, series_code=series_code) def series_header_remove(self, sync, series_code=None): self.series_header_injector(sync, series_code=series_code)