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

refactor /api endpoints to use new plugin system

Browse source
This commit is contained in:
Robert Resch 2022-03-06 18:15:03 +01:00
parent 7fe1c388dd
commit 329217ba8f
17 changed files with 447 additions and 451 deletions
Download patch file Download diff file Expand all files Collapse all files

2
bumper/confserver.py
View file

@ -106,11 +106,9 @@ class ConfServer:
# common api paths
api_v1 = {"prefix": "/v1/", "app": web.Application()} # for /v1/
portal_api = {"prefix": "/api/", "app": web.Application()} # for /api/
apis = {
WebserverSubApi.V1: api_v1,
WebserverSubApi.API: portal_api,
}
add_plugins(self._app)

81
bumper/plugins/bumper_confserver_api_dim.py
View file

@ -1,81 +0,0 @@
"""Api dim plugin module."""
import json
import logging
import random
import string
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
import bumper
from bumper.db import bot_get
from bumper.models import ERR_COMMON
from bumper.plugins import WebserverPlugin, WebserverSubApi
# pylint: disable=no-self-use
class ApiDimPlugin(WebserverPlugin):
"""Api dim plugin."""
sub_api = WebserverSubApi.API
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route(
"*",
"/dim/devmanager.do",
self._handle_dim_devmanager,
name="portal_api_dim_devmanager",
),
]
async def _handle_dim_devmanager(self, request: Request) -> Response:
# Used in EcoVacs Home App
try:
json_body = json.loads(await request.text())
randomid = "".join(random.sample(string.ascii_letters, 6))
did = ""
if "toId" in json_body: # Its a command
did = json_body["toId"]
if did != "":
bot = bot_get(did)
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}")
return web.json_response(body)
# No response, send error back
logging.error(
"No bots with DID: {} connected to MQTT".format(json_body["toId"])
)
body = {"id": randomid, "errno": ERR_COMMON, "ret": "fail"}
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"] == "HasUnreadMsg": # EcoVacs Home
body = {"ret": "ok", "unRead": False}
return web.json_response(body)
if json_body["td"] == "ReceiveShareDevice": # EcoVacs Home
body = {"ret": "ok"}
return web.json_response(body)
except Exception: # pylint: disable=broad-except
logging.error("Unexpected exception occurred", exc_info=True)
raise HTTPInternalServerError

30
bumper/plugins/bumper_confserver_portal_ecms.py
View file

@ -1,30 +0,0 @@
import logging
from aiohttp import web
from bumper import plugins
class portal_api_ecms(plugins.ConfServerApp):
def __init__(self):
self.name = "portal_api_ecms"
self.plugin_type = "sub_api"
self.sub_api = "portal_api"
self.routes = [
web.route(
"*",
"/ecms/app/ad/res",
self.handle_ad_res,
name="portal_api_ecms_ad_res",
),
]
async def handle_ad_res(self, request):
try:
body = {"code": 0, "data": [], "message": "success", "success": True}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")

77
bumper/plugins/bumper_confserver_portal_iot.py
View file

@ -1,77 +0,0 @@
import json
import logging
import random
import string
from aiohttp import web
import bumper
from bumper.db import bot_get
from bumper.plugins import ConfServerApp
class portal_api_iot(ConfServerApp):
def __init__(self):
self.name = "portal_api_iot"
self.plugin_type = "sub_api"
self.sub_api = "portal_api"
self.routes = [
web.route(
"*",
"/iot/devmanager.do",
self.handle_devmanager_botcommand,
name="portal_api_iot_devmanager",
),
]
async def handle_devmanager_botcommand(self, request):
try:
json_body = json.loads(await request.text())
randomid = "".join(random.sample(string.ascii_letters, 4))
did = ""
if "toId" in json_body: # Its a command
did = json_body["toId"]
if did != "":
bot = bot_get(did)
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",
}
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)
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 as e:
logging.exception(f"{e}")

101
bumper/plugins/bumper_confserver_portal_lg.py
View file

