Skip to content

Instantly share code, notes, and snippets.

@woshishabii
Created January 21, 2024 01:39
Show Gist options
  • Save woshishabii/4a729846b1e66422d432ab9b992f53b7 to your computer and use it in GitHub Desktop.
Save woshishabii/4a729846b1e66422d432ab9b992f53b7 to your computer and use it in GitHub Desktop.
Check hash s multithreaded and progress-bar ed
import hashlib
from pathlib import Path
from pprint import pprint
import multitasking
import signal
import tqdm
import os
MAX_THREAD = 64
signal.signal(signal.SIGINT, multitasking.killall)
multitasking.set_max_threads(MAX_THREAD)
@multitasking.task
def check(fn: str) -> bool:
global checking
path = Path(f'./media/{fn}')
checking += 1
bar.set_description(f"Checking: {checking}")
with open(path, 'rb') as f:
if path.stem == (m := hashlib.md5(f.read()).hexdigest()):
res = True
else:
res = False
bar.update(1)
checking -= 1
bar.set_description(f"Checking: {checking}")
return res
if __name__ == "__main__":
file_list = os.listdir("./media")
print(f"Starting Checking HASH, Total Files: {len(file_list)}")
bar = tqdm.tqdm(total=len(file_list), desc="Checking.. ")
checking = 0
failed = [_ for _ in file_list if not check(_)]
multitasking.wait_for_tasks()
bar.close()
print(f"Corrupted files: {len(failed)}")
pprint(failed)
print("Finished")
exit(len(failed))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment