frxz751113/iptv
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update 酒店源.py

Browse Source
This commit is contained in:
frxz751113
2024-08-29 21:05:12 +08:00
committed by GitHub
parent 5ab3595938
commit 03a44449ca
1 changed files with 74 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/酒店源.py
+74 -69
View File
@@ -53,8 +53,8 @@ def is_url_accessible(url):
response = requests.get(url, timeout=1) response = requests.get(url, timeout=1)
if 200 <= response.status_code <= 401: if 200 <= response.status_code <= 401:
return url return url
except requests.exceptions.RequestException: except requests.exceptions.RequestException as e:
pass print(f"Error checking {url}: {e}")
return None return None
@@ -66,76 +66,81 @@ for url in urls:
chrome_options.add_argument('--no-sandbox') chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--disable-dev-shm-usage') chrome_options.add_argument('--disable-dev-shm-usage')
driver = webdriver.Chrome(options=chrome_options) driver = webdriver.Chrome(options=chrome_options)
# 使用 WebDriver 访问网页 try:
driver.get(url) # 使用 WebDriver 访问网页
time.sleep(10) driver.get(url)
# 获取网页内容 time.sleep(10)
page_content = driver.page_source # 获取网页内容
# 关闭 WebDriver page_content = driver.page_source
driver.quit() # 关闭 WebDriver
# 查找所有符合指定格式的网址 driver.quit()
pattern = r"http://\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d+" # 查找所有符合指定格式的网址
urls_all = re.findall(pattern, page_content) pattern = r"http://\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d+"
urls = set(urls_all) urls_all = re.findall(pattern, page_content)
x_urls = [] urls = set(urls_all)
for url in urls: x_urls = []
url = url.strip()
ip_start_index = url.find("//") + 2
ip_end_index = url.find(":", ip_start_index)
ip_dot_start = url.find(".") + 1
ip_dot_second = url.find(".", ip_dot_start) + 1
ip_dot_three = url.find(".", ip_dot_second) + 1
base_url = url[:ip_start_index]
ip_address = url[ip_start_index:ip_dot_three]
port = url[ip_end_index:]
ip_end = "1"
modified_ip = f"{ip_address}{ip_end}"
x_url = f"{base_url}{modified_ip}{port}"
x_urls.append(x_url)
urls = set(x_urls)
valid_urls = []
# 多线程获取可用 url
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
futures = []
for url in urls: for url in urls:
url = url.strip() url = url.strip()
modified_urls = modify_urls(url) ip_start_index = url.find("//") + 2
for modified_url in modified_urls: ip_end_index = url.find(":", ip_start_index)
futures.append(executor.submit(is_url_accessible, modified_url)) ip_dot_start = url.find(".") + 1
for future in concurrent.futures.as_completed(futures): ip_dot_second = url.find(".", ip_dot_start) + 1
result = future.result() ip_dot_three = url.find(".", ip_dot_second) + 1
if result: base_url = url[:ip_start_index]
valid_urls.append(result) ip_address = url[ip_start_index:ip_dot_three]
for url in valid_urls: port = url[ip_end_index:]
print(url) ip_end = "1"
# 遍历网址列表，获取 JSON 文件并解析 modified_ip = f"{ip_address}{ip_end}"
for url in valid_urls: x_url = f"{base_url}{modified_ip}{port}"
try: x_urls.append(x_url)
json_url = f"{url}" urls = set(x_urls)
response = requests.get(json_url, timeout=1) valid_urls = []
json_data = response.content.decode('utf-8') # 多线程获取可用 url
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
futures = []
for url in urls:
url = url.strip()
modified_urls = modify_urls(url)
for modified_url in modified_urls:
futures.append(executor.submit(is_url_accessible, modified_url))
for future in concurrent.futures.as_completed(futures):
result = future.result()
if result:
valid_urls.append(result)
for url in valid_urls:
print(url)
# 遍历网址列表，获取 JSON 文件并解析
for url in valid_urls:
try: try:
lines = json_data.split('\n') json_url = f"{url}"
for line in lines: response = requests.get(json_url, timeout=1)
if 'hls' in line and ('udp' not in line or 'rtp' not in line): json_data = response.content.decode('utf-8')
line = line.strip() try:
if line: lines = json_data.split('\n')
name, channel_url = line.split(',') for line in lines:
urls = channel_url.split('/', 3) if 'hls' in line and ('udp' not in line or 'rtp' not in line):
url_data = json_url.split('/', 3) line = line.strip()
ip_port = url_data[2] if line:
urls[2] = urls[2].replace(urls[2].split('/')[2].split(':')[0], ip_port.split(':')[0]) name, channel_url = line.split(',')
urls[2] = urls[2].replace(urls[2].split('/')[2].split(':')[1], ip_port.split(':')[1]) urls = channel_url.split('/', 3)
if len(urls) >= 4: url_data = json_url.split('/', 3)
urld = (f"{urls[0]}//{urls[2]}/{urls[3]}") ip_port = url_data[2]
else: urls[2] = urls[2].replace(urls[2].split('/')[2].split(':')[0], ip_port.split(':')[0])
urld = (f"{urls[0]}//{urls[2]}") urls[2] = urls[2].replace(urls[2].split('/')[2].split(':')[1], ip_port.split(':')[1])
with open('iptv.txt', 'a', encoding='utf-8') as outfile: if len(urls) >= 4:
outfile.write(f"{name},{urld}\n") urld = (f"{urls[0]}//{urls[2]}/{urls[3]}")
except: else:
pass urld = (f"{urls[0]}//{urls[2]}")
except: with open('iptv.txt', 'a', encoding='utf-8') as outfile:
pass outfile.write(f"{name},{urld}\n")
except Exception as e:
print(f"Error processing line in JSON: {e}")
except Exception as e:
print(f"Error fetching JSON from {json_url}: {e}")
except Exception as e:
print(f"Error with URL {url}: {e}")
print("频道列表文件 iptv.txt 获取完成！") print("频道列表文件 iptv.txt 获取完成！")