import json import os import requests import logging from imdb.models import CelebrityModel, MediaModel, EpisodeModel, CelebrityMediaRelationModel from injector.models import MediaExtFieldModel, MediaStorageModel class MediaSyncStrategy: def get_flag(self): raise NotImplementedError("子类必须实现 get_flag 方法") def execute(self): raise NotImplementedError("子类必须实现 execute 方法") @classmethod def run_all_plugins(cls): """自动注册所有策略子类""" child_classes = cls.__subclasses__() strategy_map = {} for subclass in child_classes: instance = subclass() strategy_map[instance.get_flag()] = instance return strategy_map def to_ip_url(self, url: str): dn = 'senyu-img.oss-cn-hangzhou.aliyuncs.com' ip_port = '{}:{}'.format("47.110.47.81", "2152") return url.replace(dn, ip_port) class SyncAllStrategy(MediaSyncStrategy): def get_flag(self): return "all" def read_config_content(self): # 获取项目根目录路径 BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) config_path = os.path.join(BASE_DIR, 'config.json') # 读取配置文件 with open(config_path, 'r', encoding='utf-8') as f: config_data = json.load(f) # 访问字段 current_media_id = config_data.get('CURRENT_MEDIA_ID') current_actor_id = config_data.get('CURRENT_ACTOR_ID') params = {"media_id": current_media_id, "actor_id": current_actor_id} return params def write_config_content(self, actor_id, media_id): BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) config_path = os.path.join(BASE_DIR, 'config.json') # 读取配置文件 with open(config_path, 'r', encoding='utf-8') as f: config_data = json.load(f) if actor_id: config_data["CURRENT_ACTOR_ID"] = actor_id if media_id: config_data["CURRENT_MEDIA_ID"] = media_id with open(config_path, 'w', encoding='utf-8') as f: json.dump(config_data, f, ensure_ascii=False, indent=4) def execute(self, request): params = self.read_config_content() print(params) # params = {"media_id": 17520, "actor_id": 23390} data = self.query_last_media(**params) self.filter_result(data) self.save_result_to_db(data) self.save_current_Id(data) return {"code": 0, "message": "success", "data": data} def save_result_to_db(self, data): if len(data) == 0: return "没有数据" actor_list = data.get("actor_list") episode_list = data.get("episode_list") media_list = data.get("media_list") media_ext_list = data.get("media_ext_list") actor_media_list = data.get("actor_media_list") celebrity_objects = [] for actor in actor_list: celebrity = CelebrityModel( id=actor.get("id"), name=actor["name"], birthday=actor.get("birthday") or "", douban_id=actor.get("douban_id", 0), gender=actor.get("gender", ""), profession=actor.get("profession", "") ) celebrity_objects.append(celebrity) media_objects = [] for media in media_list: media_obj = MediaModel( id=media.get("id"), category=media.get("category", ""), name=media.get("name", ""), year=media.get("year", 0), # genres=media.get("genres", []), date_published=media.get("date_published", ""), douban_score=media.get("douban_score", ""), episodes=media.get("episodes"), seconds=media.get("seconds", 0), total_seconds=media.get("total_seconds", 0), regions=media.get("regions", []), languages=media.get("languages", ""), alias=media.get("alias", ""), imdb_id=media.get("imdb_id", ""), img=media.get("img", ""), storyline=media.get("storyline", ""), program_id=media.get("program_id", "0"), user_id=media.get("user_id", 0), hidden=media.get("hidden", False), is_divided=media.get("is_divided", False), partner_alias=media.get("partner_alias") ) media_objects.append(media_obj) episode_objects = [] for episode in episode_list: episode_obj = EpisodeModel( id=episode.get("id"), media_id=episode.get("media_id"), title=episode.get("title", ""), serial_num=episode.get("serial_num", 0), storyline=episode.get("storyline", ""), episode_horpic=episode.get("episode_horpic", ""), horpic_size=episode.get("horpic_size", ""), begin=episode.get("begin", ""), end=episode.get("end", "") ) episode_objects.append(episode_obj) media_ext_objects = [] for media_ext in media_ext_list: media_ext_obj = MediaExtFieldModel( id=media_ext.get("id"), media_id=media_ext.get("media_id"), media_verpic=media_ext.get("media_verpic", ""), verpic_size=media_ext.get("verpic_size", ""), verpic_length=media_ext.get("verpic_length", 0), media_horpic=media_ext.get("media_horpic", ""), horpic_size=media_ext.get("horpic_size", ""), horpic_length=media_ext.get("horpic_length", 0), rating=media_ext.get("rating", "Ⅰ"), focus=media_ext.get("focus", ""), approval_num=media_ext.get("approval_num", ""), media_verpic1=media_ext.get("media_verpic1", ""), verpic_size1=media_ext.get("verpic_size1", ""), media_horpic1=media_ext.get("media_horpic1", ""), horpic_size1=media_ext.get("horpic_size1", ""), format=media_ext.get("format", 1), definition=media_ext.get("definition", 1), len1=media_ext.get("len1", 0), len2=media_ext.get("len2", 0), update_frequency=media_ext.get("update_frequency", ""), report_id=media_ext.get("report_id", ""), prior_cdn_id=media_ext.get("prior_cdn_id", ""), copyright=media_ext.get("copyright", ""), playplat=media_ext.get("playplat", ""), telecom=media_ext.get("telecom", ""), icp=media_ext.get("icp", "") ) media_ext_objects.append(media_ext_obj) actor_media_objects = [] for relation in actor_media_list: relation_obj = CelebrityMediaRelationModel( id=relation.get("id"), media_id=relation.get("media_id"), celebrity_id=relation.get("celebrity_id"), role=relation.get("role", "") ) actor_media_objects.append(relation_obj) CelebrityModel.objects.bulk_create(celebrity_objects) MediaModel.objects.bulk_create(media_objects) EpisodeModel.objects.bulk_create(episode_objects) MediaExtFieldModel.objects.bulk_create(media_ext_objects) # CelebrityMediaRelationModel.objects.bulk_create(actor_media_objects) return "success" def query_last_media(self, media_id, actor_id): params = {"media_id": media_id, "actor_id": actor_id} resp = requests.post("http://47.110.47.81/injector/async_media_delta/", data=params) data = resp.json().get("data", {}) return data def save_current_Id(self, data): actor_list = data.get("actor_list") media_list = data.get("media_list") actor_last_id = None media_last_id = None if len(actor_list) > 0: actor_last_id = actor_list[-1]["id"] if len(media_list) > 0: media_last_id = media_list[-1]["id"] self.write_config_content(actor_last_id, media_last_id) def filter_result(self, data): if len(data) == 0: return "没有数据" media_ext_list = data.get("media_ext_list") episode_list = data.get("episode_list") # 修改海报地址 for media_ext in media_ext_list: media_ext["media_horpic"] = self.to_ip_url(media_ext["media_horpic"]) media_ext["media_verpic"] = self.to_ip_url(media_ext["media_verpic"]) for episode in episode_list: episode["episode_horpic"] = self.to_ip_url(episode["episode_horpic"]) class SyncOneStrategy(MediaSyncStrategy): def get_flag(self): return "one" def execute(self, request): obj_json = json.loads(request.body) media_id = obj_json.get("media_id") data = self.query_update_media(media_id) self.filter_result(data) logging.warning("data:{}".format(data)) self.update_media_to_db(data) return {"code": 0, "message": "success", "data": data} def query_update_media(self, media_id): params = {"media_id": media_id} resp = requests.post("http://47.110.47.81/injector/async_media_one/", data=params) data = resp.json().get("data", {}) return data def filter_result(self, data): if len(data) == 0: return "没有数据" media_ext_list = data.get("media_ext_list") episode_list = data.get("episode_list") # 修改海报地址 for media_ext in media_ext_list: media_ext["media_horpic"] = self.to_ip_url(media_ext["media_horpic"]) media_ext["media_verpic"] = self.to_ip_url(media_ext["media_verpic"]) for episode in episode_list: episode["episode_horpic"] = self.to_ip_url(episode["episode_horpic"]) def update_media_to_db(self, data): media_list = data.get("media_list") media_ext_list = data.get("media_ext_list") media_first = media_list[0] media_ext_first = media_ext_list[0] media_first.pop("genres") MediaModel.objects.update_or_create(id=media_first.get("id"), defaults=media_first) MediaExtFieldModel.objects.update_or_create(id=media_ext_first.get("id"), defaults=media_ext_first) class SyncStorageStrategy(MediaSyncStrategy): def get_flag(self): return "file" def execute(self, request): obj_json = json.loads(request.body) media_id = obj_json.get("media_id") data = self.query_storage_by_media(media_id) logging.warning("media_id data:{} ".format(media_id, data)) self.filter_result(data) self.save_storage_to_db(data) return {"code": 0, "message": "success"} def query_storage_by_media(self, media_id): params = {"media_id": media_id} resp = requests.post("http://47.110.47.81/injector/async_media_storage/", data=params) data = resp.json().get("data", {}) print(data) return data def filter_result(self, data): pass def save_storage_to_db(self, data): # 视频model 增加一个字段 storage_list = data.get("storages") origin_id_list = [storage.get("id") for storage in storage_list] existing_maps = {obj.origin_id: obj for obj in MediaStorageModel.objects.filter(origin_id__in=origin_id_list)} # 存在就更新，不存在就创建 create_list = [] update_list = [] for storage in storage_list: origin_id = storage.get("id") if origin_id in existing_maps: storage_obj = existing_maps[origin_id] for key, value in storage.items(): if key != "id" and hasattr(storage_obj, key): setattr(storage_obj, key, value) update_list.append(storage_obj) else: storage_obj = MediaStorageModel(origin_id=origin_id, **{k: v for k, v in storage.items() if k != "id"}) create_list.append(storage_obj) if create_list: MediaStorageModel.objects.bulk_create(create_list) if update_list: update_fields = list(filter(lambda x: x != "id", storage_list[0].keys())) MediaStorageModel.objects.bulk_update(update_list, update_fields, batch_size=1000) class MediaSyncStrategyFactory: """媒体同步策略工厂""" Strategy_Map = MediaSyncStrategy.run_all_plugins() @classmethod def register_strategy(cls, flag, strategy): """注册新策略""" cls.Strategy_Map[flag] = strategy @classmethod def execute(cls, flag, request): """执行策略""" strategy = cls.Strategy_Map.get(flag) if not strategy: return None return strategy.execute(request)