One post tagged with "并发编程"

import asyncio


async def task1(x):
    print("task1开始")
    # 执行到await时任务被挂起，跳到下一个任务
    await asyncio.sleep(1)
    print("task1完成")
    return x ** 2


async def task2(y):
    print("task2开始")
    await asyncio.sleep(2)
    print("task2完成")
    return y ** 2


async def main():
    # 协程对象，该对象不能直接运行
    coroutine1 = task1(2)
    coroutine2 = task2(3)
    await asyncio.gather(coroutine1, coroutine2)

    
# python3.7版本及之后实现协程
asyncio.run(main())

# python3.7之前
# # 事件循环对象
# loop = asyncio.get_event_loop()
# # 将协程对象注册到事件循环中，并执行
# ret = loop.run_until_complete(coroutine1)
# print(ret)

"""  
In order to solve the problem that github cannot be accessed,  
by replacing ip in the hosts file  
"""  
import ctypes  
import os, sys  
import traceback  
  
import aiohttp  
import asyncio  
from fake_useragent import UserAgent  
from lxml import etree  
from functools import partial  
import shutil  
  
  
ua = UserAgent(use_external_data=True)  
ip_dict = dict()  
  
  
def build_headers():  
    headers = {  
        'use_agent': ua.random  
    }  
    return headers  
  
  
async def fetch_latest_ip(url):  
    async with aiohttp.ClientSession(headers=build_headers()) as session:  
        async with session.get(url) as response:  
            result = await response.text()  
            html = etree.HTML(result)  
            ip = html.xpath("//th[text()='IPv4 Addresses']/following-sibling::td//text()")  
            print(f"正在获取ip: {url}\n解析完成: {ip}")  
            return ip  
  
          
async def main():  
    base_url = 'https://www.ipaddress.com/site/'  
    targets = (  
        'github.com',  
        'assets-cdn.github.com',  
        'github.global.ssl.fastly.net'  
    )  
    tasks = []  
    for url in targets:  
        task = asyncio.ensure_future(fetch_latest_ip(base_url + url))  
        # 回调  
        task.add_done_callback(partial(parse, url))  
        tasks.append(task)  
    await asyncio.gather(*tasks)  
  
  
def parse(url, task):  
    ip_dict[url] = task.result()  
  
  
def save():  
    host = r'C:/Windows/System32/drivers/etc/hosts'  
    temp = r'./temp'  
    try:  
        if not os.path.exists(host):  
            print("hosts文件不存在")  
            return False  
        with open(host, mode='r') as f1, open(temp, mode='w') as f2:  
            lines = f1.readlines()  
            for line in reversed(lines):  
                if 'github.' in line:  
                    lines.remove(line)  
  
            for name in ip_dict:  
                for ip in ip_dict[name]:  
                    lines.append(f"{ip} {name}\n")  
            # 提取一行中的ip和name  
            # ip, name = re.split(r"\s+", line)            f2.writelines(lines)  
  
        os.remove(host)  
        shutil.copy(temp, host)  
        os.system("ipconfig/flushdns")  
        print("DNS刷新成功")  
        return True  
    except Exception:  
        traceback.print_exc()  
        return False  
  
  
if __name__ == '__main__':  
    try:  
        is_admin = ctypes.windll.shell32.IsUserAnAdmin()  
        if not is_admin:  
            # 解决无权访问的问题，重新以管理员权限运行  
            ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, __file__, None, 1)  
        # 这里如果使用asyncio.run(main())会报错，不知道原因  
        loop = asyncio.get_event_loop()  
        loop.run_until_complete(main())  
        ret = save()  
        if ret:  
            print("Succeeded!!")  
        else:  
            print("Failed!!")  
    except Exception as e:  
        traceback.print_exc()  
        print("Failed!!")  
    finally:  
        input("\nPlease press Enter to exit!")