univk_audio.async_auth.AsyncVKAuth.__solve_captcha() - Code Metrics - Inspection of "Additional data commit" - Exponefrv1/univk_audio - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( a851d0...3dcf5c )

by Unik

created 2023-08-11 14:54 UTC

AsyncVKAuth.__solve_captcha() A

↳ Parent: univk_audio.async_auth

Complexity

Conditions

Size

Total Lines	13
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	13
rs	9.8
c	0
b	0
f	0
cc	2
nop	2

"""
Authorization module of univk_audio library.

Methods:
--------------------------------

get_auth_cookies(path: Optional[str])

Get authorization data for searching and
downloading songs.

Returns a string with cookie.

--------------------------------

close()

Close authorization session.

Returns None.

--------------------------------
"""

import re
from types import TracebackType
from typing import Optional, Dict, Type

import aiofiles
import aiohttp

from .request_data import VKAuthData
from .auth_exceptions import (
    OAuthRequestError,
    VKIDRequestError,
    ConnectAuthRequestError,
    SendCodeRequestError,
    GetCookiesRequestError,
    AuthObjectNotFound,
    AuthParsingError,
    CookieWriterError
)

__all__ = (
    "AsyncVKAuth",
    "OAuthRequestError",
    "VKIDRequestError",
    "ConnectAuthRequestError",
    "SendCodeRequestError",
    "GetCookiesRequestError",
    "AuthObjectNotFound",
    "AuthParsingError",
    "CookieWriterError"
)

