|
|
@@ -1,20 +1,30 @@
|
|
|
import logging
|
|
|
-import re
|
|
|
import sys
|
|
|
+from contextlib import asynccontextmanager
|
|
|
+from json import JSONDecodeError
|
|
|
from pathlib import Path
|
|
|
|
|
|
-import requests
|
|
|
+import aiohttp
|
|
|
from fastapi import FastAPI, Request
|
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
from fastapi.templating import Jinja2Templates
|
|
|
-from requests import JSONDecodeError
|
|
|
from selectolax.parser import HTMLParser, Node
|
|
|
|
|
|
+
|
|
|
+@asynccontextmanager
|
|
|
+async def lifespan(app: FastAPI):
|
|
|
+ global session
|
|
|
+ session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(10))
|
|
|
+ yield
|
|
|
+ await session.close()
|
|
|
+
|
|
|
+
|
|
|
ROOT_PATH = Path(__file__).parent
|
|
|
-app = FastAPI(docs_url=None, redoc_url=None)
|
|
|
+app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None)
|
|
|
app.mount("/static", StaticFiles(directory=ROOT_PATH / "static"), name="static")
|
|
|
templates = Jinja2Templates(directory=ROOT_PATH / "templates")
|
|
|
+session: aiohttp.ClientSession | None = None
|
|
|
|
|
|
|
|
|
def remove_classes(node: Node) -> Node:
|
|
|
@@ -29,24 +39,27 @@ def remove_classes(node: Node) -> Node:
|
|
|
@app.get("/{path:path}", response_class=HTMLResponse)
|
|
|
async def catch_all(response: Request):
|
|
|
"""Check all routes on Urban Dictionary and redirect if needed."""
|
|
|
- path_without_host = re.sub(r"https?://[^/]+/+", "", str(response.url))
|
|
|
- url = f"https://www.urbandictionary.com/{path_without_host}"
|
|
|
-
|
|
|
- data = requests.get(url, timeout=10)
|
|
|
+ path_without_host = (
|
|
|
+ f"{response.url.path}{f'?{response.url.query}' if response.url.query else ''}"
|
|
|
+ )
|
|
|
+ url = f"https://www.urbandictionary.com{path_without_host}"
|
|
|
|
|
|
- if data.history:
|
|
|
- return RedirectResponse(re.sub(r"https?://[^/]+", "", data.url), status_code=301)
|
|
|
+ async with session.get(url) as dict_response:
|
|
|
+ if dict_response.history:
|
|
|
+ return RedirectResponse(dict_response.url.relative(), status_code=301)
|
|
|
+ html = await dict_response.text()
|
|
|
|
|
|
results = []
|
|
|
- parser = HTMLParser(data.text)
|
|
|
+ parser = HTMLParser(html)
|
|
|
definitions = parser.css("div[data-defid]")
|
|
|
try:
|
|
|
thumbs_api_url = (
|
|
|
f'https://api.urbandictionary.com/v0/uncacheable?ids='
|
|
|
f'{",".join(d.attributes["data-defid"] for d in definitions)}'
|
|
|
)
|
|
|
- thumbs_json = requests.get(thumbs_api_url, timeout=10).json()["thumbs"]
|
|
|
- thumbs_data = {el["defid"]: el for el in thumbs_json}
|
|
|
+ async with session.get(thumbs_api_url) as thumbs_response:
|
|
|
+ thumbs_json = await thumbs_response.json()
|
|
|
+ thumbs_data = {el["defid"]: el for el in thumbs_json["thumbs"]}
|
|
|
except (KeyError, JSONDecodeError):
|
|
|
thumbs_data = {}
|
|
|
|
Zubarev Grigoriy