# -*- coding: utf-8 -*- # 4K影视插件 - 优化版本 import re import sys import json from pyquery import PyQuery as pq sys.path.append('..') from base.spider import Spider class Spider(Spider): def init(self, extend=""): pass def getName(self): return "4k影视" def isVideoFormat(self, url): pass def manualVideoCheck(self): pass def destroy(self): pass host = 'https://www.4kvm.org' headers = { 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate', 'Cache-Control': 'no-cache', 'Pragma': 'no-cache', 'Sec-Ch-Ua': '"Not/A)Brand";v="8", "Chromium";v="134", "Google Chrome";v="134"', 'Sec-Ch-Ua-Mobile': '?0', 'Sec-Ch-Ua-Platform': '"macOS"', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'none', 'Sec-Fetch-User': '?1', 'Upgrade-Insecure-Requests': '1', } # 公共方法 def _normalize_url(self, url): """标准化URL处理""" if not url: return url if url.startswith('//'): return f"https:{url}" elif url.startswith('/'): return f"{self.host}{url}" return url def _extract_video_basic(self, item): """提取视频基本信息""" try: link = self._normalize_url(item('a').attr('href') or item('h3 a').attr('href') or item('.data h3 a').attr('href')) if not link: return None title = (item('h3').text().strip() or item('.data h3').text().strip() or item('img').attr('alt') or item('a').attr('title') or '未知标题') img = self._normalize_url(item('img').attr('src') or item('img').attr('data-src')) # 简化备注提取 remarks = (item('.rating, .imdb, .vote').text().strip() or item('.year, .date, span').text().strip() or item('.type, .genre, .tag').text().strip()) return { 'vod_id': link, 'vod_name': title, 'vod_pic': img or '', 'vod_remarks': remarks, 'vod_year': '' } except: return None def _get_episode_count(self, season_data, page_html): """智能检测集数""" # 方法1: 精确容器检测 episode_container = season_data('.jujiepisodios') if episode_container: episode_links = episode_container('a') episode_numbers = [int(link.text().strip()) for link in episode_links.items() if link.text().strip().isdigit() and 1 <= int(link.text().strip()) <= 200] if episode_numbers: return max(episode_numbers) # 方法2: JavaScript数据提取 video_matches = re.findall(r'video.*?=.*?\[(.*?)\]', page_html, re.IGNORECASE | re.DOTALL) for match in video_matches: if '"name":' in match: episode_names = re.findall(r'"name"\s*:\s*(\d+)', match) if len(episode_names) >= 5: episode_numbers = sorted(set(int(name) for name in episode_names)) if episode_numbers[0] == 1 and episode_numbers[-1] - episode_numbers[0] == len(episode_numbers) - 1: return max(episode_numbers) # 方法3: 文本模式匹配 page_text = season_data.text() for pattern in [r'共(\d+)集', r'全(\d+)集', r'更新至(\d+)集', r'第(\d+)集']: matches = re.findall(pattern, page_text) if matches: return max(int(m) for m in matches if m.isdigit()) # 默认值 return 24 if season_data('iframe, video, .player') else 1 def homeContent(self, filter): try: data = self.getpq(self.fetch(self.host, headers=self.headers).text) classes = [] # 简化分类提取 nav_items = data('header .head-main-nav ul.main-header > li') for k in nav_items.items(): main_link = k.children('a').eq(0) link = main_link.attr('href') name = main_link.text().strip() if link and name and name not in ['首页', '影片下载']: link = self._normalize_url(link) class_info = {'type_name': name, 'type_id': link} if '电视剧' in name or 'tvshows' in link: class_info['filter_type'] = 'tvshows' classes.append(class_info) # 子分类 for sub_item in k('ul li').items(): sub_link = self._normalize_url(sub_item('a').attr('href')) sub_name = sub_item('a').text().strip() if sub_link and sub_name: sub_class_info = {'type_name': f"{name}-{sub_name}", 'type_id': sub_link} if '电视剧' in name or 'tvshows' in sub_link: sub_class_info['filter_type'] = 'tvshows' classes.append(sub_class_info) return {'class': classes, 'list': self.getHomeList(data)} except Exception as e: return {'class': [], 'list': []} def homeVideoContent(self): pass def categoryContent(self, tid, pg, filter, extend): try: url = tid if pg == '1' else f"{tid}/page/{pg}" if '?' not in tid else f"{tid}&page={pg}" data = self.getpq(self.fetch(url, headers=self.headers).text) video_list = self.getVideoList(data) if '电视剧' in url or 'tvshows' in url: video_list = self.filterTVShowsOnly(video_list) return {'list': video_list, 'page': int(pg), 'pagecount': 9999, 'limit': 30, 'total': 999999} except Exception as e: return {'list': [], 'page': int(pg), 'pagecount': 1, 'limit': 30, 'total': 0} def detailContent(self, ids): try: first_id = next(iter(ids)) if hasattr(ids, '__iter__') and not isinstance(ids, str) else ids data = self.getpq(self.fetch(first_id, headers=self.headers).text) # 基本信息提取 vod = { 'vod_id': first_id, 'vod_name': data('.sheader h1, h1').text().strip(), 'vod_pic': self._normalize_url(data('.sheader .poster img, .poster img').attr('src')), 'vod_content': data('.sbox .wp-content, #info .wp-content').text().strip(), 'vod_year': '', 'vod_area': '', 'vod_remarks': '', 'vod_actor': '', 'vod_director': '' } # 提取分类 genres = data('.sgeneros a') if genres: vod['type_name'] = ', '.join(g.text() for g in genres.items()) # 播放链接处理 play_options = data('#playeroptions ul li, .dooplay_player_option') if play_options: play_links = self._extract_play_options(play_options, first_id) else: season_links = data('.seasons-list a, .season-item a, .se-c a, .se-a a, .seasons a') play_links = self.getSeasonEpisodes(season_links) if season_links else [f"播放${first_id}"] vod['vod_play_from'] = '4K影视' vod['vod_play_url'] = '#'.join(play_links) return {'list': [vod]} except Exception as e: return {'list': []} def _extract_play_options(self, play_options, first_id): """提取播放选项""" play_links = [] for option in play_options.items(): title = option('.title, span.title').text().strip() or '播放' server = option('.server, span.server').text().strip() if server: title = f"{title}-{server}" data_post = option.attr('data-post') data_nume = option.attr('data-nume') data_type = option.attr('data-type') if data_post and data_nume: play_url = f"{first_id}?post={data_post}&nume={data_nume}&type={data_type}" play_links.append(f"{title}${play_url}") return play_links def searchContent(self, key, quick, pg="1"): try: search_url = f"{self.host}/xssearch?s={key}" if pg != "1": search_url += f"&p={pg}" data = self.getpq(self.fetch(search_url, headers=self.headers).text) raw_results = self.getVideoList(data) filtered_results = self.filterSearchResults(raw_results, key) return {'list': filtered_results, 'page': int(pg)} except Exception as e: return {'list': [], 'page': int(pg)} def playerContent(self, flag, id, vipFlags): try: # 解析播放参数 data_post = data_nume = data_type = None if '?' in id: base_url, params = id.split('?', 1) param_dict = dict(param.split('=', 1) for param in params.split('&') if '=' in param) data_post = param_dict.get('post') data_nume = param_dict.get('nume') data_type = param_dict.get('type') # API调用 if data_post and data_nume: try: api_url = f"{self.host}/wp-json/dooplayer/v1/post/{data_post}" api_response = self.fetch(api_url, headers=self.headers, params={'type': data_type or 'movie', 'source': data_nume}) if api_response.status_code == 200: api_data = api_response.json() if 'embed_url' in api_data: embed_url = api_data['embed_url'] parse_flag = 0 if any(ext in embed_url.lower() for ext in ['.m3u8', '.mp4', '.flv', '.avi']) else 1 return {'parse': parse_flag, 'url': embed_url, 'header': self.headers} except: pass # 页面解析回退 page_url = base_url if '?' in id else id data = self.getpq(self.fetch(page_url, headers=self.headers).text) # 查找播放源 iframe = data('iframe.metaframe, .dooplay_player iframe, .player iframe').attr('src') if iframe: iframe = self._normalize_url(iframe) parse_flag = 0 if any(ext in iframe.lower() for ext in ['.m3u8', '.mp4', '.flv']) else 1 return {'parse': parse_flag, 'url': iframe, 'header': self.headers} video_src = self._normalize_url(data('video source, video').attr('src')) if video_src: return {'parse': 0, 'url': video_src, 'header': self.headers} return {'parse': 1, 'url': page_url, 'header': self.headers} except Exception as e: return {'parse': 1, 'url': id, 'header': self.headers} def localProxy(self, param): pass def liveContent(self, url): pass def getHomeList(self, data): """获取首页推荐列表""" videos = [] items = data('article, .module .content .items .item, .movies-list article') for item in items.items(): video_info = self._extract_video_basic(item) if video_info: videos.append(video_info) return videos def getVideoList(self, data): """获取视频列表""" videos = [] items = data('article, .items article, .content article, .search-results article') for item in items.items(): video_info = self._extract_video_basic(item) if video_info: videos.append(video_info) return videos def extractVideoInfo(self, item): """兼容性方法，已优化为_extract_video_basic""" return self._extract_video_basic(item) def getpq(self, text): """创建PyQuery对象""" try: return pq(text) except: try: return pq(text.encode('utf-8')) except: return pq('') def filterSearchResults(self, results, search_key): """过滤和排序搜索结果""" if not results or not search_key: return results search_key_lower = search_key.lower().strip() search_words = search_key_lower.split() scored_results = [] for result in results: title = result.get('vod_name', '').lower() # 计算相关性分数 if search_key_lower == title: score = 100 elif search_key_lower in title: score = 80 elif title.startswith(search_key_lower): score = 70 elif all(word in title for word in search_words): score = 60 else: word_matches = sum(1 for word in search_words if word in title) if word_matches > 0: score = 30 + (word_matches * 10) else: continue # 内容类型加分 if '剧' in search_key_lower and 'tvshows' in result.get('vod_id', ''): score += 5 elif '电影' in search_key_lower and 'movies' in result.get('vod_id', ''): score += 5 scored_results.append((score, result)) # 排序和过滤 scored_results.sort(key=lambda x: x[0], reverse=True) min_score = 30 if len(search_words) > 1 else 40 filtered = [result for score, result in scored_results if score >= min_score] # 如果结果太少，放宽标准 if len(filtered) < 3 and len(scored_results) > 3: filtered = [result for score, result in scored_results[:10]] return filtered def filterTVShowsOnly(self, video_list): """过滤电视剧分类中的电影内容""" if not video_list: return video_list filtered_videos = [] movie_keywords = ['/movies/', '/movie/', 'Movie', '电影'] tvshow_keywords = ['/tvshows/', '/tvshow/', '/seasons/', 'TV', '剧', '季', '集'] for video in video_list: vod_id = video.get('vod_id', '') vod_name = video.get('vod_name', '') vod_remarks = video.get('vod_remarks', '') # 检查是否是电影 is_movie = any(keyword in vod_id for keyword in movie_keywords[:3]) if is_movie: continue # 检查是否是电视剧 is_tvshow = (any(keyword in vod_id for keyword in tvshow_keywords[:3]) or any(keyword in vod_name + vod_remarks for keyword in tvshow_keywords[3:])) if is_tvshow or not is_movie: filtered_videos.append(video) return filtered_videos def getSeasonEpisodes(self, season_links): """获取电视剧每个季的集数信息""" play_links = [] try: for season in season_links.items(): season_title = season.text().strip() or '第1季' season_url = self._normalize_url(season.attr('href')) if not season_url: continue try: season_resp = self.fetch(season_url, headers=self.headers) if season_resp.status_code == 200: season_data = self.getpq(season_resp.text) episode_count = self._get_episode_count(season_data, season_resp.text) # 限制集数范围（提高上限以支持长篇动漫） episode_count = min(max(episode_count, 1), 500) # 生成播放链接 if episode_count == 1: play_links.append(f"{season_title}${season_url}") else: clean_title = season_title.split('已完結')[0].split('更新')[0].strip() for ep_num in range(1, episode_count + 1): episode_title = f"{clean_title} 第{ep_num}集" episode_url = f"{season_url}?ep={ep_num}" play_links.append(f"{episode_title}${episode_url}") else: play_links.append(f"{season_title}${season_url}") except Exception: play_links.append(f"{season_title}${season_url}") except Exception: pass return play_links