#! /usr/bin/env python3 # -*- coding:utf-8 -*- import logging from collections import defaultdict from django.db.models import Max from apollo.settings import DEBUG from django.core.paginator import Paginator from domain.imdb.media_domain import Media from domain.injector.task_query_strategy import TaskQueryStrategyFactory from imdb.models import EpisodeEffectiveTimeModel from injector.forms import TaskModelForm, TaskCronModelForm from injector.models import TaskModel, MediaInfoSyncModel, MediaStorageModel, InjectedFieldChangeModel from injector.models import PartnerMediaStorageModel from domain.injector import factory from domain.dataset import DataSet from domain.lsp.lsp_domain import Lsp from domain.injector.value_object import TaskStatusValue, TaskModelActionValue, LspCommonValue from domain.injector.task_queue import TaskQueue from utils import date_utils from utils.exception import ParamError logger = logging.getLogger(__file__) class Task(DataSet): model = TaskModel @classmethod def get_pending_first(cls, state, count): items = cls.model.objects.order_by("-level", "id").filter(status=state).select_related("partner", "storage") return items[0: count] @classmethod def get_pending_list(cls): status = TaskStatusValue.pending() return cls.model.objects.filter(status__in=status).select_related("partner", "storage").order_by("-level") @classmethod def is_failed(cls, task_id): status = TaskStatusValue.failed() return cls.model.objects.filter(status__in=status, pk=task_id).exists() @classmethod def is_occupied_bandwidth(cls, task): """是否占用下载带宽""" if task.status == TaskStatusValue.TRANS_CODE_PENDING: return True elif task.status == TaskStatusValue.FILE_PENDING: return True return False @staticmethod def to_dict(task_obj): # storage = task_obj.storage partner = task_obj.partner try: is_show = int(PartnerMediaStorageModel.objects.filter( storage_id=task_obj.storage.id, partner_id=partner.id ).first().is_show) except AttributeError: is_show = 0 # 银河和广东的先不提供信息更新和删除 json_partner = [1, 7] if DEBUG else [1, 4] if partner.id in json_partner: in_pending = int(task_obj.status) not in ( TaskStatusValue.INFO_TRANSFERING, TaskStatusValue.INFO_COMPLETED, TaskStatusValue.INFO_FAILED ) else: in_pending = task_obj.status in TaskStatusValue.pending() # 临时 加的 分发判断，后面改成状态了，在删除了。 is_distribute = False if partner.alias == "SHANXIYIDONG" and task_obj.status == 23 and task_obj.remark != "分发成功": # 计算时间 # 这个计算时间 hours = 0.1 if DEBUG else 8 if date_utils.is_over_n_hours(task_obj.update_time, hours): is_distribute = True if task_obj.remark == "分发失败": is_distribute = True partner_domain = factory.create_by_id(task_obj.partner_id) ppid = "暂无" print("partner_domain:" + partner_domain.partner_alias) if partner_domain.partner_alias in LspCommonValue.LSPMOVIETAGLIST: ppid = partner_domain.gen_program_code_by_movie(task_obj.media) data = { "id": task_obj.id, "name": task_obj.storage.name, "media_id": task_obj.media_id, "media_name": task_obj.media.name, "episode_id": task_obj.episode_id, "episode_sn": task_obj.episode.serial_num, "episode_title": task_obj.title, "origin_name": task_obj.origin, "status": task_obj.status, "level": task_obj.level, "level_name": task_obj.get_level_display(), "remark": task_obj.remark, "storage_id": task_obj.replace if task_obj.replace and partner.id != 1 else task_obj.storage_id, "partner": Lsp.to_dict(partner), "is_show": is_show, # 是否在注入过程中 "in_pending": int(in_pending), "create_time": date_utils.dt_format(task_obj.create_time), "update_time": date_utils.dt_format(task_obj.update_time), "is_distribute": is_distribute, "action_desc": task_obj.action_description(), "is_online": task_obj.is_online() and task_obj.is_status_completed(), "is_offline": task_obj.is_offline() and task_obj.is_status_completed(), "info": ppid, } return data def active_is_show(self, task_obj): return task_obj.action == TaskModelActionValue.REGISTER and task_obj.status == TaskStatusValue.INFO_COMPLETED def is_active(self, task_obj): return task_obj.action == TaskModelActionValue.REGISTER and task_obj.status == TaskStatusValue.INFO_COMPLETED @classmethod def _get_list(cls, query, page, page_size): query = query.order_by("-id") pager = Paginator(query, page_size) obj_list = pager.get_page(page) task_list = [cls.to_dict(obj) for obj in obj_list] data = { "total_page": pager.num_pages, "data_list": task_list, "total_item": pager.count } return data @classmethod def get_list(cls, page: int = 1, page_size: int = 20): query = cls.model.objects.all().select_related("storage", "media", "episode", "partner") data = cls._get_list(query, page, page_size) return data @classmethod def get_query_params_common(cls, params, request): query_params = {} for key, default_value in params.items(): value = request.GET.get(key, default_value) if value is None or value == "" or value == 0: continue if key in ("task_status", "partner_id"): value = int(value) query_params[key] = value return query_params @classmethod def search_task_by_params(cls, params, page=1, page_size=20): # 动态构建查询参数 query_params = TaskQueryStrategyFactory.build(params) if not query_params: return cls.get_list(page, page_size) logging.warning("query_params:" + str(query_params)) query = cls.model.objects.filter(**query_params) data = cls._get_list(query, page, page_size) return data @classmethod def search(cls, partner_id=0, task_status=0, storage_name="", page=1, page_size=20, lsp_ids=None): if not (partner_id or task_status or storage_name or lsp_ids): return cls.get_list(page, page_size) query_data = {} if partner_id: query_data["partner_id"] = partner_id if task_status: query_data["status"] = int(task_status) if int(task_status) == 94: # 表示查的是remark 完成状态 分发失败 query_data["status"] = 23 query_data["remark__icontains"] = "分发失败" elif int(task_status) == 93: # 表示未分发 query_data["status"] = 23 query_data["remark__icontains"] = "SUCCESS" if storage_name: query_data["media__name__icontains"] = storage_name if lsp_ids: query_data["partner_id__in"] = lsp_ids query = cls.model.objects.filter(**query_data) data = cls._get_list(query, page, page_size) return data @staticmethod def init_status(partner): """任务初始状态""" if partner.conf.need_transcode: return TaskStatusValue.TRANS_CODE_PENDING return TaskStatusValue.FILE_PENDING @staticmethod def can_add(partner_id, media_id): partner_domain = factory.create_by_id(partner_id) if not partner_domain.conf.series_first: # 对剧头无要求 return True, None media = Media().get_by_id(media_id) if media.episodes == 1 and partner_domain.episode_no_header(media): # 电影类节目不需要处理剧头 return True, None item = MediaInfoSyncModel.objects.filter(partner_id=partner_id, media_id=media_id).first() if item and item.is_injector() and partner_domain.is_divide_series: return True, None # 对剧头有要求，查看剧头注入状态;若无记录，则先创建 if item and item.has_injected() and item.action != MediaInfoSyncModel.DELETE: return True, None return False, item @classmethod def add(cls, partner_id, storage_id, level, uid, **kwargs): # 数据库唯一键(partner_id, storage_id) partner_domain = factory.create_by_id(partner_id) partner_domain.check_args(storage_id) status = cls.init_status(partner_domain) storage = MediaStorageModel.objects.select_related("media", "episode").get(id=storage_id) flag, _ = cls.can_add(partner_id, storage.media_id) title = partner_domain.get_episode_title(storage.media, storage.episode) kwargs["title"] = title kwargs["origin_name"] = partner_domain.get_media_name_by_alias(storage.media) if flag: if level == TaskModel.CRON_LEVEL: cls.insert_cron_task(partner_id, storage, level, uid, status, **kwargs) else: cls.insert_task(partner_id, storage, level, uid, status, **kwargs) return flag @classmethod def insert_cron_task(cls, partner_id, storage, level, uid, status, **kwargs): # 校验合作方的必填字段;添加任务 info = { "media_id": storage.media_id, "episode_id": storage.episode_id, "storage_id": storage.id, "partner_id": partner_id, "level": level, "status": status, "create_user": uid, "origin": kwargs.get("origin_name", ""), "title": kwargs.get("title", "") } kwargs.pop("origin_name", None) kwargs.pop("title", None) cron_datetime = kwargs.get("datetime", None) # 新增任务 try: partner_domain = factory.create_by_id(partner_id) logger.warning("partner_domain: {}".format(partner_domain)) task = cls.model.objects.filter(partner_id=partner_id, storage_id=storage.id, level=7).first() if task: status = cls.init_status(partner_domain) update_time = date_utils.today_time() cls.model.objects.filter(id=task.id).update(status=status, create_time=update_time, update_time=update_time) else: task = cls.model.objects.create(**info) if int(partner_id) == 18: partner_domain.injector(task) else: print("定时任务 消息发送") if TaskQueue.has_field(level, task.id): raise ParamError("请勿重复发起定时任务") TaskQueue.add(level, task.id, cron_datetime) except ParamError as e: raise ParamError(e) except Exception as e: logger.warning("partner_domain error: {}".format(e)) raise ParamError(f"请勿重复发起注入任务:{e}") # 下片应在回调成功后更新is_show，目前不考虑下片 # PartnerMediaStorageModel.objects.create(partner_id=partner_id, storage_id=storage.id, is_show=1) PartnerMediaStorageModel.objects.update_or_create({"is_show": 1}, partner_id=partner_id, storage_id=storage.id) return @classmethod def insert_task(cls, partner_id, storage, level, uid, status, **kwargs): # 校验合作方的必填字段;添加任务 if int(partner_id) == 1: level = 8 TaskQueue.set_count_nx(level) l_task = TaskModel.objects.filter(partner_id=partner_id, episode_id=storage.episode_id).values( "replace").order_by("-create_time").first() # 注入一致 if not kwargs and l_task: print("kwargs:{}".format(kwargs)) kwargs["replace"] = l_task.get("replace") if int(partner_id) == 4: level = 9 TaskQueue.set_count_nx(level) if int(partner_id) == 25: level = 10 TaskQueue.set_count_nx(level) if int(partner_id) == 39: level = 11 TaskQueue.set_count_nx(level) info = { "media_id": storage.media_id, "episode_id": storage.episode_id, "storage_id": storage.id, "partner_id": partner_id, "level": level, "status": status, "create_user": uid, "origin": kwargs.get("origin_name", ""), "title": kwargs.get("title", "") } kwargs.pop("origin_name", None) kwargs.pop("title", None) try: if kwargs: for k, v in kwargs.items(): info[k] = v partner_domain = factory.create_by_id(partner_id) logger.warning("partner_domain: {}".format(partner_domain)) task = cls.model.objects.create(**info) if int(partner_id) == 18: partner_domain.injector(task) else: print("注入任务 消息发送") TaskQueue.add(level, task.id, status) except Exception as e: logger.warning("partner_domain error: {}".format(e)) raise ParamError("请勿重复发起注入任务") # 下片应在回调成功后更新is_show，目前不考虑下片 # PartnerMediaStorageModel.objects.create(partner_id=partner_id, storage_id=storage.id, is_show=1) PartnerMediaStorageModel.objects.update_or_create(partner_id=partner_id, storage_id=storage.id, defaults={"is_show": 1}) return @classmethod def re_add(cls, task_id, level): """重新注入""" task = cls.model.objects.select_related("media", "episode").filter(pk=task_id).first() # 校验合作方的必填字段;添加任务 partner_domain = factory.create_by_id(task.partner_id) partner_domain.check_args(task.storage_id) title = partner_domain.get_episode_title(task.media, task.episode) origin_name = partner_domain.get_media_name_by_alias(task.media) task.origin = origin_name task.title = title task.save() if task.status == TaskStatusValue.TRANS_CODE_FAILED: status = TaskStatusValue.TRANS_CODE_PENDING elif task.status == TaskStatusValue.FILE_FAILED: status = TaskStatusValue.FILE_PENDING else: status = TaskStatusValue.INFO_PENDING update_info = { "level": level, "status": status } cls.model.objects.filter(pk=task_id).update(**update_info) if task.partner_id == 18: task.level = level task.status = status partner_domain.injector(task) else: TaskQueue.update(level, task.id, status) return @classmethod def filter_last_task(cls, tasks): last_pk_dict = {} for task in tasks: create_time = task["create_time"] episode_id = task["episode_id"] replace = task["replace"] pk = task["pk"] if pk not in last_pk_dict: last_pk_dict[episode_id] = (create_time, pk, replace) else: if create_time > last_pk_dict[pk][0]: last_pk_dict[episode_id] = (create_time, pk, replace) if len(last_pk_dict) > 0: task_id_list = [(episode_id, val[1], val[2]) for episode_id, val in last_pk_dict.items()] return task_id_list @classmethod def re_replace_batch(cls, storage_ids, partner_id, level, uid): # # 根据介质ID 读取对应的分集ID，找到集合，[{'pk': 230656, 'episode_id': 259638}, {'pk': 230657, 'episode_id': 259639}] queryset = MediaStorageModel.objects.filter(pk__in=storage_ids).values_list("pk", "episode") # 字符为 数组记录pk episode_pk_dict = defaultdict(list) for item in queryset: episode_pk_dict[item[1]].append(item[0]) # 根据分集ID 与合作商 查找所有的Task，且只取 最近一次的task，这是先根据分集ID分组，在从每个分组 找出时间最新的task。 tasks = TaskModel.objects.filter(partner_id=1, episode_id__in=episode_pk_dict.keys()).values("pk", "episode_id", "create_time", "replace") # 遍历，获取 分集中 最新的 task 及 replace，这里默认多码率注入是 同时注入 或者替代，不会单独某码率替代，另一个不替代 task_list = cls.filter_last_task(tasks) # 根据 partner_domain = factory.create_by_id(partner_id) status = cls.init_status(partner_domain) if task_list: for episode_id, task_id, replace in task_list: storages = MediaStorageModel.objects.filter(pk__in=episode_pk_dict[episode_id]) # replace=replace 替代注入 要求一个分集一同更新 replace = cls.update_tag(replace, episode_id) print("============replace==========") for storage in storages: print(partner_id, storage, level, uid, status, replace) cls.insert_task(partner_id, storage, level, uid, status, replace=replace) @classmethod def update_tag(cls, replace, episode_id): if replace: return replace + "a" else: return "gitv{}a".format(episode_id) @classmethod def re_add_batch(cls, task_ids, level): """重新注入""" TaskModel.objects.filter(id__in=tuple(task_ids)).update(level=level) tasks = TaskModel.objects.select_related("media", "episode").filter(id__in=task_ids) print("level:{} tasks:{}".format(level, task_ids)) logger.info("level:{} tasks:{}".format(level, task_ids)) partner_domain = factory.create_by_id(tasks[0].partner_id) for task in tasks: # 校验合作方的必填字段;添加任务 partner_domain.check_args(task.storage_id) title = partner_domain.get_episode_title(task.media, task.episode) origin_name = partner_domain.get_media_name_by_alias(task.media) task.origin = origin_name task.title = title task.save() if task.status == TaskStatusValue.TRANS_CODE_FAILED: status = TaskStatusValue.TRANS_CODE_PENDING elif task.status == TaskStatusValue.FILE_FAILED: status = TaskStatusValue.FILE_PENDING else: status = TaskStatusValue.INFO_PENDING update_info = { "status": status, "level": level } cls.model.objects.filter(pk=task.id).update(**update_info) if task.partner_id == 18: task.status = status partner_domain.injector(task) else: TaskQueue.update(level, task.id, status) return @classmethod def task_replace_storage(cls, task_id, storage_id): """专为湖北移动 湖北移动百事通 设计的 介质替代功能""" task = cls.model.objects.filter(pk=task_id).first() # 校验合作方的必填字段;添加任务 partner_domain = factory.create_by_id(task.partner_id) partner_domain.check_args(task.storage_id) if task.status == TaskStatusValue.TRANS_CODE_FAILED: status = TaskStatusValue.TRANS_CODE_PENDING elif task.status == TaskStatusValue.FILE_FAILED: status = TaskStatusValue.FILE_PENDING else: status = TaskStatusValue.INFO_PENDING update_info = { "status": status, "replace": storage_id } cls.model.objects.filter(pk=task_id).update(**update_info) TaskQueue.update(task.level, task.id, status) return def update(self, query_info, update_info): self.model.objects.filter(**query_info).update(**update_info) def insert(self, data): self.model(**data).save() @staticmethod def to_cron_dict(task_obj, datetime=None): data = { "id": task_obj.id, "name": task_obj.storage.name, "media_id": task_obj.media_id, "media_name": task_obj.media.name, "level": task_obj.level, "level_name": task_obj.get_level_display(), "partner": Lsp.to_dict(task_obj.partner), "datetime": datetime, "create_time": date_utils.dt_format(task_obj.create_time) } return data @classmethod def get_cron_by_datetime(cls, date=None): cron_task_dict = TaskQueue.get_cron_queue() if date: date_cron_task_dict = {} for task_id, cron_datetime in cron_task_dict.items(): if cron_datetime.startswith(date): date_cron_task_dict[task_id] = cron_datetime return date_cron_task_dict return cron_task_dict @classmethod def delete_cron_task(cls, level, task_id): TaskQueue.delete(level, task_id) @classmethod def update_cron_task(cls, level, task_id, cron_datetime): res = TaskQueue.has_field(level, task_id) if res: TaskQueue.update(level, task_id, cron_datetime) return res @classmethod def get_cron_list(cls, partner_id, datetime, media_name, page=1, page_size=20, lsp_ids=None): query_data = {} if partner_id: query_data["partner_id"] = partner_id if media_name: query_data["media__name__icontains"] = media_name if lsp_ids: query_data["partner_id__in"] = lsp_ids cron_task_dict = cls.get_cron_by_datetime(datetime) task_ids = cron_task_dict.keys() queryset = TaskModel.objects.filter(id__in=task_ids).filter(**query_data).order_by("-id") pager = Paginator(queryset, page_size) page_task_list = pager.get_page(page) task_list = [cls.to_cron_dict(task, cron_task_dict.get(task.id)) for task in page_task_list] data = { "total_page": pager.num_pages, "data_list": task_list, "total_item": pager.count } return data @classmethod def form_data(cls, jobj): level = int(jobj.get("level", 0)) if level == TaskModel.CRON_LEVEL: form = TaskCronModelForm(jobj) else: form = TaskModelForm(jobj) return form @classmethod def distribute_task(cls, task_id): task = TaskModel.objects.filter(pk=task_id).first() partner_domain = factory.create_by_id(task.partner_id) partner_domain.exec_distribute(task, "register") @classmethod def handle_hls(cls): # 首先 判断是否存在， 如果不存在 就去240 生成切片 pass @classmethod def add_task_update(cls, params): if "params" in params: info_ids = params.get("params").get("tasks") else: info_ids = params.get("tasks") cls.model.objects.filter(id__in=info_ids).update(status=TaskStatusValue.INFO_PENDING) task = TaskModel.objects.select_related("media").filter(pk=info_ids[0]).first() category = params.get("category", None) is_charge = params.get("is_charge", None) times = params.get("times", None) partner_domain = factory.create_by_id(task.partner_id) if category: InjectedFieldChangeModel.objects.filter(media_id=task.media_id, partner_id=task.partner_id).update( category=category, is_charge=is_charge) if times: time_obj = EpisodeEffectiveTimeModel.objects.filter(media_id=task.media_id, partner_id=task.partner_id).first() if time_obj: effective_times = time_obj.effective_times for key in effective_times: effective_times[key] = times time_obj.effective_times = effective_times time_obj.save() tasks = TaskModel.objects.select_related("media").filter(pk__in=info_ids) for task_obj in tasks: title = partner_domain.get_episode_title(task_obj.media, task_obj.episode) origin_name = partner_domain.get_media_name_by_alias(task_obj.media) task.origin = origin_name task.title = title task.save() for task_id in info_ids: TaskQueue.update(21, task_id, TaskStatusValue.INFO_PENDING) @classmethod def add_task_del(cls, tasks): cls.model.objects.filter(id__in=tasks).update(status=TaskStatusValue.INFO_PENDING) for task_id in tasks: TaskQueue.update(20, task_id, TaskStatusValue.INFO_PENDING) @classmethod def get_all_status(cls): status_dict = dict(TaskModel.STATUS_CHOICES) status_dict[93] = "未分发" status_dict[94] = "分发失败" data = [{"id": key, "name": val} for key, val in status_dict.items()] for item in data: if item["name"].__contains__("失败"): item["refill"] = True return data @classmethod def get_task_extra_info(cls, task_id, partner_id): data = {"text": "空内容"} return data @classmethod def task_to_online(cls, task_id, level): task = TaskModel.objects.filter(id=task_id).first() partner_domain = factory.create_by_id(task.partner_id) res = partner_domain.active_task(task) if res == TaskModelActionValue.ACTIVE_NO_EXIST: raise ParamError("该平台没有下线功能") return res @classmethod def task_to_offline(cls, task_id, level): task = TaskModel.objects.filter(id=task_id).first() partner_domain = factory.create_by_id(task.partner_id) res = partner_domain.un_active_task(task) if res == TaskModelActionValue.ACTIVE_NO_EXIST: raise ParamError("该平台没有上线功能") return res