@ -1,101 +0,0 @@
import json
import logging
import random
import string
import xml.etree.ElementTree as ET
from aiohttp import web
import bumper
from bumper.db import bot_get
from bumper.models import ERR_COMMON
from bumper.plugins import ConfServerApp
class portal_api_lg(ConfServerApp):
def __init__(self):
self.name = "portal_api_lg"
self.plugin_type = "sub_api"
self.sub_api = "portal_api"
self.routes = [
web.route("*", "/lg/log.do", self.handle_lg_log, name="portal_api_lg_log"),
]
async def handle_lg_log(self, request): # EcoVacs Home
randomid = "".join(random.sample(string.ascii_letters, 6))
try:
json_body = json.loads(await request.text())
did = json_body["did"]
botdetails = bot_get(did)
if botdetails:
if not "cmdName" in json_body:
if "td" in json_body:
json_body["cmdName"] = json_body["td"]
if not "toId" in json_body:
json_body["toId"] = did
if not "toType" in json_body:
json_body["toType"] = botdetails["class"]
if not "toRes" in json_body:
json_body["toRes"] = botdetails["resource"]
if not "payloadType" in json_body:
json_body["payloadType"] = "x"
if not "payload" in json_body:
# json_body["payload"] = ""
if json_body["td"] == "GetCleanLogs":
json_body["td"] = "q"
json_body["payload"] = '<ctl count="30"/>'
if did != "":
bot = bot_get(did)
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}")
logs = []
logsroot = ET.fromstring(retcmd["resp"])
if logsroot.attrib["ret"] == "ok":
for l in logsroot:
cleanlog = {
"ts": l.attrib["s"],
"area": l.attrib["a"],
"last": l.attrib["l"],
"cleanType": l.attrib["t"],
# imageUrl allows for providing images of cleanings, something to look into later
# "imageUrl": "https://localhost:8007",
}
logs.append(cleanlog)
body = {
"ret": "ok",
"logs": logs,
}
else:
body = {"ret": "ok", "logs": []}
logging.debug(f"lg logs return: {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}")
body = {"id": randomid, "errno": ERR_COMMON, "ret": "fail"}
return web.json_response(body)

100
bumper/plugins/bumper_confserver_portal_neng.py
View file

@ -1,100 +0,0 @@
import logging
from aiohttp import web
from bumper import plugins
class portal_api_neng(plugins.ConfServerApp):
def __init__(self):
self.name = "portal_api_neng"
self.plugin_type = "sub_api"
self.sub_api = "portal_api"
self.routes = [
web.route(
"*",
"/neng/message/hasUnreadMsg",
self.handle_neng_hasUnreadMessage,
name="portal_api_neng_hasUnreadMessage",
),
web.route(
"*",
"/neng/message/getShareMsgs",
self.handle_neng_getShareMsgs,
name="portal_api_neng_getShareMsgs",
),
web.route(
"*",
"/neng/message/getlist",
self.handle_neng_getlist,
name="portal_api_neng_getlist",
),
]
async def handle_neng_hasUnreadMessage(self, request): # EcoVacs Home
try:
body = {"code": 0, "data": {"hasUnRead": True}}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
async def handle_neng_getShareMsgs(self, request): # EcoVacs Home
try:
body = {"code": 0, "data": {"hasNext": False, "msgs": []}}
# share msg response
# {
# "code": 0,
# "data": {
# "hasNext": False,
# "msgs": [
# {
# "action": "shareDevice",
# "deviceName": "DEEBOT 900 Series",
# "did": "DID",
# "icon": "https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839",
# "id": "0154d03a-294e-4b99-a6df-fc2dbf4146d5",
# "isRead": False,
# "message": "User user@gmail.com sent you the sharing invitation of DEEBOT 900 Series.",
# "mid": "ls1ok3",
# "resource": "grU0",
# "shareStatus": "sharing",
# "ts": 1578206187412
# }
# ]
# }
# }
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")
async def handle_neng_getlist(self, request): # EcoVacs Home
try:
body = {"code": 0, "data": {"hasNext": False, "msgs": []}}
# Sample Message
# {
# "id": "5da0ac9d636aec5107627ac4",
# "ts": 1570811036877,
# "did": "bot did",
# "cid": "ls1ok3",
# "name": "DEEBOT 900 Series",
# "icon": "https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839",
# "eventTypeId": "5aab824bb62ce30001f9a702",
# "title": "DEEBOT is off the floor.",
# "body": "DEEBOT is off the floor. Please put it back.",
# "read": false,
# "UILogicId": "D_900",
# "type": "web",
# "url": "https://portal-ww.ecouser.net/api/pim/eventdetail.html?id=5ba21e44aed83800015b9ca8" # Off the floor instructions
# }
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")

