diff --git a/scripts/ytresolver.py b/scripts/ytresolver.py index dc4aae3..6871ff0 100644 --- a/scripts/ytresolver.py +++ b/scripts/ytresolver.py @@ -29,6 +29,83 @@ _YDL_OPTS = {"format": "bestaudio/best", "noplaylist": True, "quiet": True, _ydl = yt_dlp.YoutubeDL(_YDL_OPTS) # 프로세스 상주(임포트/초기화 1회) _ydl_lock = threading.Lock() + +# ===== 유튜브 내부 API(VISIONOS) 직접 호출 ===== +# yt-dlp 를 거치지 않고 player API 한 번으로 오디오 직링크를 얻는다(실측 0.3초대). +# visitorData 는 이 프로세스가 직접 발급/캐시하며, 실패하면 즉시 재발급한다. +# 연령제한/로그인 필요 영상은 여기서 None 을 돌려 아래 기존 경로(yt-dlp+쿠키)로 폴백한다. +import urllib.request as _ur + +_VD_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" +_IT_UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 15_7_3) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/26.0 Safari/605.1.15" +_VID_RE = __import__("re").compile(r"(?:v=|youtu\.be/|/shorts/|/embed/|/live/)([A-Za-z0-9_-]{11})") +_VD_TTL = 6 * 3600 +_vd = {"value": "", "expires": 0} +_vd_lock = threading.Lock() + +def _post_player(payload, ua, extra_headers=None): + headers = {"Content-Type": "application/json", "User-Agent": ua} + if extra_headers: + headers.update(extra_headers) + req = _ur.Request("https://www.youtube.com/youtubei/v1/player?prettyPrint=false", + data=json.dumps(payload).encode(), headers=headers) + return json.loads(_ur.urlopen(req, timeout=15).read()) + +def get_visitor_data(force=False): + now = time.time() + with _vd_lock: + if not force and _vd["value"] and _vd["expires"] > now: + return _vd["value"] + body = {"videoId": "xgJ13Dt9F14", + "context": {"client": {"clientName": "WEB", "clientVersion": "2.20260101.00.00", + "hl": "ko", "gl": "KR"}}} + d = _post_player(body, _VD_UA) + vd = (d.get("responseContext") or {}).get("visitorData") or "" + if vd: + with _vd_lock: + _vd["value"] = vd + _vd["expires"] = time.time() + _VD_TTL + print("[visitorData] 갱신 완료(len=%d)" % len(vd), file=sys.stderr, flush=True) + return vd + +def _innertube_once(vid, vd): + ctx = {"clientName": "VISIONOS", "clientVersion": "1.02", "deviceMake": "Apple", + "deviceModel": "RealityDevice17,1", "osName": "visionOS", "osVersion": "26.5.23O471", + "hl": "ko", "gl": "KR", "visitorData": vd} + body = {"videoId": vid, "context": {"client": ctx}, "contentCheckOk": True, "racyCheckOk": True} + d = _post_player(body, _IT_UA, {"X-Goog-Visitor-Id": vd, + "X-Youtube-Client-Name": "101", "X-Youtube-Client-Version": "1.02"}) + status = d.get("playabilityStatus", {}).get("status") + if status != "OK": + return None, status + af = (d.get("streamingData") or {}).get("adaptiveFormats") or [] + au = [f for f in af if str(f.get("mimeType", "")).startswith("audio") and f.get("url")] + if not au: + return None, "NO_AUDIO" + au.sort(key=lambda f: f.get("bitrate", 0), reverse=True) + return au[0]["url"], status + +def innertube_resolve(target): + m = _VID_RE.search(target) + if not m: + return None # 검색어 등 영상 ID 가 없는 입력 -> 기존 경로 + vid = m.group(1) + vd = get_visitor_data() + if not vd: + return None + url, status = _innertube_once(vid, vd) + if url: + return url + # visitorData 만료로 인한 로그인 요구면 1회 재발급 후 재시도 + if status in ("LOGIN_REQUIRED", "ERROR"): + vd = get_visitor_data(force=True) + if vd: + url, status = _innertube_once(vid, vd) + if url: + return url + print("[innertube] 폴백(status=%s)" % status, file=sys.stderr, flush=True) + return None + def lib_resolve(target): # yt-dlp 를 라이브러리로 직접 호출: 매 요청 프로세스 기동(약 0.4s) 을 없앤다. q = target if (target.startswith("http") or target.startswith("ytsearch")) else "ytsearch1:" + target @@ -76,6 +153,16 @@ def first_url(out): def _do_resolve(target): + # 0순위: 내부 API 직접 호출(0.3초대). 실패하면 아래 기존 경로로 자동 폴백. + try: + _u = innertube_resolve(target) + if _u and _u.startswith("http"): + print("[result] OK(innertube) " + _u[:40], file=sys.stderr, flush=True) + with LOCK: + CACHE[target] = (_u, time.time() + TTL) + return _u + except Exception as e: + print("[it-fail] " + str(e)[:120], file=sys.stderr, flush=True) # 1순위: 라이브러리 상주 호출(가장 빠름). 실패하면 기존 subprocess 경로로 폴백. try: u = lib_resolve(target)