补充/proxy接口

app/api/v2/endpoints/proxy.py

@@ -63,4 +63,35 @@ async def Stream_Proxy_Pro(request: Request):
     """
         ## 接口文档
     """
-    return RedirectResponse("http://proxy.naihe.me/docs")
+    return RedirectResponse("http://proxy.naihe.me/docs")
+
+
+@tv.get('/ytb.m3u8')
+async def ytb(request: Request,
+              url: str = Query(..., regex="^[0-9A-Za-z]{11}$")):
+    """
+        ## 用途/Usage
+        - 代理youtube直播链接
+        - 输入参数为11位ID，如`https://www.youtube.com/watch?v=abcdEFg12H&ab_channel=SPACE%28Official%29`,ID是`abcdEFg12H`
+    """
+    url = dict(request.query_params)
+    if url.get("url"):
+        stream_id = parse(url)
+        stream_url = get_ytb(stream_id)
+        _data = get_m3u8_down(stream_url)
+        return StreamingResponse(processing(stream_url, iter(_data.split("\n"))), 200, headers=headers2)
+
+@tv.get('/ytbproxy', summary="代理youtube m3u8")
+async def ytbproxy(request: Request, url: str):
+    """
+    代理youtube链接
+    """
+    url = dict(request.query_params)
+    if url.get("url"):
+        url = url.get("url")
+        url = b64decode(url.encode("utf-8")).decode("utf-8")
+        url = url.replace('=','%3D')
+        url = url.replace(';','%3B')
+
+        _data = get_m3u8_down(url)
+        return StreamingResponse(processing(url, iter(_data.split("\n"))), 200, headers=headers2)

app/common/cache_tools.py

@@ -20,12 +20,21 @@
 async def processing(url, data):
     for _temp in data:
         if ".ts" in _temp:
-            if not is_url(_temp):
+            if is_url(_temp):
                 yield "/file.ts?x=" + b64encode(_temp.encode("utf-8")).decode("utf-8")
             else:
                 yield "/file.ts/?x=" + b64encode(urljoin(url, _temp).encode("utf-8")).decode("utf-8")
         else:
-            yield _temp
+            if _temp.startswith("#") or len(_temp)==0:
+                yield _temp
+            # 如果获取的是m3u8列表，则继续进行代理获取
+            elif is_url(_temp):
+                if "googlevideo.com" in _temp:
+                    yield "/ytbproxy?url="+b64encode(_temp.encode("utf-8")).decode("utf-8")
+                else:
+                    yield "/proxy.m3u8?url=" + _temp
+            else:
+                yield "/proxy.m3u8?url=" + urljoin(url, _temp)
         yield "\n"
 
 
@@ -48,3 +57,38 @@ def get_m3u8_down(url):
     with requests.get(url=url, headers=headers) as res:
         _data = res.text
         return _data
+
+
+def get_ytb(stream_id):    
+    ytburl = f"https://www.youtube.com/watch?v={stream_id}"
+    if ytb_stream.get(stream_id) is not None:
+        if ytb_stream[stream_id]["expire"] > time.time()-60:            
+            logger.success(f"从缓存加载直播源 {stream_id}")
+        else:
+            logger.info(f"youtube直播源已过期 {stream_id}")
+            update_ytb(ytburl,stream_id)
+    else:
+        logger.info(f"youtube直播源不存在 {stream_id}")
+        update_ytb(ytburl,stream_id)        
+    stream_url = ytb_stream[stream_id]['stream_url']
+    return stream_url
+
+
+# 获取YouTube直播源
+def update_ytb(url,stream_id):
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:104.0) Gecko/20100101 Firefox/104.0"
+    }
+    print(url)
+    response = requests.get(url, headers=headers).text
+
+    #20230115有效，直接获取链接
+    pattern = re.search(r'hlsManifestUrl.+?(https://.+?m3u8)',response)
+    stream_url = pattern.group(1)
+    pattern2 = re.search(r'expire/(\d{10})',stream_url)
+    expire = int(pattern2.group(1)) 
+
+    ytb_stream[stream_id] = {"expire":expire,"url":url,"stream_url":stream_url}
+    logger.success(f"youtube直播源已更新 {ytb_stream[stream_id]}")
+    logger.success(f"总数 {len(ytb_stream.keys())}")
+

app/conf/config.py

@@ -99,5 +99,8 @@ class Config(BaseSettings):
 ts_info = {}
 
 gdata = None
+
+ytb_stream = {}
+
 logger.info("配置加载完成")
 

app/plugins/proxy/tools.py

@@ -52,9 +52,9 @@ def md5(s):
 def is_url(url):
     regex = re.compile(config.url_regex)
     if regex.match(url):
-        return False
-    else:
         return True
+    else:
+        return False
 
 
 def parse(url):