Cách sử dụng aiohttp để Scrape Web

pip install aiohttp

import asyncio
import aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            status = response.status
            text = await response.text()
            print(f'Mã trạng thái: {status}')
            print('Nội dung phản hồi:', text)

if __name__ == '__main__':
    asyncio.run(fetch('https://httpbin.org/get'))

import asyncio
import aiohttp
from bs4 import BeautifulSoup

async def fetch_content(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def scrape_quotes():
    url = 'http://quotes.toscrape.com/'
    html = await fetch_content(url)
    soup = BeautifulSoup(html, 'html.parser')
    quotes = soup.find_all('div', class_='quote')
    for quote in quotes:
        text = quote.find('span', class_='text').get_text(strip=True)
        author = quote.find('small', class_='author').get_text(strip=True)
        print(f'{text} — {author}')

if __name__ == '__main__':
    asyncio.run(scrape_quotes())

“Thế giới mà chúng ta đã tạo ra là một quá trình tư duy của chúng ta. Nó không thể thay đổi nếu không thay đổi tư duy của chúng ta.” — Albert Einstein
“Đó là những lựa chọn của chúng ta, Harry, điều đó cho thấy chúng ta thực sự là ai, hơn nhiều so với khả năng của chúng ta.” — J.K. Rowling
... (các câu nói khác)

pip install capsolver

import asyncio
import os
import aiohttp
import capsolver

# Thiết lập khóa API CapSolver của bạn
capsolver.api_key = os.getenv("CAPSOLVER_API_KEY", "Khóa API CapSolver của bạn")
PAGE_URL = os.getenv("PAGE_URL", "https://example.com")  # URL trang có captcha
SITE_KEY = os.getenv("SITE_KEY", "SITE_KEY")             # Khóa trang web Captcha

async def solve_recaptcha_v2():
    solution = capsolver.solve({
        "type": "ReCaptchaV2TaskProxyless",
        "websiteURL": PAGE_URL,
        "websiteKey": SITE_KEY
    })
    return solution['solution']['gRecaptchaResponse']

async def access_protected_page():
    captcha_response = await solve_recaptcha_v2()
    print("Captcha đã được giải quyết!")

    async with aiohttp.ClientSession() as session:
        data = {
            'g-recaptcha-response': captcha_response,
            # Bao gồm dữ liệu biểu mẫu khác nếu trang web yêu cầu
        }
        async with session.post(PAGE_URL, data=data) as response:
            content = await response.text()
            print('Nội dung trang:', content)

if __name__ == '__main__':
    asyncio.run(access_protected_page())

import asyncio
import aiohttp

async def fetch(url, proxy):
    async with aiohttp.ClientSession() as session:
        async with session.get(url, proxy=proxy) as response:
            return await response.text()

async def main():
    proxy = 'http://username:password@proxyserver:port'
    url = 'https://httpbin.org/ip'
    content = await fetch(url, proxy)
    print('Nội dung phản hồi:', content)

if __name__ == '__main__':
    asyncio.run(main())

import asyncio
import aiohttp

async def main():
    jar = aiohttp.CookieJar()
    async with aiohttp.ClientSession(cookie_jar=jar) as session:
        await session.get('https://httpbin.org/cookies/set?name=value')
        # Hiển thị cookie
        for cookie in jar:
            print(f'{cookie.key}: {cookie.value}')

if __name__ == '__main__':
    asyncio.run(main())

import asyncio
import aiohttp

async def main():
    headers = {
        'User-Agent': 'Mozilla/5.0 (compatible)',
        'Accept-Language': 'en-US,en;q=0.5',
    }
    data = {
        'username': 'testuser',
        'password': 'testpass',
    }
    async with aiohttp.ClientSession() as session:
        async with session.post('https://httpbin.org/post', headers=headers, data=data) as response:
            json_response = await response.json()
            print('JSON phản hồi:', json_response)

if __name__ == '__main__':
    asyncio.run(main())