31
bumper/plugins/bumper_confserver_portal_rapp.py
View file

@ -1,31 +0,0 @@
import logging
from aiohttp import web
from bumper import plugins
class api_rapp(plugins.ConfServerApp):
def __init__(self):
self.name = "api_rapp"
self.plugin_type = "sub_api"
self.sub_api = "portal_api"
self.routes = [
web.route(
"*", "/rapp/sds/user/data/map/get", self.handle_map_get, name="api_rapp"
),
]
async def handle_map_get(self, request):
try:
body = {
"code": 0,
"data": {"data": {"name": "My Home"}, "tag": None},
"message": "success",
}
return web.json_response(body)
except Exception as e:
logging.exception(f"{e}")

1
bumper/web/plugins/api/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Api plugin modules."""

11
bumper/plugins/bumper_confserver_api_appsvr.py → bumper/web/plugins/api/appsvr.py
View file

@ -1,4 +1,4 @@
"""Api appsvr plugin module."""
"""Appsvr plugin module."""
import json
import logging
from typing import Iterable
@ -11,14 +11,13 @@ from aiohttp.web_routedef import AbstractRouteDef
from bumper.db import db_get, token_by_authcode, user_add_oauth
from bumper.models import include_EcoVacsHomeProducts_info
from bumper.plugins import WebserverPlugin, WebserverSubApi
from .. import WebserverPlugin
# pylint: disable=no-self-use
class ApiAppsvrPlugin(WebserverPlugin):
"""Api appsvr plugin."""
sub_api = WebserverSubApi.API
class AppsvrPlugin(WebserverPlugin):
"""Appsvr plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:

77
bumper/web/plugins/api/dim.py Normal file
View file

@ -0,0 +1,77 @@
"""Dim plugin module."""
import json
import logging
import random
import string
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
import bumper
from bumper.db import bot_get
from bumper.models import ERR_COMMON
from .. import WebserverPlugin
async def _handle_dim_devmanager(request: Request) -> Response:
# Used in EcoVacs Home App
try:
json_body = json.loads(await request.text())
randomid = "".join(random.sample(string.ascii_letters, 6))
did = ""
if "toId" in json_body: # Its a command
did = json_body["toId"]
if did != "":
bot = bot_get(did)
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}")
return web.json_response(body)
# No response, send error back
logging.error(
"No bots with DID: {} connected to MQTT".format(json_body["toId"])
)
body = {"id": randomid, "errno": ERR_COMMON, "ret": "fail"}
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"] == "HasUnreadMsg": # EcoVacs Home
body = {"ret": "ok", "unRead": False}
return web.json_response(body)
if json_body["td"] == "ReceiveShareDevice": # EcoVacs Home
body = {"ret": "ok"}
return web.json_response(body)
except Exception: # pylint: disable=broad-except
logging.error("Unexpected exception occurred", exc_info=True)
raise HTTPInternalServerError
class DimPlugin(WebserverPlugin):
"""Dim plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route(
"*",
"/dim/devmanager.do",
_handle_dim_devmanager,
),
]

30
bumper/web/plugins/api/ecms.py Normal file
View file

@ -0,0 +1,30 @@
"""Ecms plugin module."""
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from .. import WebserverPlugin
async def _handle_ad_res(_: Request) -> Response:
body = {"code": 0, "data": [], "message": "success", "success": True}
return web.json_response(body)
class EcmsPlugin(WebserverPlugin):
"""Ecms plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route(
"*",
"/ecms/app/ad/res",
_handle_ad_res,
),
]

81
bumper/web/plugins/api/iot.py Normal file
View file

