mia/bumper
1
0
Fork 0
Code Issues 47 Pull requests 11 Projects Releases Packages Wiki Activity Actions

fix pre-commit findings

Browse source
This commit is contained in:
Robert Resch 2022-05-17 22:18:56 +02:00
parent e81ed17aa3
commit 50d31344c8
11 changed files with 205 additions and 188 deletions
Download patch file Download diff file Expand all files Collapse all files

4
bumper/mqtt/helper_bot.py
View file

@ -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
)

1
bumper/web/images/__init__.py
View file

@ -1,3 +1,4 @@
"""Web image module."""
import logging
import os

3
bumper/web/plugins/__init__.py
View file

@ -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,

8
bumper/web/plugins/api/dim.py
View file

@ -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)

11
bumper/web/plugins/api/iot.py
View file

@ -31,14 +31,12 @@ 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}")
logging.debug("Send Bot - %s", json_body)
logging.debug("Bot Response - %s", 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"])
)
logging.error("No bots with DID: %s connected to MQTT", json_body["toId"])
body = {
"id": randomid,
"errno": 500,
@ -47,7 +45,6 @@ async def _handle_devmanager_bot_command(request: Request) -> Response:
}
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"}

38
bumper/web/plugins/api/lg.py
View file

@ -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}")
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)

2
bumper/web/plugins/api/neng.py
View file

@ -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": []}}

61
bumper/web/plugins/api/pim.py
View file

@ -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,53 +35,61 @@ 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):
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
except Exception as e:
logging.exception(f"{e}")
async def handle_getConfignetAll(self, request):
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
except Exception as e:
logging.exception(f"{e}")
async def handle_getConfigGroups(self, request):
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
except Exception as e:
logging.exception(f"{e}")
async def handle_getErrDetail(self, request):
async def _handle_get_err_detail(_: Request) -> Response:
"""Get error details."""
try:
body = {
"code": -1,
@ -86,18 +97,20 @@ class PimPlugin(WebserverPlugin):
"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
except Exception as e:
logging.exception(f"{e}")
async def handle_product_config_batch(self, request):
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 productConfig in productConfigBatch:
if pid == productConfig["pid"]:
data.append(productConfig)
for product_config in productConfigBatch:
if pid == product_config["pid"]:
data.append(product_config)
continue
# not found in productConfigBatch
@ -106,9 +119,9 @@ class PimPlugin(WebserverPlugin):
body = {"code": 200, "data": data, "message": "success"}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
except Exception: # pylint: disable=broad-except
logging.error("An exception occurred during handling request.", exc_info=True)
raise HTTPInternalServerError
confignetAllResponse = {

49
bumper/web/plugins/api/users.py
View file

@ -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,19 +24,19 @@ class UsersPlugin(WebserverPlugin):
web.route(
"*",
"/users/user.do",
self._handle_usersapi,
_handle_usersapi,
),
]
async def _handle_usersapi(self, request):
async def _handle_usersapi(request: Request) -> Response:
if not request.method == "GET": # Skip GET for now
try:
body = {}
postbody = {}
postbody: Mapping[str, Any]
if request.content_type == "application/x-www-form-urlencoded":
postbody = await request.post()
else:
postbody = json.loads(await request.text())
@ -45,25 +47,25 @@ class UsersPlugin(WebserverPlugin):
srvip = bumper_announce_ip
srvport = 5223
logging.info(
"Announcing EcoMsgNew Server to bot as: {}:{}".format(
srvip, srvport
"Announcing EcoMsgNew Server to bot as: %s:%d",
srvip,
srvport,
)
msgserver = json.dumps(
{"ip": srvip, "port": srvport, "result": "ok"}
)
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
# bot seems to be very picky about having no spaces, only way was with text
msgserver = msgserver.replace(" ", "")
return web.json_response(text=msgserver)
elif service == "EcoUpdate":
if service == "EcoUpdate":
srvip = "47.88.66.164" # EcoVacs Server
srvport = 8005
logging.info(
"Announcing EcoUpdate Server to bot as: {}:{}".format(
srvip, srvport
)
"Announcing EcoMsgNew Server to bot as: %s:%d",
srvip,
srvport,
)
body = {"result": "ok", "ip": srvip, "port": srvport}
@ -78,14 +80,14 @@ class UsersPlugin(WebserverPlugin):
"userId": postbody["userId"],
}
else: # EcoVacs Home LoginByITToken
loginToken = loginByItToken(postbody["token"])
if not loginToken == {}:
login_token = loginByItToken(postbody["token"])
if not login_token:
body = {
"resource": postbody["resource"],
"result": "ok",
"todo": "result",
"token": loginToken["token"],
"userId": loginToken["userid"],
"token": login_token["token"],
"userId": login_token["userid"],
}
else:
body = {"result": "fail", "todo": "result"}
@ -110,9 +112,10 @@ class UsersPlugin(WebserverPlugin):
body = {"result": "ok", "todo": "result"}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
except Exception: # pylint: disable=broad-except
logging.error(
"An exception occurred during handling request.", exc_info=True
)
# Return fail for GET
body = {"result": "fail", "todo": "result"}

5
bumper/web/plugins/v1/private/campaign.py
View file

@ -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(),
}
)

1
bumper/web/plugins/v1/private/message.py
View file

@ -1,3 +1,4 @@
"""Message plugin module."""
from typing import Iterable
from aiohttp import web