75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
from sanic import Sanic
|
|
from sanic.response import text, json
|
|
from motor.motor_asyncio import AsyncIOMotorClient
|
|
import pymongo
|
|
import os
|
|
|
|
app = Sanic(__name__)
|
|
|
|
DOORBOY_SECRET = os.environ["DOORBOY_SECRET"]
|
|
MONGO_URI = os.getenv("MONGO_URI",
|
|
"mongodb://127.0.0.1:27017/default?replicaSet=rs0")
|
|
|
|
assert len(DOORBOY_SECRET) >= 10
|
|
|
|
|
|
@app.listener("before_server_start")
|
|
async def setup_db(app, loop):
|
|
# TODO: find cleaner way to do this, for more see
|
|
# https://github.com/sanic-org/sanic/issues/919
|
|
app.ctx.db = AsyncIOMotorClient(MONGO_URI).get_default_database()
|
|
|
|
|
|
@app.route("/allowed")
|
|
async def view_doorboy_uids(request):
|
|
if request.headers.get("KEY") != DOORBOY_SECRET:
|
|
return text("how about no")
|
|
allowed_names = []
|
|
async for obj in app.ctx.db.member.find({"enabled": True}):
|
|
allowed_names.append(obj["_id"])
|
|
allowed_uids = []
|
|
flt = {
|
|
"token.uid_hash": {"$exists": True},
|
|
"inventory.owner": {"$exists": True},
|
|
"token.enabled": {"$exists": True}
|
|
}
|
|
prj = {
|
|
"inventory.owner": True,
|
|
"token.uid_hash": True
|
|
}
|
|
async for obj in app.ctx.db.inventory.find(flt, prj):
|
|
if obj["inventory"].pop("owner").get("foreign_id") in allowed_names:
|
|
del obj["_id"]
|
|
del obj["inventory"]
|
|
allowed_uids.append(obj)
|
|
return json({"allowed_uids": allowed_uids})
|
|
|
|
|
|
@app.route("/longpoll", stream=True)
|
|
async def view_longpoll(request):
|
|
response = await request.respond(content_type="text/event-stream")
|
|
if request.headers.get("KEY") != DOORBOY_SECRET:
|
|
return text("Invalid token")
|
|
|
|
await response.send("data: response-generator-started\n\n")
|
|
pipeline = [
|
|
{
|
|
"$match": {
|
|
"operationType": "insert",
|
|
}
|
|
}
|
|
]
|
|
try:
|
|
async with app.ctx.db.eventlog.watch(pipeline) as stream:
|
|
await response.send("data: watch-stream-opened\n\n")
|
|
async for event in stream:
|
|
if event["fullDocument"].get("type") == "open-door":
|
|
await response.send("data: %s\n\n" %
|
|
event["fullDocument"]["door"])
|
|
except pymongo.errors.PyMongoError:
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(debug=False, host="0.0.0.0", port=5000)
|