import datetime import json import logging import math import os import requests from domain.imdb.value_object import CelebrityRoleValue from domain.injector.value_object import DefinitionValue, TaskModelActionValue, ContentTypeValue from domain.lsp.base_injector import LspInjectorC2, TaskStatusValue, C2ApiUtil, TaskModel from imdb.models import CelebrityMediaRelationModel from utils.date_utils import datetime_now from lxml import etree from injector.models import MediaInfoSyncModel from domain.lsp.generated.common_xsd import Adi from domain.lsp.value_object import C2ApiUtil, CodeSampleFactory from domain.injector.task_queue import TaskQueue from django.core.cache import cache import pypinyin import json logger = logging.getLogger(__file__) class JiangSuYd(LspInjectorC2): Media_Required_Args = ("episodes", "name", "date_published", "category", "genres", "regions", "storyline", "seconds", "total_seconds", "languages", "year", "douban_score") MediaExt_Required_Args = ("media_horpic", "definition") Video_Required_Args = ("hls_url", "a_codec", "definition") Episode_Required_Args = ("serial_num",) def __init__(self, alias="JIANGSUYD"): super().__init__(alias) self.csp_id = self.conf.csp_id self.lsp_id = self.conf.lsp_id self.csp_id_hw = "shsytmsj" self.csp_id_zx = "shsytmsjzx" self.cast_sep = ";" self.region_sep = ";" self.genre_sep = "" self.tag_sep = ";" self.is_cdn_xml = False self.proxy_hw = False # 额外加参数控制状态 self.content_provider = "森宇童梦视界" self.cdn_ftp_url = "ftp://shsytmsj:UvtHM28SrznJZv37@112.25.63.183:2111/{}/{}" self.code_factory = CodeSampleFactory(30) self.EXPIRE_TIME = 3600 * 24 * 3 def get_media_name(self, name): return str(name).strip() def gen_media_definition(self, definition): if definition == "1": return "LD" elif definition == "2": return "SD" elif definition == "3": return "HD" elif definition == "4": return "XD" elif definition == "5": return "4K" def _get_positive(self, media_storage): return media_storage.content_type def gen_ftp_pic_url(self, media_id, pic_url): # file_name = pic_url.split("/")[-1] # return self.cdn_ftp_url.format(media_id, file_name) return pic_url def gen_ftp_movie_url(self, media_id, video_url): # file_name = "/".join(video_url.split("/")[-2:]) # return self.cdn_ftp_url.format(media_id, file_name) return video_url def transform_chinese_to_letter(self, the_str): _letter = pypinyin.pinyin(the_str, style=pypinyin.FIRST_LETTER) result = [str(_letter[index][0]).lower() for index, _ in enumerate(_letter)] return "".join(result) def gen_date_published(self, date_published): if date_published: return str(date_published).replace("-", "") return "" def gen_series_obj(self, _id, media, media_ext, storage, action): 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) 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) property_list = [ {"ActorDisplay": directors}, {"Definition": self.gen_program_definition(media_ext.definition, storage.v_codec)}, {"PictureURL1": self.gen_ftp_pic_url(media.id, self.to_ip_url(media_verpic))}, {"VerticalPoster": self.gen_ftp_pic_url(media.id, self.to_ip_url(media_verpic))}, {"Name": self.get_media_name(self.get_media_name_by_alias(media))}, {"Pinyin": self.transform_chinese_to_letter(self.get_media_name_by_alias(media))}, {"Language": media.languages}, {"WriterDisplay": actors}, {"Description": media.storyline}, {"OrgAirDate": self.gen_date_published(media.date_published)}, {"Type": self._get_category_desc(change_fields["category"])}, {"ProgramType": self._get_category_desc(change_fields["category"])}, {"ProgramType2": self.get_genre(media)}, {"ReleaseYear": media.year}, {"VolumnCount": media.episodes}, {"OriginalCountry": self.get_regions(media)}, {"Tag": self.get_genre(media)}, {"Score": media.douban_score}, {"Duration": int(media.total_seconds)}, {"ContentType": "Video"}, {"IfPreview": self._get_positive(storage)}, {"IsSeries": 0 if media.episodes == 1 else 1}, {"Status": 1}, {"ScriptWriter": writers}, {"IsCharge": change_fields["is_charge"]}, {"ProductCode": "003022cp0001"}, {"HorizontalPoster": self.gen_ftp_pic_url(media.id, self.to_ip_url(media_horpic))}, {"PictureURL2": self.gen_ftp_pic_url(media.id, self.to_ip_url(media_horpic))}, {"CopyRightBeginDate": vst}, {"CopyRightEndDate": vet} ] data = { "element_type": "Series", "id": _id, "code": _id, "action": action, "property": property_list } return data def gen_program_definition(self, definition, v_codec): if definition in (DefinitionValue.K4, DefinitionValue.P3840, DefinitionValue.P4096): if v_codec == "h264": return 4104 else: return 4205 return definition def gen_program_obj(self, _id, media, media_ext, storage, episode, action): 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) 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) actors, directors, writers = self.get_media_person(media, actor=4, director=4, writer=2) property_list = [ {"ActorDisplay": actors}, {"Definition": self.gen_program_definition(storage.definition, storage.v_codec)}, {"Name": episode_title}, {"Language": media.languages}, {"WriterDisplay": writers}, {"Description": media.storyline}, {"PictureURL1": self.gen_ftp_pic_url(media.id, self.to_ip_url(media_verpic))}, {"OrgAirDate": self.gen_date_published(media.date_published)}, {"ReleaseYear": media.year}, {"OriginalCountry": self.get_regions(media)}, {"ProgramType": self._get_category_desc(change_fields["category"])}, {"ProgramType2": self.get_genre(media)}, {"VolumnCount": episode.serial_num}, {"Duration": int(media.seconds)}, {"IsCharge": change_fields["is_charge"]}, {"Tag": self.get_genre(media)}, {"IfPreview": self._get_positive(storage)}, {"IsSeries": 0 if media.episodes == 1 else 1}, {"Status": 1}, {"BitId": self.gen_program_definition(storage.definition, storage.v_codec)}, {"IsCharge": change_fields["is_charge"]}, {"PictureURL2": self.gen_ftp_pic_url(media.id, self.to_ip_url(media_horpic))}, ] data = { "element_type": "Program", "id": _id, "code": _id, "action": action, "property": property_list } return data def gen_movie_ftp_url(self, media_id, url): s2 = str(url).split("/")[-2] s1 = str(url).split("/")[-1] file_name = "{}/{}".format(s2, s1) return self.cdn_ftp_url.format(media_id, file_name) def gen_movie_obj(self, _id, media, storage, action, args): seriesId_hw, programId_hw, movieId_hw, seriesId_zx, programId_zx, movieId_zx = args _ids = {"hwMovieId": movieId_hw, "hwProgramId": programId_hw, "zteMovieId": movieId_zx, "zteProgramId": programId_zx} ext_info = {"cdnPlayInfos": [{"cdnId": 108, "outMediaCode": f"{movieId_zx}", "playUrl": f"http://183.207.249.71:80/shsj/vod/{programId_zx}/{movieId_zx}"}, {"cdnId": 23, "outMediaCode": f"{movieId_hw}", "playUrl": f"http://zv6.ott.mobaibox.com:6060/shsytmsjzx/vod/{programId_hw}/{movieId_hw}/index.m3u8"}]} property_list = [ {"Definition": self.gen_program_definition(storage.definition, storage.v_codec)}, {"FileURL": self.gen_ftp_movie_url(media.id, storage.hls_url)}, {"PlayURL": json.dumps(_ids)}, # {"CdnPlatform": 162}, {"Dispatch": 0}, {"IsTranscode": 0}, {"Status": 1}, {"MediaType": "Clear"}, {"BitId": self.gen_program_definition(storage.definition, storage.v_codec)}, {"Md5": storage.md5}, # {"extInfo": json.dumps(ext_info)} ] data = { "element_type": "Movie", "id": _id, "code": _id, "action": action, "property": property_list } return data def gen_program_code_by_movie(self, media): psid = self.code_factory.gen_series_code(media.id) return psid def gen_sub_program(self, media, media_ext, storage, episode, action, args): # seriesId_hw, programId_hw, movieId_hw, seriesId_zx, programId_zx, movieId_zx = args # psid = seriesId_hw # ppid = programId_hw # pmid = movieId_hw psid = self.code_factory.gen_series_code(media.id) ppid = self.code_factory.gen_program_code(episode.id) pmid = self.code_factory.gen_movie_code(storage.id) series_obj = self.gen_series_obj(psid, media, media_ext, storage, action) program_obj = self.gen_program_obj(ppid, media, media_ext, storage, episode, action) movie_obj = self.gen_movie_obj(pmid, media, storage, action, args) series_program_mapping = self.property_mapping("Series", "Program", psid, ppid, sequence=episode.serial_num, action=action) program_movie_mapping = self.property_mapping("Program", "Movie", ppid, pmid, action=action) data = { "BizDomain": "514", "object_list": [series_obj, program_obj, movie_obj], "mapping_list": [series_program_mapping, program_movie_mapping] } return data def program(self, cmd_id, action, *args): seriesId_hw, programId_hw, movieId_hw, seriesId_zx, programId_zx, movieId_zx = args task_id = cmd_id.split("_")[0] task = TaskModel.objects.filter(pk=task_id).select_related("media", "storage", "episode").first() media = task.media media_ext = media.mediaextfieldmodel storage = task.storage episode = task.episode data = self.gen_sub_program(media, media_ext, storage, episode, action, args) xml = Adi.to_xml(data, flag="Nope") return xml def post_api(self, correlate_id, cmd_id): headers = {"content-type": "text/xml", "SOAPAction": "application/soap+xml;charset=utf8"} ftp_url = self.get_cmd_ftp_url(self.csp_id, cmd_id) data = C2ApiUtil.REQUESTXML.format(self.csp_id, self.lsp_id, correlate_id, ftp_url) logging.warning("{} exec_sync_header_cmd HEADER_ID: {}, Request: {}".format(self.lsp_id, cmd_id, data)) try: content = requests.post(self.conf.cms_api, headers=headers, data=data).content logging.warning("{} exec_sync_header_cmd HEADER_ID: {}, Response: {}".format(self.lsp_id, cmd_id, content)) tree = etree.fromstring(content) code = tree.find(".//Result").text.strip() msg = tree.find(".//ErrorDescription").text.strip() except Exception as e: logging.error("exception:{}".format(e)) code = -1 msg = str(e) return code, msg def exec_cmd(self, task_id, cmd_id): correlate_id = self.generate_correlate_id(task_id, "common", datetime_now()) code, msg = self.post_api(correlate_id, cmd_id) if str(code) == "0": status = TaskStatusValue.INFO_TRANSFERING msg = "" else: status = TaskStatusValue.INFO_FAILED TaskModel.objects.filter(pk=task_id).update(status=status, remark=msg) return status def cdn_injector(self, task, action="REGIST"): # 访问 Task有几个，如果1个则，表示 是第一个,就带上海报地址 count = TaskModel.objects.filter(partner_id=self.lsp_pk, media_id=task.media_id).count() storage = task.storage path = self.sub_path_from_url(storage.hls_url, "=") params = {"path": path, "mid": task.media_id, "vid": task.episode_id} if count < 2: logging.warning("发起cdn请求，带海报，task_id:{}".format(task.id)) media_ext = task.media.mediaextfieldmodel 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) params["pic1"] = self.to_ip_url(media_verpic) params["pic2"] = self.to_ip_url(media_horpic) else: logging.info("发起cdn请求，不带海报，task_id:{}".format(task.id)) resp = requests.get(self.conf.transcode_api, params=params) status = TaskStatusValue.TRANS_CODE_FAILED try: if resp.status_code == 200: if resp.json()["code"] == 200: # 成功 status = TaskStatusValue.TRANS_CODE_TRANSFERING task.status = status else: status = TaskStatusValue.TRANS_CODE_FAILED task.status = status task.remark = resp.json()["msg"] except Exception as e: task.status = status task.remark = str(e) task.save() return status def gen_offical_time(self, _date): p_date_list = str(_date).split("-") p_date = "".join(p_date_list) return p_date + "000000" def gen_media_name_search_first_letter(self, name): return self.transform_chinese_to_letter(name).upper() def build_tv_list(self, media, storage_list): # 判断当前任务 _tv_list = list() for storage, episode in storage_list: episode_title = self.get_episode_title(media, episode), file_path = self.gen_movie_ftp_url(media.id, storage.hls_url) video_id = episode.id _storage = { "cpAlbumId": media.id, "cpTvId": video_id, "tvName": episode_title, "tvIsEffective": 1, "tvIsOnline": 1, "isPurchase": 0, "playOrder": episode.serial_num, "tvUrl": file_path, "tvDuration": "", "bitRate": "", "viceoFormat": "", "tvDesc": "", "tvPriority": 2, } _tv_list.append(_storage) return _tv_list def build_post_data(self, media, storage_list, media_ext, change_fields, csp_id): _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) media_horpic = self.image_resize.get_oss_resize_params(self.partner_alias, media_ext.media_horpic, size=media_ext.horpic_size, to_ip=True) _data = dict() tv_list = self.build_tv_list(media, storage_list) v_pic_name = media_verpic.split("/")[-1] h_pic_name = media_horpic.split("/")[-1] _data["cspId"] = csp_id _data["cpAlbumId"] = media.id _data["albumName"] = self.get_media_name_by_alias(media) _data["albumOfficalTime"] = media.date_published _data["tvSets"] = media.episodes _data["updateToSet"] = media.episodes _data["duration"] = media.total_seconds _data["price"] = 0 _data["isPurchase"] = change_fields["is_charge"] _data["isEffective"] = 1 _data["isOnline"] = 1 _data["director"] = directors _data["actors"] = actors _data["screenwriter"] = writers _data["place"] = media.regions _data["language"] = media.languages _data["releaseYear"] = media.year _data["isChoiceness"] = 1 _data["albumSearchName"] = self.gen_media_name_search_first_letter(self.get_media_name_by_alias(media)) _data["albumClassifys"] = "" _data["albumTypes"] = self._get_category_desc(change_fields["category"]) _data["programTypes"] = self.get_genre(media) _data["provider"] = self.content_provider _data["albumRecommendMsgs"] = media_ext.focus _data["albumDesc"] = media.storyline _data["picVertical"] = self.cdn_ftp_url.format(media.id, v_pic_name) _data["picCross"] = self.cdn_ftp_url.format(media.id, h_pic_name) _data["tvlist"] = tv_list return _data def one_requests(self, data, csp_id, action): logging.warning("start proxy_ininjector:{}".format(csp_id)) flag = "zx" if "shsytmsjzx" == csp_id else "hw" try: storage_api = self.conf.storage_api.format(cspId=csp_id) if action == "DELETE": storage_api = storage_api.replace("Data", "Delete") resp = requests.post(storage_api, json=data) logging.warning("proxy csp_id-resp:{} {}".format(flag, resp.json())) code = resp.json().get("code") if code == "A000000": status = TaskStatusValue.FILE_TRANSFERING remark = "{}-success".format(flag) else: status = TaskStatusValue.FILE_FAILED remark = "{}-fail".format(flag) except Exception as e: status = TaskStatusValue.FILE_FAILED remark = "{}-{}".format(flag, str(e)) return status, remark def proxy_injector(self, task, action): # 代理注入 media = task.media media_ext = media.mediaextfieldmodel # 暂时先不控制个数，而是查询所有 quyerset = TaskModel.objects.filter(partner_id=self.lsp_pk, media_id=task.media_id, status=TaskStatusValue.FILE_PENDING).select_related("storage", "episode") logging.warning("quyerset: {}".format(quyerset)) storage_list = [(task.storage, task.episode) for task in quyerset] param_dict = {"category": media.category, "is_charge": 1} change_fields = self.injected_change.get_injected_fields(media.id, self.lsp_pk, param_dict) data = self.build_post_data(media, storage_list, media_ext, change_fields, self.csp_id_hw) data_zx = self.build_post_data(media, storage_list, media_ext, change_fields, self.csp_id_zx) logging.warning("proxy_injector data: {}".format(data)) # task_cache = cache.get("proxy_task_{}".format(task.id)) # if task_cache: status, remark = self.one_requests(data, self.csp_id_hw, action) if status == TaskStatusValue.FILE_TRANSFERING: status, remark = self.one_requests(data_zx, self.csp_id_zx, action) # cache.set("proxy_task_{}".format(task.id), task.id, self.EXPIRE_TIME) if status == TaskStatusValue.FILE_TRANSFERING: for task in quyerset: # task.status = status # task.remark = remark # task.save() status = TaskStatusValue.INFO_PENDING task.status = status task.save() TaskQueue.update(task.level, task.id, status) return status def get_continue_to_injector(self, status): """获取是否继续注入流程flag，C2为一个流程,所以直接返回False""" return status not in TaskStatusValue.stop_injector() def query_proxy_result(self, media_id, video_id, csp_id): params = { "cspId": csp_id, "cpAlbumId": media_id, "cpTvId": video_id } url = "http://jscp.agency.gitv.tv/injectResult/v2/queryTv" r = requests.post(url, params=params) resp_json = r.json() logging.warning("csp_id:{} query_proxy_result:{}".format(csp_id, resp_json)) code = resp_json["code"] message = resp_json["message"] if code == "A000000": data = resp_json["data"] for item in data: cpAlbumId = item["cpAlbumId"] cpTvId = item["cpTvId"] taskStatus = item["taskStatus"] seriesId = item["seriesId"] programId = item["programId"] movieId = item["movieId"] return (taskStatus, seriesId, programId, movieId) return None def requests_soap(self, task, action, *args): # seriesId_hw, programId_hw, movieId_hw, seriesId_zx, programId_zx, movieId_zx = args _task_id = self.generate_correlate_id(task.id) self.save_program_xml_file(self.csp_id, _task_id, action, *args) status = self.exec_cmd(task.id, _task_id) return status def fusion_injector(self, task, action): t_status = task.status resp = self.query_proxy_result(task.media_id, task.episode_id, self.csp_id_hw) resp_zx = self.query_proxy_result(task.media_id, task.episode_id, self.csp_id_zx) status = status_zx = False remark = None if resp: taskStatus, seriesId_hw, programId_hw, movieId_hw = resp if taskStatus in (2, 6): status = True # if action == "DELETE" and taskStatus == 6: # status = True elif taskStatus in (-1, -3, -4, -6, -7, -9): # 表示 注入失败了 remark = "hw代理注入失败" t_status = TaskStatusValue.FILE_FAILED TaskModel.objects.filter(pk=task.id).update(status=t_status, remark=remark) TaskQueue.update(task.level, task.id, t_status) if resp_zx: taskStatus, seriesId_zx, programId_zx, movieId_zx = resp if taskStatus in (2, 6): status_zx = True # if action == "DELETE" and taskStatus == 6: # status_zx = True elif taskStatus in (-1, -3, -4, -6, -7, -9): # 表示 注入失败了 if remark is None: remark = "zh代理注入失败" else: remark = "hw-zx代理注入失败" t_status = TaskStatusValue.FILE_FAILED TaskModel.objects.filter(pk=task.id).update(status=t_status, remark=remark) TaskQueue.update(task.level, task.id, t_status) if status and status_zx: logging.warning("执行soap注入") args = (seriesId_hw, programId_hw, movieId_hw, seriesId_zx, programId_zx, movieId_zx) _status = self.requests_soap(task, action, *args) t_status = _status return t_status # 进行 融合注入 # r2 = requests.post(url, params=params_zx) # resp_json2 = r2.json() # logging.warning("resp_json_zx:{}".format(resp_json2)) # 一会解析，只要收不到id，就不不改变状态， 消息也不去掉。一直是信息注入发起状态 # 子集注入 def injector(self, task, action="REGIST"): status = 0 if task.status == TaskStatusValue.TRANS_CODE_PENDING: status = self.cdn_injector(task, action) if task.status in (TaskStatusValue.FILE_PENDING,): status = self.proxy_injector(task, action) if task.status in (TaskStatusValue.INFO_PENDING,): status = self.fusion_injector(task, action) flag = self.get_continue_to_injector(status) return status, flag def update(self, task): self.injector(task, action="UPDATE") self.update_storage(task.partner_id, task.storage_id) def remove(self, task): # 要先删除融合注入， status = TaskStatusValue.INFO_PENDING task.status = status task.save() # TaskQueue.update(task.level, task.id, status) self.injector(task, action="DELETE") self.off_line_storage(task.partner_id, task.storage_id) def callback_cdn(self, request_body): logging.warning("callback_cdn:{}".format(request_body)) resp_json = json.loads(request_body) code = resp_json["code"] msg = resp_json["msg"] vid = resp_json["vid"] task = TaskModel.objects.filter(partner_id=self.lsp_pk, episode_id=vid).order_by("-pk").first() if code == 200: # 表示上传文件成功 status = TaskStatusValue.FILE_PENDING task.remark = msg else: status = TaskStatusValue.TRANS_CODE_FAILED task.remark = msg task.status = status task.save() TaskQueue.update(task.level, task.id, status) print("callback_ftp: ", resp_json) return {"code": 1, "message": "success"}