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

anilife.live 사이트 구현

Browse Source 
다른 버그도 고침
This commit is contained in:
projectdx
2025-12-28 19:38:18 +09:00
parent e6e8c45f5a
commit 6dbeff13d3
14 changed files with 1576 additions and 347 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
lib/ytdlp_downloader.py
View File

@@ -16,11 +16,13 @@ logger = logging.getLogger(__name__)
class YtdlpDownloader:
"""yt-dlp 기반 다운로더"""
def __init__(self, url, output_path, headers=None, callback=None):
def __init__(self, url, output_path, headers=None, callback=None, proxy=None, cookies_file=None):
self.url = url
self.output_path = output_path
self.headers = headers or {}
self.callback = callback # 진행 상황 콜백
self.proxy = proxy
self.cookies_file = cookies_file # CDN 세션 쿠키 파일 경로
self.cancelled = False
self.process = None
self.error_output = [] # 에러 메시지 저장
@@ -30,6 +32,7 @@ class YtdlpDownloader:
self.current_speed = ""
self.elapsed_time = ""
self.percent = 0
def format_time(self, seconds):
"""시간을 읽기 좋은 형식으로 변환"""
@@ -57,12 +60,7 @@ class YtdlpDownloader:
return f"{bytes_per_sec / (1024 * 1024):.2f} MB/s"
def download(self):
"""yt-dlp Python 모듈로 다운로드 수행"""
try:
import yt_dlp
except ImportError:
return False, "yt-dlp를 찾을 수 없습니다. pip install yt-dlp 로 설치해주세요."
"""yt-dlp CLI를 통한 브라우저 흉내(Impersonate) 방식 다운로드 수행"""
try:
self.start_time = time.time()
@@ -71,86 +69,118 @@ class YtdlpDownloader:
if output_dir and not os.path.exists(output_dir):
os.makedirs(output_dir)
# 진행률 콜백
def progress_hook(d):
# URL 전처리: 확장자 힌트(?dummy=.m3u8) 사용
# (m3u8: 접두사나 #.m3u8보다 호환성이 높음. HLS 인식 강제용)
current_url = self.url
if 'master.txt' in current_url:
concat_char = '&' if '?' in current_url else '?'
current_url = f"{current_url}{concat_char}dummy=.m3u8"
# 1. 기본 명령어 구성 (Impersonate & HLS 강제)
cmd = [
'yt-dlp',
'--newline',
'--no-playlist',
'--no-part',
'--hls-prefer-ffmpeg',
'--hls-use-mpegts',
'--no-check-certificate',
'--progress',
'--verbose', # 디버깅용 상세 로그
'--impersonate', 'chrome-120', # 정밀한 크롬-120 지문 사용
'--extractor-args', 'generic:force_hls', # HLS 강제 추출
'-o', self.output_path,
]
# 2. 프록시 설정
if self.proxy:
cmd += ['--proxy', self.proxy]
# 2.5 쿠키 파일 설정 (CDN 세션 인증용)
if self.cookies_file and os.path.exists(self.cookies_file):
cmd += ['--cookies', self.cookies_file]
logger.info(f"Using cookies file: {self.cookies_file}")
# 3. 필수 헤더 구성
# --impersonate가 기본적인 Sec-Fetch를 처리하지만,
# X-Requested-With와 정확한 Referer/Origin은 명시적으로 주는 것이 안전합니다.
has_referer = False
for k, v in self.headers.items():
if k.lower() == 'referer':
cmd += ['--referer', v]
has_referer = True
elif k.lower() == 'user-agent':
# impersonate가 설정한 UA를 명시적 UA로 덮어씀 (필요시)
cmd += ['--user-agent', v]
else:
cmd += ['--add-header', f"{k}:{v}"]
# cdndania 전용 헤더 보강
if 'cdndania.com' in current_url:
if not has_referer:
cmd += ['--referer', 'https://cdndania.com/']
cmd += ['--add-header', 'Origin:https://cdndania.com']
cmd += ['--add-header', 'X-Requested-With:XMLHttpRequest']
cmd.append(current_url)
logger.info(f"Executing refined browser-impersonated yt-dlp CLI (v16): {' '.join(cmd)}")
# 4. subprocess 실행 및 파싱
self.process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
bufsize=1
)
# [download] 10.5% of ~100.00MiB at 2.45MiB/s
prog_re = re.compile(r'\[download\]\s+(?P<percent>[\d\.]+)%\s+of\s+.*?\s+at\s+(?P<speed>.*?)(\s+ETA|$)')
for line in self.process.stdout:
if self.cancelled:
raise Exception("Cancelled")
self.process.terminate()
return False, "Cancelled"
if d['status'] == 'downloading':
# 진행률 추출
total = d.get('total_bytes') or d.get('total_bytes_estimate') or 0
downloaded = d.get('downloaded_bytes', 0)
speed = d.get('speed', 0)
if total > 0:
self.percent = (downloaded / total) * 100
self.current_speed = self.format_speed(speed) if speed else ""
if self.start_time:
elapsed = time.time() - self.start_time
self.elapsed_time = self.format_time(elapsed)
# 콜백 호출
if self.callback:
self.callback(
percent=int(self.percent),
current=int(self.percent),
total=100,
speed=self.current_speed,
elapsed=self.elapsed_time
)
line = line.strip()
if not line: continue
elif d['status'] == 'finished':
logger.info(f"yt-dlp download finished: {d.get('filename', '')}")
match = prog_re.search(line)
if match:
try:
self.percent = float(match.group('percent'))
self.current_speed = match.group('speed').strip()
if self.start_time:
elapsed = time.time() - self.start_time
self.elapsed_time = self.format_time(elapsed)
if self.callback:
self.callback(percent=int(self.percent), current=int(self.percent), total=100, speed=self.current_speed, elapsed=self.elapsed_time)
except: pass
elif 'error' in line.lower() or 'security' in line.lower() or 'unable' in line.lower():
logger.warning(f"yt-dlp output notice: {line}")
self.error_output.append(line)
self.process.wait()
# yt-dlp 옵션 설정
ydl_opts = {
'outtmpl': self.output_path,
'progress_hooks': [progress_hook],
'quiet': False,
'no_warnings': False,
'noprogress': False,
}
# 헤더 추가
http_headers = {}
if self.headers:
if self.headers.get('Referer'):
http_headers['Referer'] = self.headers['Referer']
if self.headers.get('User-Agent'):
http_headers['User-Agent'] = self.headers['User-Agent']
if http_headers:
ydl_opts['http_headers'] = http_headers
logger.info(f"yt-dlp downloading: {self.url}")
logger.info(f"Output path: {self.output_path}")
logger.info(f"Headers: {http_headers}")
# 다운로드 실행
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.download([self.url])
# 파일 존재 확인
if os.path.exists(self.output_path):
if self.process.returncode == 0 and os.path.exists(self.output_path):
# 가짜 파일(보안 에러 텍스트) 체크
file_size = os.path.getsize(self.output_path)
if file_size < 2000:
try:
with open(self.output_path, 'r') as f:
text = f.read().lower()
if "security error" in text or not text:
os.remove(self.output_path)
return False, f"CDN 보안 차단(가짜 파일 다운로드됨: {file_size}B)"
except: pass
return True, "Download completed"
else:
# yt-dlp가 확장자를 변경했을 수 있음
base_name = os.path.splitext(self.output_path)[0]
for ext in ['.mp4', '.mkv', '.webm', '.ts']:
possible_path = base_name + ext
if os.path.exists(possible_path):
if possible_path != self.output_path:
os.rename(possible_path, self.output_path)
return True, "Download completed"
return False, "Output file not found"
except Exception as e:
error_msg = str(e)
logger.error(f"yt-dlp download error: {error_msg}")
error_msg = "\n".join(self.error_output[-3:]) if self.error_output else f"Exit code {self.process.returncode}"
return False, f"yt-dlp 실패: {error_msg}"
except Exception as e:
logger.error(f"yt-dlp download exception: {e}")
return False, f"yt-dlp download exception: {str(e)}"
def cancel(self):
"""다운로드 취소"""