From c735b45a256e53d32a157371f594596350a1af83 Mon Sep 17 00:00:00 2001 From: Ross Date: Mon, 24 Nov 2025 10:54:42 +0000 Subject: [PATCH] refactor login handling: replace inline logic with perform_login function for improved clarity and maintainability --- nice.py | 113 +++++++++++++++++++++++++++++--------------------------- 1 file changed, 59 insertions(+), 54 deletions(-) diff --git a/nice.py b/nice.py index 5efdfcf..61b8566 100644 --- a/nice.py +++ b/nice.py @@ -36,7 +36,7 @@ from starlette.middleware.base import BaseHTTPMiddleware import typer import logging -import traceback +# traceback was previously imported for debugging but is unused # TODO fix ssl VERIFY_SSL = False @@ -449,6 +449,49 @@ def upload_files(q, rqst): ) +def perform_login(user: str, pw: str) -> dict: + """Module-level blocking login function suitable for run.cpu_bound. + + Returns a dict with keys: + - error: str on error + - session: requests.Session on success + - token: csrf token if found + - success: bool + """ + s = requests.session() + try: + r = s.get(LOGIN_URL, verify=VERIFY_SSL, timeout=10) + except Exception as e: + logger.debug(e) + return {"error": str(e)} + + try: + token = BeautifulSoup(r.content, "html.parser").find("input").attrs.get("value") + except Exception: + token = None + + data = { + "username": user, + "password": pw, + "csrfmiddlewaretoken": token, + "next": "/", + } + + s.headers["Referer"] = LOGIN_URL + + try: + r2 = s.post(LOGIN_URL, data=data, verify=VERIFY_SSL, timeout=10) + except Exception as e: + logger.debug(e) + return {"error": str(e)} + + soup = BeautifulSoup(r2.content, "html.parser") + if soup.find("button") and soup.find("button").find(string="Login"): + return {"session": s, "success": False} + else: + return {"session": s, "success": True, "token": token} + + @@ -457,68 +500,30 @@ def login() -> Optional[RedirectResponse]: dark = ui.dark_mode() dark.enable() - def try_login() -> ( - None - ): # local function to avoid passing username and password as arguments - # Because I haven't gotten around to implementing a proper login system yet - # we just log in via requests and store teh session in the global rqst variable + async def try_login() -> None: + # Run blocking network calls off the UI thread to keep the page responsive. user = username.value pw = password.value global rqst, LOGIN_SUCCESS, LOGGED_IN_USER - rqst = requests.session() - try: - rsp = rqst.get(LOGIN_URL, verify=VERIFY_SSL) - except requests.exceptions.ConnectionError as e: + + # Use the module-level perform_login function (picklable) to avoid + # multiprocessing pickling errors when using run.cpu_bound. + result = await run.cpu_bound(perform_login, user, pw) + + if result.get("error"): ui.notify("Connection error!", color="negative") - logger.debug(e) - print(traceback.format_exc()) + logger.debug(result.get("error")) return - token = ( - BeautifulSoup(rsp.content, "html.parser").find("input").attrs["value"] - ) # , attr={"name": "csrfmiddlewaretoken"}).attrs("value") - - # token = rsp.cookies["csrftoken"] - # header = {"X-CSRFToken": token} - # cookies = {"csrftoken": token} - - data = { - "username": user, - "password": pw, - "csrfmiddlewaretoken": token, - "next": "/", - } - - rqst.headers["Referer"] = LOGIN_URL - - rsp = rqst.post( - LOGIN_URL, - data=data, - # headers=header, - # cookies=cookies - verify=VERIFY_SSL - ) - - print(rsp.content) - - soup = BeautifulSoup(rsp.content, "html.parser") - - if soup.find("button") and soup.find("button").find(string="Login"): - print("login fail") - pass - else: - print("login success") + if result.get("success"): + rqst = result.get("session") + token = result.get("token") + if token: + rqst.headers["X-CSRFToken"] = token LOGIN_SUCCESS = True LOGGED_IN_USER = user user_info_ui.refresh() - # user_label.clear() - # user_label.text = f"User: {user}" - # print(rqst.headers) - - rqst.headers["X-CSRFToken"] = token - - if LOGIN_SUCCESS: ui.notify("Logged in!", color="positive") ui.navigate.to("/") else: @@ -561,7 +566,7 @@ def main_page(): globals(), "LOGIN_SUCCESS", backward=lambda v: not v ) - updates = ui.list() + # placeholder for update list (previously assigned but unused) loaded_series_ui()