status_check.py

'''
Author : Devansh Batham (Asm0d3us)
Modules required : aiohttp , asyncio 

Note : Requires Python 3.7+


How to use : 

mkdir Status_check
cd Status_check
sudo apt install python3.7 python3-venv python3.7-venv
python3.7 -m venv py37-venv
. py37-venv/bin/activate
Create a file named raw_urls.txt (all urls must be either http/https)
python status_check.py
'''


import time
import os
import sys
import aiohttp
import asyncio
from aiohttp import *
start_time = time.time()


def main():
    asyncio.run(download_url_contents())

async def download_url_contents():
    async with aiohttp.ClientSession() as session:
        await gen_url_tasks(session, 'raw_urls.txt')
async def fetch_url(session, url):
    # Get the response text of the supplied url
    try:
        async with session.get(url) as response:
            reason = response.reason
            status = response.status

            if status != 200:
                # "Error Handling"
                print(f"\u001b[31;1m[{status}] : {url}  {reason}\u001b[0m")
                pass
            else:
                # print(f"{url} : {status} {reason}")
                print(f"\u001b[32;1m[{status}]  :  {url} \u001b[0m")
    except ClientConnectorError:
        return (url, 500) 
    except ClientOSError:
        return (url, 500)
    except ServerDisconnectedError:
        return (url,500)
    except asyncio.TimeoutError:
        return (url, 500)
    except UnicodeDecodeError:
        return (url, 500)
    except TooManyRedirects:
        return (url, 500)
    except ServerTimeoutError:
        return (url, 500)
    except ServerConnectionError:
        return (url, 500)
    except RuntimeError:
        return (url, 500)

async def gen_url_tasks(session, url_list):
    # Generate the tasks for each url in supplied url list.

    with open(url_list) as f:
        urls = f.read().splitlines()
    tasks = []
    for url in urls:
        task = asyncio.ensure_future(fetch_url(session, url))
        tasks.append(task)
    results = await asyncio.gather(*tasks)
    print(f"\n    \u001b[36;1m[+] URLs Checked  : {len(urls)}\u001b[0m\n")
    return results

if __name__ == "__main__":
    if os.name =="nt":
        os.system("cls")
    assert sys.version_info >= (3, 7), "Use Python 3.7+"
    main()
    print("\n \u001b[31m [!] Total execution time    : %ss\u001b[0m" % str((time.time() - start_time))[:-12])