import argparse
import logging
import os
import smtplib
import time
import httpx
import trio
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import tabulate
logger = logging.getLogger('MyLog')
logger.setLevel(logging.INFO)
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)-5.5s - %(message)s', "%Y-%m-%d %H:%M:%S")
file_handler = logging.FileHandler(
'MyLog.log', 'a', encoding='utf-8')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
year, month = time.strftime('%Y:%m').split(":")
def validfile(parser, fn):
if not os.path.isfile(fn):
parser.error(f"File {fn} Doesn't Exist")
else:
return fn
result = []
async def checker(channel):
async with channel:
async for client, ticker in channel:
while True:
try:
r = await client.head(f"https://www.hiddenurl.com/{year}/{month}/{ticker}")
break
except httpx.RequestError as e:
logger.error('Url: {:75}, Reason: {}'.format(
r.url, type(e).__name__))
continue
status = "Url: {:75}, Status: {}".format(r.url, r.status_code)
logger.info(status)
if r.status_code == 200:
print(status)
result.append(r.url)
async def mailer():
myemail = "my@my.com"
HTML = tabulate(result, tablefmt="html")
msg = MIMEMultipart("alternative", None, [MIMEText(HTML, 'html')])
msg['Subject'] = f'We Got {len(result)} Result!'
msg['From'] = myemail
msg['To'] = "hidden@hidden.com"
try:
with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp:
smtp.login(myemail, 'mypass')
smtp.send_message(msg)
logger.info('Email Sent! ---> {}'.format(result))
print(f"Yay! Email Sent With {len(result)} Records!")
except (smtplib.SMTPException, smtplib.socket.error) as e:
status="Unable To Send Email Due To --> {}".format(type(e).__name__)
logger.error(status)
print(status)
async def main(links):
headers = {
'User-Agent': 'Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)'
}
async with httpx.AsyncClient(timeout=None) as client, trio.open_nursery() as nurse, await trio.open_file(links) as f:
client.headers.update(headers)
sender, receiver = trio.open_memory_channel(0)
async with receiver:
for _ in range(50):
nurse.start_soon(checker, receiver.clone())
async with sender:
async for ticker in f:
ticker = ticker.lower().rstrip()
if ticker and not ticker.startswith("https://codereview.stackexchange.com/questions/259917/#"):
await sender.send([client, ticker])
if result:
await mailer()
else:
logger.info('No Active Ticker Detected!')
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description='Checking Tickers Response Code')
parser.add_argument('tickers', help='File Include Tickers',
type=lambda x: validfile(parser, x))
trio.run(main, parser.parse_args().tickers)
Это мой текущий код для проверки списка URL-адресов для кода ответа в том же домене.
Я хотел бы знать, есть ли способ, по которому я могу лучше повысить производительность своего кода.
Заранее спасибо.