71 lines
1.8 KiB
Python
71 lines
1.8 KiB
Python
import asyncio, logging, time, aiohttp
|
|
from datetime import datetime
|
|
from threading import Thread
|
|
from typing import Optional, Tuple, Sequence
|
|
from app import crud
|
|
from app.database import SessionLocal
|
|
|
|
|
|
logging.basicConfig(format='%(levelname)s:%(message)s', filename='example.log', filemode='w', level=logging.INFO)
|
|
|
|
db=SessionLocal()
|
|
|
|
URLS = [
|
|
"https://python.org",
|
|
"https://google.com",
|
|
"https://stackoverflow.com",
|
|
'https://dotya.ml',
|
|
'https://git.dotya.ml',
|
|
'https://drone.dotya.ml',
|
|
'https://netdata.dotya.ml',
|
|
'https://tew0x00.dotya.ml',
|
|
'https://speedtest.dotya.ml',
|
|
'https://testdotya.dotya.ml',
|
|
"https://ubuntu.com",
|
|
]
|
|
URL = crud.get_urls(db=db)
|
|
|
|
def start_background_loop(loop: asyncio.AbstractEventLoop) -> None:
|
|
asyncio.set_event_loop(loop)
|
|
loop.run_forever()
|
|
|
|
|
|
async def fetch(url: str, session: aiohttp.ClientSession = None) -> Tuple[str, str]:
|
|
async def _fetch(url: str, session: aiohttp.ClientSession):
|
|
URL = crud.get_urls(db=db)
|
|
s = time.time()
|
|
async with session.head(url) as response:
|
|
f = time.time() - s
|
|
m = ('[ %f ]\t<%s>\ttook: %f\t(%s)' %(time.time(),response.status,f,url))
|
|
print(m)
|
|
logging.info(m)
|
|
return url
|
|
|
|
|
|
if session:
|
|
return await _fetch(url, session)
|
|
else:
|
|
async with aiohttp.ClientSession() as session:
|
|
return await _fetch(url, session)
|
|
|
|
async def fetch_urls(loop: asyncio.AbstractEventLoop) -> Sequence[Tuple[str, str]]:
|
|
async with aiohttp.ClientSession() as session:
|
|
tasks = [loop.create_task(fetch(url, session)) for url in URL]
|
|
results = await asyncio.gather(*tasks)
|
|
return results
|
|
|
|
def main() -> None:
|
|
loop = asyncio.new_event_loop()
|
|
t = Thread(target=start_background_loop, args=(loop,), daemon=True)
|
|
t.start()
|
|
|
|
|
|
|
|
while True:
|
|
task = asyncio.run_coroutine_threadsafe(fetch_urls(loop), loop)
|
|
time.sleep(10)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|