import logging import random import requests import xml.etree.ElementTree as ET from domain.imdb.value_object import CelebrityRoleValue from domain.lsp.generated.common_xsd import Adi from domain.lsp.value_object import CodeSampleFactory, C2ApiUtil from imdb.models import CelebrityMediaRelationModel from injector.models import MediaInfoSyncModel, TaskModel from .base_injector import LspInjectorC2 from domain.injector.value_object import TaskStatusValue, DefinitionValue, ContentTypeValue from utils.date_utils import datetime_now import hashlib import json import ftplib from lxml import etree logger = logging.getLogger(__file__) # 定义 id、 code 模式 class CodeFactory1: Series = "ps" # 基础媒资信息id类别，供剧头使用 Program = "p" # 基础媒资信息id类别 Movie = "m" # 基础媒资分集id类别 def __init__(self, csp_code=""): self.csp_code = csp_code def gen_series_code(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_program_code(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_movie_code(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.Movie, self.csp_code, flag_0, series_id, movie_id) # 定义陕西移动统一注入类 class ShanXiTongYi(LspInjectorC2): def __init__(self, alias="SHANXITONGYI"): 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.desc_prefix = "（来源：喀什森茂）" self.code_factory = CodeSampleFactory(self.csp_id, max_len=30) self.try_count = 3 self.type_num_mapper = {"H.264": 1, "MPEG4": 2, "AVS": 3, "MPEG2": 4, "MP3": 5, "H.265": 7, "AVS2": 8} self.resolution_num_mapper = {"QCIF": 1, "QVGA": 2, "2/3 D1": 3, "3/4 D1": 4, "D1": 5, "720P": 6, "1080i": 7, "1080P": 8, "4K(3840*2160)": 9, "4K(4096*2160)": 10} 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 get_content_type(self, content_type): return { ContentTypeValue.POSITIVE: "正片", ContentTypeValue.CLIP: "花絮", ContentTypeValue.TRAILER: "预告片" }.get(content_type, "正片") # 清晰度（1:高清HD、2:流畅LD、3:标清SD、4：XD，默认SD） def _get_bit_rate_type(self, definition): if definition == DefinitionValue.P320: return "2" elif definition == DefinitionValue.P480: return "3" elif definition == DefinitionValue.P720: return "4" elif definition == DefinitionValue.P1080: return "5" elif definition == DefinitionValue.K4: return "4205" return "3" def _get_bit_rate_id(self, definition, vcodec=None): if definition == DefinitionValue.P320: return "2" elif definition == DefinitionValue.P480: return "3" elif definition == DefinitionValue.P720: return "4" elif definition == DefinitionValue.P1080: return "5" elif definition == DefinitionValue.K4: if vcodec == "h264": return "4104" else: return "4205" return "3" 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_tags(self, media_obj): genres = media_obj.genres if genres and len(genres): names = [genre["name"] for genre in genres] return self.tag_sep.join(names) return "" def property_series(self, ps_id, media, media_storage, media_ext, action="REGIST"): actors, directors, writers = self.get_media_person(media, actor=4, director=4, writer=2) vst, vet = self.get_media_effective_time(media.id) # 这个需要value_object.py文件中 LspExtendConfig类中配置 GUANGXIYIDONG属性 定义横图大小、竖图大小、分类 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) the_property = [ {"ActorDisplay": directors}, # 要写导演 一会改动 {"Definition": definition}, {"PictureURL1": self.to_ip_url(media_verpic)}, {"Name": self.get_media_name_by_alias(media)}, {"Language": media.languages}, {"WriterDisplay": actors}, # 这里是主演 {"Description": self.make_desc_prefix(media.storyline)}, {"ProgramType": self._get_category_desc(change_fields["category"])}, {"ProgramType2": self.get_genre(media)}, {"ReleaseYear": media.year}, {"VolumnCount": media.episodes}, {"OriginalCountry": self.get_regions(media)}, {"IsCharge": str(change_fields["is_charge"])}, {"OrgAirDate": ""}, {"Duration": int(media.total_seconds / 60) + 1}, {"OnlineTime": f"{media.date_published} 23:59:59"}, {"Recommendation": media_ext.focus}, {"HorizontalPoster": self.to_ip_url(media_horpic)}, {"ContentType": "Video"}, {"IfPreview": 0}, {"scriptWriter": writers}, # 这里要写编剧 {"ProgramCount": media.episodes}, {"IsSeries": 0 if media.episodes == 1 else 1}, {"IsShow": 1}, {"LicensingWindowStart": vst}, {"LicensingWindowEnd": vet} ] return { "element_type": "Series", "id": ps_id, "code": ps_id, "action": action, "property": the_property } # 广西移动，要求简介 有前缀 def make_desc_prefix(self, content): return self.desc_prefix + content # 广西移动只支持一个国家 目前处理是取第一个国家 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 property_program(self, ppid, media, media_storage, media_ext, episode, action="REGIST"): 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) episode_title = self.get_episode_title(media, episode) p_program = { "element_type": "Program", "id": ppid, "code": ppid, "action": action, "property": [ {"Definition": definition}, {"PictureURL1": self.to_ip_url(media_horpic)}, {"Name": episode_title}, {"VolumnCount": episode.serial_num}, {"Duration": int(media_storage.seconds / 60) + 1}, {"ProgramType": self._get_category_desc(change_fields["category"])}, {"Description": ""}, {"ProgramType": self._get_category_desc(change_fields["category"])}, {"ProgramType2": self.get_genre(media)}, {"IfPreview": 0}, {"IsCharge": str(change_fields["is_charge"])}, {"IsShow": 1} ] } return p_program def change_storage_url(self, media_storage): return media_storage.hls_url def gen_md5(self, url): _hash_md5 = hashlib.md5() buffer = requests.get(url).content _hash_md5.update(buffer) md5 = _hash_md5.hexdigest() print("md5:{}".format(md5)) return md5 def property_movie(self, _id, media, media_storage, _type=1, action="REGIST"): 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) return { "element_type": "Movie", "id": _id, "code": _id, "action": action, "property": [ {"FileURL": self.change_storage_url(media_storage)}, {"Definition": definition}, {"BitRateId": self._get_bit_rate_id(media_storage.definition, media_storage.v_codec)}, {"Definition": self._get_bit_rate_type(media_storage.definition)}, {"CdnPlatform": "101,102"}, {"Dispatch": 1}, {"Md5": self.gen_md5(media_storage.hls_url)}, {"ProgramType": self._get_category_desc(change_fields["category"])}, ] } # 定义分集信息 def _album_sub_series(self, media_storage, media, episode, media_ext, _episodes, action="REGIST"): # 定义 media的 id psid = self.code_factory.gen_series_code(media.id) p_series = self.property_series(psid, media, media_storage, media_ext, action=action) ppid = self.code_factory.gen_program_code(media_storage.id) p_program = self.property_program(ppid, media, media_storage, media_ext, episode, action=action) # 定义 program 对象部分 pmid = self.code_factory.gen_movie_code(media_storage.id) p_movie = self.property_movie(pmid, media, media_storage, action=action) mapping_program = self.property_mapping("Series", "Program", psid, ppid, sequence=episode.serial_num, action=action) mapping_movie = self.property_mapping("Program", "Movie", ppid, pmid, action=action) # if media.episodes == 1: # return None # else: # TaskModel.objects.filter(media_id=media.id, partner_id=self.lsp_pk). if episode.serial_num > 1: return { "object_list": [p_program, p_movie], "mapping_list": [mapping_program, mapping_movie], } return { "object_list": [p_series, p_program, p_movie], "mapping_list": [mapping_program, mapping_movie], } # 定义节目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 # 媒体文件补充信息 # 目前的设计是 任何媒介都要有1集 电影也要设置为1集 _episodes = int(media.episodes) if media.episodes else 0 if _episodes > 1: # 如果大于1 表示 连续剧 则该介质是某集电视剧 data = self._album_sub_series(media_storage, media, episode, media_ext, _episodes, action) else: # 电影 # data = self._album_movie(media_storage, media, episode, media_ext, action) data = self._album_sub_series(media_storage, media, episode, media_ext, _episodes, action) xml = Adi.to_xml(data, flag="Nope") logger.warning(xml) return xml def handler_resp_url(self, res_ftp_url): try: r = res_ftp_url.split("//")[1].split("/") info = r[0] path = r[1:] dirs_path = path[:-1] file_name = path[-1] username, passwd = info.split("@")[0].split(":") host = info.split("@")[1] port = "21" if host.find(":") != -1: host, port = host.split(":") ftp = ftplib.FTP() ftp.connect(host, int(port)) ftp.login(username, passwd) print("login success") for path in dirs_path: ftp.cwd(path) print("cwd success") file_path = "/home/admin/xml_file/" + file_name with open(file_path, "wb") as fp: ftp.retrbinary('RETR {}'.format(file_name), fp.write) with open(file_path, "rb") as fp: tree = etree.fromstring(fp.read()) print("read xml") property_list = tree.findall(".//Property") for item in property_list: if item.attrib.get("Name") == "Description": return item.text except Exception as e: print(e) return "" def callback_program(self, code, task_id, res_furl): if int(code) == 0: task_status = TaskStatusValue.INFO_COMPLETED remark = "" else: task_status = TaskStatusValue.INFO_FAILED logger.error( "{} 子集|电影 回调失败: task_id: {}, result_file_url: {}".format(self.partner_alias, task_id, res_furl)) remark = self.handler_resp_url(res_furl) TaskModel.objects.filter(pk=task_id).update(status=task_status, remark=remark) code = 0 desc = "成功" return code, desc # 剧头注入 向LSP发起xml内容 def exec_sync_header_cmd(self, sync, count=0): count += 1 _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) msg = tree.find(".//ErrorDescription").text if msg.lower() == "success": status = MediaInfoSyncModel.SYNCING else: if str(msg).lower().__contains__("could not connect") and count < self.try_count: return self.exec_sync_header_cmd(sync, count) status = MediaInfoSyncModel.FAILED the_sync = MediaInfoSyncModel.objects.filter(pk=sync.id) if the_sync.first().status == MediaInfoSyncModel.PENDING: the_sync.update(status=status) return # 介质注入 向LSP发起xml内容 def exec_cmd(self, task_id, count=0): count += 1 _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 msg = tree.find(".//ErrorDescription").text if msg.lower() == "success": status = TaskStatusValue.FILE_TRANSFERING remark = "" else: if str(msg).lower().__contains__("could not connect") and count < self.try_count: return self.exec_cmd(task_id, count) status = TaskStatusValue.FILE_FAILED remark = msg TaskModel.objects.filter(pk=task_id).update(status=status, remark=remark) return status # 介质注入 def injector(self, task): status, flag = 0, 0 # 如果工单状态 在 pending 在就绪状态，我们就继续 if task.status in (TaskStatusValue.FILE_PENDING, TaskStatusValue.INFO_PENDING): # 组成一新的task_id 在全局中唯一 taskId_00 _task_id = self.generate_correlate_id(task.id) # 生成新的xml 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 update(self, task): if task.status in (TaskStatusValue.FILE_PENDING, TaskStatusValue.INFO_PENDING): pass # 介质注入删除 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))