Блог

Я уже несколько дней пытаюсь использовать aiohttp для входа на веб-сайт, а затем перейти в административную область, чтобы получить содержимое. После того, как я опубликую сеанс, я не знаю, что делать, чтобы получить контент со страницы администратора. Я также попытался сделать это, захватив файлы cookie из сеанса, но я не уверен, что делать после захвата файлов cookie. Этот раздел кода был закомментирован, поскольку это не самый предпочтительный способ.

 async def do_task(session, credentials):
    try:
     async with session.get(credentials['domain']) as r:
         url = r.url #follow redirect to login page
         login_data = {"log": credentials['username'], "pwd": credentials['password']}

         # Please help with below
         await session.post(url, json=login_data)
         return await r.get(f'{url}admin').text()

        # #Attempt with cookies
        #  async with session.post(url, json=login_data) as login:
        #      session.cookie_jar.update_cookies(login.cookies)
        #      return await login.get(f'{url}admin').text()

    except Exception as e:
     print(e)

async def tasks(session, dict_list):
    tasks = []
    for credentials in dict_list:
        task = asyncio.create_task(do_task(session, credentials))
        tasks.append(task)
    results = await asyncio.gather(*tasks)
    return results

async def main(x):
    async with aiohttp.ClientSession() as session:
        data = await tasks(session, x)
        return data

if __name__ == '__main__':
    dict_list = ({
        "username": 'test',
        "domain": 'http://url.com/admin',
        "password": 'enter'
    },
    )

    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) #for windows
    results = asyncio.run(main(dict_list))
 

Сообщение об ошибке, которое я получаю, выглядит так 'ClientResponse' object has no attribute 'get'

Я ранее делал то же самое с запросами, используя приведенный ниже код, но я пытаюсь ускорить процесс с помощью aiohttp.

             with requests.Session() as login_request:
                login_data = {"log": x['username'], "pwd": x['password']
                              }
                login_request.post(url, data=login_data)
                source_code = login_request.get(url).content
 

Главная проблема здесь:

r.get(f'{url}/admin’).текст()

В вашем случае r это ClientResponse (не ClientSession является async with session.get(credentials['domain']) результатом).

Вы можете получать файлы cookie асинхронно, а затем использовать их для асинхронной очистки. Должно быть что-то вроде этого:

 async def login(session, domain: str, login_data: dict):
    async with session.post(domain   '/admin', data=login_data) as resp:
        # I don't know how you check success login...
        # data = await resp.json() or data await resp.text()
        # if data... blablabla
        return [domain, resp.cookies]

async def process_page(session, url: str, cookie: SimpleCookie):
    async with session.get(url, cookie_jar=cookie) as resp:
        content = await resp.text()
        # do something   return...

# example logins = [{'domain': '...', 'login_data': {...}}, ...]
# get cookies for all users and domains
cookies = await asyncio.gather(*[
    login(session, l['domain'], t['login_data'])
    for l in logins
])
      
# processing '/page1' for all domains and users 
result = await asyncio.gather(*[
    process_page(session, c[0]   '/page1', c[1])
    for c in cookies
])