Devbranch

Dockerfile.simple

@@ -0,0 +1,14 @@
+FROM python:3.9-slim
+LABEL maintainer="Naihe <239144498@qq.com>"
+WORKDIR /code
+
+COPY ./requirements.txt /code/requirements.txt
+
+RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && \
+    echo 'Asia/Shanghai' >/etc/timezone && \
+    pip install --no-cache-dir --upgrade -r /code/requirements.txt
+COPY ./app /code/app
+
+EXPOSE 8080
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]

app/api/v2/endpoints/more.py

@@ -12,7 +12,7 @@
 from fastapi.responses import FileResponse
 from fastapi.requests import Request
 from starlette.responses import StreamingResponse
-from app.plugins.a4gtv.more_util import parse, processing, splicing
+from app.plugins.a4gtv.more_util import parse, processing, splicing, get_ytb
 from app.conf import config
 from app.conf.config import headers
 from app.scheams.response import Response200, Response400
@@ -77,6 +77,26 @@ async def pdl(request: Request, url: str = Query(...)):
             return Response(content=await res.read(), status_code=200, headers=headers, media_type='video/MP2T')
 
 
+@more.get('/ytb', summary="获取youtube直播源")
+async def ytb(request: Request, url: str = Query(...)):
+    """
+    从youtube直播页面获取直播源，输入url作为参数即可
+
+    例如，直播页面地址为"https://www.youtube.com/watch?v=c6lkorBJ1LY&ab_channel=SPACE%28Official%29"
+
+    使用/ytb接口即可作为iptv直播源，如"http://域名:8080/ytb?url=https://www.youtube.com/watch?v=c6lkorBJ1LY&ab_channel=SPACE%28Official%29"
+    - **url**: 视频链接
+    """
+    header = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:104.0) Gecko/20100101 Firefox/104.0",
+        "Accept": "*/*",
+        "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",
+        "Upgrade-Insecure-Requests": "1",
+    }
+    data = get_ytb(url,headers)
+    print(data)
+    return Response(data, status_code=200)
+
 @more.get('/count', summary="统计")
 async def count1():
     """

app/conf/config.py

@@ -138,4 +138,6 @@ class Config(BaseSettings):
 if localhost and "http" not in localhost:
     logger.warning("localhost配置错误，具体查看教程https://www.cnblogs.com/1314h/p/16651157.html")
 
+ytb_stream = {}
+
 logger.info("配置加载完成")

app/plugins/a4gtv/more_util.py

@@ -5,10 +5,13 @@
 # @File    : more_util.py
 # @Software: PyCharm
 import re
+import time
+import requests
 from base64 import b64encode
-from urllib.parse import urlencode, unquote, quote_plus, urlparse, parse_qsl, urlunparse, urljoin
+from urllib.parse import urlencode, unquote, quote_plus, urlparse, parse_qsl, urlunparse, urljoin, parse_qs
 from app.conf import config
-
+from loguru import logger
+from app.conf.config import ytb_stream
 
 def is_url(url):
     regex = re.compile(config.url_regex)
@@ -26,15 +29,58 @@ def parse(url):
 
 def processing(url, data):
     for _temp in data:
-        if ".ts" in _temp:
+        if ".ts" in _temp or "googlevideo.com/videoplayback" in _temp :
             if is_url(_temp):
                 yield "/pdl?url=" + b64encode(_temp.encode("utf-8")).decode("utf-8")
             else:
                 yield "/pdl/?url=" + b64encode(urljoin(url, _temp).encode("utf-8")).decode("utf-8")
+        # 补充/proxy接口，如果获取的是不含域名的m3u8列表，则拼出完整地址        
+        elif len(_temp)>4:
+            if _temp[0] != '#' and _temp[:4] != 'http':
+                yield "/proxy?url=" + urljoin(url, _temp)
+            else:
+                yield _temp
         else:
             yield _temp
         yield "\n"
 
+def update_ytb(url,headers,stream_id):
+    response = requests.get(url, headers=headers).text
+    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())}")
+
+
+def get_ytb(url,headers):
+    try:
+        query = urlparse(url).query
+        query_dict = parse_qs(query)
+        stream_id = query_dict['v'][0]
+    except:
+        msg = f"url格式解析失败 {url}"
+        logger.error(msg)
+        return msg
+    if ytb_stream.get(stream_id) is not None:
+        if ytb_stream[stream_id]["expire"] > time.time()-60:            
+            logger.success(f"从缓存加载直播源 {url}")
+        else:
+            logger.info(f"youtube直播源已过期 {url}")
+            update_ytb(url,headers,stream_id)
+    else:
+        logger.info(f"youtube直播源不存在 {url}")
+        update_ytb(url,headers,stream_id)
+
+    stream_url = "/proxy?url="+ytb_stream[stream_id]['stream_url']
+    data = "#EXTM3U\n#EXTINF:\n"+stream_url
+    return data
+
+
+
+
 
 def splicing(url, query_params):
     url_parsed = list(urlparse(url))