1124

肥猫/api/HMDJ.py

@@ -1,380 +1 @@
-# -*- coding: utf-8 -*-
-import requests
-import re
-import json
-import traceback
-import sys
-from urllib.parse import quote
-
-sys.path.append('../../')
-try:
-    from base.spider import Spider
-except ImportError:
-    # 定义一个基础接口类，用于本地测试
-    class Spider:
-        def init(self, extend=""):
-            pass
-
-class Spider(Spider):
-    def __init__(self):
-        self.siteUrl = "https://www.kuaikaw.cn"
-        self.cateManual = {
-            "甜宠": "462",
-            "古装仙侠": "1102",
-            "现代言情": "1145",
-            "青春": "1170",
-            "豪门恩怨": "585",
-            "逆袭": "417-464",
-            "重生": "439-465",
-            "系统": "1159",
-            "总裁": "1147",
-            "职场商战": "943"
-        }
-        self.headers = {
-            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0",
-            "Referer": self.siteUrl,
-            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
-            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
-        }
-        
-    def getName(self):
-        return "河马短剧"
-    
-    def init(self, extend=""):
-        return
-    
-    def fetch(self, url, headers=None, retry=2):
-        """统一的网络请求接口"""
-        if headers is None:
-            headers = self.headers
-        
-        for i in range(retry + 1):
-            try:
-                response = requests.get(url, headers=headers, timeout=10, allow_redirects=True)
-                response.raise_for_status()
-                return response
-            except Exception as e:
-                if i == retry:
-                    print(f"请求异常: {url}, 错误: {str(e)}")
-                    return None
-                continue
-    
-    def isVideoFormat(self, url):
-        video_formats = ['.mp4', '.mkv', '.avi', '.wmv', '.m3u8', '.flv', '.rmvb']
-        return any(format in url.lower() for format in video_formats)
-    
-    def manualVideoCheck(self):
-        return False
-    
-    def homeContent(self, filter):
-        result = {}
-        classes = [{'type_name': k, 'type_id': v} for k, v in self.cateManual.items()]
-        result['class'] = classes
-        
-        try:
-            result['list'] = self.homeVideoContent()['list']
-        except:
-            result['list'] = []
-        return result
-    
-    def homeVideoContent(self):
-        videos = []
-        try:
-            response = self.fetch(self.siteUrl)
-            if not response:
-                return {'list': []}
-                
-            html_content = response.text
-            next_data_pattern = r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>'
-            next_data_match = re.search(next_data_pattern, html_content, re.DOTALL)
-            if not next_data_match:
-                return {'list': []}
-                
-            next_data_json = json.loads(next_data_match.group(1))
-            page_props = next_data_json.get("props", {}).get("pageProps", {})
-            
-            # 处理轮播图数据
-            if "bannerList" in page_props:
-                for banner in page_props["bannerList"]:
-                    if banner.get("bookId"):
-                        videos.append({
-                            "vod_id": f"/drama/{banner['bookId']}",
-                            "vod_name": banner.get("bookName", ""),
-                            "vod_pic": banner.get("coverWap", ""),
-                            "vod_remarks": f"{banner.get('statusDesc', '')} {banner.get('totalChapterNum', '')}集".strip()
-                        })
-            
-            # 处理SEO分类推荐
-            if "seoColumnVos" in page_props:
-                for column in page_props["seoColumnVos"]:
-                    for book in column.get("bookInfos", []):
-                        if book.get("bookId"):
-                            videos.append({
-                                "vod_id": f"/drama/{book['bookId']}",
-                                "vod_name": book.get("bookName", ""),
-                                "vod_pic": book.get("coverWap", ""),
-                                "vod_remarks": f"{book.get('statusDesc', '')} {book.get('totalChapterNum', '')}集".strip()
-                            })
-            
-            # 去重处理
-            seen = set()
-            unique_videos = []
-            for video in videos:
-                key = (video["vod_id"], video["vod_name"])
-                if key not in seen:
-                    seen.add(key)
-                    unique_videos.append(video)
-        
-        except Exception as e:
-            print(f"获取首页推荐内容出错: {e}")
-            unique_videos = []
-        
-        return {'list': unique_videos}
-    
-    def categoryContent(self, tid, pg, filter, extend):
-        result = {'list': [], 'page': pg, 'pagecount': 1, 'limit': 20, 'total': 0}
-        url = f"{self.siteUrl}/browse/{tid}/{pg}"
-        
-        response = self.fetch(url)
-        if not response:
-            return result
-            
-        html_content = response.text
-        next_data_match = re.search(r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>', html_content, re.DOTALL)
-        if not next_data_match:
-            return result
-            
-        try:
-            next_data_json = json.loads(next_data_match.group(1))
-            page_props = next_data_json.get("props", {}).get("pageProps", {})
-            
-            current_page = page_props.get("page", 1)
-            total_pages = page_props.get("pages", 1)
-            book_list = page_props.get("bookList", [])
-            
-            videos = []
-            for book in book_list:
-                if book.get("bookId"):
-                    videos.append({
-                        "vod_id": f"/drama/{book['bookId']}",
-                        "vod_name": book.get("bookName", ""),
-                        "vod_pic": book.get("coverWap", ""),
-                        "vod_remarks": f"{book.get('statusDesc', '')} {book.get('totalChapterNum', '')}集".strip()
-                    })
-            
-            result.update({
-                'list': videos,
-                'page': int(current_page),
-                'pagecount': total_pages,
-                'limit': len(videos),
-                'total': len(videos) * total_pages if videos else 0
-            })
-            
-        except Exception as e:
-            print(f"分类内容获取出错: {e}")
-            
-        return result
-    
-    def searchContent(self, key, quick, pg=1):
-        return self.searchContentPage(key, quick, pg)
-    
-    def searchContentPage(self, key, quick, pg=1):
-        result = {'list': [], 'page': pg, 'pagecount': 1, 'limit': 20, 'total': 0}
-        search_url = f"{self.siteUrl}/search?searchValue={quote(key)}&page={pg}"
-        
-        response = self.fetch(search_url)
-        if not response:
-            return result
-            
-        html_content = response.text
-        next_data_match = re.search(r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>', html_content, re.DOTALL)
-        if not next_data_match:
-            return result
-            
-        try:
-            next_data_json = json.loads(next_data_match.group(1))
-            page_props = next_data_json.get("props", {}).get("pageProps", {})
-            
-            total_pages = page_props.get("pages", 1)
-            book_list = page_props.get("bookList", [])
-            
-            videos = []
-            for book in book_list:
-                if book.get("bookId"):
-                    videos.append({
-                        "vod_id": f"/drama/{book['bookId']}",
-                        "vod_name": book.get("bookName", ""),
-                        "vod_pic": book.get("coverWap", ""),
-                        "vod_remarks": f"{book.get('statusDesc', '')} {book.get('totalChapterNum', '')}集".strip()
-                    })
-            
-            result.update({
-                'list': videos,
-                'pagecount': total_pages,
-                'total': len(videos) * total_pages if videos else 0
-            })
-            
-        except Exception as e:
-            print(f"搜索内容出错: {e}")
-            
-        return result
-
-    def detailContent(self, ids):
-        result = {'list': []}
-        if not ids:
-            return result
-            
-        vod_id = ids[0]
-        if not vod_id.startswith('/drama/'):
-            vod_id = f'/drama/{vod_id}'
-        
-        drama_url = f"{self.siteUrl}{vod_id}"
-        response = self.fetch(drama_url)
-        if not response:
-            return result
-            
-        html = response.text
-        next_data_match = re.search(r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>', html, re.DOTALL)
-        if not next_data_match:
-            return result
-            
-        try:
-            next_data = json.loads(next_data_match.group(1))
-            page_props = next_data.get("props", {}).get("pageProps", {})
-            book_info = page_props.get("bookInfoVo", {})
-            chapter_list = page_props.get("chapterList", [])
-            
-            if not book_info.get("bookId"):
-                return result
-            
-            # 基本信息
-            categories = [c.get("name", "") for c in book_info.get("categoryList", [])]
-            performers = [p.get("name", "") for p in book_info.get("performerList", [])]
-            
-            vod = {
-                "vod_id": vod_id,
-                "vod_name": book_info.get("title", ""),
-                "vod_pic": book_info.get("coverWap", ""),
-                "type_name": ",".join(categories),
-                "vod_year": "",
-                "vod_area": book_info.get("countryName", ""),
-                "vod_remarks": f"{book_info.get('statusDesc', '')} {book_info.get('totalChapterNum', '')}集".strip(),
-                "vod_actor": ", ".join(performers),
-                "vod_director": "",
-                "vod_content": book_info.get("introduction", "")
-            }
-            
-            # 处理剧集
-            play_urls = self.processEpisodes(vod_id, chapter_list)
-            if play_urls:
-                vod['vod_play_from'] = '河马剧场'
-                vod['vod_play_url'] = '$$$'.join(play_urls)
-            
-            result['list'] = [vod]
-            
-        except Exception as e:
-            print(f"详情页解析出错: {e}")
-            traceback.print_exc()
-            
-        return result
-    
-    def processEpisodes(self, vod_id, chapter_list):
-        play_urls = []
-        episodes = []
-        
-        for chapter in chapter_list:
-            chapter_id = chapter.get("chapterId", "")
-            chapter_name = chapter.get("chapterName", "")
-            
-            if not chapter_id or not chapter_name:
-                continue
-                
-            # 尝试获取直接视频链接
-            video_url = self.getDirectVideoUrl(chapter)
-            if video_url:
-                episodes.append(f"{chapter_name}${video_url}")
-                continue
-                
-            # 回退方案
-            episodes.append(f"{chapter_name}${vod_id}${chapter_id}${chapter_name}")
-        
-        if episodes:
-            play_urls.append("#".join(episodes))
-        
-        return play_urls
-    
-    def getDirectVideoUrl(self, chapter):
-        if "chapterVideoVo" not in chapter or not chapter["chapterVideoVo"]:
-            return None
-            
-        video_info = chapter["chapterVideoVo"]
-        for key in ["mp4", "mp4720p", "vodMp4Url"]:
-            if key in video_info and video_info[key] and ".mp4" in video_info[key].lower():
-                return video_info[key]
-        return None
-
-    def playerContent(self, flag, id, vipFlags):
-        result = {
-            "parse": 0,
-            "url": id,
-            "header": json.dumps(self.headers)
-        }
-        
-        # 如果已经是视频链接直接返回
-        if 'http' in id and ('.mp4' in id or '.m3u8' in id):
-            return result
-            
-        # 解析参数
-        parts = id.split('$')
-        if len(parts) < 2:
-            return result
-            
-        drama_id = parts[0].replace('/drama/', '')
-        chapter_id = parts[1]
-        
-        # 尝试获取视频链接
-        video_url = self.getEpisodeVideoUrl(drama_id, chapter_id)
-        if video_url:
-            result["url"] = video_url
-            
-        return result
-    
-    def getEpisodeVideoUrl(self, drama_id, chapter_id):
-        episode_url = f"{self.siteUrl}/episode/{drama_id}/{chapter_id}"
-        response = self.fetch(episode_url)
-        if not response:
-            return None
-            
-        html = response.text
-        
-        # 方法1: 从NEXT_DATA提取
-        next_data_match = re.search(r'<script id="__NEXT_DATA__".*?>(.*?)</script>', html, re.DOTALL)
-        if next_data_match:
-            try:
-                next_data = json.loads(next_data_match.group(1))
-                page_props = next_data.get("props", {}).get("pageProps", {})
-                chapter_info = page_props.get("chapterInfo", {})
-                
-                if chapter_info and "chapterVideoVo" in chapter_info:
-                    video_info = chapter_info["chapterVideoVo"]
-                    for key in ["mp4", "mp4720p", "vodMp4Url"]:
-                        if key in video_info and video_info[key] and ".mp4" in video_info[key].lower():
-                            return video_info[key]
-            except:
-                pass
-                
-        # 方法2: 直接从HTML提取
-        mp4_matches = re.findall(r'(https?://[^"\']+\.mp4)', html)
-        if mp4_matches:
-            for url in mp4_matches:
-                if chapter_id in url or drama_id in url:
-                    return url
-            return mp4_matches[0]
-            
-        return None
-
-    def localProxy(self, param):
-        return [200, "video/MP2T", {}, param]
-
-    def destroy(self):
-        pass
+[session-a2446591] Route error: The Repository has been blocked. see: https://gitee.com/PizazzXS/another-d

肥猫/api/LSYS.py

@@ -1,279 +1 @@
-# coding=utf-8
-# !/usr/bin/python
-# by嗷呜(finally)
-import sys
-import os
-sys.path.append("..")
-import re
-import hashlib
-import hmac
-import random
-import string
-from Crypto.Util.Padding import unpad
-from concurrent.futures import ThreadPoolExecutor
-from Crypto.PublicKey import RSA
-from Crypto.Cipher import PKCS1_v1_5, AES
-from base64 import b64encode, b64decode
-import json
-import time
-from base.spider import Spider
-
-class Spider(Spider):
-
-    def getName(self):
-        return "电影猎手"
-
-    def init(self, extend=""):
-        self.device = self.device_id()
-        self.host = self.gethost()
-        pass
-
-    def isVideoFormat(self, url):
-        pass
-
-    def manualVideoCheck(self):
-        pass
-
-    def action(self, action):
-        pass
-
-    def destroy(self):
-        pass
-
-    t = str(int(time.time()))
-
-    def homeContent(self, filter):
-        result = {}
-        filters = {}
-        classes = []
-        bba = self.url()
-        data = self.fetch(f"{self.host}/api/v1/app/config?pack={bba[0]}&signature={bba[1]}", headers=self.header()).text
-        data1 = self.aes(data)
-        dy = {"class":"类型","area":"地区","lang":"语言","year":"年份","letter":"字母","by":"排序","sort":"排序"}
-        data1['data']['movie_screen']['sort'].pop(0)
-        for item in data1['data']['movie_screen']['sort']:
-            item['n'] = item.pop('name')
-            item['v'] = item.pop('value')
-        for item in data1['data']['movie_screen']['filter']:
-            has_non_empty_field = False
-            classes.append({"type_name": item["name"], "type_id": str(item["id"])})
-            for key in dy:
-                if key in item and item[key]:
-                    has_non_empty_field = True
-                    break
-            if has_non_empty_field:
-                filters[str(item["id"])] = []
-                filters[str(item["id"])].append(
-                    {"key": 'sort', "name": '排序', "value": data1['data']['movie_screen']['sort']})
-                for dkey in item:
-                    if dkey in dy and item[dkey]:
-                        item[dkey].pop(0)
-                        value_array = [
-                            {"n": value.strip(), "v": value.strip()}
-                            for value in item[dkey]
-                            if value.strip() != ""
-                        ]
-                        filters[str(item["id"])].append(
-                            {"key": dkey, "name": dy[dkey], "value": value_array}
-                        )
-        result["class"] = classes
-        result["filters"] = filters
-        return result
-
-    def homeVideoContent(self):
-        bba = self.url()
-        url = f'{self.host}/api/v1/movie/index_recommend?pack={bba[0]}&signature={bba[1]}'
-        data = self.fetch(url, headers=self.header()).json()
-        videos = []
-        for item in data['data']:
-            if len(item['list']) > 0:
-                for it in item['list']:
-                    try:
-                        videos.append(self.voides(it))
-                    except Exception as e:
-                        continue
-        result = {"list": videos}
-        return result
-
-    def categoryContent(self, tid, pg, filter, extend):
-        body = {"type_id": tid, "sort": extend.get("sort", "by_default"), "class": extend.get("class", "类型"),
-                "area": extend.get("area", "地区"), "year": extend.get("year", "年份"), "page": str(pg),
-                "pageSize": "21"}
-        result = {}
-        list = []
-        bba = self.url(body)
-        url = f"{self.host}/api/v1/movie/screen/list?pack={bba[0]}&signature={bba[1]}"
-        data = self.fetch(url, headers=self.header()).json()['data']['list']
-        for item in data:
-            list.append(self.voides(item))
-        result["list"] = list
-        result["page"] = pg
-        result["pagecount"] = 9999
-        result["limit"] = 90
-        result["total"] = 999999
-        return result
-
-    def detailContent(self, ids):
-        body = {"id": ids[0]}
-        bba = self.url(body)
-        url = f'{self.host}/api/v1/movie/detail?pack={bba[0]}&signature={bba[1]}'
-        data = self.fetch(url, headers=self.header()).json()['data']
-        video = {'vod_name': data.get('name'),'type_name': data.get('type_name'),'vod_year': data.get('year'),'vod_area': data.get('area'),'vod_remarks': data.get('dynami'),'vod_content': data.get('content')}
-        play = []
-        names = []
-        tasks = []
-        for itt in data["play_from"]:
-            name = itt["name"]
-            a = []
-            if len(itt["list"]) > 0:
-                names.append(name)
-                play.append(self.playeach(itt['list']))
-            else:
-                tasks.append({"movie_id": ids[0], "from_code": itt["code"]})
-                names.append(name)
-        if tasks:
-            with ThreadPoolExecutor(max_workers=len(tasks)) as executor:
-                results = executor.map(self.playlist, tasks)
-                for result in results:
-                    if result:
-                        play.append(result)
-                    else:
-                        play.append("")
-        video["vod_play_from"] = "$$$".join(names)
-        video["vod_play_url"] = "$$$".join(play)
-        result = {"list": [video]}
-        return result
-
-    def searchContent(self, key, quick, pg=1):
-        body = {"keyword": key, "sort": "", "type_id": "0", "page": str(pg), "pageSize": "10",
-                "res_type": "by_movie_name"}
-        bba = self.url(body)
-        url = f"{self.host}/api/v1/movie/search?pack={bba[0]}&signature={bba[1]}"
-        data = self.fetch(url, headers=self.header()).json()['data'].get('list')
-        videos = []
-        for it in data:
-            try:
-                videos.append(self.voides(it))
-            except Exception as e:
-                continue
-        result = {"list": videos, "page": pg}
-        return result
-
-    def playerContent(self, flag, id, vipFlags):
-        url = id
-        if "m3u8" not in url and "mp4" not in url:
-            try:
-                add = id.split('|||')
-                data = {"from_code": add[0], "play_url": add[1], "episode_id": add[2], "type": "play"}
-                bba = self.url(data)
-                data2 = self.fetch(f"{self.host}/api/v1/movie_addr/parse_url?pack={bba[0]}&signature={bba[1]}",
-                                   headers=self.header()).json()['data']
-                url = data2.get('play_url') or data2.get('download_url')
-                try:
-                    url1 = self.fetch(url, headers=self.header(), allow_redirects=False).headers['Location']
-                    if url1 and "http" in url1:
-                        url = url1
-                except:
-                    pass
-            except Exception as e:
-                pass
-        if '.jpg' in url or '.jpeg' in url or '.png' in url:
-            url = self.getProxyUrl() + "&url=" + b64encode(url.encode('utf-8')).decode('utf-8') + "&type=m3u8"
-        result = {}
-        result["parse"] = 0
-        result["url"] = url
-        result["header"] = {'user-agent': 'okhttp/4.9.2'}
-        return result
-
-    def localProxy(self, param):
-        url = b64decode(param["url"]).decode('utf-8')
-        durl = url[:url.rfind('/')]
-        data = self.fetch(url, headers=self.header()).content.decode("utf-8")
-        lines = data.strip().split('\n')
-        for index, string in enumerate(lines):
-            # if 'URI="' in string and 'http' not in string:
-            #     lines[index] = index
-            # 暂时预留，貌似用不到
-            if '#EXT' not in string and 'http' not in string:
-                lines[index] = durl + ('' if string.startswith('/') else '/') + string
-        data = '\n'.join(lines)
-        return [200, "application/vnd.apple.mpegur", data]
-
-    def device_id(self):
-        characters = string.ascii_lowercase + string.digits
-        random_string = ''.join(random.choices(characters, k=32))
-        return random_string
-
-    def gethost(self):
-        headers = {
-            'User-Agent': 'okhttp/4.9.2',
-            'Connection': 'Keep-Alive',
-        }
-        response = self.fetch('https://app-site.ecoliving168.com/domain_v5.json', headers=headers).json()
-        url = response['api_service'].replace('/api/', '')
-        return url
-
-    def header(self):
-        headers = {
-            'User-Agent': 'Android',
-            'Accept': 'application/prs.55App.v2+json',
-            'timestamp': self.t,
-            'x-client-setting': '{"pure-mode":1}',
-            'x-client-uuid': '{"device_id":' + self.device + '}, "type":1,"brand":"Redmi", "model":"M2012K10C", "system_version":30, "sdk_version":"3.1.0.7"}',
-            'x-client-version': '3096 '
-        }
-        return headers
-
-    def url(self, id=None):
-        if not id:
-            id = {}
-        id["timestamp"] = self.t
-        public_key = 'MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA02F/kPg5A2NX4qZ5JSns+bjhVMCC6JbTiTKpbgNgiXU+Kkorg6Dj76gS68gB8llhbUKCXjIdygnHPrxVHWfzmzisq9P9awmXBkCk74Skglx2LKHa/mNz9ivg6YzQ5pQFUEWS0DfomGBXVtqvBlOXMCRxp69oWaMsnfjnBV+0J7vHbXzUIkqBLdXSNfM9Ag5qdRDrJC3CqB65EJ3ARWVzZTTcXSdMW9i3qzEZPawPNPe5yPYbMZIoXLcrqvEZnRK1oak67/ihf7iwPJqdc+68ZYEmmdqwunOvRdjq89fQMVelmqcRD9RYe08v+xDxG9Co9z7hcXGTsUquMxkh29uNawIDAQAB'
-        encrypted_text = json.dumps(id)
-        public_key = RSA.import_key(b64decode(public_key))
-        cipher = PKCS1_v1_5.new(public_key)
-        encrypted_message = cipher.encrypt(encrypted_text.encode('utf-8'))
-        encrypted_message_base64 = b64encode(encrypted_message).decode('utf-8')
-        result = encrypted_message_base64.replace('+', '-').replace('/', '_').replace('=', '')
-        key = '635a580fcb5dc6e60caa39c31a7bde48'
-        sign = hmac.new(key.encode(), result.encode(), hashlib.md5).hexdigest()
-        return result, sign
-
-    def playlist(self, body):
-        try:
-            bba = self.url(body)
-            url = f'{self.host}/api/v1/movie_addr/list?pack={bba[0]}&signature={bba[1]}'
-            data = self.fetch(url, headers=self.header()).json()['data']
-            return self.playeach(data)
-        except Exception:
-            return []
-
-    def playeach(self,data):
-        play_urls = []
-        for it in data:
-            if re.search(r"mp4|m3u8", it["play_url"]):
-                play_urls.append(f"{it['episode_name']}${it['play_url']}")
-            else:
-                play_urls.append(
-                    f"{it['episode_name']}${it['from_code']}|||{it['play_url']}|||{it['episode_id']}"
-                )
-        return '#'.join(play_urls)
-    
-    def voides(self, item):
-        if item['name'] or item['title']:
-            voide = {
-                "vod_id": item.get('id') or item.get('click'),
-                'vod_name': item.get('name') or item.get('title'),
-                'vod_pic': item.get('cover') or item.get('image'),
-                'vod_year': item.get('year') or item.get('label'),
-                'vod_remarks': item.get('dynamic') or item.get('sub_title')
-            }
-            return voide
-
-    def aes(self, text):
-        text = text.replace('-', '+').replace('_', '/') + '=='
-        key = b"e6d5de5fcc51f53d"
-        iv = b"2f13eef7dfc6c613"
-        cipher = AES.new(key, AES.MODE_CBC, iv)
-        pt = unpad(cipher.decrypt(b64decode(text)), AES.block_size).decode("utf-8")
-        return json.loads(pt)
+[session-5b43329a] Route error: The Repository has been blocked. see: https://gitee.com/PizazzXS/another-d