class AsyncVKAuth:

    """
    Main auth module class.

    AsyncVKAuth(login: str, password: str, user_agent: Optional[str])

    Check module docstring for additional info.
    """

    __slots__ = (
        "_captcha_solver",
        "_req_data",
        "_session"
    )

    def __init__(
            self,
            login: str,
            password: str,
            user_agent: Optional[str] = None
        ) -> "AsyncVKAuth":

        req_data = VKAuthData(login, password, user_agent)
        session = aiohttp.ClientSession(cookie_jar=aiohttp.CookieJar())

        self._req_data = req_data
        self._session = session


    async def __aenter__(self) -> "AsyncVKAuth":
        return self


    async def __send_oauth_request(self) -> None:
        try:
            async with self._session.get(
                self._req_data.links[0],

                headers=self._req_data.main_headers

            ):
                cookies = self._session.cookie_jar.filter_cookies("http://oauth.vk.com/")
                cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
                self._req_data.main_headers.update({"Cookie": cookies})
        except Exception as err:
            raise OAuthRequestError("Failed to send/process 'OAuth' request") from err


    async def __send_vk_id_auth_request(self) -> None:
        try:
            async with self._session.get(
                self._req_data.links[1],

                headers=self._req_data.main_headers

            ) as vk_id_request:
                if vk_id_request.status == 200:
                    response_content = await vk_id_request.text()
                    await self.__parse_auth_part(response_content)
                else:
                    raise VKIDRequestError(
                            f"VK ID returned an error request status code: {vk_id_request.status}"

                        )
        except VKIDRequestError as expected_err:
            raise expected_err
        except Exception as err:
            raise VKIDRequestError("Failed to send/process 'VK ID' request") from err


    async def __parse_auth_part(self, response_content: str) -> None:
        try:
            matches = re.findall(self._req_data.auth_part_pattern, response_content)
            if matches:
                access_token = matches[0][0]
                cookies = self._session.cookie_jar.filter_cookies("https://id.vk.com/")
                cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
                self._req_data.connect_auth_data.update({"auth_token": access_token})
                self._req_data.main_headers["Cookie"] = cookies
                self._req_data.main_headers.update(
                        {

                            "Origin": "https://id.vk.com",
                            "Referer": "https://id.vk.com/"
                        }
                    )
            else:
                raise AuthObjectNotFound("Could not find 'auth' object in response content")
        except AuthObjectNotFound as expected_err:
            raise expected_err
        except Exception as err:
            raise AuthParsingError("Failed to parse auth part from response content") from err


    async def __solve_captcha_request(self) -> Dict[str, str]:
        async with self._session.post(
            self._req_data.links[2],

            headers=self._req_data.main_headers,

            data=self._req_data.connect_auth_data

        ) as connect_auth_request:
            return await connect_auth_request.json()


    async def __solve_captcha(self, access_token_data) -> Dict[str, str]:
        while access_token_data.get("type") == "captcha":
            print("Please solve the captcha:\n", access_token_data.get("captcha_img"))
            answer = str(input("Solving: "))
            self._req_data.connect_auth_data.update({
                "captcha_key": answer,
                "captcha_sid": access_token_data.get("captcha_sid"),
                "is_refresh_enabled": access_token_data.get("is_refresh_enabled"),
                "captcha_ts": access_token_data.get("captcha_ts"),
                "captcha_attempt": access_token_data.get("captcha_attempt")
            })
            access_token_data = await self.__solve_captcha_request()
            return access_token_data


    async def __process_access_token_data(self, access_token_data) -> None:
        if access_token_data.get("type") == "captcha":
            access_token_data = await self.__solve_captcha(access_token_data)
        if access_token_data.get("type") == "error":
            error_code = access_token_data.get("error_code")
            raise ConnectAuthRequestError(
                    "Failed to send/process 'Connect Auth' request. " \

                    f"Error code: {error_code}"
                )
        if access_token_data.get("type") == "okay":
            access_token = access_token_data.get("data").get("access_token")
            cookies = self._session.cookie_jar.filter_cookies("https://login.vk.com/")
            cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
            self._req_data.oauth_code_data.update({"access_token": access_token})
            self._req_data.main_headers.update({"Cookie": cookies})


    async def __send_connect_auth_request(self) -> None:
        try:
            async with self._session.post(
                self._req_data.links[2],

                headers=self._req_data.main_headers,

                data=self._req_data.connect_auth_data

            ) as connect_auth_request:
                if connect_auth_request.status == 200:
                    access_token_data = await connect_auth_request.json()
                    await self.__process_access_token_data(access_token_data)
                else:
                    raise ConnectAuthRequestError(
                            f"Connect Auth returned an error request status code: {connect_auth_request.status}"

                        )
        except ConnectAuthRequestError as expected_err:
            raise expected_err
        except Exception as err:
            raise ConnectAuthRequestError("Failed to send/process 'Connect Auth' request") from err


    async def __send_oauth_code_request(self) -> None:
        try:
            async with self._session.post(
                self._req_data.links[3],

                headers=self._req_data.main_headers,

                data=self._req_data.oauth_code_data

            ) as oauth_code_request:
                code_data = await oauth_code_request.json()
                self._req_data.code = code_data.get("response")
        except Exception as err:
            raise SendCodeRequestError("Failed to send/process 'OAuth Code' request") from err


    async def __send_get_cookies_request(self) -> str:
        try:
            async with self._session.get(
                self._req_data.links[4],

                params={"code": self._req_data.code}

            ):
                cookies = self._session.cookie_jar.filter_cookies("http://luxvk.com/")
                cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
                return cookies
        except Exception as err:
            raise GetCookiesRequestError("Failed to send/process 'Get Cookies' request") from err


    async def __write_cookies(self, path: str, cookies: str) -> None:
        try:
            async with aiofiles.open(path, mode="w") as file:
                await file.write(cookies)
        except FileNotFoundError as expected_err:
            expected_err.strerror = "Failed to write cookie string into the file. " \
            f"No such file or directory: '{path}'"
            raise expected_err
        except Exception as err:
            raise CookieWriterError("Failed to write cookie string into the file") from err


    async def get_auth_cookies(self, path: Optional[str] = None) -> str:

        """
            Returns a string with cookie.
        """

        try:
            await self.__send_oauth_request()
            await self.__send_vk_id_auth_request()
            await self.__send_connect_auth_request()
            await self.__send_oauth_code_request()
            cookie_string = await self.__send_get_cookies_request()
            if path is not None:
                await self.__write_cookies(path, cookie_string)
            await self.close()
            return cookie_string
        except Exception as err:
            await self.close()
            raise err


    async def close(self) -> None:
        """
            Close session method.
            Returns Nothing.
        """
        await self._session.close()


    async def __aexit__(
            self,
            exc_type: Optional[Type[BaseException]],
            exc_val: Optional[BaseException],
            exc_tb: Optional[TracebackType]
        ) -> None:
        await self.close()


