76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
import os
|
|
import time
|
|
from collections import defaultdict, deque
|
|
from threading import Lock
|
|
|
|
from flask import Flask, jsonify, render_template, request
|
|
from scraper import ScrapeError, get_rating_distribution, validate_maps_url
|
|
|
|
app = Flask(__name__)
|
|
app.config["MAX_CONTENT_LENGTH"] = 4096
|
|
_scrape_lock = Lock() # jeden Chromium naraz, żeby nie zamęczyć RPi
|
|
_rate_lock = Lock()
|
|
_requests = defaultdict(deque)
|
|
|
|
|
|
def _rate_limited(ip, limit=6, window=60):
|
|
now = time.monotonic()
|
|
with _rate_lock:
|
|
attempts = _requests[ip]
|
|
while attempts and attempts[0] < now - window:
|
|
attempts.popleft()
|
|
if len(attempts) >= limit:
|
|
return True
|
|
attempts.append(now)
|
|
return False
|
|
|
|
|
|
@app.after_request
|
|
def headers(response):
|
|
response.headers["Content-Security-Policy"] = (
|
|
"default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; "
|
|
"object-src 'none'; base-uri 'none'; frame-ancestors 'none'; form-action 'self'"
|
|
)
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["Referrer-Policy"] = "no-referrer"
|
|
response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
|
|
return response
|
|
|
|
|
|
@app.errorhandler(413)
|
|
def too_large(_error):
|
|
return jsonify(error="Żądanie jest zbyt duże."), 413
|
|
|
|
|
|
@app.get("/")
|
|
def index():
|
|
return render_template("index.html")
|
|
|
|
|
|
@app.post("/api/reviews")
|
|
def reviews():
|
|
if _rate_limited(request.remote_addr or "unknown"):
|
|
return jsonify(error="Za dużo prób. Poczekaj minutę."), 429
|
|
data = request.get_json(silent=True)
|
|
if not isinstance(data, dict) or not isinstance(data.get("url"), str):
|
|
return jsonify(error="Podaj link do miejsca w Google Maps."), 400
|
|
try:
|
|
url = validate_maps_url(data["url"])
|
|
except ValueError as exc:
|
|
return jsonify(error=str(exc)), 400
|
|
if not _scrape_lock.acquire(blocking=False):
|
|
return jsonify(error="Trwa już inne sprawdzanie. Spróbuj za chwilę."), 429
|
|
try:
|
|
return jsonify(get_rating_distribution(url))
|
|
except ScrapeError as exc:
|
|
return jsonify(error=str(exc)), 422
|
|
except Exception:
|
|
app.logger.exception("Unexpected scraper failure")
|
|
return jsonify(error="Google Maps nie odpowiedziało poprawnie. Spróbuj później."), 502
|
|
finally:
|
|
_scrape_lock.release()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host=os.environ.get("HOST", "127.0.0.1"), port=int(os.environ.get("PORT", 8080)), debug=False)
|