# -*- coding: utf-8 -*- import re import sys import os import gzip import json from pathlib import Path from itertools import chain import shutil sys.path.append('..') from base.spider import Spider ''' @BNTB ''' CACHE_SUFFIX = ".gzcache" def json_compress_dump(obj): compact_str = json.dumps(obj, separators=(",", ":"), ensure_ascii=False) return gzip.compress(compact_str.encode("utf-8"), compresslevel=9) def json_decompress_load(raw_bytes): json_bytes = gzip.decompress(raw_bytes) return json.loads(json_bytes.decode("utf-8")) class Spider(Spider): def getName(self): return "本地m3u" def init(self, extend=""): self.con=json.loads(extend) self.seachid='' self.path = self.initpath(self.con['path']) self.urls=self.con['urls'].split('#') #self.path2 = self.con['path2'] self.file_names_type = {} self.user_extend = self.con['filters'] self.data = {} self.all_classes = self.con['classes'] self.f1p = self.path+'/temp/data' + CACHE_SUFFIX self.f2p = self.path+'/temp/file_names_type' + CACHE_SUFFIX self.init_ok() def initpath(self,p): if 'http:' in p and '/file/' in p: return '/sdcard'+p.split('file',1)[-1] return p def parse_live(self, text,fn): data = [] cur_genre='默认' self.file_names_type[fn] = [] lines=text.split('\n') for line in lines: s = line.strip() if not s: continue if s.endswith(",#genre#"): cur_genre = s.split(",")[0] self.file_names_type[fn].append(cur_genre) continue if "," in s: name, url = s.split(",", 1) data.append({ "vod_name": name, "vod_id": url, "type_name": cur_genre, "vod_remarks": fn }) self.data[fn] = data def header(self): return { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36' } def init_ok(self): if not os.path.exists(self.f1p) or not os.path.exists(self.f2p): for url in self.urls: try: self.parse_all_m3u_ok(url.strip()) except: pass cache_dir = os.path.dirname(self.f1p) os.makedirs(cache_dir, exist_ok=True) # 写入压缩后的缓存文件 with open(self.f1p, "wb") as f1: f1.write(json_compress_dump(self.data)) with open(self.f2p, "wb") as f2: f2.write(json_compress_dump(self.file_names_type)) else: # 读取解压缓存 with open(self.f1p, "rb") as f1: self.data = json_decompress_load(f1.read()) with open(self.f2p, "rb") as f2: self.file_names_type = json_decompress_load(f2.read()) def parse_all_m3u_ok(self,url): r=self.fetch(url,headers=self.header(),timeout=10) M3U_DIR = Path(self.path) M3U_PATTERN = re.compile( r'#EXTINF:-1.*?' r'(?:tvg-name="([^"]+)"|tvg-name=([^\s,]+))?.*?' r'(?:group-title="([^"]+)"|group-title=([^\s,]+))?,([^\n]*)' r'\n(https?://\S+)', re.IGNORECASE | re.DOTALL ) seen = set() if not M3U_DIR.exists(): os.makedirs(M3U_DIR, exist_ok=True) result = [] type_set = set() fn=url.rsplit('/', 1)[-1] if re.search("#EXTINF", r.text): if '.m3u' not in fn: fn=fn+'.m3u' match_list = M3U_PATTERN.findall(r.text) for item in match_list: tvg_name_q, tvg_name_nq, group_q, group_nq, title, url = item tvg_name = (tvg_name_q or tvg_name_nq or "").strip() group_title = (group_q or group_nq or "未分类").strip() title = title.strip() url = url.strip() type_set.add(group_title) final_name = tvg_name or title if url and url not in seen: seen.add(url) result.append({ "vod_id": url, "vod_name": final_name, "type_name": group_title, "vod_remarks": fn }) self.file_names_type[fn] = list(type_set) self.data[fn] = result elif re.search("#genre#", r.text): if '.txt' not in fn: fn=fn+'.txt' self.parse_live(r.text,fn) def slice_page(self, arr, page, size=2000): start = (page - 1) * size end = start + size return arr[start:end] def search_video(self, key, results, p): if not key or not results: return {"list": [], "page": 1, "pagecount": 0} key = key.lower() result = [] sum_long = len(results) max_page = (sum_long + 1999) // 2000 while not result and p <= max_page: slice_data = self.slice_page(results, p) result = [item for item in slice_data if key in item.get("vod_name", "").lower()] if result: return {"list": result, "page": p, 'pagecount': max_page} p += 1 return {"list": [], "page": p - 1, "pagecount": max_page} def mod_list(self,result,p): sum_long = len(result) max_page = (sum_long + 1999) // 2000 return {"list": self.slice_page(result,p), "page": p, 'pagecount': max_page} def get_all_m3u8(self, text): pattern = r'https?://[^\s<>"]+\.m3u8[^\s<>"]*' return re.findall(pattern, text) def bulid_classe_fliter(self): classes=[{"type_id": 'all', "type_name": '所有'}] classes.extend([{"type_id": n + '__', "type_name": n,"type_flag":1} for n in self.data.keys()]) classes.extend([{"type_id": n, "type_name": n} for n in self.all_classes]) filters = {} for filename, typename_list in self.file_names_type.items(): filters[filename+'__'] = [{ "key": "type", "name": "类型", "value": [{"n": '★'+t, "v": t} for t in typename_list] }] for filename, extend_val in self.user_extend.items(): filters[filename + '__'].append({ "key": "keys", "name": "关键字", "value": [{"n": t, "v": t} for t in extend_val] }) return [classes, filters] def homeContent(self, filter): result = {"class": [], "filters": {}, "list": []} result["class"], result["filters"] = self.bulid_classe_fliter() return result def homeVideoContent(self): return self.homeContent(False) def categoryContent(self, tid, pg, filter, extend): p = int(pg) s = str(tid) if '__' in s: file_key = s.strip('__') self.seachid=file_key source_list = self.data.get(file_key, []) if extend: select_type = extend.get("type", "") select_keys = extend.get("keys", "") if select_type: if select_keys: videos = [row for row in source_list if row.get("type_name") == select_type] return self.search_video(select_keys, videos, p) else: videos = [row for row in source_list if row.get("type_name") == select_type] return {"list": videos, "page": p, "pagecount": 1} if select_keys: return self.search_video(select_keys, source_list, p) return self.mod_list(source_list, p) elif 'all'==s: self.seachid='all' all_total = [{"vod_id": '清空缓存',"vod_name": '点击清空缓存',"type_name": '清空缓存',"vod_remarks": '清空缓存'}]+list(chain.from_iterable(self.data.values())) return self.mod_list(all_total, p) else: all_total = list(chain.from_iterable(self.data.values())) return self.search_video(s,all_total , p) def searchContent(self, key, quick, pg="1"): p = int(pg) if self.seachid in self.data.keys(): return self.search_video(key, self.data[self.seachid], p) else: sun_list = list(chain.from_iterable(self.data.values())) return self.search_video(key, sun_list, p) return {"list": [], "page": p} def reset_folder(self,folder_path): folder = Path(folder_path) if folder.exists(): shutil.rmtree(folder) # 重建空文件夹 folder.mkdir(parents=True, exist_ok=True) def detailContent(self, ids): vid = str(ids[0]) if vid=='清空缓存': self.reset_folder(self.path+'/temp/') return { "list": [{ "vod_id": '无数据', "vod_name": '请返回无数据', "vod_content": '播放链接'+vid, "vod_play_from": '无数据', "vod_play_url": '无数据$缓冲被清空' }] } return { "list": [{ "vod_id": vid, "vod_name": '', "vod_pic": '', "vod_content": '播放链接'+vid, "vod_play_from": '本地文件播放', "vod_play_url": vid + '$' + vid }] } def playerContent(self, flag, id, vipFlags): if '.m3u8' in id and '.ts' in id: return {"parse": 0, "url": id} return {"parse": 1, "url": id} def localProxy(self, params): return [200, "video/MP2T", ""]