Python多线程外链自动发布源码

import time
import requests
from concurrent.futures import ThreadPoolExecutor
from urllib3 import disable_warnings

disable_warnings()


class Links:
    def __init__(self, websiteurl, workers=16):
        self.timeout = 5
        self.path = 'links.txt'
        self.websiteurl = websiteurl
        self.workers = workers

    def push(self, url: str):
        """分发单条链接
        :param url: 待分发的网址
        """
        url_new = url.format(url=self.websiteurl)
        try:
            with requests.get(url=url_new, timeout=self.timeout, verify=False, stream=True) as response:
                code = response.status_code
            return code, url_new
        except Exception:
            return 500, url_new

    def run(self):
        with open(self.path, 'r', encoding='utf-8') as f:
            urls = f.read().splitlines()

        pool = ThreadPoolExecutor(self.workers)

        allcount = len(urls)
        count = 1

        code_report = {
            '2xx': 0,
            '3xx': 0,
            '4xx': 0,
            '5xx': 0
        }  # 代表各个状态码的链接数量

        results = pool.map(self.push, urls)
        for code, url in results:
            self.update_code_dict(code_report, code)
            print(f'{count / allcount:.2%} >>> {count} / {allcount} >>> {code} {url}')
            count += 1

        print('\n>>> 已全部分发完毕~')
        print('>>> 各个状态码的情况分布：')
        for k, v in code_report.items():
            print(f'{k}：{v}')

    @staticmethod
    def update_code_dict(code_dict: dict, code: int):
        """code_dict代表各个状态码的数量
        code代表状态码"""
        if code < 300:
            code_dict['2xx'] += 1
        elif code < 400:
            code_dict['3xx'] += 1
        elif code < 500:
            code_dict['4xx'] += 1
        else:
            code_dict['5xx'] += 1


if __name__ == '__main__':
    print('定制谷歌外链\n'
          '作者：代码控\n'
          '作者：禁止用于违法！\n'
          f'{"-" * 10}')

    websiteurl = input('输入你的网站（不要带http）：')
    workers = input('请输入工作线程（不写则默认为10）:')
    print('-' * 10)

    if workers == '':
        workers = 10

    start_time = time.time()

    link = Links(websiteurl, int(workers))
    link.run()

    print(f'>>> 运行时间：{time.time() - start_time:.2f}秒')

    input('\n\n按任意键退出……')