我正在尝试使用 Pyppeteer 和 asyncio 编写一个程序,该程序将截取不同网站的屏幕截图。但是每次我在大约 20 秒后运行程序时都会发生错误(运行时错误:会话已关闭。很可能页面已关闭“)。后来我发现这是 pyppeteer 的一个已知错误并找到了它的补丁github:
def patch_pyppeteer():
import pyppeteer.connection
original_method = pyppeteer.connection.websockets.client.connect
def new_method(*args, **kwargs):
kwargs['ping_interval'] = None
kwargs['ping_timeout'] = None
return original_method(*args, **kwargs)
pyppeteer.connection.websockets.client.connect = new_method
patch_pyppeteer()
然而,即使错误停止发生,浏览器在同一时间段后仍然失去与互联网的连接,并且 page.goto(url) 函数从未返回控制权。还有什么办法可以解决这个问题吗?
这是我自己的代码的一部分:
async def take_screenshot(page, domain, ip_addr, counter, browser):
url = 'https://' + domain
fname = 'Screenshots\\'+str(counter) + "_"+domain+'_'+ip_addr + '.png'
try:
await page.goto('https://'+domain)
await page.screenshot({'path':fname})
except Exception as e:
print('Error while taking a screenshot')
print(str(e))
exit_program(browser)
async def main():
counter = 0
domains, black_list, output_file = await open_source_files(sys.argv)
print('Launching browser')
browser = await launch(headless=False)
page = await browser.newPage()
while(True):
domain = await read_domain(domains, browser)
if (domain == -1):
break
print('Getting the amount of pages')
amount = await get_pages(page, browser, domain)
print('Reading IP addresses')
ip_list = await get_ipaddr(domain, page, browser)
await delete_domain_from_hosts(domain, browser)
for ip in ip_list:
print('Flushing DNS cache')
flush_dns()
print('Writing in hosts')
lin = await write_in_hosts(ip, domain, browser)
hosts_flag = ip
print('Taking screenshot')
await take_screenshot(page, domain, ip, counter, browser)
print('Deleting from hosts')
await delete_from_hosts(lin, browser)
hosts_flag = None
counter+=1
await browser.close()
domains.close()
black_list.close()
output_file.close()
exit(0)
patch_pyppeteer()
asyncio.get_event_loop().run_until_complete(main())