univk_audio.async_client.AsyncVKMusic.__search() - Code Metrics - Exponefrv1/univk_audio - Measure and Improve Code Quality continuously with Scrutinizer

univk_audio.async_client.AsyncVKMusic.__search() A
last analyzed 2024-11-19 00:14 UTC

↳ Parent: univk_audio.async_client

Complexity

Conditions

Size

Total Lines	20
Code Lines	18

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	18
dl	0
loc	20
rs	9.0333
c	0
b	0
f	0
cc	5
nop	2

"""
Client module of univk_audio library.

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

search(query: str)

Get songs from vk.

Returns a dict with song's title and download link.

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

download(link: str, path: str)

Download songs.
Need to specify the download link of the song.

Returns True, if song is downloaded.

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

close()

Close client session.

Returns None.

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

import re
import asyncio
from types import TracebackType
from typing import (
    Optional,
    Tuple,
    Dict,
    Type
)

import aiohttp
import aiofiles
import httpx
from bs4 import BeautifulSoup

from .request_data import VKMusicData
from .client_exceptions import (
    InvalidQuery,
    InvalidPath,
    SearchSongError,
    ParserError,
    ParseLinkError,
    DownloaderRequestError,
    DownloaderWriteError
)

__all__ = (
    "AsyncVKMusic",
    "InvalidQuery",
    "InvalidPath",
    "SearchSongError",
    "ParserError",
    "ParseLinkError",
    "DownloaderRequestError",
    "DownloaderWriteError"
)


class AsyncVKMusic:

    """
    Main client module class.

    AsyncVKMusic(cookies: str, user_agent: Optional[str])

    Check module docstring for additional info.
    """

    __slots__ = (
        "_req_data",
        "_httpx_session",
        "_aiohttp_session",
        "_loop"
    )

    def __init__(
        self,

        cookies: str,

        user_agent: Optional[str] = None

    ) -> None:
        req_data = VKMusicData(cookies, user_agent)
        httpx_session = httpx.AsyncClient()
        aiohttp_session = aiohttp.ClientSession()
        loop = asyncio.get_event_loop()

        self._req_data = req_data
        self._httpx_session = httpx_session
        self._aiohttp_session = aiohttp_session
        self._loop = loop


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


    def __parse_search(self, data: str) -> Dict[str, str]:
class Foo:
    def some_method(self, x, y):
        return x + y;
        try:
            soup = BeautifulSoup(data, 'lxml')
            div = soup.find("div", {"class": "col pl"})
            ul_list_group = div.find("ul", {"class": "sm2-playlist-bd list-group"})
            elements = ul_list_group.find_all(
                "li",
                {"class": "list-group-item justify-content-between list-group-item-action"}
            )
            search_results = {}
            for element in elements:
                link = element.find("a", {"target": "_blank"})
                search_results.update({element.get_text(): link.get("href")})
            return search_results
        except Exception as err:
            raise ParserError("Failed to parse song data") from err


    async def __get_song_link(self, download_link: str) -> str:
        try:
            download_request = await self._httpx_session.get(
                    self._req_data.base_url + download_link,

                    headers=self._req_data.download_headers

                )
            song_link = str(download_request.headers.get("Location"))
            await self._httpx_session.get(
                    song_link,

                    headers=self._req_data.download_headers

                )
            regex = r"https:\/\/dl01\.dtmp3\.pw\/cs\d+-\d+v\d+\.vkuseraudio\.net\/s\/v1\/ac\/"
            song_link = re.sub(regex, "https://ts01.flac.pw/dl/", song_link)
            song_link = song_link.replace("/index.m3u8?siren=1", ".mp3")
            return song_link
        except Exception as err:
            raise ParseLinkError("Failed to get song download link") from err


    async def __download_song_request(self, song_link: str) -> Tuple[bytes, int]:
        try:
            async with self._aiohttp_session.get(
                song_link,

                headers=self._req_data.download_headers

            ) as song_content:
                content_bytes = await song_content.read()
                content_length = int(song_content.headers['Content-Length'])
                return (content_bytes, content_length)
        except aiohttp.client_exceptions.ClientConnectorError as expected_err:
            raise DownloaderRequestError(
                    "Failed to send/process download request. Cannot connect to music source. "

                    "Try to change your ip adress and location"
                ) from expected_err
        except Exception as err:
            raise DownloaderRequestError("Failed to send/process download request") from err


    async def __search(self, query: str):
        try:
            if len(query) == 0 or len(query) > 25:
                raise InvalidQuery("Invalid query provided")
            params = {"q": query, "p": 1}
            search_request = await self._httpx_session.get(
                self._req_data.base_url,
                headers=self._req_data.main_headers,
                params=params
            )
            search_results = await self._loop.run_in_executor(
                None,
                self.__parse_search,
                search_request.text
            )
            return search_results
        except InvalidQuery as expected_err:
            raise expected_err
        except Exception as err:
            raise SearchSongError(f"Failed to search the '{query}'") from err


    async def __download(self, link: str, path: str) -> bool:
        try:
            if len(path) == 0:
                raise InvalidPath("Invalid path provided")
            song_link = await self.__get_song_link(link)
            length = 0
            content = b""
            while length == 0:
                await asyncio.sleep(3)
                content_data = await self.__download_song_request(song_link)
                content, length = content_data[0], content_data[1]
            async with aiofiles.open(path, mode="wb") as file:
                await file.write(content)
            return True
        except InvalidPath as expected_err:
            raise expected_err
        except FileNotFoundError as expected_err:
            expected_err.strerror = "Failed to write song content into the file. " \
            f"No such file or directory: '{path}'"
            raise expected_err
        except Exception as err:
            raise DownloaderWriteError("Failed to write song content into the file") from err


    async def search(self, query: str) -> Dict[str, str]:

        """
            Returns a dictionary in following format:
            {"song title": "link to pass in the download function"}
        """

        try:
            search_results = await self.__search(query)
            return search_results
        except Exception as err:
            await self.close()
            raise err


    async def download(self, link: str, path: str) -> bool:

        """
            Returns "True" if file is downloaded,
            otherwise raises DownloaderWriteError exception.
            Works much better with a VPN (if you're from Russia)
        """

        try:
            is_saved_successfully = await self.__download(link, path)
            return is_saved_successfully
        except Exception as err:
            await self.close()
            raise err


    async def close(self) -> None:

        await self._httpx_session.aclose()
        await self._aiohttp_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			Client module of univk_audio library.
3
4			Methods:
5			--------------------------------
6
7			search(query: str)
8
9			Get songs from vk.
10
11			Returns a dict with song's title and download link.
12
13			--------------------------------
14
15			download(link: str, path: str)
16
17			Download songs.
18			Need to specify the download link of the song.
19
20			Returns True, if song is downloaded.
21
22			--------------------------------
23
24			close()
25
26			Close client session.
27
28			Returns None.
29
30			--------------------------------
31			"""
32
33			import re
34			import asyncio
35			from types import TracebackType
36			from typing import (
37			Optional,
38			Tuple,
39			Dict,
40			Type
41			)
42
43			import aiohttp
44			import aiofiles
45			import httpx
46			from bs4 import BeautifulSoup
47
48			from .request_data import VKMusicData
49			from .client_exceptions import (
50			InvalidQuery,
51			InvalidPath,
52			SearchSongError,
53			ParserError,
54			ParseLinkError,
55			DownloaderRequestError,
56			DownloaderWriteError
57			)
58
59			__all__ = (
60			"AsyncVKMusic",
61			"InvalidQuery",
62			"InvalidPath",
63			"SearchSongError",
64			"ParserError",
65			"ParseLinkError",
66			"DownloaderRequestError",
67			"DownloaderWriteError"
68			)
69
70
71			class AsyncVKMusic:
72
73			"""
74			Main client module class.
75
76			AsyncVKMusic(cookies: str, user_agent: Optional[str])
77
78			Check module docstring for additional info.
79			"""
80
81			__slots__ = (
82			"_req_data",
83			"_httpx_session",
84			"_aiohttp_session",
85			"_loop"
86			)
87
88			def __init__(
89			self,
			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...
90			cookies: str,
			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...
91			user_agent: Optional[str] = None
			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...
92			) -> None:
93			req_data = VKMusicData(cookies, user_agent)
94			httpx_session = httpx.AsyncClient()
95			aiohttp_session = aiohttp.ClientSession()
96			loop = asyncio.get_event_loop()
97
98			self._req_data = req_data
99			self._httpx_session = httpx_session
100			self._aiohttp_session = aiohttp_session
101			self._loop = loop
102
103
104			async def __aenter__(self) -> "AsyncVKMusic":
105			return self
106
107
108			def __parse_search(self, data: str) -> Dict[str, str]:
			0 ignored issues – show Coding Style introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
109			try:
110			soup = BeautifulSoup(data, 'lxml')
111			div = soup.find("div", {"class": "col pl"})
112			ul_list_group = div.find("ul", {"class": "sm2-playlist-bd list-group"})
113			elements = ul_list_group.find_all(
114			"li",
115			{"class": "list-group-item justify-content-between list-group-item-action"}
116			)
117			search_results = {}
118			for element in elements:
119			link = element.find("a", {"target": "_blank"})
120			search_results.update({element.get_text(): link.get("href")})
121			return search_results
122			except Exception as err:
123			raise ParserError("Failed to parse song data") from err
124
125
126			async def __get_song_link(self, download_link: str) -> str:
127			try:
128			download_request = await self._httpx_session.get(
129			self._req_data.base_url + download_link,
			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...
130			headers=self._req_data.download_headers
			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...
131			)
132			song_link = str(download_request.headers.get("Location"))
133			await self._httpx_session.get(
134			song_link,
			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...
135			headers=self._req_data.download_headers
			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			)
137			regex = r"https:\/\/dl01\.dtmp3\.pw\/cs\d+-\d+v\d+\.vkuseraudio\.net\/s\/v1\/ac\/"
138			song_link = re.sub(regex, "https://ts01.flac.pw/dl/", song_link)
139			song_link = song_link.replace("/index.m3u8?siren=1", ".mp3")
140			return song_link
141			except Exception as err:
142			raise ParseLinkError("Failed to get song download link") from err
143
144
145			async def __download_song_request(self, song_link: str) -> Tuple[bytes, int]:
146			try:
147			async with self._aiohttp_session.get(
148			song_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...
149			headers=self._req_data.download_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...
150			) as song_content:
151			content_bytes = await song_content.read()
152			content_length = int(song_content.headers['Content-Length'])
153			return (content_bytes, content_length)
154			except aiohttp.client_exceptions.ClientConnectorError as expected_err:
155			raise DownloaderRequestError(
156			"Failed to send/process download request. Cannot connect to music source. "
			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...
157			"Try to change your ip adress and location"
158			) from expected_err
159			except Exception as err:
160			raise DownloaderRequestError("Failed to send/process download request") from err
161
162
163			async def __search(self, query: str):
164			try:
165			if len(query) == 0 or len(query) > 25:
166			raise InvalidQuery("Invalid query provided")
167			params = {"q": query, "p": 1}
168			search_request = await self._httpx_session.get(
169			self._req_data.base_url,
170			headers=self._req_data.main_headers,
171			params=params
172			)
173			search_results = await self._loop.run_in_executor(
174			None,
175			self.__parse_search,
176			search_request.text
177			)
178			return search_results
179			except InvalidQuery as expected_err:
180			raise expected_err
181			except Exception as err:
182			raise SearchSongError(f"Failed to search the '{query}'") from err
183
184
185			async def __download(self, link: str, path: str) -> bool:
186			try:
187			if len(path) == 0:
188			raise InvalidPath("Invalid path provided")
189			song_link = await self.__get_song_link(link)
190			length = 0
191			content = b""
192			while length == 0:
193			await asyncio.sleep(3)
194			content_data = await self.__download_song_request(song_link)
195			content, length = content_data[0], content_data[1]
196			async with aiofiles.open(path, mode="wb") as file:
197			await file.write(content)
198			return True
199			except InvalidPath as expected_err:
200			raise expected_err
201			except FileNotFoundError as expected_err:
202			expected_err.strerror = "Failed to write song content into the file. " \
203			f"No such file or directory: '{path}'"
204			raise expected_err
205			except Exception as err:
206			raise DownloaderWriteError("Failed to write song content into the file") from err
207
208
209			async def search(self, query: str) -> Dict[str, str]:
210
211			"""
212			Returns a dictionary in following format:
213			{"song title": "link to pass in the download function"}
214			"""
215
216			try:
217			search_results = await self.__search(query)
218			return search_results
219			except Exception as err:
220			await self.close()
221			raise err
222
223
224			async def download(self, link: str, path: str) -> bool:
225
226			"""
227			Returns "True" if file is downloaded,
228			otherwise raises DownloaderWriteError exception.
229			Works much better with a VPN (if you're from Russia)
230			"""
231
232			try:
233			is_saved_successfully = await self.__download(link, path)
234			return is_saved_successfully
235			except Exception as err:
236			await self.close()
237			raise err
238
239
240			async def close(self) -> None:
			0 ignored issues – show introduced 2023-08-10 20:11 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
241			await self._httpx_session.aclose()
242			await self._aiohttp_session.close()
243
244
245			async def __aexit__(
246			self,
			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...
247			exc_type: Optional[Type[BaseException]],
			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...
248			exc_val: Optional[BaseException],
			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...
249			exc_tb: Optional[TracebackType]
			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...
250			) -> None:
251			await self.close()
252

Exponefrv1 / univk_audio

univk_audio.async_client.AsyncVKMusic.__search() A last analyzed 2024-11-19 00:14 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

univk_audio.async_client.AsyncVKMusic.__search() A
last analyzed 2024-11-19 00:14 UTC