From 8bfd5c241ae6ebc533063776830d67b1c38a4dd9 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sun, 28 Aug 2022 18:24:05 +0200 Subject: [PATCH] fix some pylint findings --- bumper/models.py | 23 --- bumper/mqtt/proxy.py | 3 +- bumper/mqtt/server.py | 76 +++++---- bumper/web/auth_util.py | 262 +++++++++++-------------------- bumper/web/middlewares.py | 20 +-- bumper/web/plugins/api/appsvr.py | 4 +- bumper/web/server.py | 140 ++++++++--------- 7 files changed, 215 insertions(+), 313 deletions(-) diff --git a/bumper/models.py b/bumper/models.py index ea40790..907835a 100644 --- a/bumper/models.py +++ b/bumper/models.py @@ -1,5 +1,4 @@ """Models module.""" -import json import uuid from datetime import datetime, timedelta from typing import Any @@ -87,28 +86,6 @@ class VacBotClient: } -class EcoVacs_Login: - """Ecovacs login.""" - - accessToken = "" - country = "" - email = "" - uid = "" - username = "" - - def toJSON(self) -> str: - """Convert to json.""" - return json.dumps(self, default=lambda o: o.__dict__, sort_keys=False) - - -class EcoVacsHome_Login(EcoVacs_Login): - """Ecovacs home login.""" - - loginName = "" - mobile: str | None = "" - ucUid = "" - - class OAuth: """Oauth.""" diff --git a/bumper/mqtt/proxy.py b/bumper/mqtt/proxy.py index d6a856e..3d04362 100644 --- a/bumper/mqtt/proxy.py +++ b/bumper/mqtt/proxy.py @@ -90,7 +90,7 @@ class ProxyClient: ) bumper.mqtt_helperbot.publish(topic, message.data) - except Exception: # pylint: disable=broad-except + except Exception: _LOGGER.error( "An error occurred during handling a message", exc_info=True ) @@ -109,6 +109,7 @@ class ProxyClient: class _NoCertVerifyClient(MQTTClient): # type:ignore[misc] + # pylint: disable=all """ Mqtt client, which is not verify the certificate. diff --git a/bumper/mqtt/server.py b/bumper/mqtt/server.py index 8556454..664332d 100644 --- a/bumper/mqtt/server.py +++ b/bumper/mqtt/server.py @@ -89,9 +89,10 @@ class MQTTServer: return self._broker.transitions.state @property - def broker(self) -> Broker: - """Get MQTT broker.""" - return self._broker + def sessions(self) -> list[Session]: + """Get sessions.""" + # pylint: disable-next=protected-access + return [session for (session, _) in self._broker._sessions.values()] async def start(self) -> None: """Start MQTT server.""" @@ -151,28 +152,30 @@ class BumperMQTTServerPlugin: return True if "@" in client_id: - didsplit = str(client_id).split("@") - if "ecouser" not in didsplit[1]: + client_id_split = str(client_id).split("@") + client_details_split = client_id_split[1].split("/") + if "ecouser" not in client_id_split[1]: # if ecouser aren't in details it is a bot - tmpbotdetail = str(didsplit[1]).split("/") bot_add( username, - didsplit[0], - tmpbotdetail[0], - tmpbotdetail[1], + client_id_split[0], + client_details_split[0], + client_details_split[1], "eco-ng", ) _LOGGER.info( "Bumper Authentication Success - Bot - SN: %s - DID: %s - Class: %s", username, - didsplit[0], - tmpbotdetail[0], + client_id_split[0], + client_details_split[0], ) if bumper.bumper_proxy_mqtt: mqtt_server = await dns.resolve("mq-ww.ecouser.net") _LOGGER_PROXY.info( - f"MQTT Proxy Mode - Using server {mqtt_server} for client {client_id}" + "MQTT Proxy Mode - Using server %s for client %s", + mqtt_server, + client_id, ) proxy = ProxyClient( client_id, mqtt_server, config={"check_hostname": False} @@ -182,13 +185,12 @@ class BumperMQTTServerPlugin: return True - tmpclientdetail = str(didsplit[1]).split("/") - userid = didsplit[0] - realm = tmpclientdetail[0] - resource = tmpclientdetail[1] - - if check_authcode(didsplit[0], password) or not bumper.use_auth: - client_add(userid, realm, resource) + if check_authcode(client_id_split[0], password) or not bumper.use_auth: + client_add( + client_id_split[0], + client_details_split[0], + client_details_split[1], + ) _LOGGER.info( "Bumper Authentication Success - Client - Username: %s - ClientID: %s", username, @@ -260,20 +262,22 @@ class BumperMQTTServerPlugin: if client_id in self._proxy_clients: await self._proxy_clients[client_id].subscribe(topic, qos) _LOGGER_PROXY.info( - f"MQTT Proxy Mode - New MQTT Topic Subscription - Client: {client_id} - Topic: {topic}" + "MQTT Proxy Mode - New MQTT Topic Subscription - Client: %s - Topic: %s", + client_id, + topic, ) elif client_id != HELPER_BOT_CLIENT_ID: _LOGGER_PROXY.warning( - f"MQTT Proxy Mode - No proxy client found! - Client: {client_id} - Topic: {topic}" + "MQTT Proxy Mode - No proxy client found! - Client: %s - Topic: %s", + client_id, + topic, ) async def on_broker_client_connected(self, client_id: str) -> None: """On client connected.""" self._set_client_connected(client_id, True) - def _set_client_connected( # pylint: disable=no-self-use - self, client_id: str, connected: bool - ) -> None: + def _set_client_connected(self, client_id: str, connected: bool) -> None: didsplit = str(client_id).split("@") bot = bot_get(didsplit[0]) @@ -286,7 +290,7 @@ class BumperMQTTServerPlugin: if client: client_set_mqtt(client["resource"], connected) - async def on_broker_message_received( # pylint: disable=no-self-use + async def on_broker_message_received( self, message: IncomingApplicationMessage, client_id: str ) -> None: """On message received.""" @@ -316,28 +320,34 @@ class BumperMQTTServerPlugin: ) if ttopic[6] == "": _LOGGER_PROXY.warning( - "Request mapper is missing entry, " - f"probably request took to long... Client_id: {client_id}" - f" - Request_id: {ttopic[10]}" + "Request mapper is missing entry, probably request took to" + " long... Client_id: %s - Request_id: %s", + client_id, + ttopic[10], ) return ttopic_join = "/".join(ttopic) _LOGGER_PROXY.info( - f"Bot Message Converted Topic From {message.topic} TO {ttopic_join} " - f"with message: {data_decoded}" + "Bot Message Converted Topic From %s TO %s with message: %s", + message.topic, + ttopic_join, + data_decoded, ) else: ttopic_join = message.topic _LOGGER_PROXY.info( - f"Bot Message From {ttopic_join} with message: {data_decoded}" + "Bot Message From %s with message: %s", + ttopic_join, + data_decoded, ) try: # Send back to ecovacs _LOGGER_PROXY.info( - "Proxy Forward Message to Ecovacs - Topic:" - f" {ttopic_join} - Message: {data_decoded}" + "Proxy Forward Message to Ecovacs - Topic: %s - Message: %s", + ttopic_join, + data_decoded, ) await self._proxy_clients[client_id].publish( ttopic_join, data_decoded.encode(), message.qos diff --git a/bumper/web/auth_util.py b/bumper/web/auth_util.py index 4b99c64..0d3f736 100644 --- a/bumper/web/auth_util.py +++ b/bumper/web/auth_util.py @@ -1,5 +1,4 @@ """Auth util module.""" -import json import logging import uuid from typing import Any @@ -27,8 +26,6 @@ from bumper.models import ( ERR_TOKEN_INVALID, ERR_USER_NOT_ACTIVATED, RETURN_API_SUCCESS, - EcoVacs_Login, - EcoVacsHome_Login, ) from bumper.util import get_current_time_as_millis from bumper.web.plugins import get_success_response @@ -55,78 +52,51 @@ async def login(request: Request) -> Response: user_devid = request.match_info.get("devid", "") countrycode = request.match_info.get("country", "us") apptype = request.match_info.get("apptype", "") - _LOGGER.info(f"client with devid {user_devid} attempting login") + _LOGGER.info("Client with devid %s attempting login", user_devid) if use_auth: - if ( - not user_devid == "" - ): # Performing basic "auth" using devid, super insecure + if user_devid != "": + # Performing basic "auth" using devid, super insecure user = user_by_device_id(user_devid) if user: if "checkLogin" in request.path: - _check_token( - apptype, countrycode, user, request.query["accessToken"] + return web.json_response( + _check_token( + apptype, countrycode, user, request.query["accessToken"] + )[1] ) - else: - login_details: EcoVacsHome_Login | EcoVacs_Login - if "global_" in apptype: # EcoVacs Home - login_details = EcoVacsHome_Login() - login_details.ucUid = "fuid_{}".format(user["userid"]) - login_details.loginName = "fusername_{}".format( - user["userid"] - ) - login_details.mobile = None - else: - login_details = EcoVacs_Login() + # Deactivate old tokens and authcodes + user_revoke_expired_tokens(user["userid"]) - # Deactivate old tokens and authcodes - user_revoke_expired_tokens(user["userid"]) + body = { + "code": API_ERRORS[RETURN_API_SUCCESS], + "data": _get_login_details( + apptype, countrycode, user, _generate_token(user) + ), + "msg": "操作成功", + "time": get_current_time_as_millis(), + } + return web.json_response(body) - login_details.accessToken = _generate_token(user) - login_details.uid = "fuid_{}".format(user["userid"]) - login_details.username = "fusername_{}".format(user["userid"]) - login_details.country = countrycode - login_details.email = "null@null.com" - - body = { - "code": API_ERRORS[RETURN_API_SUCCESS], - "data": json.loads(login_details.toJSON()), - # { - # "accessToken": self.generate_token(tmpuser), # Generate a token - # "country": countrycode, - # "email": "null@null.com", - # "uid": "fuid_{}".format(tmpuser["userid"]), - # "username": "fusername_{}".format(tmpuser["userid"]), - # }, - "msg": "操作成功", - "time": get_current_time_as_millis(), - } - - return web.json_response(body) - - body = { - "code": ERR_USER_NOT_ACTIVATED, - "data": None, - "msg": "当前密码错误", - "time": get_current_time_as_millis(), - } - - return web.json_response(body) - - else: return web.json_response( - _auth_any(user_devid, apptype, countrycode, request) + { + "code": ERR_USER_NOT_ACTIVATED, + "data": None, + "msg": "当前密码错误", + "time": get_current_time_as_millis(), + } ) - except Exception as e: - _LOGGER.exception(f"{e}") + return web.json_response(_auth_any(user_devid, apptype, countrycode, request)) + except Exception: # pylint: disable=broad-except + _LOGGER.exception("An exception occurred", exc_info=True) raise HTTPInternalServerError async def get_authcode(request: Request) -> Response: """Get auth code.""" - try: + try: # pylint: disable=too-many-nested-blocks user_devid = request.match_info.get("devid", None) # Ecovacs if not user_devid: user_devid = request.query["deviceId"] # Ecovacs Home @@ -170,134 +140,86 @@ async def get_authcode(request: Request) -> Response: def _check_token( apptype: str, countrycode: str, user: dict[str, Any], token: str -) -> Response: - try: - if db.check_token(user["userid"], token): - login_details: EcoVacsHome_Login | EcoVacs_Login - if "global_" in apptype: # EcoVacs Home - login_details = EcoVacsHome_Login() - login_details.ucUid = "fuid_{}".format(user["userid"]) - login_details.loginName = "fusername_{}".format(user["userid"]) - login_details.mobile = None - else: - login_details = EcoVacs_Login() - - login_details.accessToken = token - login_details.uid = "fuid_{}".format(user["userid"]) - login_details.username = "fusername_{}".format(user["userid"]) - login_details.country = countrycode - login_details.email = "null@null.com" - - body = { +) -> tuple[bool, dict[str, Any]]: + if db.check_token(user["userid"], token): + return ( + True, + { "code": RETURN_API_SUCCESS, - "data": json.loads(login_details.toJSON()), - # { - # "accessToken": self.generate_token(tmpuser), # Generate a token - # "country": countrycode, - # "email": "null@null.com", - # "uid": "fuid_{}".format(tmpuser["userid"]), - # "username": "fusername_{}".format(tmpuser["userid"]), - # }, + "data": _get_login_details(apptype, countrycode, user, token), "msg": "操作成功", "time": get_current_time_as_millis(), - } - return web.json_response(body) + }, + ) - else: - body = { - "code": ERR_TOKEN_INVALID, - "data": None, - "msg": "当前密码错误", - "time": get_current_time_as_millis(), - } - return web.json_response(body) - - except Exception as e: - _LOGGER.exception(f"{e}") - - raise HTTPInternalServerError + return ( + False, + { + "code": ERR_TOKEN_INVALID, + "data": None, + "msg": "当前密码错误", + "time": get_current_time_as_millis(), + }, + ) def _auth_any( devid: str, apptype: str, country: str, request: Request ) -> dict[str, Any]: - try: - user_devid = devid - countrycode = country - user = user_by_device_id(user_devid) - bots = _db_get().table("bots").all() - login_details: EcoVacs_Login | EcoVacsHome_Login + user_devid = devid + countrycode = country + user = user_by_device_id(user_devid) + bots = _db_get().table("bots").all() - if user: # Default to user 0 - tmpuser = user - if "global_" in apptype: # EcoVacs Home - login_details = EcoVacsHome_Login() - login_details.ucUid = "fuid_{}".format(tmpuser["userid"]) - login_details.loginName = "fusername_{}".format(tmpuser["userid"]) - login_details.mobile = None - else: - login_details = EcoVacs_Login() + if not user: + user_add("tmpuser") # Add a new user + tmp = user_get("tmpuser") + assert tmp + user = tmp - login_details.accessToken = _generate_token(tmpuser) - login_details.uid = "fuid_{}".format(tmpuser["userid"]) - login_details.username = "fusername_{}".format(tmpuser["userid"]) - login_details.country = countrycode - login_details.email = "null@null.com" - user_add_device(tmpuser["userid"], user_devid) + token = _generate_token(user) + user_add_device(user["userid"], user_devid) + + for bot in bots: # Add all bots to the user + if "did" in bot: + user_add_bot(user["userid"], bot["did"]) else: - user_add("tmpuser") # Add a new user - tmp = user_get("tmpuser") - assert tmp - tmpuser = tmp - if "global_" in apptype: # EcoVacs Home - login_details = EcoVacsHome_Login() - login_details.ucUid = "fuid_{}".format(tmpuser["userid"]) - login_details.loginName = "fusername_{}".format(tmpuser["userid"]) - login_details.mobile = None - else: - login_details = EcoVacs_Login() + _LOGGER.error("No DID for bot: %s", bot) - login_details.accessToken = _generate_token(tmpuser) - login_details.uid = "fuid_{}".format(tmpuser["userid"]) - login_details.username = "fusername_{}".format(tmpuser["userid"]) - login_details.country = countrycode - login_details.email = "null@null.com" - user_add_device(tmpuser["userid"], user_devid) + if "checkLogin" in request.path: # If request was to check a token do so + (success, body) = _check_token( + apptype, countrycode, user, request.query["accessToken"] + ) + if success: + return body - for bot in bots: # Add all bots to the user - if "did" in bot: - user_add_bot(tmpuser["userid"], bot["did"]) - else: - _LOGGER.error(f"No DID for bot: {bot}") + # Deactivate old tokens and authcodes + user_revoke_expired_tokens(user["userid"]) - if "checkLogin" in request.path: # If request was to check a token do so - checkToken = _check_token( - apptype, countrycode, tmpuser, request.query["accessToken"] - ) - assert checkToken.text - isGood: dict[str, Any] = json.loads(checkToken.text) - if isGood["code"] == "0000": - return isGood + body = { + "code": RETURN_API_SUCCESS, + "data": _get_login_details(apptype, countrycode, user, token), + "msg": "操作成功", + "time": get_current_time_as_millis(), + } - # Deactivate old tokens and authcodes - user_revoke_expired_tokens(tmpuser["userid"]) + return body - body = { - "code": RETURN_API_SUCCESS, - "data": json.loads(login_details.toJSON()), - # { - # "accessToken": self.generate_token(tmpuser), # Generate a token - # "country": countrycode, - # "email": "null@null.com", - # "uid": "fuid_{}".format(tmpuser["userid"]), - # "username": "fusername_{}".format(tmpuser["userid"]), - # }, - "msg": "操作成功", - "time": get_current_time_as_millis(), - } - return body +def _get_login_details( + apptype: str, countrycode: str, user: dict[str, Any], token: str +) -> dict[str, Any]: + details: dict[str, Any] = { + "accessToken": token, + "uid": f"fuid_{user['userid']}", + "username": f"fusername_{user['userid']}", + "country": countrycode, + "email": "null@null.com", + } - except Exception as e: - _LOGGER.exception(f"{e}") - return {} + if "global_" in apptype: + details.update( + {"ucUid": details["uid"], "loginName": details["username"], "mobile": None} + ) + + return details diff --git a/bumper/web/middlewares.py b/bumper/web/middlewares.py index 760a9a5..f49c665 100644 --- a/bumper/web/middlewares.py +++ b/bumper/web/middlewares.py @@ -16,11 +16,11 @@ _LOGGER = get_logger("webserver_requests") class CustomEncoder(json.JSONEncoder): """Custom json encoder, which supports set.""" - def default(self, obj: Any) -> Any: + def default(self, o: Any) -> Any: """Convert objects, which are not supported by the default JSONEncoder.""" - if isinstance(obj, set): - return list(obj) - return json.JSONEncoder.default(self, obj) + if isinstance(o, set): + return list(o) + return json.JSONEncoder.default(self, o) _EXCLUDE_FROM_LOGGING = [ @@ -32,7 +32,9 @@ _EXCLUDE_FROM_LOGGING = [ @web.middleware -async def log_all_requests(request: Request, handler: Handler) -> StreamResponse: +async def log_all_requests( # pylint: disable=too-many-branches + request: Request, handler: Handler +) -> StreamResponse: """Middleware to log all requests.""" if ( not request.match_info.route.resource @@ -45,7 +47,7 @@ async def log_all_requests(request: Request, handler: Handler) -> StreamResponse "url": str(request.url), "path": request.path, "query_string": request.query_string, - "headers": {h for h in request.headers.items()}, + "headers": set(request.headers.items()), "route_resource": request.match_info.route.resource.canonical, } } @@ -56,7 +58,7 @@ async def log_all_requests(request: Request, handler: Handler) -> StreamResponse if request.content_type == "application/json": to_log["request"]["body"] = await request.json() else: - to_log["request"]["body"] = {h for h in await request.post()} + to_log["request"]["body"] = set(await request.post()) except Exception: _LOGGER.exception( "An exception occurred during logging the request.", exc_info=True @@ -75,7 +77,7 @@ async def log_all_requests(request: Request, handler: Handler) -> StreamResponse to_log["response"] = { "status": f"{response.status}", - "headers": {h for h in response.headers.items()}, + "headers": set(response.headers.items()), } if isinstance(response, Response) and response.body: @@ -93,7 +95,7 @@ async def log_all_requests(request: Request, handler: Handler) -> StreamResponse raise except web.HTTPNotFound: - _LOGGER.debug(f"Request path {request.raw_path} not found") + _LOGGER.debug("Request path %s not found", request.raw_path) raise finally: diff --git a/bumper/web/plugins/api/appsvr.py b/bumper/web/plugins/api/appsvr.py index 0b9016c..bcd6ec8 100644 --- a/bumper/web/plugins/api/appsvr.py +++ b/bumper/web/plugins/api/appsvr.py @@ -10,7 +10,6 @@ from aiohttp.web_exceptions import HTTPInternalServerError from aiohttp.web_request import Request from aiohttp.web_response import Response from aiohttp.web_routedef import AbstractRouteDef -from amqtt.session import Session import bumper from bumper.db import _db_get, token_by_authcode, user_add_oauth @@ -136,8 +135,7 @@ def _include_product_iot_map_info(bot: dict[str, Any]) -> dict[str, Any]: ) # mqtt_connection is not always set correctly, therefore workaround until fixed properly - session: Session - for (session, _) in bumper.mqtt_server.broker._sessions.values(): + for session in bumper.mqtt_server.sessions: did = session.client_id.split("@")[0] if did == bot["did"] and session.transitions.state == "connected": result["status"] = 1 diff --git a/bumper/web/server.py b/bumper/web/server.py index e207663..805fdf3 100644 --- a/bumper/web/server.py +++ b/bumper/web/server.py @@ -5,6 +5,7 @@ import json import logging import os import ssl +from typing import Any import aiohttp import aiohttp_jinja2 @@ -22,22 +23,21 @@ from bumper.web.middlewares import log_all_requests from bumper.web.plugins import add_plugins -class _aiohttp_filter(logging.Filter): +class _AiohttpFilter(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: if record.name == "aiohttp.access" and record.levelno == 20: # Filters aiohttp.access log to switch it from INFO to DEBUG record.levelno = 10 record.levelname = "DEBUG" - if record.levelno == 10 and get_logger("confserver").getEffectiveLevel() == 10: - return True - else: - return False + return ( + record.levelno == 10 and get_logger("confserver").getEffectiveLevel() == 10 + ) _LOGGER = get_logger("webserver") # Add logging filter above to aiohttp.access -logging.getLogger("aiohttp.access").addFilter(_aiohttp_filter()) +logging.getLogger("aiohttp.access").addFilter(_AiohttpFilter()) _LOGGER_PROXY = logging.getLogger("web_proxy") @@ -129,9 +129,9 @@ class WebServer: ) await site.start() - except Exception as e: - _LOGGER.exception(f"{e}") - raise e + except Exception: + _LOGGER.exception("An exception occurred", exc_info=True) + raise async def shutdown(self) -> None: """Shutdown server.""" @@ -143,15 +143,16 @@ class WebServer: self._runners.clear() await self._app.shutdown() - except Exception as e: - _LOGGER.exception(f"{e}") + except Exception: + _LOGGER.exception("An exception occurred", exc_info=True) + raise async def _handle_base(self, request: Request) -> Response: try: bots = _db_get().table("bots").all() clients = _db_get().table("clients").all() mq_sessions = [] - for (session, _) in bumper.mqtt_server.broker._sessions.values(): + for session in bumper.mqtt_server.sessions: mq_sessions.append( { "username": session.username, @@ -159,7 +160,7 @@ class WebServer: "state": session.transitions.state, } ) - all = { + context = { "bots": bots, "clients": clients, "helperbot": {"connected": bumper.mqtt_helperbot.is_connected}, @@ -172,9 +173,11 @@ class WebServer: }, "xmpp_server": bumper.xmpp_server, } - return aiohttp_jinja2.render_template("home.jinja2", request, context=all) - except Exception as e: - _LOGGER.exception(f"{e}") + return aiohttp_jinja2.render_template( + "home.jinja2", request, context=context + ) + except Exception: # pylint: disable=broad-except + _LOGGER.exception("An exception occurred", exc_info=True) raise HTTPInternalServerError @@ -208,9 +211,10 @@ class WebServer: return web.json_response({"status": "complete"}) return web.json_response({"status": "invalid service"}) - except Exception as e: - _LOGGER.exception(f"{e}") - raise + except Exception: # pylint: disable=broad-except + _LOGGER.exception("An exception occurred", exc_info=True) + + raise HTTPInternalServerError async def _handle_remove_bot(self, request: Request) -> Response: try: @@ -218,11 +222,11 @@ class WebServer: bot_remove(did) if bot_get(did): return web.json_response({"status": "failed to remove bot"}) - else: - return web.json_response({"status": "successfully removed bot"}) - except Exception as e: - _LOGGER.exception(f"{e}") + return web.json_response({"status": "successfully removed bot"}) + + except Exception: # pylint: disable=broad-except + _LOGGER.exception("An exception occurred", exc_info=True) raise HTTPInternalServerError @@ -232,11 +236,11 @@ class WebServer: client_remove(resource) if client_get(resource): return web.json_response({"status": "failed to remove client"}) - else: - return web.json_response({"status": "successfully removed client"}) - except Exception as e: - _LOGGER.exception(f"{e}") + return web.json_response({"status": "successfully removed client"}) + + except Exception: # pylint: disable=broad-except + _LOGGER.exception("An exception occurred", exc_info=True) raise HTTPInternalServerError @@ -255,22 +259,18 @@ class WebServer: srvip = bumper.bumper_announce_ip srvport = 5223 _LOGGER.info( - "Announcing EcoMsgNew Server to bot as: {}:{}".format( - srvip, srvport - ) + "Announcing EcoMsgNew Server to bot as: %s:%d", srvip, srvport ) server = json.dumps({"ip": srvip, "port": srvport, "result": "ok"}) # bot seems to be very picky about having no spaces, only way was with text server = server.replace(" ", "") return web.json_response(text=server) - elif service == "EcoUpdate": + if service == "EcoUpdate": srvip = "47.88.66.164" # EcoVacs Server srvport = 8005 _LOGGER.info( - "Announcing EcoUpdate Server to bot as: {}:{}".format( - srvip, srvport - ) + "Announcing EcoUpdate Server to bot as: %s:%d", srvip, srvport ) return web.json_response( {"result": "ok", "ip": srvip, "port": srvport} @@ -278,8 +278,8 @@ class WebServer: return web.json_response({}) - except Exception as e: - _LOGGER.exception(f"{e}") + except Exception: # pylint: disable=broad-except + _LOGGER.exception("An exception occurred", exc_info=True) raise HTTPInternalServerError @@ -297,8 +297,8 @@ class WebServer: return web.json_response(body) - except Exception as e: - _LOGGER.exception(f"{e}") + except Exception: # pylint: disable=broad-except + _LOGGER.exception("An exception occurred", exc_info=True) raise HTTPInternalServerError @@ -316,62 +316,54 @@ class WebServer: verify_ssl=False, resolver=get_resolver_with_public_nameserver() ), ) as session: + data: Any = None + json_data: Any = None if request.content.total_bytes > 0: read_body = await request.read() _LOGGER_PROXY.info( - f"HTTP Proxy Request to EcoVacs (body=true) (URL:{request.url}) - {read_body.decode('utf-8')}" + "HTTP Proxy Request to EcoVacs (body=true) (URL:%s) - %s", + request.url, + read_body.decode("utf-8"), ) if request.content_type == "application/x-www-form-urlencoded": # android apps use form - fdata = await request.post() - async with session.request( - request.method, request.url, data=fdata - ) as resp: - response = await resp.text() - _LOGGER_PROXY.info( - f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - {response}" - ) + data = await request.post() else: # handle json - async with session.request( - request.method, request.url, json=await request.json() - ) as resp: - response = await resp.text() - _LOGGER_PROXY.info( - f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - {response}" - ) + json_data = await request.json() else: _LOGGER_PROXY.info( - f"HTTP Proxy Request to EcoVacs (body=false) (URL:{request.url})" + "HTTP Proxy Request to EcoVacs (body=false) (URL:%s)", + request.url, ) - async with session.request(request.method, request.url) as resp: - if resp.content_type == "application/octet-stream": - _LOGGER_PROXY.info( - f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - " - ) - return web.Response(body=await resp.read()) - else: - response = await resp.text() - _LOGGER_PROXY.info( - f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - {response}" - ) - if resp.status == 200: - if resp.content_type == "application/json": - response = json.loads(response) - return web.json_response(response) + async with session.request( + request.method, request.url, data=data, json=json_data + ) as resp: if resp.content_type == "application/octet-stream": - return web.Response(body=response) + _LOGGER_PROXY.info( + "HTTP Proxy Response from EcoVacs (URL: %s) - (Status: %d) - ", + request.url, + resp.status, + ) + return web.Response(body=await resp.read()) - return web.Response(text=response) + response = await resp.text() + _LOGGER_PROXY.info( + "HTTP Proxy Response from EcoVacs (URL: %s) - (Status: %d) - %s", + request.url, + resp.status, + response, + ) + return web.Response(text=response) except asyncio.CancelledError: _LOGGER_PROXY.exception( - f"Request cancelled or timeout - {request.url}", exc_info=True + "Request cancelled or timeout - %s", request.url, exc_info=True ) raise - except Exception: + except Exception: # pylint: disable=broad-except _LOGGER_PROXY.exception("An exception occurred", exc_info=True) raise HTTPInternalServerError