проверить список кодов ответов URL

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-адресов для кода ответа в том же домене.

Я хотел бы знать, есть ли способ, по которому я могу лучше повысить производительность своего кода.

Заранее спасибо.

0

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *