import time from domain.injector.factory import create_by_id from domain.injector.value_object import TaskStatusValue from imdb.models import MediaModel from injector.models import LSP, TaskModel, PartnerMediaStorageModel from django_redis import get_redis_connection from apollo.settings import DEBUG import threading import traceback class MediaOffline: KEY1 = "md_{}_{}" # name:media_id KEY2 = "st_{}_{}" # name:status KEY3 = "pt_{}" # partner_id:key1 FIELD_OK = "all" LIST = "nls_{}_{}" MSG = {"0": "待执行", "1": "任务已执行毕", "2": "任务执行失败", "3": "任务全部完成", "4": "任务执行未完成", "5": "节目已执行", "6": "节目不存在"} def __init__(self, uid, partner=None, name_list=[]): self.name_list = name_list self.partner = partner self.uid = uid self.conn = get_redis_connection("default") def encode(self, s): return s.encode("utf-8") def decode(self, bs): return bs.decode("utf-8") def query_db_by_name(self): data = MediaModel.objects.filter(name__in=self.name_list).values("id", "name") partner_id = self.partner r_dict = {} for item in data: if item['name'] in r_dict: count = TaskModel.objects.filter(partner_id=partner_id, media_id=item['id']).count() # 防止多md的情况 if count > 0: r_dict[item['name']] = item['id'] else: r_dict[item['name']] = item['id'] n_dict = {} for name in self.name_list: if name not in r_dict: n_dict[name] = "0" else: n_dict[name] = r_dict[name] return n_dict def is_storage_del(self, s_id, p_id): ps = PartnerMediaStorageModel.objects.filter(storage_id=s_id, partner_id=p_id).first() if ps and ps.is_show: return False return True def do_remove(self, task_list): count1 = 0 count2 = 0 for task in task_list: # 判断该介质是否已经删除 if self.is_storage_del(task.storage_id, task.partner_id) and task.status == TaskStatusValue.INFO_COMPLETED: count1 += 1 else: self.remove(task.id) if DEBUG: time.sleep(3) TaskModel.objects.filter(pk=task.id).update(status=TaskStatusValue.INFO_COMPLETED) count2 += 1 return count1, count2 def offline(self, r_dict, partner_id): for key, media_id in r_dict.items(): print("key:{}".format(key)) task_list = TaskModel.objects.filter(media_id=media_id, partner_id=partner_id) print(task_list) if task_list: try: count, _ = self.do_remove(task_list) if count == len(task_list): self.set_hash(MediaOffline.KEY2.format(self.uid, self.partner), self.encode(key), "1") else: self.set_hash(MediaOffline.KEY2.format(self.uid, self.partner), self.encode(key), "5") except Exception as e: print(e) traceback.print_exc() print("执行任务时，失败:{}".format(key)) self.set_hash(MediaOffline.KEY2.format(self.uid, self.partner), self.encode(key), "2") # 节目下载失败 else: self.set_hash(MediaOffline.KEY2.format(self.uid, self.partner), self.encode(key), "6") # 没有找到任务，节目不存在 print("没有找到任务，节目有错误:{}".format(key)) self.set_hash(MediaOffline.KEY2.format(self.uid, self.partner), MediaOffline.FIELD_OK, "1") # 所有任务都执行完 s_dict = self.conn.hgetall(MediaOffline.KEY2.format(self.uid, partner_id)) print(s_dict) return True, "下架完成" def remove(self, task_id): update_dic = {"status": TaskStatusValue.INFO_PENDING} TaskModel.objects.filter(id=task_id).update(**update_dic) task = TaskModel.objects.filter(id=task_id).first() partner_domain = create_by_id(task.partner_id) partner_domain.check_args(task.storage_id) partner_domain.remove(task) print("下架完成:{}:{}".format(task_id, task.media.name)) def set_hash(self, key, field, val): self.conn.hset(key, field, val) self.conn.expire(key, 60 * 60 * 24) def del_hash(self, key): self.conn.delete(key) def afresh(self): print("afresh" * 30) n_dict = self.conn.hgetall(MediaOffline.KEY1.format(self.uid, self.partner)) s_dict = self.conn.hgetall(MediaOffline.KEY2.format(self.uid, self.partner)) r_dict = {} for n, i in n_dict.items(): if i != "0" and s_dict[n] == "0": r_dict[self.decode(n)] = i return self.offline(r_dict, self.partner) def ready(self): self.del_hash(MediaOffline.KEY1.format(self.uid, self.partner)) self.del_hash(MediaOffline.KEY2.format(self.uid, self.partner)) r_dict = self.query_db_by_name() print(r_dict) # 结果存储到redis中 # 存储id：name 存储 id：状态 # 存储 key到列表中 for k, v in r_dict.items(): self.set_hash(MediaOffline.KEY1.format(self.uid, self.partner), self.encode(k), v) self.set_hash(MediaOffline.KEY2.format(self.uid, self.partner), self.encode(k), "0") self.set_hash(MediaOffline.KEY3.format(self.uid), self.partner, MediaOffline.KEY1.format(self.uid, self.partner)) threading.Thread(target=self.offline, args=(r_dict, self.partner)).start() # return self.offline(r_dict, self.partner) return True, "开始下架" def worker(self, going=False): if going: return self.afresh() if self.conn.hexists(MediaOffline.KEY3.format(self.uid), self.partner) and not self.conn.hexists( MediaOffline.KEY2.format(self.uid, self.partner), MediaOffline.FIELD_OK): return False, "该平台还有未完成的节目下线，请等待..." return self.ready() def offline_list(self): partner_dic = self.conn.hgetall(MediaOffline.KEY3.format(self.uid)) data = [] status = True for p, k in partner_dic.items(): p_name = LSP.objects.filter(id=p).first().name md_dict = self.conn.hgetall(k) s_dict = self.conn.hgetall(MediaOffline.KEY2.format(self.uid, self.decode(p))) for n, i in md_dict.items(): data.append( {"partner": p_name, "partner_id": self.decode(p), "name": self.decode(n), "status": self.decode(s_dict.get(n, "0"))}) status = status & bool( self.conn.hget(MediaOffline.KEY2.format(self.uid, self.decode(p)), MediaOffline.FIELD_OK)) status = 3 if status else 4 return data, status def clear_data(self, uid): partner_dic = self.conn.hgetall(MediaOffline.KEY3.format(uid)) for p, k in partner_dic.items(): self.conn.delete(k) self.conn.delete(MediaOffline.KEY2.format(uid, self.decode(p))) self.conn.delete(MediaOffline.KEY3.format(uid))