Skip to content

Commit

Permalink
feat:fofa multicast
Browse files Browse the repository at this point in the history
  • Loading branch information
@Guovin
Guovin committed Sep 3, 2024
1 parent daa9fd1 commit c6fb90d
Show file tree
Hide file tree
Showing 10 changed files with 283 additions and 177 deletions.
4 changes: 3 additions & 1 deletion config/config.ini
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,12 @@ url_keywords_blacklist =
open_subscribe = False
subscribe_urls = https://m3u.ibert.me/txt/fmml_dv6.txt,https://m3u.ibert.me/txt/o_cn.txt,https://m3u.ibert.me/txt/j_iptv.txt,https://github.moeyy.xyz/https://raw.githubusercontent.com/PizazzGY/TVBox/main/live.txt
open_multicast = True
open_multicast_tonkiang = True
open_multicast_fofa = True
multicast_region_list = 全部
multicast_page_num = 3
open_proxy = False
open_driver = False
open_driver = True
open_hotel = True
open_hotel_tonkiang = False
open_hotel_fofa = True
Expand Down
64 changes: 33 additions & 31 deletions docs/config.md
View file

Large diffs are not rendered by default.

64 changes: 33 additions & 31 deletions docs/config_en.md
View file

Large diffs are not rendered by default.

88 changes: 44 additions & 44 deletions main.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,10 +95,10 @@ async def visit_page(self, channel_names=None):
task_func(subscribe_urls, callback=self.update_progress)
)
elif setting == "open_hotel_tonkiang" or setting == "open_hotel_fofa":
task = asyncio.create_task(task_func(self.update_progress))
task = asyncio.create_task(task_func(callback=self.update_progress))
else:
task = asyncio.create_task(
task_func(channel_names, self.update_progress)
task_func(channel_names, callback=self.update_progress)
)
self.tasks.append(task)
setattr(self, result_attr, await task)
Expand Down Expand Up @@ -155,48 +155,48 @@ async def main(self):
self.channel_data,
callback=sort_callback,
)
no_result_cate_names = [
(cate, name)
for cate, channel_obj in self.channel_data.items()
for name, info_list in channel_obj.items()
if len(info_list) < 3
]
no_result_names = [name for (_, name) in no_result_cate_names]
if no_result_names:
print(
f"Not enough url found for {', '.join(no_result_names)}, try a supplementary multicast search..."
)
sup_results = await get_channels_by_multicast(
no_result_names, self.update_progress
)
sup_channel_items = defaultdict(lambda: defaultdict(list))
for cate, name in no_result_cate_names:
data = sup_results.get(name)
if data:
sup_channel_items[cate][name] = data
self.total = len(
[
url
for obj in sup_channel_items.values()
for url_list in obj.values()
for url in url_list
]
)
if self.total > 0 and config.getboolean("Settings", "open_sort"):
self.update_progress(
f"正在对补充频道测速排序, 共{len([name for obj in sup_channel_items.values() for name in obj.keys()])}个频道, 含{self.total}个接口",
0,
)
self.start_time = time()
self.pbar = tqdm_asyncio(total=self.total, desc="Sorting")
sup_channel_items = await process_sort_channel_list(
sup_channel_items,
callback=sort_callback,
)
self.channel_data = merge_objects(
self.channel_data, sup_channel_items
)
self.total = self.get_urls_len()
# no_result_cate_names = [
# (cate, name)
# for cate, channel_obj in self.channel_data.items()
# for name, info_list in channel_obj.items()
# if len(info_list) < 3
# ]
# no_result_names = [name for (_, name) in no_result_cate_names]
# if no_result_names:
# print(
# f"Not enough url found for {', '.join(no_result_names)}, try a supplementary multicast search..."
# )
# sup_results = await get_channels_by_multicast(
# no_result_names, self.update_progress
# )
# sup_channel_items = defaultdict(lambda: defaultdict(list))
# for cate, name in no_result_cate_names:
# data = sup_results.get(name)
# if data:
# sup_channel_items[cate][name] = data
# self.total = len(
# [
# url
# for obj in sup_channel_items.values()
# for url_list in obj.values()
# for url in url_list
# ]
# )
# if self.total > 0 and config.getboolean("Settings", "open_sort"):
# self.update_progress(
# f"正在对补充频道测速排序, 共{len([name for obj in sup_channel_items.values() for name in obj.keys()])}个频道, 含{self.total}个接口",
# 0,
# )
# self.start_time = time()
# self.pbar = tqdm_asyncio(total=self.total, desc="Sorting")
# sup_channel_items = await process_sort_channel_list(
# sup_channel_items,
# callback=sort_callback,
# )
# self.channel_data = merge_objects(
# self.channel_data, sup_channel_items
# )
# self.total = self.get_urls_len()
self.pbar = tqdm(total=self.total, desc="Writing")
self.start_time = time()
write_channel_to_file(
Expand Down
53 changes: 36 additions & 17 deletions updates/fofa/request.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from utils.tools import merge_objects, get_pbar_remaining
from updates.proxy import get_proxy, get_proxy_next
from requests_custom.utils import get_source_requests, close_session
from collections import defaultdict

timeout = 10

Expand All @@ -22,7 +23,7 @@ def get_fofa_urls_from_region_list():
# region_list = config.get("Settings", "hotel_region_list").split(",")
urls = []
region_url = getattr(fofa_map, "region_url")
# if "all" in region_list or "全部" in region_list:
# if "all" in region_list or "ALL" in region_list or "全部" in region_list:
urls = [url for url_list in region_url.values() for url in url_list if url]
# else:
# for region in region_list:
Expand All @@ -31,25 +32,35 @@ def get_fofa_urls_from_region_list():
return urls


async def get_channels_by_fofa(callback):
async def get_channels_by_fofa(urls=None, multicast=False, callback=None):
"""
Get the channel by FOFA
"""
fofa_urls = get_fofa_urls_from_region_list()
fofa_urls = urls if urls else get_fofa_urls_from_region_list()
fofa_urls_len = len(fofa_urls)
pbar = tqdm_asyncio(total=fofa_urls_len, desc="Processing fofa")
pbar = tqdm_asyncio(
total=fofa_urls_len,
desc=f"Processing fofa {'for multicast' if multicast else 'for hotel'}",
)
start_time = time()
fofa_results = {}
callback(f"正在获取Fofa源更新, 共{fofa_urls_len}个地区", 0)
mode_name = {"组播" if multicast else "酒店"}
if callback:
callback(
f"正在获取Fofa{mode_name}源更新, 共{fofa_urls_len}个查询地址",
0,
)
proxy = None
open_proxy = config.getboolean("Settings", "open_proxy")
open_driver = config.getboolean("Settings", "open_driver")
if open_proxy:
proxy = await get_proxy(fofa_urls[0], best=True, with_test=True)
test_url = fofa_urls[0][0] if multicast else fofa_urls[0]
proxy = await get_proxy(test_url, best=True, with_test=True)

def process_fofa_channels(fofa_url):
def process_fofa_channels(fofa_info):
nonlocal proxy, fofa_urls_len, open_driver
results = {}
fofa_url = fofa_info[0] if multicast else fofa_info
results = defaultdict(lambda: defaultdict(list))
try:
if open_driver:
driver = setup_driver(proxy)
Expand All @@ -69,11 +80,18 @@ def process_fofa_channels(fofa_url):
)
fofa_source = re.sub(r"<!--.*?-->", "", page_source, flags=re.DOTALL)
urls = set(re.findall(r"https?://[\w\.-]+:\d+", fofa_source))

with ThreadPoolExecutor(max_workers=100) as executor:
futures = [executor.submit(process_fofa_json_url, url) for url in urls]
for future in futures:
results = merge_objects(results, future.result())
if multicast:
region = fofa_info[1]
type = fofa_info[2]
multicast_result = [(url, None, None) for url in urls]
results[region][type] = multicast_result
else:
with ThreadPoolExecutor(max_workers=100) as executor:
futures = [
executor.submit(process_fofa_json_url, url) for url in urls
]
for future in futures:
results = merge_objects(results, future.result())
except Exception as e:
print(e)
finally:
Expand All @@ -82,10 +100,11 @@ def process_fofa_channels(fofa_url):
driver.quit()
pbar.update()
remain = fofa_urls_len - pbar.n
callback(
f"正在获取Fofa源更新, 剩余{remain}个地区待获取, 预计剩余时间: {get_pbar_remaining(n=pbar.n, total=pbar.total, start_time=start_time)}",
int((pbar.n / fofa_urls_len) * 100),
)
if callback:
callback(
f"正在获取Fofa{mode_name}源更新, 剩余{remain}个查询地址待获取, 预计剩余时间: {get_pbar_remaining(n=pbar.n, total=pbar.total, start_time=start_time)}",
int((pbar.n / fofa_urls_len) * 100),
)
return results

max_workers = 3 if open_driver else 10
Expand Down
27 changes: 16 additions & 11 deletions updates/hotel/request.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
import updates.fofa.fofa_map as fofa_map


async def get_channels_by_hotel(callback):
async def get_channels_by_hotel(callback=None):
"""
Get the channels by multicase
"""
Expand All @@ -36,7 +36,7 @@ async def get_channels_by_hotel(callback):
open_driver = config.getboolean("Settings", "open_driver")
page_num = config.getint("Settings", "hotel_page_num")
region_list = config.get("Settings", "hotel_region_list").split(",")
if "all" in region_list or "全部" in region_list:
if "all" in region_list or "ALL" in region_list or "全部" in region_list:
fofa_region_name_list = list(getattr(fofa_map, "region_url").keys())
region_list = fofa_region_name_list
if open_proxy:
Expand All @@ -51,7 +51,10 @@ def process_region_by_hotel(region):
if open_driver:
driver = setup_driver(proxy)
try:
retry_func(lambda: driver.get(pageUrl), name=f"hotel search:{name}")
retry_func(
lambda: driver.get(pageUrl),
name=f"Tonkiang hotel search:{name}",
)
except Exception as e:
if open_proxy:
proxy = get_proxy_next()
Expand All @@ -67,7 +70,7 @@ def process_region_by_hotel(region):
try:
page_soup = retry_func(
lambda: get_soup_requests(pageUrl, data=post_form, proxy=proxy),
name=f"hotel search:{name}",
name=f"Tonkiang hotel search:{name}",
)
except Exception as e:
if open_proxy:
Expand Down Expand Up @@ -141,15 +144,17 @@ def process_region_by_hotel(region):
driver.close()
driver.quit()
pbar.update()
callback(
f"正在进行酒店源更新, 剩余{region_list_len - pbar.n}个地区待查询, 预计剩余时间: {get_pbar_remaining(n=pbar.n, total=pbar.total, start_time=start_time)}",
int((pbar.n / region_list_len) * 100),
)
if callback:
callback(
f"正在进行Tonkiang酒店源更新, 剩余{region_list_len - pbar.n}个地区待查询, 预计剩余时间: {get_pbar_remaining(n=pbar.n, total=pbar.total, start_time=start_time)}",
int((pbar.n / region_list_len) * 100),
)
return {"region": region, "type": type, "data": info_list}

region_list_len = len(region_list)
pbar = tqdm_asyncio(total=region_list_len, desc="Hotel search")
callback(f"正在进行酒店源更新, 共{region_list_len}个地区", 0)
pbar = tqdm_asyncio(total=region_list_len, desc="Tonkiang hotel search")
if callback:
callback(f"正在进行Tonkiang酒店源更新, 共{region_list_len}个地区", 0)
search_region_result = defaultdict(list)
with ThreadPoolExecutor(max_workers=3) as executor:
futures = {
Expand All @@ -174,7 +179,7 @@ def process_region_by_hotel(region):
for url, _, _ in result
]
channels = await get_channels_by_subscribe_urls(
urls, retry=False, error_print=False, with_cache=True
urls, hotel=True, retry=False, error_print=False, with_cache=True
)
if not open_driver:
close_session()
Expand Down
Loading

0 comments on commit c6fb90d

Please sign in to comment.