prehistoric-systems/web
1
0
Fork 0
mirror of https://codeberg.org/Freedium-cfd/web.git synced 2026-03-11 09:04:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

CORE: refactor function name

Browse source
This commit is contained in:
ZhymabekRoman 2024-04-25 13:58:00 +05:00
parent 276105b518
commit 6539d3a63e
2 changed files with 11 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
core/medium_parser/core.py
View file

@ -15,7 +15,7 @@ from .exceptions import InvalidMediumPostID, InvalidMediumPostURL, InvalidURL, M
from .medium_api import query_post_by_id
from .models.html_result import HtmlResult
from .time import convert_datetime_to_human_readable
from .utils import get_medium_post_id_by_url, getting_percontage_of_match, is_valid_medium_post_id_hexadecimal, is_valid_medium_url, is_valid_url, sanitize_url
from .utils import resolve_medium_url, getting_percontage_of_match, is_valid_medium_post_id_hexadecimal, is_valid_medium_url, is_valid_url, sanitize_url
class MediumParser:
@ -34,7 +34,7 @@ class MediumParser:
if is_valid_url(url) and not await is_valid_medium_url(sanitized_url, timeout):
raise InvalidURL(f"Invalid Medium URL: {sanitized_url}")
post_id = await get_medium_post_id_by_url(sanitized_url, timeout)
post_id = await resolve_medium_url(sanitized_url, timeout)
if not post_id:
raise InvalidMediumPostURL(f"Could not find Medium post ID for URL: {sanitized_url}")

18
core/medium_parser/utils.py
View file

@ -111,10 +111,10 @@ async def resolve_medium_short_link_v1(short_url_id: str, timeout: int = 5) -> s
allow_redirects=False,
)
post_url = request.headers["Location"]
return await get_medium_post_id_by_url(post_url)
return await resolve_medium_url(post_url)
async def get_medium_post_id_by_url(url: str, timeout: int = 5) -> str:
async def resolve_medium_url(url: str, timeout: int = 5) -> str:
parsed_url = urlparse(url)
if parsed_url.path.startswith("/p/"):
post_id = parsed_url.path.rsplit("/p/")[1]
@ -122,34 +122,34 @@ async def get_medium_post_id_by_url(url: str, timeout: int = 5) -> str:
parsed_query = parse_qs(parsed_url.query)
if parsed_query.get("u") and len(parsed_query["u"]) == 1:
post_url = parsed_query["u"][0]
return await get_medium_post_id_by_url(post_url)
return await resolve_medium_url(post_url)
return False
elif parsed_url.netloc == "webcache.googleusercontent.com" and parsed_url.path.startswith("/search"):
parsed_query = parse_qs(parsed_url.query)
if parsed_query.get("q") and len(parsed_query["q"]) == 1:
post_url = parsed_query["q"][0].removeprefix("cache:")
return await get_medium_post_id_by_url(post_url)
return await resolve_medium_url(post_url)
return False
elif parsed_url.netloc == "www.google.com" and parsed_url.path.startswith("/url"):
parsed_query = parse_qs(parsed_url.query)
if parsed_query.get("url") and len(parsed_query["url"]) == 1:
post_url = parsed_query["url"][0]
return await get_medium_post_id_by_url(post_url)
return await resolve_medium_url(post_url)
elif parsed_query.get("q") and len(parsed_query["q"]) == 1:
post_url = parsed_query["q"][0]
return await get_medium_post_id_by_url(post_url)
return await resolve_medium_url(post_url)
return False
elif parsed_url.netloc == "12ft.io":
parsed_query = parse_qs(parsed_url.query)
if parsed_query.get("q") and len(parsed_query["q"]) == 1:
post_url = parsed_query["q"][0]
return await get_medium_post_id_by_url(post_url)
return await resolve_medium_url(post_url)
return False
elif parsed_url.path.startswith("/m/global-identity-2"):
parsed_query = parse_qs(parsed_url.query)
if parsed_query.get("redirectUrl") and len(parsed_query["redirectUrl"]) == 1:
post_url = parsed_query["redirectUrl"][0]
return await get_medium_post_id_by_url(post_url)
return await resolve_medium_url(post_url)
return False
elif parsed_url.netloc == "link.medium.com":
short_url_id = parsed_url.path.removeprefix("/")
@ -164,7 +164,7 @@ async def get_medium_post_id_by_url(url: str, timeout: int = 5) -> str:
return post_id
async def get_medium_post_id_by_url_old(url: str, timeout: int = 5) -> str:
async def resolve_medium_url_old(url: str, timeout: int = 5) -> str:
async with aiohttp.ClientSession() as session:
retry_client = RetryClient(client_session=session, raise_for_status=False, retry_options=retry_options)
request = await retry_client.get(url, timeout=timeout)