1			"""
2			Authorization module of univk_audio library.
3
4			Methods:
5			--------------------------------
6
7			get_auth_cookies(path: Optional[str])
8
9			Get authorization data for searching and
10			downloading songs.
11
12			Returns a string with cookie.
13
14			--------------------------------
15
16			close()
17
18			Close authorization session.
19
20			Returns None.
21
22			--------------------------------
23			"""
24
25			import re
26			from types import TracebackType
27			from typing import Optional, Dict, Type
28
29			import aiofiles
30			import aiohttp
31
32			from .request_data import VKAuthData
33			from .auth_exceptions import (
34			OAuthRequestError,
35			VKIDRequestError,
36			ConnectAuthRequestError,
37			SendCodeRequestError,
38			GetCookiesRequestError,
39			AuthObjectNotFound,
40			AuthParsingError,
41			CookieWriterError
42			)
43
44			__all__ = (
45			"AsyncVKAuth",
46			"OAuthRequestError",
47			"VKIDRequestError",
48			"ConnectAuthRequestError",
49			"SendCodeRequestError",
50			"GetCookiesRequestError",
51			"AuthObjectNotFound",
52			"AuthParsingError",
53			"CookieWriterError"
54			)
55
56			class AsyncVKAuth:
57
58			"""
59			Main auth module class.
60
61			AsyncVKAuth(login: str, password: str, user_agent: Optional[str])
62
63			Check module docstring for additional info.
64			"""
65
66			__slots__ = (
67			"_captcha_solver",
68			"_req_data",
69			"_session"
70			)
71
72			def __init__(
73			self,
74			login: str,
75			password: str,
76			user_agent: Optional[str] = None
77			) -> "AsyncVKAuth":
78
79			req_data = VKAuthData(login, password, user_agent)
80			session = aiohttp.ClientSession(cookie_jar=aiohttp.CookieJar())
81
82			self._req_data = req_data
83			self._session = session
84
85
86			async def __aenter__(self) -> "AsyncVKAuth":
87			return self
88
89
90			async def __send_oauth_request(self) -> None:
91			try:
92			async with self._session.get(
93			self._req_data.links[0],
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
94			headers=self._req_data.main_headers
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
95			):
96			cookies = self._session.cookie_jar.filter_cookies("http://oauth.vk.com/")
97			cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
98			self._req_data.main_headers.update({"Cookie": cookies})
99			except Exception as err:
100			raise OAuthRequestError("Failed to send/process 'OAuth' request") from err
101
102
103			async def __send_vk_id_auth_request(self) -> None:
104			try:
105			async with self._session.get(
106			self._req_data.links[1],
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
107			headers=self._req_data.main_headers
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
108			) as vk_id_request:
109			if vk_id_request.status == 200:
110			response_content = await vk_id_request.text()
111			await self.__parse_auth_part(response_content)
112			else:
113			raise VKIDRequestError(
114			f"VK ID returned an error request status code: {vk_id_request.status}"
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 4 spaces). Loading history...
115			)
116			except VKIDRequestError as expected_err:
117			raise expected_err
118			except Exception as err:
119			raise VKIDRequestError("Failed to send/process 'VK ID' request") from err
120
121
122			async def __parse_auth_part(self, response_content: str) -> None:
123			try:
124			matches = re.findall(self._req_data.auth_part_pattern, response_content)
125			if matches:
126			access_token = matches[0][0]
127			cookies = self._session.cookie_jar.filter_cookies("https://id.vk.com/")
128			cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
129			self._req_data.connect_auth_data.update({"auth_token": access_token})
130			self._req_data.main_headers["Cookie"] = cookies
131			self._req_data.main_headers.update(
132			{
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 4 spaces). Loading history...
133			"Origin": "https://id.vk.com",
134			"Referer": "https://id.vk.com/"
135			}
136			)
137			else:
138			raise AuthObjectNotFound("Could not find 'auth' object in response content")
139			except AuthObjectNotFound as expected_err:
140			raise expected_err
141			except Exception as err:
142			raise AuthParsingError("Failed to parse auth part from response content") from err
143
144
145			async def __solve_captcha_request(self) -> Dict[str, str]:
146			async with self._session.post(
147			self._req_data.links[2],
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
148			headers=self._req_data.main_headers,
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
149			data=self._req_data.connect_auth_data
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
150			) as connect_auth_request:
151			return await connect_auth_request.json()
152
153
154			async def __solve_captcha(self, access_token_data) -> Dict[str, str]:
155			while access_token_data.get("type") == "captcha":
156			print("Please solve the captcha:\n", access_token_data.get("captcha_img"))
157			answer = str(input("Solving: "))
158			self._req_data.connect_auth_data.update({
159			"captcha_key": answer,
160			"captcha_sid": access_token_data.get("captcha_sid"),
161			"is_refresh_enabled": access_token_data.get("is_refresh_enabled"),
162			"captcha_ts": access_token_data.get("captcha_ts"),
163			"captcha_attempt": access_token_data.get("captcha_attempt")
164			})
165			access_token_data = await self.__solve_captcha_request()
166			return access_token_data
167
168
169			async def __process_access_token_data(self, access_token_data) -> None:
170			if access_token_data.get("type") == "captcha":
171			access_token_data = await self.__solve_captcha(access_token_data)
172			if access_token_data.get("type") == "error":
173			error_code = access_token_data.get("error_code")
174			raise ConnectAuthRequestError(
175			"Failed to send/process 'Connect Auth' request. " \
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 4 spaces). Loading history...
176			f"Error code: {error_code}"
177			)
178			if access_token_data.get("type") == "okay":
179			access_token = access_token_data.get("data").get("access_token")
180			cookies = self._session.cookie_jar.filter_cookies("https://login.vk.com/")
181			cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
182			self._req_data.oauth_code_data.update({"access_token": access_token})
183			self._req_data.main_headers.update({"Cookie": cookies})
184
185
186			async def __send_connect_auth_request(self) -> None:
187			try:
188			async with self._session.post(
189			self._req_data.links[2],
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
190			headers=self._req_data.main_headers,
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
191			data=self._req_data.connect_auth_data
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
192			) as connect_auth_request:
193			if connect_auth_request.status == 200:
194			access_token_data = await connect_auth_request.json()
195			await self.__process_access_token_data(access_token_data)
196			else:
197			raise ConnectAuthRequestError(
198			f"Connect Auth returned an error request status code: {connect_auth_request.status}"
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 4 spaces). Loading history... Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (112/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
199			)
200			except ConnectAuthRequestError as expected_err:
201			raise expected_err
202			except Exception as err:
203			raise ConnectAuthRequestError("Failed to send/process 'Connect Auth' request") from err
204
205
206			async def __send_oauth_code_request(self) -> None:
207			try:
208			async with self._session.post(
209			self._req_data.links[3],
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
210			headers=self._req_data.main_headers,
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
211			data=self._req_data.oauth_code_data
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
212			) as oauth_code_request:
213			code_data = await oauth_code_request.json()
214			self._req_data.code = code_data.get("response")
215			except Exception as err:
216			raise SendCodeRequestError("Failed to send/process 'OAuth Code' request") from err
217
218
219			async def __send_get_cookies_request(self) -> str:
220			try:
221			async with self._session.get(
222			self._req_data.links[4],
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
223			params={"code": self._req_data.code}
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Wrong hanging indentation before block (add 4 spaces). Loading history...
224			):
225			cookies = self._session.cookie_jar.filter_cookies("http://luxvk.com/")
226			cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
227			return cookies
228			except Exception as err:
229			raise GetCookiesRequestError("Failed to send/process 'Get Cookies' request") from err
230
231
232			async def __write_cookies(self, path: str, cookies: str) -> None:
233			try:
234			async with aiofiles.open(path, mode="w") as file:
235			await file.write(cookies)
236			except FileNotFoundError as expected_err:
237			expected_err.strerror = "Failed to write cookie string into the file. " \
238			f"No such file or directory: '{path}'"
239			raise expected_err
240			except Exception as err:
241			raise CookieWriterError("Failed to write cookie string into the file") from err
242
243
244			async def get_auth_cookies(self, path: Optional[str] = None) -> str:
245
246			"""
247			Returns a string with cookie.
248			"""
249
250			try:
251			await self.__send_oauth_request()
252			await self.__send_vk_id_auth_request()
253			await self.__send_connect_auth_request()
254			await self.__send_oauth_code_request()
255			cookie_string = await self.__send_get_cookies_request()
256			if path is not None:
257			await self.__write_cookies(path, cookie_string)
258			await self.close()
259			return cookie_string
260			except Exception as err:
261			await self.close()
262			raise err
263
264
265			async def close(self) -> None:
266			"""
267			Close session method.
268			Returns Nothing.
269			"""
270			await self._session.close()
271
272
273			async def __aexit__(
274			self,
275			exc_type: Optional[Type[BaseException]],
276			exc_val: Optional[BaseException],
277			exc_tb: Optional[TracebackType]
278			) -> None:
279			await self.close()
280

Exponefrv1 / univk_audio

Push — main ( a851d0...3dcf5c )

AsyncVKAuth.__solve_captcha() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like