univk_audio.async_auth.AsyncVKAuth.__send_oauth_request() - Code Metrics - Exponefrv1/univk_audio - Measure and Improve Code Quality continuously with Scrutinizer

AsyncVKAuth.__send_oauth_request() A
last analyzed 2024-11-19 00:14 UTC

↳ Parent: univk_audio.async_auth

Complexity

Conditions

Size

Total Lines	15
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	15
dl	0
loc	15
rs	9.65
c	0
b	0
f	0
cc	3
nop	1

"""
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__ = (
        "_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

            ) as oauth_response:
                cookies = self._session.cookie_jar.filter_cookies("http://oauth.vk.com/")
                cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
                redirect_url = str(oauth_response.url)
                auth_hash = redirect_url.split("=")[8].replace("&scope", "")
                self._req_data.vkid_auth_link = redirect_url
                self._req_data.oauth_code_data.update({"hash": auth_hash})
                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.vkid_auth_link,

                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[1],

            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[1],

                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[2],

                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[3],

                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			"_req_data",
68			"_session"
69			)
70
71			def __init__(
72			self,
73			login: str,
74			password: str,
75			user_agent: Optional[str] = None
76			) -> "AsyncVKAuth":
77
78			req_data = VKAuthData(login, password, user_agent)
79			session = aiohttp.ClientSession(cookie_jar=aiohttp.CookieJar())
80
81			self._req_data = req_data
82			self._session = session
83
84
85			async def __aenter__(self) -> "AsyncVKAuth":
86			return self
87
88
89			async def __send_oauth_request(self) -> None:
90			try:
91			async with self._session.get(
92			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...
93			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...
94			) as oauth_response:
95			cookies = self._session.cookie_jar.filter_cookies("http://oauth.vk.com/")
96			cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
97			redirect_url = str(oauth_response.url)
98			auth_hash = redirect_url.split("=")[8].replace("&scope", "")
99			self._req_data.vkid_auth_link = redirect_url
100			self._req_data.oauth_code_data.update({"hash": auth_hash})
101			self._req_data.main_headers.update({"Cookie": cookies})
102			except Exception as err:
103			raise OAuthRequestError("Failed to send/process 'OAuth' request") from err
104
105
106			async def __send_vk_id_auth_request(self) -> None:
107			try:
108			async with self._session.get(
109			self._req_data.vkid_auth_link,
			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...
110			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...
111			) as vk_id_request:
112			if vk_id_request.status == 200:
113			response_content = await vk_id_request.text()
114			await self.__parse_auth_part(response_content)
115			else:
116			raise VKIDRequestError(
117			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...
118			)
119			except VKIDRequestError as expected_err:
120			raise expected_err
121			except Exception as err:
122			raise VKIDRequestError("Failed to send/process 'VK ID' request") from err
123
124
125			async def __parse_auth_part(self, response_content: str) -> None:
126			try:
127			matches = re.findall(self._req_data.auth_part_pattern, response_content)
128			if matches:
129			access_token = matches[0][0]
130			cookies = self._session.cookie_jar.filter_cookies("https://id.vk.com/")
131			cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
132			self._req_data.connect_auth_data.update({"auth_token": access_token})
133			self._req_data.main_headers["Cookie"] = cookies
134			self._req_data.main_headers.update(
135			{
			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...
136			"Origin": "https://id.vk.com",
137			"Referer": "https://id.vk.com/"
138			}
139			)
140			else:
141			raise AuthObjectNotFound("Could not find 'auth' object in response content")
142			except AuthObjectNotFound as expected_err:
143			raise expected_err
144			except Exception as err:
145			raise AuthParsingError("Failed to parse auth part from response content") from err
146
147
148			async def __solve_captcha_request(self) -> Dict[str, str]:
149			async with self._session.post(
150			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...
151			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...
152			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...
153			) as connect_auth_request:
154			return await connect_auth_request.json()
155
156
157			async def __solve_captcha(self, access_token_data) -> Dict[str, str]:
158			while access_token_data.get("type") == "captcha":
159			print("Please, solve the captcha:\n", access_token_data.get("captcha_img"))
160			answer = str(input("Solving: "))
161			self._req_data.connect_auth_data.update({
162			"captcha_key": answer,
163			"captcha_sid": access_token_data.get("captcha_sid"),
164			"is_refresh_enabled": access_token_data.get("is_refresh_enabled"),
165			"captcha_ts": access_token_data.get("captcha_ts"),
166			"captcha_attempt": access_token_data.get("captcha_attempt")
167			})
168			access_token_data = await self.__solve_captcha_request()
169			return access_token_data
170
171
172			async def __process_access_token_data(self, access_token_data) -> None:
173			if access_token_data.get("type") == "captcha":
174			access_token_data = await self.__solve_captcha(access_token_data)
175			if access_token_data.get("type") == "error":
176			error_code = access_token_data.get("error_code")
177			raise ConnectAuthRequestError(
178			"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...
179			f"Error code: {error_code}"
180			)
181			if access_token_data.get("type") == "okay":
182			access_token = access_token_data.get("data").get("access_token")
183			cookies = self._session.cookie_jar.filter_cookies("https://login.vk.com/")
184			cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
185			self._req_data.oauth_code_data.update({"access_token": access_token})
186			self._req_data.main_headers.update({"Cookie": cookies})
187
188
189			async def __send_connect_auth_request(self) -> None:
190			try:
191			async with self._session.post(
192			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...
193			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...
194			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...
195			) as connect_auth_request:
196			if connect_auth_request.status == 200:
197			access_token_data = await connect_auth_request.json()
198			await self.__process_access_token_data(access_token_data)
199			else:
200			raise ConnectAuthRequestError(
201			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...
202			)
203			except ConnectAuthRequestError as expected_err:
204			raise expected_err
205			except Exception as err:
206			raise ConnectAuthRequestError("Failed to send/process 'Connect Auth' request") from err
207
208
209			async def __send_oauth_code_request(self) -> None:
210			try:
211			async with self._session.post(
212			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...
213			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...
214			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...
215			) as oauth_code_request:
216			code_data = await oauth_code_request.json()
217			self._req_data.code = code_data.get("response")
218			except Exception as err:
219			raise SendCodeRequestError("Failed to send/process 'OAuth Code' request") from err
220
221
222			async def __send_get_cookies_request(self) -> str:
223			try:
224			async with self._session.get(
225			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...
226			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...
227			):
228			cookies = self._session.cookie_jar.filter_cookies("http://luxvk.com/")
229			cookies = "; ".join([str(x)+"="+str(y) for x, y in cookies.items()])
230			return cookies
231			except Exception as err:
232			raise GetCookiesRequestError("Failed to send/process 'Get Cookies' request") from err
233
234
235			async def __write_cookies(self, path: str, cookies: str) -> None:
236			try:
237			async with aiofiles.open(path, mode="w") as file:
238			await file.write(cookies)
239			except FileNotFoundError as expected_err:
240			expected_err.strerror = "Failed to write cookie string into the file. " \
241			f"No such file or directory: '{path}'"
242			raise expected_err
243			except Exception as err:
244			raise CookieWriterError("Failed to write cookie string into the file") from err
245
246
247			async def get_auth_cookies(self, path: Optional[str] = None) -> str:
248
249			"""
250			Returns a string with cookie.
251			"""
252
253			try:
254			await self.__send_oauth_request()
255			await self.__send_vk_id_auth_request()
256			await self.__send_connect_auth_request()
257			await self.__send_oauth_code_request()
258			cookie_string = await self.__send_get_cookies_request()
259			if path is not None:
260			await self.__write_cookies(path, cookie_string)
261			await self.close()
262			return cookie_string
263			except Exception as err:
264			await self.close()
265			raise err
266
267
268			async def close(self) -> None:
269			"""
270			Close session method.
271			Returns Nothing.
272			"""
273			await self._session.close()
274
275
276			async def __aexit__(
277			self,
278			exc_type: Optional[Type[BaseException]],
279			exc_val: Optional[BaseException],
280			exc_tb: Optional[TracebackType]
281			) -> None:
282			await self.close()
283

Exponefrv1 / univk_audio

AsyncVKAuth.__send_oauth_request() A last analyzed 2024-11-19 00:14 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

AsyncVKAuth.__send_oauth_request() A
last analyzed 2024-11-19 00:14 UTC