From 6179a871d1781186b6540ca03ddca992dd1bcbb5 Mon Sep 17 00:00:00 2001 From: cluntop <85211716+cluntop@users.noreply.github.com> Date: Fri, 27 Feb 2026 02:35:57 +0800 Subject: [PATCH] Update Up --- py/sinparty.py | 94 +++++++++++++++++++++++++++++++------------------- 1 file changed, 59 insertions(+), 35 deletions(-) diff --git a/py/sinparty.py b/py/sinparty.py index 69cdb6ca4..679c2fa09 100644 --- a/py/sinparty.py +++ b/py/sinparty.py @@ -19,13 +19,14 @@ async def fetch_m3u8(session: aiohttp.ClientSession, name: str, link: str): if match: m3u8_url = match.group(1).replace('\\/', '/') # 【可视化新增】：实时输出成功嗅探到底层流的数据 - print(f" [√ 流捕获] {name.ljust(15)} -> {m3u8_url}") + print(f" [√ 流捕获成功] {name.ljust(15)} -> {m3u8_url}") return name, m3u8_url, link - except Exception: + except Exception as e: + print(f" [x] 网络请求异常 ({name}): {e}") pass # 【可视化新增】：实时输出未命中底层流，降级处理的数据 - print(f" [- 未嗅探] {name.ljust(15)} -> 未发现直接 m3u8，保留原始跳转链接") + print(f" [- 嗅探失败/降级] {name.ljust(15)} -> 未发现直接 m3u8，保留原始跳转链接") return name, None, link async def main(): @@ -33,21 +34,28 @@ async def main(): # === 阶段 1：Playwright 全站分页抓取 === async with async_playwright() as p: + print("[*] 正在启动 Chromium 无头浏览器...") browser = await p.chromium.launch(headless=True) page = await browser.new_page(ignore_https_errors=True) # 注入反爬虫绕过脚本，抹除自动化特征，欺骗防御蜘蛛的探测 await page.add_init_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})") - print("\n[*] 已挂载 Stealth 反检测脚本，webdriver 特征已抹除。") + print("[*] 已挂载 Stealth 反检测脚本，webdriver 特征已抹除，准备渗透。") page_num = 1 while True: - print(f"\n{'='*50}") - print(f"[*] 正在加载并抓取第 {page_num} 页数据... (URL: https://sinparty.com/?page={page_num})") + print(f"\n{'='*60}") + print(f"[*] [抓取网站流程] 正在加载并抓取第 {page_num} 页... ") + print(f"[*] [目标 URL] https://sinparty.com/?page={page_num}") # 动态改变 page=&& 参数 url = f"https://sinparty.com/?page={page_num}" # 移除不可靠的 networkidle，使用默认导航机制 - await page.goto(url) + response = await page.goto(url) + + if response: + print(f"[*] [网络状态] 页面响应码: {response.status} (200为正常)") + else: + print(f"[!] [网络状态] 未获取到页面响应信息，可能存在拦截。") # 强制屏蔽拦截遮罩，破坏防御弹窗，恢复“可操作、可活动、可点击”状态 await page.add_style_tag(content=''' @@ -58,60 +66,75 @@ async def main(): await page.evaluate('''() => { document.querySelectorAll('.app-modal__overlay, .modal-auth__inner').forEach(el => el.remove()); }''') - print("[*] 防御拦截层 (.app-modal__overlay) 已被强制物理摧毁，DOM 可点击限制已解除。") + print("[*] [防御突破] 拦截层 (.app-modal__overlay) 已强制物理摧毁，DOM 可点击限制解除。") # 定位目标：跳过 skeleton 骨架屏，直接锁定在线主播节点 try: # 显式等待真实数据的 CSS 节点渲染到 DOM 中（最长容忍 20 秒） - # 统一
数组 和
- print("[*] 正在等待目标外层节点渲染: .content-gallery--live-listing .content-gallery__item") - await page.wait_for_selector(".content-gallery content-gallery--live-listing", timeout=20000) + print("[*] [DOM 解析] 正在等待目标外层节点渲染: .content-gallery--live-listing ...") + # 修复了原来类名中间缺漏的点 + await page.wait_for_selector(".content-gallery.content-gallery--live-listing", timeout=20000) + print("[√] [DOM 解析] 外层数据容器已成功渲染！") except Exception: # 如果 20 秒后目标节点仍未出现，说明确实到达了没有数据的最后一页 - print(f"[!] 第 {page_num} 页未检测到有效在线主播数据，翻页结束。") + print(f"[!] [状态反馈] 第 {page_num} 页未检测到有效数据渲染，判断为到达最后一页，结束翻页。") break # 此时 DOM 中必定已有数据，安全执行并集提取 - # 每二次数组截胡
elements = await page.locator(".content-gallery--live-listing .content-gallery__item").all() - print(f"[*] 成功截胡当前页卡片数组，共包含 {len(elements)} 个目标节点。开始提取详细数据...") + print(f"[*] [数据截胡] 成功获取当前页卡片数组，共包含 {len(elements)} 个目标节点。开始深度提取...\n") - for element in elements: - # 抓取标题与名字：兼容 .cam-tile__title 或 .cam-tile__details + for idx, element in enumerate(elements, 1): + print(f" --- 正在分析第 {idx}/{len(elements)} 个节点 ---") + + # 【可视化新增】：抓取并显示底层 HTML 数据结构 (截取前150个字符防刷屏) + raw_html = await element.inner_html() + clean_html = raw_html.replace('\n', '').replace(' ', '') + print(f" [RAW HTML] 源码探针: {clean_html[:150]}...") + + # 抓取标题与名字 title_loc = element.locator(".cam-tile__title") if await title_loc.count() > 0: title = await title_loc.first.inner_text() + title = title.strip() else: - title = "未知用户" + title = "" - # 抓取 href 跳转链接：对应 class="cam-tile" 等于跳转链接 + # 抓取 href 跳转链接 a_loc = element.locator("a.cam-tile") if await a_loc.count() > 0: href = await a_loc.first.get_attribute("href") else: href = "" - if href: - if href.startswith("/"): - href = f"https://sinparty.com{href}" - - # 【可视化新增】：实时输出当前从 DOM 树中剥离出来的详细键值对 - print(f" [+ 数据提取] 节点: a.cam-tile | 标题: {title.strip().ljust(15)} | 跳转链: {href}") + if href and href.startswith("/"): + href = f"https://sinparty.com{href}" + + # 【可视化新增】：判定成功与否，正常不正常 + if not title or not href: + print(f" [x] 状态: 异常/失败！") + if not title: print(" -> 原因: 找不到 class='.cam-tile__title' 的名称节点") + if not href: print(" -> 原因: 找不到 class='a.cam-tile' 的跳转链接") + else: + print(f" [√] 状态: 正常/成功！") + print(f" [+] 提取结果 -> 标题: {title.ljust(15)} | 链接: {href}") results.append({ - "name": title.strip(), + "name": title, "link": href }) + print(" " + "-"*40) page_num += 1 await browser.close() + print("\n[*] 浏览器引擎已关闭，第一阶段 DOM 提取结束。") # === 阶段 2：AIOHTTP 高性能并发抓取 m3u8 流 === - print(f"\n{'='*50}") - print(f"[*] 全站遍历完毕，共提取 {len(results)} 个直播间链接。") - print("[*] 启动 aiohttp 协程池 (TCPConnector limit=100)，开始高并发底层嗅探...") - print(f"{'='*50}\n") + print(f"\n{'='*60}") + print(f"[*] [流程进度] 全站遍历完毕，累计提取有效直播间链接数: {len(results)}") + print("[*] [并发调度] 启动 aiohttp 协程池 (TCPConnector limit=100)，开始底层高并发嗅探...") + print(f"{'='*60}\n") connector = aiohttp.TCPConnector(limit=100, ssl=False) async with aiohttp.ClientSession(connector=connector) as session: @@ -136,17 +159,18 @@ async def main(): m3u_content = "\n".join(m3u_lines) - print(f"\n{'='*50}") - print("=== 转换格式 M3U 输出 ===") + print(f"\n{'='*60}") + print("=== [数据写入] 转换格式 M3U 最终输出 ===") print(m3u_content) - print(f"{'='*50}") + print(f"{'='*60}") with open("lib/party.m3u", "w", encoding="utf-8") as f: f.write(m3u_content) - print(f"\n[*] 并发处理彻底完成！") - print(f"[*] 成功解析并提取 {success_count} 个底层视频流。") - print(f"[*] 总计写入 {len(results)} 条格式化数据至 lib/party.m3u。") + print(f"\n[*] [任务终点] 所有并发处理彻底完成！") + print(f"[*] [统计] 成功解析底层视频流: {success_count} 个") + print(f"[*] [统计] 降级保留跳转链接: {len(results) - success_count} 个") + print(f"[*] [统计] 总计写入数据: {len(results)} 条 -> 已保存至 lib/party.m3u") if __name__ == "__main__": asyncio.run(main())