@ -0,0 +1,81 @@
"""Iot plugin module."""
import json
import logging
import random
import string
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
import bumper
from bumper.db import bot_get
from .. import WebserverPlugin
async def _handle_devmanager_bot_command(request: Request) -> Response:
try:
json_body = json.loads(await request.text())
randomid = "".join(random.sample(string.ascii_letters, 4))
did = ""
if "toId" in json_body: # Its a command
did = json_body["toId"]
if did != "":
bot = bot_get(did)
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",
}
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)
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)
raise HTTPInternalServerError
class IotPlugin(WebserverPlugin):
"""Iot plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route(
"*",
"/iot/devmanager.do",
_handle_devmanager_bot_command,
),
]

103
bumper/web/plugins/api/lg.py Normal file
View file

@ -0,0 +1,103 @@
"""Lg plugin module."""
import json
import logging
import random
import string
import xml.etree.ElementTree as ET
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
import bumper
from bumper.db import bot_get
from bumper.models import ERR_COMMON
from .. import WebserverPlugin
async def _handle_lg_log(request: Request)->Response:
# EcoVacs Home
randomid = "".join(random.sample(string.ascii_letters, 6))
try:
json_body = json.loads(await request.text())
did = json_body["did"]
botdetails = bot_get(did)
if botdetails:
if not "cmdName" in json_body:
if "td" in json_body:
json_body["cmdName"] = json_body["td"]
if not "toId" in json_body:
json_body["toId"] = did
if not "toType" in json_body:
json_body["toType"] = botdetails["class"]
if not "toRes" in json_body:
json_body["toRes"] = botdetails["resource"]
if not "payloadType" in json_body:
json_body["payloadType"] = "x"
if not "payload" in json_body:
# json_body["payload"] = ""
if json_body["td"] == "GetCleanLogs":
json_body["td"] = "q"
json_body["payload"] = '<ctl count="30"/>'
if did != "":
bot = bot_get(did)
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}")
logs = []
logsroot = ET.fromstring(retcmd["resp"])
if logsroot.attrib["ret"] == "ok":
for l in logsroot:
cleanlog = {
"ts": l.attrib["s"],
"area": l.attrib["a"],
"last": l.attrib["l"],
"cleanType": l.attrib["t"],
# imageUrl allows for providing images of cleanings, something to look into later
# "imageUrl": "https://localhost:8007",
}
logs.append(cleanlog)
body = {
"ret": "ok",
"logs": logs,
}
else:
body = {"ret": "ok", "logs": []}
logging.debug(f"lg logs return: {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}")
body = {"id": randomid, "errno": ERR_COMMON, "ret": "fail"}
return web.json_response(body)
class LgPlugin(WebserverPlugin):
"""Lg plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route("*", "/lg/log.do", _handle_lg_log),
]

94
bumper/web/plugins/api/neng.py Normal file
View file

