diff --git a/bumper/mqtt/helper_bot.py b/bumper/mqtt/helper_bot.py index f1685f0..4ce7060 100644 --- a/bumper/mqtt/helper_bot.py +++ b/bumper/mqtt/helper_bot.py @@ -2,7 +2,7 @@ import asyncio import json import ssl -from typing import Any, MutableMapping, Union +from typing import Any, MutableMapping from cachetools import TTLCache from gmqtt import Client, Subscription @@ -60,7 +60,7 @@ class HelperBot: data_decoded = str(payload.decode()) if topic_split[10] in self._commands: self._commands[topic_split[10]].add_response(data_decoded) - except Exception: + except Exception: # pylint: disable=broad-except _LOGGER.error( "An exception occurred during handling message.", exc_info=True ) diff --git a/bumper/web/images/__init__.py b/bumper/web/images/__init__.py index fd32b16..51d9290 100644 --- a/bumper/web/images/__init__.py +++ b/bumper/web/images/__init__.py @@ -1,3 +1,4 @@ +"""Web image module.""" import logging import os diff --git a/bumper/web/plugins/__init__.py b/bumper/web/plugins/__init__.py index 0e80435..db93938 100644 --- a/bumper/web/plugins/__init__.py +++ b/bumper/web/plugins/__init__.py @@ -31,6 +31,7 @@ def _add_routes( if not module.__name__.startswith(plugin_module_name): return + assert module.__file__ is not None if module.__file__.endswith("__init__.py"): sub_app = web.Application() else: @@ -53,6 +54,7 @@ def _add_routes( def _import_plugins(module: ModuleType) -> None: + assert module.__file__ is not None for file in glob(join(dirname(module.__file__), "**/*.py"), recursive=True): if not isfile(file) or file == module.__file__: continue @@ -77,6 +79,7 @@ def add_plugins(app: web.Application) -> None: def get_success_response(data: Any) -> Response: + """Get success response with provided data.""" body = { "code": RETURN_API_SUCCESS, "data": data, diff --git a/bumper/web/plugins/api/dim.py b/bumper/web/plugins/api/dim.py index b55182c..8fc4c96 100644 --- a/bumper/web/plugins/api/dim.py +++ b/bumper/web/plugins/api/dim.py @@ -33,14 +33,12 @@ async def _handle_dim_devmanager(request: Request) -> Response: if bot["company"] == "eco-ng" and bot["mqtt_connection"]: retcmd = await bumper.mqtt_helperbot.send_command(json_body, randomid) body = retcmd - logging.debug(f"Send Bot - {json_body}") - logging.debug(f"Bot Response - {body}") + logging.debug("Send Bot - %s", json_body) + logging.debug("Bot Response - %s", body) return web.json_response(body) # No response, send error back - logging.error( - "No bots with DID: {} connected to MQTT".format(json_body["toId"]) - ) + logging.error("No bots with DID: %s connected to MQTT", json_body["toId"]) body = {"id": randomid, "errno": ERR_COMMON, "ret": "fail"} return web.json_response(body) diff --git a/bumper/web/plugins/api/iot.py b/bumper/web/plugins/api/iot.py index 6c50e96..04b6ab9 100644 --- a/bumper/web/plugins/api/iot.py +++ b/bumper/web/plugins/api/iot.py @@ -31,35 +31,32 @@ async def _handle_devmanager_bot_command(request: Request) -> Response: if bot["company"] == "eco-ng": retcmd = await bumper.mqtt_helperbot.send_command(json_body, randomid) body = retcmd - logging.debug(f"Send Bot - {json_body}") - logging.debug(f"Bot Response - {body}") - return web.json_response(body) - else: - # No response, send error back - logging.error( - "No bots with DID: {} connected to MQTT".format(json_body["toId"]) - ) - body = { - "id": randomid, - "errno": 500, - "ret": "fail", - "debug": "wait for response timed out", - } + logging.debug("Send Bot - %s", json_body) + logging.debug("Bot Response - %s", body) return web.json_response(body) - else: - if "td" in json_body: # Seen when doing initial wifi config - if json_body["td"] == "PollSCResult": - body = {"ret": "ok"} - return web.json_response(body) + # No response, send error back + logging.error("No bots with DID: %s connected to MQTT", json_body["toId"]) + body = { + "id": randomid, + "errno": 500, + "ret": "fail", + "debug": "wait for response timed out", + } + return web.json_response(body) - if json_body["td"] == "HasUnreadMsg": # EcoVacs Home - body = {"ret": "ok", "unRead": False} - return web.json_response(body) + if "td" in json_body: # Seen when doing initial wifi config + if json_body["td"] == "PollSCResult": + body = {"ret": "ok"} + return web.json_response(body) - if json_body["td"] == "PreWifiConfig": # EcoVacs Home - body = {"ret": "ok"} - return web.json_response(body) + if json_body["td"] == "HasUnreadMsg": # EcoVacs Home + body = {"ret": "ok", "unRead": False} + return web.json_response(body) + + if json_body["td"] == "PreWifiConfig": # EcoVacs Home + body = {"ret": "ok"} + return web.json_response(body) except Exception: # pylint: disable=broad-except logging.error("Unexpected exception occurred", exc_info=True) diff --git a/bumper/web/plugins/api/lg.py b/bumper/web/plugins/api/lg.py index ed765c4..2ca6955 100644 --- a/bumper/web/plugins/api/lg.py +++ b/bumper/web/plugins/api/lg.py @@ -29,23 +29,23 @@ async def _handle_lg_log(request: Request) -> Response: botdetails = bot_get(did) if botdetails: - if not "cmdName" in json_body: + if "cmdName" not in json_body: if "td" in json_body: json_body["cmdName"] = json_body["td"] - if not "toId" in json_body: + if "toId" not in json_body: json_body["toId"] = did - if not "toType" in json_body: + if "toType" not in json_body: json_body["toType"] = botdetails["class"] - if not "toRes" in json_body: + if "toRes" not in json_body: json_body["toRes"] = botdetails["resource"] - if not "payloadType" in json_body: + if "payloadType" not in json_body: json_body["payloadType"] = "x" - if not "payload" in json_body: + if "payload" not in json_body: # json_body["payload"] = "" if json_body["td"] == "GetCleanLogs": json_body["td"] = "q" @@ -56,17 +56,17 @@ async def _handle_lg_log(request: Request) -> Response: if bot["company"] == "eco-ng": retcmd = await bumper.mqtt_helperbot.send_command(json_body, randomid) body = retcmd - logging.debug(f"Send Bot - {json_body}") - logging.debug(f"Bot Response - {body}") + logging.debug("Send Bot - %s", json_body) + logging.debug("Bot Response - %s", body) logs = [] logsroot = ET.fromstring(retcmd["resp"]) if logsroot.attrib["ret"] == "ok": - for l in logsroot: + for log_line in logsroot: cleanlog = { - "ts": l.attrib["s"], - "area": l.attrib["a"], - "last": l.attrib["l"], - "cleanType": l.attrib["t"], + "ts": log_line.attrib["s"], + "area": log_line.attrib["a"], + "last": log_line.attrib["l"], + "cleanType": log_line.attrib["t"], # imageUrl allows for providing images of cleanings, something to look into later # "imageUrl": "https://localhost:8007", } @@ -78,15 +78,13 @@ async def _handle_lg_log(request: Request) -> Response: else: body = {"ret": "ok", "logs": []} - logging.debug(f"lg logs return: {json.dumps(body)}") + logging.debug("lg logs return: %s", json.dumps(body)) return web.json_response(body) - else: - # No response, send error back - logging.error( - "No bots with DID: {} connected to MQTT".format(json_body["toId"]) - ) - except Exception as e: - logging.exception(f"{e}") + + # No response, send error back + logging.error("No bots with DID: %s connected to MQTT", json_body["toId"]) + except Exception: # pylint: disable=broad-except + logging.error("An unknown exception occurred", exc_info=True) body = {"id": randomid, "errno": ERR_COMMON, "ret": "fail"} return web.json_response(body) diff --git a/bumper/web/plugins/api/neng.py b/bumper/web/plugins/api/neng.py index b1e619d..4609221 100644 --- a/bumper/web/plugins/api/neng.py +++ b/bumper/web/plugins/api/neng.py @@ -16,6 +16,7 @@ async def _handle_neng_has_unread_message(_: Request) -> Response: async def handle_neng_get_share_msgs(_: Request) -> Response: + """Return shared messages.""" # EcoVacs Home body = {"code": 0, "data": {"hasNext": False, "msgs": []}} @@ -46,6 +47,7 @@ async def handle_neng_get_share_msgs(_: Request) -> Response: async def handle_neng_get_list(_: Request) -> Response: + """Get messages.""" # EcoVacs Home body = {"code": 0, "data": {"hasNext": False, "msgs": []}} diff --git a/bumper/web/plugins/api/pim.py b/bumper/web/plugins/api/pim.py index ab6c771..ea19543 100644 --- a/bumper/web/plugins/api/pim.py +++ b/bumper/web/plugins/api/pim.py @@ -4,6 +4,9 @@ import logging from typing import Iterable from aiohttp import web +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 bumper.models import RETURN_API_SUCCESS, EcoVacsHomeProducts @@ -22,7 +25,7 @@ class PimPlugin(WebserverPlugin): web.route( "*", "/pim/product/getProductIotMap", - self.handle_getProductIotMap, + _handle_get_product_iot_map, ), web.route( "*", @@ -32,83 +35,93 @@ class PimPlugin(WebserverPlugin): web.route( "*", "/pim/product/getConfignetAll", - self.handle_getConfignetAll, + _handle_get_confignet_all, ), web.route( "*", "/pim/product/getConfigGroups", - self.handle_getConfigGroups, + _handle_get_config_groups, ), web.route( "*", "/pim/dictionary/getErrDetail", - self.handle_getErrDetail, + _handle_get_err_detail, ), web.route( "*", "/pim/product/software/config/batch", - self.handle_product_config_batch, + _handle_product_config_batch, ), ] - async def handle_getProductIotMap(self, request): - try: - body = { - "code": RETURN_API_SUCCESS, - "data": EcoVacsHomeProducts, - } - return web.json_response(body) - except Exception as e: - logging.exception(f"{e}") +async def _handle_get_product_iot_map(_: Request) -> Response: + """Get product iot map.""" + try: + body = { + "code": RETURN_API_SUCCESS, + "data": EcoVacsHomeProducts, + } + return web.json_response(body) + except Exception: # pylint: disable=broad-except + logging.error("An exception occurred during handling request.", exc_info=True) + raise HTTPInternalServerError - async def handle_getConfignetAll(self, request): - try: - body = confignetAllResponse - return web.json_response(body) - except Exception as e: - logging.exception(f"{e}") +async def _handle_get_confignet_all(_: Request) -> Response: + """Get config net all.""" + try: + body = confignetAllResponse + return web.json_response(body) + except Exception: # pylint: disable=broad-except + logging.error("An exception occurred during handling request.", exc_info=True) + raise HTTPInternalServerError - async def handle_getConfigGroups(self, request): - try: - body = configGroupsResponse - return web.json_response(body) - except Exception as e: - logging.exception(f"{e}") +async def _handle_get_config_groups(_: Request) -> Response: + """Get config groups.""" + try: + body = configGroupsResponse + return web.json_response(body) + except Exception: # pylint: disable=broad-except + logging.error("An exception occurred during handling request.", exc_info=True) + raise HTTPInternalServerError - async def handle_getErrDetail(self, request): - try: - body = { - "code": -1, - "data": [], - "msg": "This errcode's detail is not exists", - } - return web.json_response(body) - except Exception as e: - logging.exception(f"{e}") +async def _handle_get_err_detail(_: Request) -> Response: + """Get error details.""" + try: + body = { + "code": -1, + "data": [], + "msg": "This errcode's detail is not exists", + } + return web.json_response(body) + except Exception: # pylint: disable=broad-except + logging.error("An exception occurred during handling request.", exc_info=True) + raise HTTPInternalServerError - async def handle_product_config_batch(self, request): - try: - json_body = json.loads(await request.text()) - data = [] - for pid in json_body["pids"]: - for productConfig in productConfigBatch: - if pid == productConfig["pid"]: - data.append(productConfig) - continue - # not found in productConfigBatch - # some devices don't have any product configuration - data.append({"cfg": {}, "pid": pid}) +async def _handle_product_config_batch(request: Request) -> Response: + """Handle product config batch.""" + try: + json_body = json.loads(await request.text()) + data = [] + for pid in json_body["pids"]: + for product_config in productConfigBatch: + if pid == product_config["pid"]: + data.append(product_config) + continue - body = {"code": 200, "data": data, "message": "success"} - return web.json_response(body) + # not found in productConfigBatch + # some devices don't have any product configuration + data.append({"cfg": {}, "pid": pid}) - except Exception as e: - logging.exception(f"{e}") + body = {"code": 200, "data": data, "message": "success"} + return web.json_response(body) + except Exception: # pylint: disable=broad-except + logging.error("An exception occurred during handling request.", exc_info=True) + raise HTTPInternalServerError confignetAllResponse = { diff --git a/bumper/web/plugins/api/users.py b/bumper/web/plugins/api/users.py index ecd787a..09ba132 100644 --- a/bumper/web/plugins/api/users.py +++ b/bumper/web/plugins/api/users.py @@ -1,9 +1,11 @@ """Users plugin module.""" import json import logging -from typing import Iterable +from typing import Any, Iterable, Mapping from aiohttp import web +from aiohttp.abc import Request +from aiohttp.web_response import Response from aiohttp.web_routedef import AbstractRouteDef from bumper import bumper_announce_ip @@ -22,98 +24,99 @@ class UsersPlugin(WebserverPlugin): web.route( "*", "/users/user.do", - self._handle_usersapi, + _handle_usersapi, ), ] - async def _handle_usersapi(self, request): - if not request.method == "GET": # Skip GET for now - try: - body = {} - postbody = {} - if request.content_type == "application/x-www-form-urlencoded": - postbody = await request.post() +async def _handle_usersapi(request: Request) -> Response: + if not request.method == "GET": # Skip GET for now + try: - else: - postbody = json.loads(await request.text()) + body = {} + postbody: Mapping[str, Any] + if request.content_type == "application/x-www-form-urlencoded": + postbody = await request.post() + else: + postbody = json.loads(await request.text()) - todo = postbody["todo"] - if todo == "FindBest": - service = postbody["service"] - if service == "EcoMsgNew": - srvip = bumper_announce_ip - srvport = 5223 - logging.info( - "Announcing EcoMsgNew Server to bot as: {}:{}".format( - srvip, srvport - ) - ) - msgserver = {"ip": srvip, "port": srvport, "result": "ok"} - msgserver = json.dumps(msgserver) - msgserver = msgserver.replace( - " ", "" - ) # bot seems to be very picky about having no spaces, only way was with text + todo = postbody["todo"] + if todo == "FindBest": + service = postbody["service"] + if service == "EcoMsgNew": + srvip = bumper_announce_ip + srvport = 5223 + logging.info( + "Announcing EcoMsgNew Server to bot as: %s:%d", + srvip, + srvport, + ) + msgserver = json.dumps( + {"ip": srvip, "port": srvport, "result": "ok"} + ) + # bot seems to be very picky about having no spaces, only way was with text + msgserver = msgserver.replace(" ", "") - return web.json_response(text=msgserver) + return web.json_response(text=msgserver) - elif service == "EcoUpdate": - srvip = "47.88.66.164" # EcoVacs Server - srvport = 8005 - logging.info( - "Announcing EcoUpdate Server to bot as: {}:{}".format( - srvip, srvport - ) - ) - body = {"result": "ok", "ip": srvip, "port": srvport} + if service == "EcoUpdate": + srvip = "47.88.66.164" # EcoVacs Server + srvport = 8005 + logging.info( + "Announcing EcoMsgNew Server to bot as: %s:%d", + srvip, + srvport, + ) + body = {"result": "ok", "ip": srvip, "port": srvport} - elif todo == "loginByItToken": - if "userId" in postbody: - if check_authcode(postbody["userId"], postbody["token"]): - body = { - "resource": postbody["resource"], - "result": "ok", - "todo": "result", - "token": postbody["token"], - "userId": postbody["userId"], - } - else: # EcoVacs Home LoginByITToken - loginToken = loginByItToken(postbody["token"]) - if not loginToken == {}: - body = { - "resource": postbody["resource"], - "result": "ok", - "todo": "result", - "token": loginToken["token"], - "userId": loginToken["userid"], - } - else: - body = {"result": "fail", "todo": "result"} + elif todo == "loginByItToken": + if "userId" in postbody: + if check_authcode(postbody["userId"], postbody["token"]): + body = { + "resource": postbody["resource"], + "result": "ok", + "todo": "result", + "token": postbody["token"], + "userId": postbody["userId"], + } + else: # EcoVacs Home LoginByITToken + login_token = loginByItToken(postbody["token"]) + if not login_token: + body = { + "resource": postbody["resource"], + "result": "ok", + "todo": "result", + "token": login_token["token"], + "userId": login_token["userid"], + } + else: + body = {"result": "fail", "todo": "result"} - elif todo == "GetDeviceList": - body = { - "devices": db_get().table("bots").all(), - "result": "ok", - "todo": "result", - } + elif todo == "GetDeviceList": + body = { + "devices": db_get().table("bots").all(), + "result": "ok", + "todo": "result", + } - elif todo == "SetDeviceNick": - bot_set_nick(postbody["did"], postbody["nick"]) - body = {"result": "ok", "todo": "result"} + elif todo == "SetDeviceNick": + bot_set_nick(postbody["did"], postbody["nick"]) + body = {"result": "ok", "todo": "result"} - elif todo == "AddOneDevice": - bot_set_nick(postbody["did"], postbody["nick"]) - body = {"result": "ok", "todo": "result"} + elif todo == "AddOneDevice": + bot_set_nick(postbody["did"], postbody["nick"]) + body = {"result": "ok", "todo": "result"} - elif todo == "DeleteOneDevice": - bot_remove(postbody["did"]) - body = {"result": "ok", "todo": "result"} + elif todo == "DeleteOneDevice": + bot_remove(postbody["did"]) + body = {"result": "ok", "todo": "result"} - return web.json_response(body) + return web.json_response(body) + except Exception: # pylint: disable=broad-except + logging.error( + "An exception occurred during handling request.", exc_info=True + ) - except Exception as e: - logging.exception(f"{e}") - - # Return fail for GET - body = {"result": "fail", "todo": "result"} - return web.json_response(body) + # Return fail for GET + body = {"result": "fail", "todo": "result"} + return web.json_response(body) diff --git a/bumper/web/plugins/v1/private/campaign.py b/bumper/web/plugins/v1/private/campaign.py index ef88e95..64305b4 100644 --- a/bumper/web/plugins/v1/private/campaign.py +++ b/bumper/web/plugins/v1/private/campaign.py @@ -29,14 +29,15 @@ class CampaignPlugin(WebserverPlugin): async def _handle_home_page_alert(_: Request) -> Response: - nextAlert = convert_to_millis((datetime.now() + timedelta(hours=12)).timestamp()) return get_success_response( { "clickSchemeUrl": None, "clickWebUrl": None, "hasCampaign": "N", "imageUrl": None, - "nextAlertTime": nextAlert, + "nextAlertTime": convert_to_millis( + (datetime.now() + timedelta(hours=12)).timestamp() + ), "serverTime": get_current_time_as_millis(), } ) diff --git a/bumper/web/plugins/v1/private/message.py b/bumper/web/plugins/v1/private/message.py index cf04c9e..0a447f6 100644 --- a/bumper/web/plugins/v1/private/message.py +++ b/bumper/web/plugins/v1/private/message.py @@ -1,3 +1,4 @@ +"""Message plugin module.""" from typing import Iterable from aiohttp import web