eric/gitlab-instance-0a899031_douyinyoutube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

's'

Browse Source
This commit is contained in:
eric
2026-03-28 17:34:28 -05:00
parent 46fb194a02
commit e050afbac1
8 changed files with 392 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/http_clients/async_http.py
View File

@@ -2,19 +2,19 @@
from __future__ import annotations
import httpx
from typing import Dict, Any
from typing import Any, Dict, List, Optional, Union
from .. import utils
OptionalStr = str | None
OptionalDict = Dict[str, Any] | None
OptionalStr = Optional[str]
OptionalDict = Optional[Dict[str, Any]]
async def async_req(
url: str,
proxy_addr: OptionalStr = None,
headers: OptionalDict = None,
data: dict | bytes | None = None,
json_data: dict | list | None = None,
data: Union[dict, bytes, None] = None,
json_data: Union[dict, List, None] = None,
timeout: int = 20,
redirect_url: bool = False,
return_cookies: bool = False,
@@ -23,7 +23,7 @@ async def async_req(
content_conding: str = 'utf-8',
verify: bool = False,
http2: bool = True
) -> OptionalDict | OptionalStr | tuple:
) -> Union[OptionalDict, OptionalStr, tuple]:
if headers is None:
headers = {}
try:

9
src/http_clients/sync_http.py
View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import gzip
from typing import List, Optional, Union
import urllib.parse
import urllib.error
import requests
@@ -15,16 +16,16 @@ opener = urllib.request.build_opener(no_proxy_handler)
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
OptionalStr = str | None
OptionalDict = dict | None
OptionalStr = Optional[str]
OptionalDict = Optional[dict]
def sync_req(
url: str,
proxy_addr: OptionalStr = None,
headers: OptionalDict = None,
data: dict | bytes | None = None,
json_data: dict | list | None = None,
data: Union[dict, bytes, None] = None,
json_data: Union[dict, List, None] = None,
timeout: int = 20,
redirect_url: bool = False,
abroad: bool = False,

18
src/spider.py
View File

@@ -19,7 +19,7 @@ import uuid
from operator import itemgetter
import urllib.parse
import urllib.error
from typing import List
from typing import List, Optional, Union
import httpx
import ssl
import re
@@ -37,8 +37,8 @@ from .ab_sign import ab_sign
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
OptionalStr = str | None
OptionalDict = dict | None
OptionalStr = Optional[str]
OptionalDict = Optional[dict]
def get_params(url: str, params: str) -> OptionalStr:
@@ -285,7 +285,7 @@ async def get_douyin_stream_data(url: str, proxy_addr: OptionalStr = None, cooki
@trace_error_decorator
async def get_tiktok_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict | None:
async def get_tiktok_stream_data(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> Optional[dict]:
headers = {
'referer': 'https://www.tiktok.com/',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
@@ -364,7 +364,7 @@ async def get_kuaishou_stream_data(url: str, proxy_addr: OptionalStr = None, coo
@trace_error_decorator
async def get_kuaishou_stream_data2(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> dict | None:
async def get_kuaishou_stream_data2(url: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> Optional[dict]:
headers = {
'User-Agent': 'ios/7.830 (ios 17.0; ; iPhone 15 (A2846/A3089/A3090/A3092))',
'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',
@@ -964,7 +964,7 @@ async def get_sooplive_cdn_url(broad_no: str, proxy_addr: OptionalStr = None, co
@trace_error_decorator
async def get_sooplive_tk(url: str, rtype: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> str | tuple:
async def get_sooplive_tk(url: str, rtype: str, proxy_addr: OptionalStr = None, cookies: OptionalStr = None) -> Union[str, tuple]:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0',
'Origin': 'https://play.sooplive.co.kr',
@@ -1557,7 +1557,7 @@ def get_looklive_secret_data(text) -> tuple:
charset = '1234567890abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ!@#$%^&*()_+-=[]{}|;:,.<>?'
return ''.join(secrets.choice(charset) for _ in range(size)).encode('utf-8')
def aes_encrypt(_text: str | bytes, _sec_key: str | bytes) -> bytes:
def aes_encrypt(_text: Union[str, bytes], _sec_key: Union[str, bytes]) -> bytes:
if isinstance(_text, str):
_text = _text.encode('utf-8')
if isinstance(_sec_key, str):
@@ -1570,7 +1570,7 @@ def get_looklive_secret_data(text) -> tuple:
encoded_ciphertext = base64.b64encode(ciphertext)
return encoded_ciphertext
def rsa_encrypt(_text: str | bytes, pub_key: str, mod: str) -> str:
def rsa_encrypt(_text: Union[str, bytes], pub_key: str, mod: str) -> str:
if isinstance(_text, str):
_text = _text.encode('utf-8')
text_reversed = _text[::-1]
@@ -2256,7 +2256,7 @@ async def get_liveme_stream_url(url: str, proxy_addr: OptionalStr = None, cookie
return result
async def get_huajiao_sn(url: str, cookies: OptionalStr = None, proxy_addr: OptionalStr = None) -> tuple | None:
async def get_huajiao_sn(url: str, cookies: OptionalStr = None, proxy_addr: OptionalStr = None) -> Optional[tuple]:
headers = {
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'referer': 'https://www.huajiao.com/',

3
src/stream.py
View File

@@ -20,6 +20,7 @@ import re
from operator import itemgetter
import urllib.parse
import urllib.request
from typing import Union
from .utils import trace_error_decorator
from .spider import (
get_douyu_stream_data, get_bilibili_stream_data
@@ -412,7 +413,7 @@ async def get_netease_stream_url(json_data: dict, video_quality: str) -> dict:
async def get_stream_url(json_data: dict, video_quality: str, url_type: str = 'm3u8', spec: bool = False,
hls_extra_key: str | int = None, flv_extra_key: str | int = None) -> dict:
hls_extra_key: Union[str, int, None] = None, flv_extra_key: Union[str, int, None] = None) -> dict:
if not json_data['is_live']:
return json_data

20
src/utils.py
View File

@@ -11,15 +11,15 @@ import functools
import hashlib
import re
import traceback
from typing import Any
from typing import Any, List, Optional, Union
from urllib.parse import parse_qs, urlparse
from collections import OrderedDict
import execjs
from .logger import logger
import configparser
OptionalStr = str | None
OptionalDict = dict | None
OptionalStr = Optional[str]
OptionalDict = Optional[dict]
class Color:
@@ -53,7 +53,7 @@ def trace_error_decorator(func: callable) -> callable:
return wrapper
def check_md5(file_path: str | Path) -> str:
def check_md5(file_path: Union[str, Path]) -> str:
with open(file_path, 'rb') as fp:
file_md5 = hashlib.md5(fp.read()).hexdigest()
return file_md5
@@ -64,7 +64,7 @@ def dict_to_cookie_str(cookies_dict: dict) -> str:
return cookie_str
def read_config_value(file_path: str | Path, section: str, key: str) -> str | None:
def read_config_value(file_path: Union[str, Path], section: str, key: str) -> Optional[str]:
config = configparser.ConfigParser()
try:
@@ -84,7 +84,7 @@ def read_config_value(file_path: str | Path, section: str, key: str) -> str | No
return None
def update_config(file_path: str | Path, section: str, key: str, new_value: str) -> None:
def update_config(file_path: Union[str, Path], section: str, key: str, new_value: str) -> None:
config = configparser.ConfigParser()
try:
@@ -137,7 +137,7 @@ def remove_emojis(text: str, replace_text: str = '') -> str:
return emoji_pattern.sub(replace_text, text)
def remove_duplicate_lines(file_path: str | Path) -> None:
def remove_duplicate_lines(file_path: Union[str, Path]) -> None:
unique_lines = OrderedDict()
text_encoding = 'utf-8-sig'
with open(file_path, 'r', encoding=text_encoding) as input_file:
@@ -148,7 +148,7 @@ def remove_duplicate_lines(file_path: str | Path) -> None:
output_file.write(line + '\n')
def check_disk_capacity(file_path: str | Path, show: bool = False) -> float:
def check_disk_capacity(file_path: Union[str, Path], show: bool = False) -> float:
absolute_path = os.path.abspath(file_path)
directory = os.path.dirname(absolute_path)
disk_usage = shutil.disk_usage(directory)
@@ -188,7 +188,7 @@ def jsonp_to_json(jsonp_str: str) -> OptionalDict:
raise Exception("No JSON data found in JSONP response.")
def replace_url(file_path: str | Path, old: str, new: str) -> None:
def replace_url(file_path: Union[str, Path], old: str, new: str) -> None:
with open(file_path, 'r', encoding='utf-8-sig') as f:
content = f.read()
if old in content:
@@ -196,7 +196,7 @@ def replace_url(file_path: str | Path, old: str, new: str) -> None:
f.write(content.replace(old, new))
def get_query_params(url: str, param_name: OptionalStr) -> dict | list[str]:
def get_query_params(url: str, param_name: OptionalStr) -> Union[dict, List[str]]:
parsed_url = urlparse(url)
query_params = parse_qs(parsed_url.query)