ff_plugins/anime_downloader
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

v0.6.14: Ubuntu Docker performance optimization for Ohli24

Browse Source
This commit is contained in:
projectdx
2026-01-07 17:07:46 +09:00
parent 49aea1bb54
commit 8759d1e1c8
5 changed files with 221 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
lib/botasaurus_ohli24.py
View File

@@ -9,6 +9,7 @@ import sys
import json
import os
import time
import traceback
from typing import Dict, Any, Optional
# 봇사우루스 디버깅 일시정지 방지 및 자동 종료 설정
@@ -16,19 +17,22 @@ os.environ["BOTASAURUS_ENV"] = "production"
def fetch_html(url: str, headers: Optional[Dict[str, str]] = None, proxy: Optional[str] = None) -> Dict[str, Any]:
result: Dict[str, Any] = {"success": False, "html": "", "elapsed": 0}
start_time: float = time.time()
max_retries = 2
try:
from botasaurus.request import request as b_request
# raise_exception=True는 에러 시 exception을 발생시키게 함
# close_on_crash=True는 에러 발생 시 대기하지 않고 즉시 종료 (배포 환경용)
@b_request(proxy=proxy, raise_exception=True, close_on_crash=True)
# use_stealth=True 추가하여 탐지 회피 강화
@b_request(
proxy=proxy,
raise_exception=True,
close_on_crash=True
)
def fetch_url(request: Any, data: Dict[str, Any]) -> str:
target_url = data.get('url')
headers = data.get('headers') or {}
# 기본적인 헤더 보강 (Ohli24 대응 - Cloudflare 우회 시도)
# 기본적인 헤더 보강 (Ohli24 대응 - Cloudflare/TLS Fingerprinting 대응)
default_headers = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.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",
@@ -50,37 +54,69 @@ def fetch_html(url: str, headers: Optional[Dict[str, str]] = None, proxy: Option
if k not in headers and k.lower() not in [hk.lower() for hk in headers]:
headers[k] = v
return request.get(target_url, headers=headers, timeout=30)
return request.get(target_url, headers=headers, timeout=20)
# 봇사우루스는 실패 시 자동 재시도 등을 하기도 함.
# 여기서는 단발성 요청이므로 직접 호출.
b_resp: str = fetch_url({'url': url, 'headers': headers})
elapsed: float = time.time() - start_time
if b_resp and len(b_resp) > 10:
result.update({
"success": True,
"html": b_resp,
"elapsed": round(elapsed, 2)
})
else:
result["error"] = f"Short response: {len(b_resp) if b_resp else 0} bytes"
result["elapsed"] = round(elapsed, 2)
for attempt in range(max_retries + 1):
start_time = time.time()
try:
b_resp: str = fetch_url({'url': url, 'headers': headers})
elapsed = time.time() - start_time
# 리스트 페이지는 보통 수백KB 이상 (최소 500바이트 체크)
if b_resp and len(b_resp) > 500:
result.update({
"success": True,
"html": b_resp,
"elapsed": round(elapsed, 2),
"attempt": attempt + 1
})
return result
else:
reason = f"Short response ({len(b_resp) if b_resp else 0} bytes)"
if attempt < max_retries:
time.sleep(1)
continue
result["error"] = reason
result["elapsed"] = round(time.time() - start_time, 2)
except Exception as inner_e:
if attempt < max_retries:
time.sleep(1)
continue
result["error"] = str(inner_e)
result["elapsed"] = round(time.time() - start_time, 2)
except Exception as e:
result["error"] = str(e)
result["elapsed"] = round(time.time() - start_time, 2)
result["error"] = f"Botasaurus init/import error: {str(e)}"
result["elapsed"] = 0
return result
if __name__ == "__main__":
if len(sys.argv) < 2:
print(json.dumps({"success": False, "error": "Usage: python botasaurus_ohli24.py <url> [headers_json] [proxy]"}))
sys.exit(1)
# 모든 stdout을 stderr로 리다이렉트 (라이브러리 로그가 stdout을 오염시키는 것 방지)
original_stdout = sys.stdout
sys.stdout = sys.stderr
target_url: str = sys.argv[1]
headers_arg: Optional[Dict[str, str]] = json.loads(sys.argv[2]) if len(sys.argv) > 2 and sys.argv[2] else None
proxy_arg: Optional[str] = sys.argv[3] if len(sys.argv) > 3 and sys.argv[3] else None
res: Dict[str, Any] = fetch_html(target_url, headers_arg, proxy_arg)
print(json.dumps(res, ensure_ascii=False))
try:
if len(sys.argv) < 2:
# 에러 메시지는 출력해야 하므로 다시 복구 후 출력
sys.stdout = original_stdout
print(json.dumps({"success": False, "error": "Usage: script.py <url> [headers] [proxy]"}))
sys.exit(1)
target_url: str = sys.argv[1]
headers_arg: Optional[Dict[str, str]] = json.loads(sys.argv[2]) if len(sys.argv) > 2 and sys.argv[2] else None
proxy_arg: Optional[str] = sys.argv[3] if len(sys.argv) > 3 and sys.argv[3] else None
res: Dict[str, Any] = fetch_html(target_url, headers_arg, proxy_arg)
# 최종 결과 출력 전에만 stdout 복구
sys.stdout = original_stdout
print(json.dumps(res, ensure_ascii=False))
except Exception as fatal_e:
# 에러 발생 시에도 JSON 형태로 출력하도록 보장
sys.stdout = original_stdout
print(json.dumps({
"success": False,
"error": f"Fatal execution error: {str(fatal_e)}",
"traceback": traceback.format_exc()
}, ensure_ascii=False))

61
lib/zendriver_daemon.py
View File

@@ -174,6 +174,16 @@ async def ensure_browser() -> Any:
log_debug("[ZendriverDaemon] No browser candidates found!")
return None
# 리눅스/도커 성능 분석용 로그
import platform
if platform.system() == "Linux":
try:
shm_size = os.statvfs('/dev/shm')
free_shm = (shm_size.f_bavail * shm_size.f_frsize) / (1024 * 1024)
log_debug(f"[ZendriverDaemon] Linux detected. /dev/shm free: {free_shm:.1f} MB")
except Exception as shm_e:
log_debug(f"[ZendriverDaemon] Failed to check /dev/shm: {shm_e}")
# 사용자 데이터 디렉토리 설정 (Mac/Root 권한 이슈 대응)
import tempfile
uid = os.getuid() if hasattr(os, 'getuid') else 'win'
@@ -204,14 +214,24 @@ async def ensure_browser() -> Any:
"--safebrowsing-disable-auto-update",
"--remote-allow-origins=*",
"--blink-settings=imagesEnabled=false",
"--disable-blink-features=AutomationControlled",
# 추가적인 도커 최적화 플래그
"--disable-features=IsolateOrigins,site-per-process",
"--no-zygote",
"--disable-extensions",
"--wasm-tier-up=false",
]
# 추가적인 리소스 블로킹 설정
# Note: zendriver supports direct CDP commands
for exec_path in candidates:
user_data_dir = os.path.join(tempfile.gettempdir(), f"zd_daemon_{uid}_{os.path.basename(exec_path).replace(' ', '_')}")
os.makedirs(user_data_dir, exist_ok=True)
try:
log_debug(f"[ZendriverDaemon] Trying browser at: {exec_path}")
start_time_init = time.time()
browser = await zd.start(
headless=True,
browser_executable_path=exec_path,
@@ -219,7 +239,7 @@ async def ensure_browser() -> Any:
user_data_dir=user_data_dir,
browser_args=browser_args
)
log_debug(f"[ZendriverDaemon] Browser started successfully with: {exec_path}")
log_debug(f"[ZendriverDaemon] Browser started successfully in {time.time() - start_time_init:.2f}s using: {exec_path}")
return browser
except Exception as e:
log_debug(f"[ZendriverDaemon] Failed to start {exec_path}: {e}")
@@ -242,25 +262,39 @@ async def fetch_with_browser(url: str, timeout: int = 30) -> Dict[str, Any]:
start_time: float = time.time()
try:
init_start = time.time()
await ensure_browser()
init_elapsed = time.time() - init_start
if browser is None:
result["error"] = "Browser not available"
return result
# zendriver의 browser.get(url)은 이미 열린 탭이 있으면 거기서 열려고 시도함.
# 하지만 모든 탭이 닫히면 StopIteration이 발생할 수 있음.
log_debug(f"[ZendriverDaemon] Fetching URL: {url}")
log_debug(f"[ZendriverDaemon] Fetching URL: {url} (Init: {init_elapsed:.2f}s)")
# StopIteration 방지를 위해 페이지 이동 시도
try:
nav_start = time.time()
# browser.get(url)은 새 탭을 열거나 기존 탭을 사용함
page: Any = await browser.get(url)
nav_elapsed = time.time() - nav_start
# 리소스 블로킹 (CDP 활용) - CSS, 폰트, 이미지 등 차단으로 속도 향상
block_start = time.time()
try:
await page.send(zd.cdp.network.set_blocked_urls(urls=[
"*.jpg", "*.jpeg", "*.png", "*.gif", "*.svg", "*.webp", "*.ico",
"*.css", "*.woff", "*.woff2", "*.ttf", "*.eot",
"*ads*", "*google-analytics*", "*googletagmanager*", "*doubleclick*"
]))
await page.send(zd.cdp.network.enable())
except Exception as e:
log_debug(f"[ZendriverDaemon] Resource blocking enable failed: {e}")
block_elapsed = time.time() - block_start
# 페이지 로드 대기 - 지능형 폴링 (최대 10초)
# 1. 리스트 페이지는 바로 반환, 2. 에피소드 페이지는 플레이어 로딩 대기
max_wait = 10
poll_interval = 0.2 # 1.0s -> 0.2s로 단축하여 반응속도 향상
poll_interval = 0.1 # 0.2s -> 0.1s로 더 빠르게 체크
waited = 0
html_content = ""
@@ -279,18 +313,25 @@ async def fetch_with_browser(url: str, timeout: int = 30) -> Dict[str, Any]:
log_debug(f"[ZendriverDaemon] Player detected in {waited:.1f}s")
break
elapsed: float = time.time() - start_time
poll_elapsed = time.time() - poll_start
total_elapsed = time.time() - start_time
if html_content and len(html_content) > 100:
result.update({
"success": True,
"html": html_content,
"elapsed": round(elapsed, 2)
"elapsed": round(total_elapsed, 2),
"metrics": {
"init": round(init_elapsed, 2),
"nav": round(nav_elapsed, 2),
"block": round(block_elapsed, 2),
"poll": round(poll_elapsed, 2)
}
})
log_debug(f"[ZendriverDaemon] Fetch success in {elapsed:.2f}s (Length: {len(html_content)})")
log_debug(f"[ZendriverDaemon] Success in {total_elapsed:.2f}s (Nav: {nav_elapsed:.2f}s, Poll: {poll_elapsed:.2f}s)")
else:
result["error"] = f"Short response: {len(html_content) if html_content else 0} bytes"
result["elapsed"] = round(elapsed, 2)
result["elapsed"] = round(total_elapsed, 2)
log_debug(f"[ZendriverDaemon] Fetch failure: Short response ({len(html_content) if html_content else 0} bytes)")
# 여기서 page.close()를 하지 않음! (탭을 하나라도 남겨두어야 StopIteration 방지 가능)