import logging import random import requests from domain.imdb.value_object import CelebrityRoleValue from domain.injector.value_object import TaskStatusValue, ContentTypeValue, DefinitionValue from domain.lsp.base_injector import LspInjectorC2 from domain.lsp.value_object import C2ApiUtil from imdb.models import CelebrityMediaRelationModel from injector.models import TaskModel import xml.etree.ElementTree as ET from utils.date_utils import datetime_now from domain.lsp.generated.common_xsd import Adi from utils.defines import ResponseCode import os from domain.injector.task_queue import TaskQueue logger = logging.getLogger("django") class CodeFactory: Series = "ps" # 基础媒资信息id类别，供剧头使用 Program = "p" # 基础媒资信息id类别 Movie = "m" # 基础媒资分集id类别 Picture = "pic" # 基础图片id类别 Package = "pkg" Category = "cate" Chars = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" 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, episode_id): flag_0 = (32 - len(str(series_id)) - len(str(episode_id)) - len(self.Program) - len(self.csp_code) - 3) * "0" return "{}_{}{}_{}_{}".format(self.Program, self.csp_code, flag_0, series_id, episode_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) def gen_picture_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.Picture, self.csp_code, flag_0, series_id, movie_id) def gen_package_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.Package, self.csp_code, flag_0, series_id, movie_id) def gen_category_code(self, series_id): flag_0 = (32 - len(str(series_id)) - len(self.Program) - len(self.csp_code) - 3) * "0" return "{}_{}{}_{}".format(self.Category, self.csp_code, flag_0, series_id) def gen_package_order(self): flag_0 = random.sample(self.Chars, 6) return "{}_{}".format("senyuapp", "".join(flag_0)) class FuJianYiDong(LspInjectorC2): Video_Required_Args = ("url",) def __init__(self, alias="FUJIANYIDONG"): 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 = "senyuapp" # 这个找LSP确认 self.code_factory = CodeFactory(self.cps_code) self.is_cdn_xml = True self.scatter_csp_id = "FJ_SHSY" def get_content_type(self, content_type): if ContentTypeValue.POSITIVE == content_type: return ContentTypeValue.POSITIVE return 2 # 清晰度（1:高清HD、2:流畅LD、3:标清SD、4：XD，默认SD） 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 "XD" elif definition == DefinitionValue.K4: return "XD" return "SD" def get_duration_change(self, duration): ts = [] for i in range(3): p = duration % 60 duration = duration // 60 p = str(p) if len(p) == 1: p = "0" + p ts.append(p) return "".join(ts[::-1]) def get_regions(self, media_obj): regions = media_obj.regions if regions and len(regions): names = ["内地" if region["id"] == 1 else region["name"] for region in regions] return self.region_sep.join(names) return "" def property_movie(self, _id, media_storage, _type=1, action="REGIST"): return { "element_type": "Movie", "id": _id, "code": _id, "action": action, "property": [ {"FileURL": media_storage.url}, {"Type": self.get_content_type(media_storage.content_type)}, {"Duration": self.get_duration_change(media_storage.seconds)}, {"FileSize": media_storage.size}, ] } def property_series(self, ps_id, media_storage, media, media_ext, action="REGIST"): _actors = CelebrityMediaRelationModel.objects.filter( media_id=media.id, role=CelebrityRoleValue.ACTOR).select_related("celebrity")[:4] _directors = CelebrityMediaRelationModel.objects.filter( media_id=media.id, role=CelebrityRoleValue.DIRECTOR).select_related("celebrity")[:4] _writers = CelebrityMediaRelationModel.objects.filter( media_id=media.id, role=CelebrityRoleValue.AUTHOR).select_related("celebrity")[:2] actors, directors, writers = self.join_names(_actors), self.join_names(_directors), self.join_names( _writers) # media_verpic = self.image_resize.get_oss_resize_params(self.partner_alias, media_ext.media_verpic, # size=media_ext.verpic_size, to_ip=True) param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) # definition = self._get_bit_rate_type(media_storage.definition) vst, vet = self.get_media_effective_time(media.id) the_property = [ {'Code': ps_id}, {"Director": directors}, # 要写导演 {"Name": self.get_media_name_by_alias(media)}, {"SearchName": self.transform_chinese_to_letter(self.get_media_name_by_alias(media))}, {"Language": media.languages}, {"Kpeople": actors}, # 这里是主演 {"Description": media.storyline}, {"Type": self._get_category_desc(change_fields["category"])}, {"ReleaseYear": media.year}, {"VolumnCount": media.episodes}, {"OrgAirDate": ""}, {"Tags": self.get_genre(media)}, {"Status": 1}, {"OriginalCountry": self.get_regions(media)}, {"Score": media.douban_score}, # {"Duration": int(media.total_seconds / 60) + 1}, {"LicensingWindowStart": vst}, {"LicensingWindowEnd": vet}, {"Price": 0}, ] return { "element_type": "Series", "id": ps_id, "code": ps_id, "action": action, "property": the_property } def _album_sub_series(self, media_storage, media, episode, media_ext, _episodes, action="REGIST"): # 还需要添加媒资信息 _actors = CelebrityMediaRelationModel.objects.filter( media_id=media.id, role=CelebrityRoleValue.ACTOR).select_related("celebrity")[:4] _directors = CelebrityMediaRelationModel.objects.filter( media_id=media.id, role=CelebrityRoleValue.DIRECTOR).select_related("celebrity")[:4] _writers = CelebrityMediaRelationModel.objects.filter( media_id=media.id, role=CelebrityRoleValue.AUTHOR).select_related("celebrity")[:2] actors, directors, writers = self.join_names(_actors), self.join_names(_directors), self.join_names( _writers) param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) # pcid = self.code_factory.gen_category_code(media.id) pcid = "cate_senyuapp0000000000000009999" # 生活栏目 # p_cate = { # "element_type": "Category", # "id": pcid, # "code": pcid, # "action": action, # "parentcode": "0", # "property": [ # {"Name": "森宇（打散）"}, # {"Status": 1}, # ] # } pcid1 = "cate_senyuapp0000000000000008888" # 生活栏目 # p_cate1 = { # "element_type": "Category", # "id": pcid1, # "code": pcid1, # "action": action, # "parentcode": pcid, # "property": [ # {"Name": "生活"}, # {"Status": 1}, # ] # } # 定义 media的 id psid = self.code_factory.gen_series_code(media.id) media_ext = media.mediaextfieldmodel p_series = self.property_series(psid, media_storage, media, media_ext, action=action) # 分集的 父id 也就是分集所在媒体id 然后拼接成 一个字符串 psid = self.code_factory.gen_series_code(media.id) # 返回 清晰度 转成需要的LD SD HD XD definition = self._get_bit_rate_type(media_storage.definition) episode_title = self.get_episode_title(media, episode) ppid = self.code_factory.gen_program_code(media.id, episode.id) pkg_id = self.code_factory.gen_package_code(media.id, media_storage.id) vst, vet = self.get_media_effective_time(media.id) # 定义 program 对象部分 p_program = { "element_type": "Program", "id": ppid, "code": ppid, "action": action, "property": [ {"Name": episode_title}, {"OriginalCountry": self.get_regions(media)}, {"Language": media.languages}, {"ReleaseYear": media.year}, {"LicensingWindowStart": vst}, {"LicensingWindowEnd": vet}, {"Director": directors}, # 要写导演 {"Score": media.douban_score}, {"Type": self._get_category_desc(change_fields["category"])}, {"Duration": int(media_storage.seconds / 60) + 1}, {"Description": episode.storyline if episode.storyline else media.storyline}, {'Seriesflag': 0 if media.episodes == 1 else 1}, {"Status": 1}, {"PriceTaxIn": 0}, ] } # 定义program与series的关系 mapping_program = { "parent_type": "Series", "parent_id": psid, "element_type": "Program", "element_id": ppid, "parent_code": psid, "element_code": ppid, "action": action } sub_property = list() if episode.serial_num is not None: sub_property.extend([{"Sequence": episode.serial_num}, ]) if sub_property: mapping_program["property"] = sub_property # 定义 program 与movie的关系 pmid = self.code_factory.gen_movie_code(media.id, media_storage.id) p_movie = self.property_movie(pmid, media_storage, action=action) mapping_movie = self.property_mapping("Program", "Movie", ppid, pmid, action=action) mapping_category = self.property_mapping("Category", "Series", pcid1, psid, action=action) return { "domain": 2, "object_list": [p_series, p_program, p_movie], "mapping_list": [mapping_program, mapping_movie, mapping_category], } 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 data = self._album_sub_series(media_storage, media, episode, media_ext, _episodes, action) xml = Adi.to_xml(data, flag="Nope") return xml def generate_correlate_id_csp_id(self, task_id, tail): return "{}_{}_{}_{}".format("sywh", task_id, tail, datetime_now()) # 介质注入 向LSP发起xml内容 def exec_cmd(self, task_id): _task_id = self.generate_correlate_id(task_id) t_task_id = self.generate_correlate_id_csp_id(task_id, "01") cmdfileUrl = self.get_cmd_ftp_url(self.csp_id, _task_id) # ftp协议 xml = C2ApiUtil.REQUESTXML.format(self.csp_id, self.lsp_id, t_task_id, cmdfileUrl) logger.info("cms_api : {}".format(self.cms_api)) logger.info("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.info("EXEC_CMD Response: {}".format(response)) tree = ET.fromstring(response) code = tree.find(".//Result").text msg = tree.find(".//ErrorDescription").text if int(code) == 0: status = TaskStatusValue.FILE_TRANSFERING remark = "" else: status = TaskStatusValue.FILE_FAILED remark = msg TaskModel.objects.filter(pk=task_id).update(status=status, remark=remark) return status def callback_cdn(self, body): return self.callback(body) def collection_xml(self, media_storage, media, episode, media_ext, _episodes, action): _actors = CelebrityMediaRelationModel.objects.filter( media_id=media.id, role=CelebrityRoleValue.ACTOR).select_related("celebrity")[:4] _directors = CelebrityMediaRelationModel.objects.filter( media_id=media.id, role=CelebrityRoleValue.DIRECTOR).select_related("celebrity")[:4] _writers = CelebrityMediaRelationModel.objects.filter( media_id=media.id, role=CelebrityRoleValue.AUTHOR).select_related("celebrity")[:2] actors, directors, writers = self.join_names(_actors), self.join_names(_directors), self.join_names( _writers) # 定义 media的 id ps_id = self.code_factory.gen_series_code(media.id) 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.id, episode.id) episode_horpic = self.image_resize.get_oss_resize_params(self.partner_alias, episode.episode_horpic, size=episode.horpic_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) media_verpic = self.image_resize.get_oss_resize_params(self.partner_alias, media_ext.media_verpic, size=media_ext.verpic_size, to_ip=True) vst, vet = self.get_media_effective_time(media.id) pmid = self.code_factory.gen_movie_code(media.id, media_storage.id) # 声明Series 标签 the_property = [ {'Code': ps_id}, {"ActorDisplay": directors}, # 要写导演 {"Definition": self._get_bit_rate_type(media_storage.definition)}, {"PictureURL1": self.to_ip_url(media_verpic)}, {"Name": self.get_media_name_by_alias(media)}, {"Language": media.languages}, {"WriterDisplay": actors}, # 这里是主演 {"Description": media.storyline}, {"ProgramType": self._get_category_desc(change_fields["category"])}, {"ProgramType2": self.get_genre(media)}, {"ReleaseYear": media.year}, {"VolumnCount": media.episodes}, {"OrgAirDate": ""}, {"Tag": ""}, {"Status": 1}, {"OriginalCountry": self.get_regions(media)}, {"Score": media.douban_score}, {"Duration": int(media.total_seconds / 60) + 1}, {"Recommendation": media_ext.focus}, {"HorizontalPoster": self.to_ip_url(media_horpic)}, {"ContentType": "Video"}, {"IfPreview": 0}, {"scriptWriter": writers}, # {"LicensingWindowStart": vst}, # {"LicensingWindowEnd": vet}, # {"Price": 0}, ] p_series = { "element_type": "Series", "id": ps_id, "code": ps_id, "action": action, "property": the_property } # 声明 Mapping标签 # 定义 program 对象部分 p_program = { "element_type": "Program", "id": ppid, "code": ppid, "action": action, "property": [ {"Definition": self._get_bit_rate_type(media_storage.definition)}, {"Name": episode_title}, {"PictureURL1": self.to_ip_url(media_verpic)}, {"VolumnCount": episode.serial_num}, {"ProgramType": self._get_category_desc(change_fields["category"])}, {"ProgramType2": self.get_genre(media)}, {"Tag": ""}, {"IfPreview": 0}, {"VideoType": self.get_content_type(media_storage.content_type)}, {"introduction": episode.storyline if episode.storyline else media.storyline}, {"skipStartTime": 0}, {"skipEndTime": 0}, {"isCharge": str(change_fields["is_charge"])}, # {"LicensingWindowStart": vst}, # {"LicensingWindowEnd": vet}, ] } # 声明Movie标签 p_movie = { "element_type": "Movie", "id": pmid, "code": pmid, "action": action, "property": [ {"FileURL": media_storage.url}, {"PlayURL": "http://ip:port/icntv/{}:{}".format(ps_id, ppid)}, {"Definition": self._get_bit_rate_type(media_storage.definition)}, {"CdnPlatform": 125}, {"Dispatch": 1}, {"Md5": media_storage.md5}, ] } # 定义program与series的关系 mapping_program = { "parent_type": "Series", "parent_id": ps_id, "element_type": "Program", "element_id": ppid, "parent_code": ps_id, "element_code": ppid, "action": action } sub_property = list() if episode.serial_num is not None: sub_property.extend([{"Sequence": episode.serial_num}, ]) if sub_property: mapping_program["property"] = sub_property # 定义 program 与movie的关系 mapping_movie = self.property_mapping("Program", "Movie", ppid, pmid, action=action) return { "domain": 2, "object_list": [p_series, p_program, p_movie], "mapping_list": [mapping_program, mapping_movie], } def save_satter_xml_file(self, csp_id, task_id, action="REGIST"): 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 data = self.collection_xml(media_storage, media, episode, media_ext, _episodes, action) xml = Adi.to_xml(data, flag="Nope") 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, str(task_id) + "_02") with open(file_path, "w") as f: f.write(xml) def exec_scatter_cmd(self, task_id): t_task_id = self.generate_correlate_id_csp_id(task_id, "02") cmdfileUrl = self.get_cmd_ftp_url(self.csp_id, str(task_id) + "_02") # ftp协议 logger.info("exec_scatter_cmd: {}".format(cmdfileUrl)) xml = C2ApiUtil.REQUESTXML.format(self.scatter_csp_id, self.conf.scatter_lsp, t_task_id, cmdfileUrl) logger.info("scatter_api : {}".format(self.conf.scatter_api)) logger.info("EXEC_CMD Request: {}".format(xml)) headers = {"content-type": "text/xml", "SOAPAction": "application/soap+xml;charset=utf8"} response = requests.post(self.conf.scatter_api, data=xml, headers=headers).content logger.info("EXEC_CMD Response: {}".format(response)) tree = ET.fromstring(response) code = tree.find(".//Result").text msg = tree.find(".//ErrorDescription").text if int(code) == 0: status = TaskStatusValue.INFO_TRANSFERING remark = "" else: status = TaskStatusValue.INFO_FAILED remark = msg TaskModel.objects.filter(pk=task_id).update(status=status, remark=remark) return status def get_continue_to_injector(self, status): return status not in TaskStatusValue.stop_injector() def callback(self, request_body): logger.info("回执:{} {}".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"] logger.info("{} re_response_xml {}".format(self.partner_alias, re_response_xml)) try: tree = ET.fromstring(request_body) _correlate_id = tree.find(re_correlate_id).text code = tree.find(re_cmd_result).text if not code: code = 0 result_file_url = tree.find(re_result_file_url).text csp_id, cor_id, flag, _ = _correlate_id.split("_") text_flag = "介质注入" if flag == "01" else "打散注入" if cor_id: if int(code) == 0: task_status = TaskStatusValue.FILE_COMPLETED if flag == "01" else TaskStatusValue.INFO_COMPLETED logger.info("{} {} 回执成功".format(self.partner_alias, text_flag)) else: task_status = TaskStatusValue.FILE_FAILED if flag == "01" else TaskStatusValue.INFO_FAILED logger.error( "{} {}回调失败: task_id: {}, result_file_url: {}".format(self.partner_alias, text_flag, cor_id, result_file_url)) TaskModel.objects.filter(pk=cor_id).update(status=task_status) code = ResponseCode.SUCCESS.value desc = ResponseCode.SUCCESS.desc if flag == "01": task = TaskModel.objects.filter(pk=cor_id).first() TaskModel.objects.filter(pk=cor_id).update(status=TaskStatusValue.INFO_PENDING) TaskQueue.update(task.level, task.id, TaskStatusValue.INFO_PENDING) logger.info("{} {} 回调结束".format(self.partner_alias, text_flag)) 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 injector(self, task, action="REGIST"): status, flag = 0, 0 # 如果工单状态 在 pending 在就绪状态，我们就继续 if task.status in (TaskStatusValue.FILE_PENDING,): logger.info("开始C2注入status:{}".format(task.status)) _task_id = self.generate_correlate_id(task.id) self.save_album_xml_file(self.csp_id, _task_id, param="program", action=action) status = self.exec_cmd(task.id) flag = self.get_continue_to_injector(status) if task.status in (TaskStatusValue.INFO_PENDING,): logger.info("开始打散注入status:{}".format(task.status)) self.save_satter_xml_file(self.csp_id, task.id, action=action) status = self.exec_scatter_cmd(task.id) flag = self.get_continue_to_injector(status) logger.info("执行完注入:status:{} flag:{}".format(status, flag)) return status, flag # 介质更新 def update(self, task): TaskModel.objects.filter(pk=task.id).update(status=TaskStatusValue.FILE_PENDING) return self.injector(task, action="UPDATE") # 介质删除 def remove(self, task): TaskModel.objects.filter(pk=task.id).update(status=TaskStatusValue.FILE_PENDING) return self.injector(task, action="DELETE")