79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
from flask import Flask, Response, jsonify, request
|
|
|
|
|
|
CONFIG_FILE = Path(__file__).with_name("config.yaml")
|
|
with CONFIG_FILE.open(encoding="utf-8") as stream:
|
|
CONFIG = yaml.safe_load(stream) or {}
|
|
|
|
LOG_FILE = Path(CONFIG["HOME_CONTROL_LOG"])
|
|
if not LOG_FILE.is_absolute():
|
|
LOG_FILE = CONFIG_FILE.parent / LOG_FILE
|
|
IP_ADDRESS = str(CONFIG.get("ip_address", "127.0.0.1"))
|
|
PORT = int(CONFIG.get("port", 5000))
|
|
URL_PREFIX = "/" + str(CONFIG.get("url_prefix", "")).strip("/")
|
|
if URL_PREFIX == "/":
|
|
URL_PREFIX = ""
|
|
|
|
app = Flask(__name__)
|
|
|
|
|
|
def _tail(path: Path, line_count: int = 50) -> str:
|
|
try:
|
|
with path.open("rb") as logfile:
|
|
logfile.seek(0, os.SEEK_END)
|
|
end = logfile.tell()
|
|
data = b""
|
|
|
|
while end > 0 and data.count(b"\n") <= line_count:
|
|
size = min(4096, end)
|
|
end -= size
|
|
logfile.seek(end)
|
|
data = logfile.read(size) + data
|
|
|
|
output = []
|
|
for raw_line in data.splitlines()[-line_count:]:
|
|
line = raw_line.decode("utf-8", "replace")
|
|
try:
|
|
event = json.loads(line).get("event", {})
|
|
output.append(str(event["text"]))
|
|
except (json.JSONDecodeError, AttributeError, KeyError, TypeError):
|
|
output.append(line)
|
|
return "\n".join(output)
|
|
except FileNotFoundError:
|
|
return ""
|
|
|
|
|
|
@app.route(f"{URL_PREFIX}/event", methods=["GET", "POST"])
|
|
def add_event():
|
|
data = request.get_json(silent=True)
|
|
if data is None:
|
|
data = request.values.to_dict(flat=False)
|
|
if not data:
|
|
data = request.get_data(as_text=True)
|
|
|
|
entry = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"remote_addr": request.remote_addr,
|
|
"event": data,
|
|
}
|
|
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
with LOG_FILE.open("a", encoding="utf-8") as logfile:
|
|
logfile.write(json.dumps(entry, ensure_ascii=False, separators=(",", ":")) + "\n")
|
|
|
|
return jsonify(ok=True), 201
|
|
|
|
|
|
@app.get(URL_PREFIX or "/", strict_slashes=False)
|
|
def show_log():
|
|
return Response(_tail(LOG_FILE), mimetype="text/plain")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host=IP_ADDRESS, port=PORT)
|