@ -0,0 +1,94 @@
"""Neng plugin module."""
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from .. import WebserverPlugin
async def _handle_neng_has_unread_message(_: Request) -> Response:
# EcoVacs Home
body = {"code": 0, "data": {"hasUnRead": True}}
return web.json_response(body)
async def handle_neng_get_share_msgs(_: Request) -> Response:
# EcoVacs Home
body = {"code": 0, "data": {"hasNext": False, "msgs": []}}
# share msg response
# {
# "code": 0,
# "data": {
# "hasNext": False,
# "msgs": [
# {
# "action": "shareDevice",
# "deviceName": "DEEBOT 900 Series",
# "did": "DID",
# "icon": "https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839",
# "id": "0154d03a-294e-4b99-a6df-fc2dbf4146d5",
# "isRead": False,
# "message": "User user@gmail.com sent you the sharing invitation of DEEBOT 900 Series.",
# "mid": "ls1ok3",
# "resource": "grU0",
# "shareStatus": "sharing",
# "ts": 1578206187412
# }
# ]
# }
# }
return web.json_response(body)
async def handle_neng_get_list(_: Request) -> Response:
# EcoVacs Home
body = {"code": 0, "data": {"hasNext": False, "msgs": []}}
# Sample Message
# {
# "id": "5da0ac9d636aec5107627ac4",
# "ts": 1570811036877,
# "did": "bot did",
# "cid": "ls1ok3",
# "name": "DEEBOT 900 Series",
# "icon": "https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839",
# "eventTypeId": "5aab824bb62ce30001f9a702",
# "title": "DEEBOT is off the floor.",
# "body": "DEEBOT is off the floor. Please put it back.",
# "read": false,
# "UILogicId": "D_900",
# "type": "web",
# "url": "https://portal-ww.ecouser.net/api/pim/eventdetail.html?id=5ba21e44aed83800015b9ca8" # Off the floor instructions
# }
return web.json_response(body)
class NengPlugin(WebserverPlugin):
"""Neng plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route(
"*",
"/neng/message/hasUnreadMsg",
_handle_neng_has_unread_message,
),
web.route(
"*",
"/neng/message/getShareMsgs",
handle_neng_get_share_msgs,
),
web.route(
"*",
"/neng/message/getlist",
handle_neng_get_list,
),
]

25
bumper/plugins/bumper_confserver_portal_pim.py → bumper/web/plugins/api/pim.py
View file

@ -1,55 +1,54 @@
"""Pim plugin module."""
import json
import logging
import os
from typing import Iterable
from aiohttp import web
from aiohttp.web_routedef import AbstractRouteDef
from bumper import bumper_dir, plugins
from bumper import bumper_dir
from bumper.models import RETURN_API_SUCCESS, EcoVacsHomeProducts
from .. import WebserverPlugin
class portal_api_pim(plugins.ConfServerApp):
def __init__(self):
self.name = "portal_api_pim"
self.plugin_type = "sub_api"
self.sub_api = "portal_api"
self.routes = [
class PimPlugin(WebserverPlugin):
"""Pim plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route(
"*",
"/pim/product/getProductIotMap",
self.handle_getProductIotMap,
name="portal_api_pim_getProductIotMap",
),
web.route(
"*",
"/pim/file/get/{id}",
self.handle_pimFile,
name="portal_api_pim_file",
),
web.route(
"*",
"/pim/product/getConfignetAll",
self.handle_getConfignetAll,
name="portal_api_pim_getConfignetAll",
),
web.route(
"*",
"/pim/product/getConfigGroups",
self.handle_getConfigGroups,
name="portal_api_pim_getConfigGroups",
),
web.route(
"*",
"/pim/dictionary/getErrDetail",
self.handle_getErrDetail,
name="portal_api_pim_getErrDetail",
),
web.route(
"*",
"/pim/product/software/config/batch",
self.handle_product_config_batch,
name="portal_api_pim_product_config_batch",
),
]

30
bumper/web/plugins/api/rapp.py Normal file
View file

@ -0,0 +1,30 @@
"""Rapp plugin module."""
from typing import Iterable
from aiohttp import web
from aiohttp.web_request import Request
from aiohttp.web_response import Response
from aiohttp.web_routedef import AbstractRouteDef
from .. import WebserverPlugin
async def _handle_map_get(_: Request) -> Response:
body = {
"code": 0,
"data": {"data": {"name": "My Home"}, "tag": None},
"message": "success",
}
return web.json_response(body)
class RappPlugin(WebserverPlugin):
"""Rapp plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route("*", "/rapp/sds/user/data/map/get", _handle_map_get),
]

24
bumper/plugins/bumper_confserver_portal_users.py → bumper/web/plugins/api/users.py
View file

@ -1,28 +1,32 @@
"""Users plugin module."""
import json
import logging
from typing import Iterable
from aiohttp import web
from aiohttp.web_routedef import AbstractRouteDef
from bumper import bumper_announce_ip, plugins
from bumper import bumper_announce_ip
from bumper.db import bot_remove, bot_set_nick, check_authcode, db_get, loginByItToken
from .. import WebserverPlugin
class portal_api_users(plugins.ConfServerApp):
def __init__(self):
self.name = "portal_api_users"
self.plugin_type = "sub_api"
self.sub_api = "portal_api"
self.routes = [
class UsersPlugin(WebserverPlugin):
"""Users plugin."""
@property
def routes(self) -> Iterable[AbstractRouteDef]:
"""Plugin routes."""
return [
web.route(
"*",
"/users/user.do",
self.handle_usersapi,
name="portal_api_users_user",
self._handle_usersapi,
),
]
async def handle_usersapi(self, request):
async def _handle_usersapi(self, request):
if not request.method == "GET": # Skip GET for now
try: