From 03cf44cba3691898065c6f237f44fa84f8080cab Mon Sep 17 00:00:00 2001 From: zand3r Date: Mon, 17 Oct 2022 00:00:11 -0400 Subject: [PATCH 1/5] switched from queue to manager.list for better control of what goes in the queue when --- inb4404.py | 235 +++++++++++++++++++++++++++++++---------------------- 1 file changed, 138 insertions(+), 97 deletions(-) diff --git a/inb4404.py b/inb4404.py index 6d60325..4c2cb91 100755 --- a/inb4404.py +++ b/inb4404.py @@ -1,13 +1,24 @@ #!/usr/bin/python3 import urllib.request, urllib.error, urllib.parse, argparse, logging import os, re, time -import http.client +import http.client import fileinput from multiprocessing import Process +from multiprocessing import Lock, Queue, current_process, Manager +import queue + log = logging.getLogger('inb4404') workpath = os.path.dirname(os.path.realpath(__file__)) args = None +call_download_thread_while_loop_sleep_time = .20 +download_from_file_while_loop_sleep_time = .20 +queue_cleanup_timer = 300 #in seconds, how often to check for dead links and mark them dead in the config file +thread_check_timer = 20 #in seconds, how often to queue up all threads to check for new content +manager = Manager() +tasks_to_accomplish = manager.list() +links_to_remove = manager.list() #queue used to keep track of threads to remove from config + def main(): global args @@ -19,12 +30,13 @@ def main(): parser.add_argument('-n', '--use-names', action='store_true', help='use thread names instead of the thread ids (...4chan.org/board/thread/thread-id/thread-name)') parser.add_argument('-r', '--reload', action='store_true', help='reload the queue file every 5 minutes') parser.add_argument('-t', '--title', action='store_true', help='save original filenames') + parser.add_argument('-p', '--parallel-threads', type=int, default=4, help='Number of parallel threads to run at once. Default is 4') args = parser.parse_args() if args.date: logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(message)s', datefmt='%Y-%m-%d %I:%M:%S %p') else: - logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(message)s', datefmt='%I:%M:%S %p') + logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(message)s', datefmt='%I:%M:%S %p') thread = args.thread[0].strip() if thread[:4].lower() == 'http': @@ -43,6 +55,7 @@ def load(url): req = urllib.request.Request(url, headers={'User-Agent': '4chan Browser'}) return urllib.request.urlopen(req).read() + def get_title_list(html_content): ret = list() @@ -59,127 +72,155 @@ def get_title_list(html_content): return ret -def call_download_thread(thread_link, args): - try: - download_thread(thread_link, args) - except KeyboardInterrupt: - pass + +def call_download_thread(que, links_to_remove, running_tasks): + while True: + try: + #time.sleep(call_download_thread_while_loop_sleep_time) + #thread_link, args = que.get() + if len(que) == 0: + time.sleep(call_download_thread_while_loop_sleep_time) + continue + thread_link = que.pop(0) + running_tasks.append(thread_link) + download_thread(thread_link, links_to_remove) + running_tasks.remove(thread_link) + except KeyboardInterrupt: + break + except queue.Empty: + pass return -def download_thread(thread_link, args): + +def download_thread(thread_link, links_to_remove): + print("running!") board = thread_link.split('/')[3] thread = thread_link.split('/')[5].split('#')[0] if len(thread_link.split('/')) > 6: thread_tmp = thread_link.split('/')[6].split('#')[0] - if args.use_names or os.path.exists(os.path.join(workpath, 'downloads', board, thread_tmp)): + if args.use_names or os.path.exists(os.path.join(workpath, 'downloads', board, thread_tmp)): thread = thread_tmp - while True: - try: - regex = '(\/\/i(?:s|)\d*\.(?:4cdn|4chan)\.org\/\w+\/(\d+\.(?:jpg|png|gif|webm)))' - html_result = load(thread_link).decode('utf-8') - regex_result = list(set(re.findall(regex, html_result))) + try: + regex = '(\/\/i(?:s|)\d*\.(?:4cdn|4chan)\.org\/\w+\/(\d+\.(?:jpg|png|gif|webm)))' + html_result = load(thread_link).decode('utf-8') + regex_result = list(set(re.findall(regex, html_result))) - directory = os.path.join(workpath, 'downloads', board, thread) - if not os.path.exists(directory): - os.makedirs(directory) + directory = os.path.join(workpath, 'downloads', board, thread) + if not os.path.exists(directory): + os.makedirs(directory) - regex_result = sorted(regex_result, key=lambda tup: tup[1]) - regex_result_len = len(regex_result) - regex_result_cnt = 1 + regex_result = sorted(regex_result, key=lambda tup: tup[1]) + regex_result_len = len(regex_result) + regex_result_cnt = 1 - if args.title: - all_titles = get_title_list(html_result) + if args.title: + all_titles = get_title_list(html_result) - for enum_index, enum_tuple in enumerate(regex_result): - link, img = enum_tuple + for enum_index, enum_tuple in enumerate(regex_result): + link, img = enum_tuple - if args.title: - img = all_titles[enum_index] + if args.title: + img = all_titles[enum_index] + + img_path = os.path.join(directory, img) + if not os.path.exists(img_path): + data = load('https:' + link) + + output_text = board + '/' + thread + '/' + img + if args.with_counter: + output_text = '[' + str(regex_result_cnt).rjust(len(str(regex_result_len))) + '/' + str(regex_result_len) + '] ' + output_text + + log.info(output_text) + + with open(img_path, 'wb') as f: + f.write(data) + + ################################################################################## + # saves new images to a seperate directory + # if you delete them there, they are not downloaded again + # if you delete an image in the 'downloads' directory, it will be downloaded again + copy_directory = os.path.join(workpath, 'new', board, thread) + if not os.path.exists(copy_directory): + os.makedirs(copy_directory) + copy_path = os.path.join(copy_directory, img) + with open(copy_path, 'wb') as f: + f.write(data) + ################################################################################## + regex_result_cnt += 1 + + except urllib.error.HTTPError: + time.sleep(10) + try: + load(thread_link) + except urllib.error.HTTPError: + log.info('%s 404\'d', thread_link) + links_to_remove.append(thread_link) + except (urllib.error.URLError, http.client.BadStatusLine, http.client.IncompleteRead): + log.fatal(thread_link + ' crashed!') + raise - img_path = os.path.join(directory, img) - if not os.path.exists(img_path): - data = load('https:' + link) + if not args.less: + log.info('Checking ' + board + '/' + thread) - output_text = board + '/' + thread + '/' + img - if args.with_counter: - output_text = '[' + str(regex_result_cnt).rjust(len(str(regex_result_len))) + '/' + str(regex_result_len) + '] ' + output_text + return True - log.info(output_text) - with open(img_path, 'wb') as f: - f.write(data) +def download_from_file(filename): + running_links = [] + last_config_reload = time.time() + last_queue_check = time.time() + running_tasks = [] - ################################################################################## - # saves new images to a seperate directory - # if you delete them there, they are not downloaded again - # if you delete an image in the 'downloads' directory, it will be downloaded again - copy_directory = os.path.join(workpath, 'new', board, thread) - if not os.path.exists(copy_directory): - os.makedirs(copy_directory) - copy_path = os.path.join(copy_directory, img) - with open(copy_path, 'wb') as f: - f.write(data) - ################################################################################## - regex_result_cnt += 1 + processes = [] - except urllib.error.HTTPError: - time.sleep(10) - try: - load(thread_link) - except urllib.error.HTTPError: - log.info('%s 404\'d', thread_link) - break - continue - except (urllib.error.URLError, http.client.BadStatusLine, http.client.IncompleteRead): - log.fatal(thread_link + ' crashed!') - raise - - if not args.less: - log.info('Checking ' + board + '/' + thread) - time.sleep(20) + for w in range(args.parallel_threads): + p = Process(target=call_download_thread, args=(tasks_to_accomplish, links_to_remove, running_tasks)) + processes.append(p) + p.start() -def download_from_file(filename): - running_links = [] - while True: - processes = [] - for link in [_f for _f in [line.strip() for line in open(filename) if line[:4] == 'http'] if _f]: - if link not in running_links: - running_links.append(link) - log.info('Added ' + link) - - process = Process(target=call_download_thread, args=(link, args, )) - process.start() - processes.append([process, link]) - - if len(processes) == 0: - log.warning(filename + ' empty') - - if args.reload: - time.sleep(60 * 5) # 5 minutes - links_to_remove = [] - for process, link in processes: - if not process.is_alive(): - links_to_remove.append(link) - else: - process.terminate() - - for link in links_to_remove: - for line in fileinput.input(filename, inplace=True): - print(line.replace(link, '-' + link), end='') - running_links.remove(link) - log.info('Removed ' + link) - if not args.less: - log.info('Reloading ' + args.thread[0]) # thread = filename here; reloading on next loop - else: - break + try: + while True: + + for link in [_f for _f in [line.strip() for line in open(filename) if line[:4] == 'http'] if _f]: + if link not in running_links: + running_links.append(link) + log.info('Added ' + link) + #tasks_to_accomplish.put((link,args)) + tasks_to_accomplish.append(link) + + if time.time() >= (last_queue_check + thread_check_timer): + for i in running_links: + if i not in tasks_to_accomplish: + tasks_to_accomplish.append(i) + last_queue_check = time.time() + + if args.reload and time.time() >= (last_config_reload + queue_cleanup_timer): # Non blocking 5 minute interval check + #links_to_remove = [] + + + for link in links_to_remove: + for line in fileinput.input(filename, inplace=True): + print(line.replace(link, '-' + link), end='') + running_links.remove(link) + log.info('Removed ' + link) + if not args.less: + log.info('Reloading ' + args.thread[0]) # thread = filename here; reloading on next loop + last_config_reload = time.time() + + time.sleep(.25) + print(len(tasks_to_accomplish)) + except KeyboardInterrupt: + for p in processes: #close processes + p.terminate() + pass + return if __name__ == '__main__': try: main() except KeyboardInterrupt: pass - From a8993b4930e6bf541e8a3ec0ac7817f51670610b Mon Sep 17 00:00:00 2001 From: zand3r Date: Mon, 17 Oct 2022 17:15:54 -0400 Subject: [PATCH 2/5] added logic to check that there are always the specified number of threads running --- inb4404.py | 42 ++++++++++++++++++++---------------------- 1 file changed, 20 insertions(+), 22 deletions(-) diff --git a/inb4404.py b/inb4404.py index 4c2cb91..0d248c5 100755 --- a/inb4404.py +++ b/inb4404.py @@ -4,8 +4,8 @@ import http.client import fileinput from multiprocessing import Process -from multiprocessing import Lock, Queue, current_process, Manager -import queue +from multiprocessing import Lock, current_process, Manager +#import queue log = logging.getLogger('inb4404') @@ -13,10 +13,10 @@ args = None call_download_thread_while_loop_sleep_time = .20 download_from_file_while_loop_sleep_time = .20 -queue_cleanup_timer = 300 #in seconds, how often to check for dead links and mark them dead in the config file +queue_cleanup_timer = 30 #in seconds, how often to check for dead links and mark them dead in the config file thread_check_timer = 20 #in seconds, how often to queue up all threads to check for new content manager = Manager() -tasks_to_accomplish = manager.list() +tasks_to_accomplish = manager.list() #queue for threads to pull work out of links_to_remove = manager.list() #queue used to keep track of threads to remove from config @@ -73,28 +73,23 @@ def get_title_list(html_content): return ret -def call_download_thread(que, links_to_remove, running_tasks): +def call_download_thread(que, links_to_remove): while True: try: - #time.sleep(call_download_thread_while_loop_sleep_time) - #thread_link, args = que.get() - if len(que) == 0: - time.sleep(call_download_thread_while_loop_sleep_time) + if len(que) == 0: #check if there are any jobs waiting + time.sleep(call_download_thread_while_loop_sleep_time) #sleep to prevent while loop from dominating CPU continue thread_link = que.pop(0) - running_tasks.append(thread_link) download_thread(thread_link, links_to_remove) - running_tasks.remove(thread_link) except KeyboardInterrupt: break - except queue.Empty: + except: pass return def download_thread(thread_link, links_to_remove): - print("running!") board = thread_link.split('/')[3] thread = thread_link.split('/')[5].split('#')[0] if len(thread_link.split('/')) > 6: @@ -160,7 +155,7 @@ def download_thread(thread_link, links_to_remove): links_to_remove.append(thread_link) except (urllib.error.URLError, http.client.BadStatusLine, http.client.IncompleteRead): log.fatal(thread_link + ' crashed!') - raise + #raise #commenting this out to test my theory if not args.less: log.info('Checking ' + board + '/' + thread) @@ -169,15 +164,14 @@ def download_thread(thread_link, links_to_remove): def download_from_file(filename): - running_links = [] + running_links = [] #4chan threads to check periodically last_config_reload = time.time() last_queue_check = time.time() - running_tasks = [] processes = [] for w in range(args.parallel_threads): - p = Process(target=call_download_thread, args=(tasks_to_accomplish, links_to_remove, running_tasks)) + p = Process(target=call_download_thread, args=(tasks_to_accomplish, links_to_remove)) processes.append(p) p.start() @@ -188,7 +182,6 @@ def download_from_file(filename): if link not in running_links: running_links.append(link) log.info('Added ' + link) - #tasks_to_accomplish.put((link,args)) tasks_to_accomplish.append(link) if time.time() >= (last_queue_check + thread_check_timer): @@ -198,20 +191,25 @@ def download_from_file(filename): last_queue_check = time.time() if args.reload and time.time() >= (last_config_reload + queue_cleanup_timer): # Non blocking 5 minute interval check - #links_to_remove = [] - - for link in links_to_remove: for line in fileinput.input(filename, inplace=True): print(line.replace(link, '-' + link), end='') running_links.remove(link) + links_to_remove.remove(link) log.info('Removed ' + link) if not args.less: log.info('Reloading ' + args.thread[0]) # thread = filename here; reloading on next loop last_config_reload = time.time() + while len(processes) != args.parallel_threads: + p = Process(target=call_download_thread, args=(tasks_to_accomplish, links_to_remove)) + processes.append(p) + p.start() + + time.sleep(.25) - print(len(tasks_to_accomplish)) + + except KeyboardInterrupt: for p in processes: #close processes p.terminate() From e45ba3c489fbe986f9e67feee692bd5745419261 Mon Sep 17 00:00:00 2001 From: zand3r Date: Mon, 17 Oct 2022 17:47:31 -0400 Subject: [PATCH 3/5] added some comments --- inb4404.py | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/inb4404.py b/inb4404.py index 0d248c5..9c857b5 100755 --- a/inb4404.py +++ b/inb4404.py @@ -11,11 +11,14 @@ log = logging.getLogger('inb4404') workpath = os.path.dirname(os.path.realpath(__file__)) args = None -call_download_thread_while_loop_sleep_time = .20 -download_from_file_while_loop_sleep_time = .20 +### +# Danger: making these sleep timers too low can cause their corresponding while loops to execute too quickly, bogging down the CPU +call_download_thread_while_loop_sleep_time = .25 +download_from_file_while_loop_sleep_time = .25 +### queue_cleanup_timer = 30 #in seconds, how often to check for dead links and mark them dead in the config file thread_check_timer = 20 #in seconds, how often to queue up all threads to check for new content -manager = Manager() +manager = Manager() #getting a manager object we can use to create managed data types tasks_to_accomplish = manager.list() #queue for threads to pull work out of links_to_remove = manager.list() #queue used to keep track of threads to remove from config @@ -86,8 +89,6 @@ def call_download_thread(que, links_to_remove): except: pass - return - def download_thread(thread_link, links_to_remove): board = thread_link.split('/')[3] @@ -184,12 +185,14 @@ def download_from_file(filename): log.info('Added ' + link) tasks_to_accomplish.append(link) + # if enough time has passed, recheck list of running threads if time.time() >= (last_queue_check + thread_check_timer): for i in running_links: - if i not in tasks_to_accomplish: + if i not in tasks_to_accomplish: # check if the link we're adding is already in the queue. only add if it isnt tasks_to_accomplish.append(i) last_queue_check = time.time() + # check if there are any links that have died, and mark them as dead so they are no longer checked if args.reload and time.time() >= (last_config_reload + queue_cleanup_timer): # Non blocking 5 minute interval check for link in links_to_remove: for line in fileinput.input(filename, inplace=True): @@ -201,13 +204,15 @@ def download_from_file(filename): log.info('Reloading ' + args.thread[0]) # thread = filename here; reloading on next loop last_config_reload = time.time() + # if, for some reason, we do not have the required amount of threads running, spin up new threads while len(processes) != args.parallel_threads: p = Process(target=call_download_thread, args=(tasks_to_accomplish, links_to_remove)) processes.append(p) p.start() - - time.sleep(.25) + # check for any threads that have completed + for process in processes: + process.join(download_from_file_while_loop_sleep_time) #this will clean up any processes that exited/crashed somehow, while also blocking for .25 seconds except KeyboardInterrupt: From f73c6405b6536c869ab2ea00fa8b540b435b3046 Mon Sep 17 00:00:00 2001 From: zand3r Date: Mon, 17 Oct 2022 18:07:06 -0400 Subject: [PATCH 4/5] added loop for single thread downloader --- inb4404.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/inb4404.py b/inb4404.py index 9c857b5..8a34c06 100755 --- a/inb4404.py +++ b/inb4404.py @@ -43,7 +43,9 @@ def main(): thread = args.thread[0].strip() if thread[:4].lower() == 'http': - download_thread(thread, args) + while True: + download_thread(thread, args) + time.sleep(20) else: download_from_file(thread) From 837055dcb0750a2dc5cb8565f402ce47c8bf4c71 Mon Sep 17 00:00:00 2001 From: zand3r Date: Mon, 17 Oct 2022 18:12:50 -0400 Subject: [PATCH 5/5] removed unused import queue --- inb4404.py | 1 - 1 file changed, 1 deletion(-) diff --git a/inb4404.py b/inb4404.py index 8a34c06..999ddc7 100755 --- a/inb4404.py +++ b/inb4404.py @@ -5,7 +5,6 @@ import fileinput from multiprocessing import Process from multiprocessing import Lock, current_process, Manager -#import queue log = logging.getLogger('inb4404')