v0.6.23: Fix Linkkf download - CDP Headers wrapper, yt-dlp --add-header support

- Fix zendriver_daemon CDP Headers bug (wrap dict with zd.cdp.network.Headers())
- Fix HTML entity decoding in iframe URLs (use html.unescape())
- Simplify GDM source_type to always use 'linkkf'
This commit is contained in:
2026-01-08 01:29:36 +09:00
parent d1866111c7
commit 24217712a6
7 changed files with 181 additions and 79 deletions

View File

@@ -1,61 +1,85 @@
import asyncio
import zendriver as zd
import requests
import json
import re
import sys
import os
import subprocess
async def test():
print("=== Zendriver Google Chrome Debug (v0.5.14) ===")
def test_fetch():
url = "https://playv2.sub3.top/r2/play.php?&id=n20&url=405686s1"
headers = {
"Referer": "https://linkkf.live/",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36"
}
# Check possible paths
bin_paths = ["/usr/bin/google-chrome", "/usr/bin/google-chrome-stable", "/usr/bin/chromium-browser"]
daemon_url = "http://127.0.0.1:19876/fetch"
payload = {
"url": url,
"headers": headers,
"timeout": 30
}
for browser_bin in bin_paths:
if not os.path.exists(browser_bin):
continue
print(f"\n>>> Testing binary: {browser_bin}")
print(f"Fetching {url} via daemon...")
try:
resp = requests.post(daemon_url, json=payload, timeout=40)
if resp.status_code != 200:
print(f"Error: HTTP {resp.status_code}")
print(resp.text)
return
# 1. Version Check
try:
out = subprocess.check_output([browser_bin, "--version"], stderr=subprocess.STDOUT).decode()
print(f"Version: {out.strip()}")
except Exception as e:
print(f"Version check failed: {e}")
if hasattr(e, 'output'):
print(f"Output: {e.output.decode()}")
# 2. Minimum execution test (Headless + No Sandbox)
print("--- Direct Execution Test ---")
try:
cmd = [browser_bin, "--headless", "--no-sandbox", "--disable-gpu", "--user-data-dir=/tmp/test_chrome", "--about:blank"]
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
await asyncio.sleep(3)
if proc.poll() is None:
print("SUCCESS: Browser process is alive!")
proc.terminate()
data = resp.json()
if not data.get("success"):
print(f"Fetch failed: {data.get('error')}")
return
html = data.get("html", "")
print(f"Fetch success. Length: {len(html)}")
# Save for inspection
with open("linkkf_player_test.html", "w", encoding="utf-8") as f:
f.write(html)
print("Saved to linkkf_player_test.html")
# Try regex patterns from mod_linkkf.py
patterns = [
r"url:\s*['\"]([^'\"]*\.m3u8[^'\"]*)['\"]",
r"<source[^>]+src=['\"]([^'\"]*\.m3u8[^'\"]*)['\"]",
r"src\s*=\s*['\"]([^'\"]*\.m3u8[^'\"]*)['\"]",
r"url\s*:\s*['\"]([^'\"]+)['\"]"
]
found = False
for p in patterns:
match = re.search(p, html, re.IGNORECASE)
if match:
url_found = match.group(1)
if ".m3u8" in url_found or "m3u8" in p:
print(f"Pattern '{p}' found: {url_found}")
found = True
if not found:
print("No m3u8 found with existing patterns.")
# Search for any .m3u8
any_m3u8 = re.findall(r"['\"]([^'\"]*\.m3u8[^'\"]*)['\"]", html)
if any_m3u8:
print(f"Generic search found {len(any_m3u8)} m3u8 links:")
for m in any_m3u8[:5]:
print(f" - {m}")
else:
stdout, stderr = proc.communicate()
print(f"FAIL: Browser process died (code {proc.returncode})")
print(f"STDERR: {stderr.decode()}")
except Exception as e:
print(f"Execution test failed: {e}")
print("No .m3u8 found in generic search either.")
# Check for other video extensions or potential indicators
if "Artplayer" in html:
print("Artplayer detected.")
if "video" in html:
print("Video tag found.")
# Check for 'cache/'
if "cache/" in html:
print("Found 'cache/' keyword.")
cache_links = re.findall(r"['\"]([^'\"]*cache/[^'\"]*)['\"]", html)
for c in cache_links:
print(f" - Possible cache link: {c}")
# 3. Zendriver Test
print("--- Zendriver Integration Test ---")
try:
browser = await zd.start(
browser_executable_path=browser_bin,
headless=True,
sandbox=False
)
print("SUCCESS: Zendriver connected!")
await browser.stop()
# If we found one that works, we can stop
print("\n!!! This path works. Set this in the plugin settings or leave empty if it is the first found.")
except Exception as e:
print(f"Zendriver failed: {e}")
except Exception as e:
print(f"Exception: {e}")
if __name__ == "__main__":
asyncio.run(test())
test_fetch()