"""Server for the Elegance Barbershop site: the page, its bookings, its content and its visitor statistics. Python standard library only. python serve.py [port] # default 4173 python serve.py --set-password NEW # set the admin password and exit python serve.py --backup DIR [KEEP] # safe copy of the database into DIR, # keeping the newest KEEP copies (14) Environment ELEGANCE_HOST=127.0.0.1 address to listen on (default: this machine only; Apache or nginx in front passes visitors through) TRUST_PROXY=1 read the visitor's IP from the proxy's header ELEGANCE_DATA=/path keep the data folder somewhere else What it does / the page (never cached, so every reload is what is on disk) /admin.html the admin panel (sign in with the admin password) /api/content the editable page content (services, prices, contact...) /api/slots free times for a service on a day /api/bookings a client books a chair /api/track visit statistics: IP, network, place (looked up on ipwho.is, cached 30 days), device, source, time on page, clicks /api/admin/* bookings, content and statistics, behind the sign-in Everything is kept in ./data, which the server never hands out as a file: data/elegance.db bookings and visit events (SQLite) data/content.json the page content, edited from the admin panel data/admin.json the admin password, salted and hashed """ import datetime as dt import hashlib import hmac import http.server import json import os import re import secrets import socketserver import sqlite3 import sys import threading import time from urllib.parse import parse_qs, unquote, urlparse ROOT = os.path.dirname(os.path.abspath(__file__)) DATA = os.environ.get("ELEGANCE_DATA") or os.path.join(ROOT, "data") DB_PATH = os.path.join(DATA, "elegance.db") CONTENT_PATH = os.path.join(DATA, "content.json") DEFAULT_CONTENT_PATH = os.path.join(DATA, "content.default.json") ADMIN_PATH = os.path.join(DATA, "admin.json") UPLOADS_DIR = os.path.join(DATA, "uploads") SESSION_HOURS = 12 BARBER_SESSION_HOURS = 24 * 14 MAX_BODY = 256 * 1024 MAX_UPLOAD = 60 * 1024 * 1024 os.makedirs(DATA, exist_ok=True) os.makedirs(UPLOADS_DIR, exist_ok=True) _lock = threading.RLock() # re-entrant: a visit write also queues its place lookup # ------------------------------------------------------------------ storage def db(): con = sqlite3.connect(DB_PATH, timeout=10) con.row_factory = sqlite3.Row return con def init_db(): with db() as con: con.executescript( """ CREATE TABLE IF NOT EXISTS bookings ( id INTEGER PRIMARY KEY AUTOINCREMENT, ref TEXT UNIQUE NOT NULL, created TEXT NOT NULL, service TEXT NOT NULL, minutes INTEGER NOT NULL, price INTEGER NOT NULL, date TEXT NOT NULL, time TEXT NOT NULL, name TEXT NOT NULL, phone TEXT NOT NULL, email TEXT, notes TEXT, status TEXT NOT NULL DEFAULT 'pending' ); CREATE INDEX IF NOT EXISTS bookings_date ON bookings(date); CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, day TEXT NOT NULL, sid TEXT NOT NULL, vid TEXT NOT NULL, type TEXT NOT NULL, name TEXT, value REAL, ref TEXT, device TEXT, lang TEXT ); CREATE INDEX IF NOT EXISTS events_day ON events(day); CREATE TABLE IF NOT EXISTS visits ( sid TEXT PRIMARY KEY, vid TEXT NOT NULL, ts TEXT NOT NULL, day TEXT NOT NULL, hour INTEGER NOT NULL, ip TEXT, isp TEXT, country TEXT, cc TEXT, region TEXT, city TEXT, tz TEXT, ref TEXT, ref_url TEXT, device TEXT, os TEXT, browser TEXT, app TEXT, lang TEXT, screen TEXT, client_tz TEXT, duration INTEGER NOT NULL DEFAULT 0, clicks INTEGER NOT NULL DEFAULT 0, actions TEXT NOT NULL DEFAULT '', sections INTEGER NOT NULL DEFAULT 0, booked INTEGER NOT NULL DEFAULT 0, is_return INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS visits_day ON visits(day); CREATE INDEX IF NOT EXISTS visits_ip ON visits(ip); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL, updated TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS content_versions ( id INTEGER PRIMARY KEY AUTOINCREMENT, saved TEXT NOT NULL, value TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS sessions ( token_hash TEXT PRIMARY KEY, expires REAL NOT NULL ); CREATE TABLE IF NOT EXISTS ip_geo ( ip TEXT PRIMARY KEY, ts REAL NOT NULL, country TEXT, cc TEXT, region TEXT, city TEXT, tz TEXT, isp TEXT ); CREATE TABLE IF NOT EXISTS barbers ( id INTEGER PRIMARY KEY AUTOINCREMENT, created TEXT NOT NULL, name TEXT NOT NULL, username TEXT UNIQUE NOT NULL, pw_salt TEXT NOT NULL, pw_hash TEXT NOT NULL, cin TEXT NOT NULL DEFAULT '', cin_photo TEXT NOT NULL DEFAULT '', diploma_photo TEXT NOT NULL DEFAULT '', avatar_photo TEXT NOT NULL DEFAULT '', start_date TEXT NOT NULL DEFAULT '', end_date TEXT NOT NULL DEFAULT '', level INTEGER NOT NULL DEFAULT 1, phone TEXT NOT NULL DEFAULT '', email TEXT NOT NULL DEFAULT '', address TEXT NOT NULL DEFAULT '', emergency_contact TEXT NOT NULL DEFAULT '', commission_pct INTEGER NOT NULL DEFAULT 0, base_salary INTEGER NOT NULL DEFAULT 0, specialty TEXT NOT NULL DEFAULT '', character_notes TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'active' ); CREATE TABLE IF NOT EXISTS barber_sessions ( token_hash TEXT PRIMARY KEY, barber_id INTEGER NOT NULL, expires REAL NOT NULL ); CREATE TABLE IF NOT EXISTS barber_attendance ( id INTEGER PRIMARY KEY AUTOINCREMENT, barber_id INTEGER NOT NULL, day TEXT NOT NULL, check_in TEXT NOT NULL DEFAULT '', check_out TEXT NOT NULL DEFAULT '', minutes_worked INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'present', notes TEXT NOT NULL DEFAULT '', created TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS barber_attendance_day ON barber_attendance(day); CREATE INDEX IF NOT EXISTS barber_attendance_barber ON barber_attendance(barber_id, day); CREATE TABLE IF NOT EXISTS barber_works ( id INTEGER PRIMARY KEY AUTOINCREMENT, barber_id INTEGER NOT NULL, day TEXT NOT NULL, time TEXT NOT NULL, client_name TEXT NOT NULL DEFAULT '', services_json TEXT NOT NULL, services_text TEXT NOT NULL, total_minutes INTEGER NOT NULL DEFAULT 0, standard_price INTEGER NOT NULL DEFAULT 0, final_price INTEGER NOT NULL DEFAULT 0, client_notes TEXT NOT NULL DEFAULT '', created TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS barber_works_day ON barber_works(day); CREATE INDEX IF NOT EXISTS barber_works_barber ON barber_works(barber_id, day); CREATE TABLE IF NOT EXISTS product_requests ( id INTEGER PRIMARY KEY AUTOINCREMENT, barber_id INTEGER NOT NULL, product_name TEXT NOT NULL, quantity INTEGER NOT NULL DEFAULT 1, urgency TEXT NOT NULL DEFAULT 'low', notes TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'pending', created TEXT NOT NULL, updated TEXT NOT NULL DEFAULT '' ); CREATE TABLE IF NOT EXISTS leave_requests ( id INTEGER PRIMARY KEY AUTOINCREMENT, barber_id INTEGER NOT NULL, leave_type TEXT NOT NULL, date_from TEXT NOT NULL, date_to TEXT NOT NULL, half_period TEXT NOT NULL DEFAULT '', time_from TEXT NOT NULL DEFAULT '', time_to TEXT NOT NULL DEFAULT '', reason TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'pending', admin_note TEXT NOT NULL DEFAULT '', created TEXT NOT NULL, updated TEXT NOT NULL DEFAULT '' ); CREATE TABLE IF NOT EXISTS training_videos ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, category TEXT NOT NULL DEFAULT 'Technique', target_level INTEGER NOT NULL DEFAULT 0, video_url TEXT NOT NULL, thumbnail_url TEXT NOT NULL DEFAULT '', description TEXT NOT NULL DEFAULT '', pinned INTEGER NOT NULL DEFAULT 0, created TEXT NOT NULL ); """ ) def get_setting(key): with db() as con: r = con.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone() return json.loads(r["value"]) if r else None def put_setting(key, value, con=None): sql = "INSERT OR REPLACE INTO settings (key, value, updated) VALUES (?,?,?)" args = (key, json.dumps(value, ensure_ascii=False), now_local().isoformat(timespec="seconds")) if con is not None: con.execute(sql, args) else: with db() as c: c.execute(sql, args) def migrate_files(): """Bring the content and the password that used to live in JSON files into the database, once. The old files are kept, renamed, as a backup.""" with db() as con: has = {r["key"] for r in con.execute("SELECT key FROM settings").fetchall()} if "content" not in has: src = CONTENT_PATH if os.path.exists(CONTENT_PATH) else DEFAULT_CONTENT_PATH with open(src, encoding="utf-8") as f: save_content(json.load(f)) if os.path.exists(CONTENT_PATH): os.replace(CONTENT_PATH, CONTENT_PATH.replace(".json", ".imported.json")) print(" content moved into the database", flush=True) if "admin" not in has and os.path.exists(ADMIN_PATH): with open(ADMIN_PATH, encoding="utf-8") as f: put_setting("admin", json.load(f)) os.replace(ADMIN_PATH, ADMIN_PATH.replace(".json", ".imported.json")) print(" admin password moved into the database", flush=True) def load_content(): c = get_setting("content") if c is None: # brand-new database: start from the built-in default with open(DEFAULT_CONTENT_PATH, encoding="utf-8") as f: c = json.load(f) return c KEEP_VERSIONS = 30 def save_content(content): """Save the page content, keeping the last versions so a change can be undone.""" with _lock, db() as con: put_setting("content", content, con) con.execute("INSERT INTO content_versions (saved, value) VALUES (?, ?)", (now_local().isoformat(timespec="seconds"), json.dumps(content, ensure_ascii=False))) con.execute("DELETE FROM content_versions WHERE id NOT IN (SELECT id FROM content_versions ORDER BY id DESC LIMIT ?)", (KEEP_VERSIONS,)) # ------------------------------------------------------------- admin auth def hash_password(password, salt=None): salt = salt or secrets.token_hex(16) digest = hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), 200_000).hex() return {"salt": salt, "hash": digest} def set_password(password): put_setting("admin", hash_password(password)) with db() as con: # a new password signs every session out con.execute("DELETE FROM sessions") def check_password(password): stored = get_setting("admin") if not stored: return False return hmac.compare_digest(hash_password(password, stored["salt"])["hash"], stored["hash"]) def ensure_admin(): if not get_setting("admin"): pw = secrets.token_urlsafe(9) set_password(pw) print(f"\n admin password (first run): {pw}\n change it in the admin panel, or: python serve.py --set-password NEW\n", flush=True) # sign-in sessions live in the database (only a hash of each token), so a # server restart does not sign the admin out def _th(token): return hashlib.sha256(token.encode()).hexdigest() def session_new(hours): tok = secrets.token_urlsafe(32) with db() as con: con.execute("DELETE FROM sessions WHERE expires < ?", (time.time(),)) con.execute("INSERT INTO sessions (token_hash, expires) VALUES (?, ?)", (_th(tok), time.time() + hours * 3600)) return tok def session_ok(token): if not token: return False with db() as con: r = con.execute("SELECT expires FROM sessions WHERE token_hash=?", (_th(token),)).fetchone() return bool(r) and r["expires"] > time.time() def session_end(token): if token: with db() as con: con.execute("DELETE FROM sessions WHERE token_hash=?", (_th(token),)) LOGIN_FAILS = {} # ip -> [timestamps] RATE = {} # (bucket, ip) -> [timestamps] def rate_ok(bucket, ip, limit, window): now = time.time() with _lock: hits = [t for t in RATE.get((bucket, ip), []) if now - t < window] if len(hits) >= limit: RATE[(bucket, ip)] = hits return False hits.append(now) RATE[(bucket, ip)] = hits return True # ---------------------------------------------------------- visitor details def parse_ua(ua): """(device, os, browser, app) from a User-Agent string.""" u = ua or "" low = u.lower() if "ipad" in low or ("tablet" in low and "android" in low): device = "Tablet" elif any(k in low for k in ("iphone", "ipod", "android", "mobile")): device = "Mobile" elif u: device = "Desktop" else: device = "Unknown" if "iphone" in low or "ipad" in low or "ipod" in low: os_ = "iOS" elif "android" in low: os_ = "Android" elif "windows" in low: os_ = "Windows" elif "mac os x" in low or "macintosh" in low: os_ = "macOS" elif "cros" in low: os_ = "ChromeOS" elif "linux" in low: os_ = "Linux" else: os_ = "Unknown" app = "" for key, name in (("instagram", "Instagram App"), ("fban", "Facebook App"), ("fbav", "Facebook App"), ("musical_ly", "TikTok App"), ("bytedancewebview", "TikTok App"), ("tiktok", "TikTok App"), ("snapchat", "Snapchat App"), ("whatsapp", "WhatsApp"), ("telegram", "Telegram"), ("linkedinapp", "LinkedIn App"), ("pinterest", "Pinterest App"), ("twitter", "X App")): if key in low: app = name break if "edg/" in low or "edga/" in low or "edgios" in low: browser = "Edge" elif "opr/" in low or "opera" in low: browser = "Opera" elif "samsungbrowser" in low: browser = "Samsung Internet" elif "firefox" in low or "fxios" in low: browser = "Firefox" elif "crios" in low or "chrome" in low: browser = "Chrome" elif "safari" in low: browser = "Safari" elif any(b in low for b in ("bot", "spider", "crawl", "headless")): browser = "Bot" else: browser = "Unknown" return device, os_, browser, app PRIVATE_RE = re.compile(r"^(127\.|10\.|192\.168\.|172\.(1[6-9]|2\d|3[01])\.|::1$|fc|fd|fe80|localhost)") GEO_TTL = 30 * 24 * 3600 _geo_queue = [] _geo_event = threading.Event() def geo_cached(ip): with db() as con: r = con.execute("SELECT * FROM ip_geo WHERE ip=?", (ip,)).fetchone() return dict(r) if r and time.time() - r["ts"] < GEO_TTL else None def apply_geo(ip, g): with _lock, db() as con: con.execute( "UPDATE visits SET country=?, cc=?, region=?, city=?, tz=?, isp=? WHERE ip=? AND country IS NULL", (g.get("country"), g.get("cc"), g.get("region"), g.get("city"), g.get("tz"), g.get("isp"), ip), ) def lookup_geo(ip): """Country, city and network for an IP (ipwho.is, cached 30 days).""" if PRIVATE_RE.match(ip): return {"country": "Local network", "cc": "", "region": "", "city": "", "tz": "", "isp": "Local"} import urllib.request url = f"https://ipwho.is/{ip}?fields=success,country,country_code,region,city,timezone,connection" req = urllib.request.Request(url, headers={"User-Agent": "EleganceStats/1.0"}) with urllib.request.urlopen(req, timeout=6) as r: d = json.loads(r.read().decode()) if not d.get("success"): return {"country": "Unknown", "cc": "", "region": "", "city": "", "tz": "", "isp": ""} conn = d.get("connection") or {} return { "country": d.get("country") or "Unknown", "cc": d.get("country_code") or "", "region": d.get("region") or "", "city": d.get("city") or "", "tz": (d.get("timezone") or {}).get("id", ""), "isp": conn.get("isp") or conn.get("org") or "", } def geo_worker(): while True: _geo_event.wait() while True: with _lock: if not _geo_queue: _geo_event.clear() break ip = _geo_queue.pop(0) try: g = geo_cached(ip) if not g: g = lookup_geo(ip) with _lock, db() as con: con.execute( "INSERT OR REPLACE INTO ip_geo (ip, ts, country, cc, region, city, tz, isp) VALUES (?,?,?,?,?,?,?,?)", (ip, time.time(), g["country"], g["cc"], g["region"], g["city"], g["tz"], g["isp"]), ) time.sleep(0.4) # stay well inside the free lookup limits apply_geo(ip, g) except Exception as e: # offline or rate limited: try again on the next visit print(f" geo lookup failed for {ip}: {e}", flush=True) def queue_geo(ip): with _lock: if ip and ip not in _geo_queue: _geo_queue.append(ip) _geo_event.set() # ------------------------------------------------------------- services def all_services(content): """Every bookable service: name -> (minutes, price, group).""" out = {} vip = content.get("vip") or {} if vip.get("name"): out[vip["name"]] = (int(vip.get("min", 60)), int(vip.get("dh", 0)), "Signature") for p in content.get("packages", []): out[f"{p['tag']} โ€” {p['name']}"] = (int(p["min"]), int(p["dh"]), "Packages") for g in content.get("menu", []): for it in g.get("items", []): out[it["name"]] = (int(it["min"]), int(it["dh"]), g["title"]) return out def to_min(hhmm): h, m = hhmm.split(":") return int(h) * 60 + int(m) def fmt_min(m): return f"{m // 60:02d}:{m % 60:02d}" def now_local(): # Morocco keeps UTC+1 (outside Ramadan); good enough for opening hours return dt.datetime.utcnow() + dt.timedelta(hours=1) def free_slots(content, date_str, minutes): try: day = dt.date.fromisoformat(date_str) except ValueError: return [] bk = content.get("booking", {}) today = now_local().date() if day < today or day > today + dt.timedelta(days=int(bk.get("daysAhead", 30))): return [] if date_str in (bk.get("closedDates") or []): return [] hours = content.get("hours", []) h = next((x for x in hours if int(x["day"]) == day.weekday()), None) if not h or h.get("closed"): return [] open_m, close_m = to_min(h["open"]), to_min(h["close"]) step = max(10, int(bk.get("interval", 30))) chairs = max(1, int(bk.get("chairs", 1))) earliest = -1 if day == today: n = now_local() earliest = n.hour * 60 + n.minute + int(bk.get("leadMinutes", 60)) with db() as con: rows = con.execute( "SELECT time, minutes FROM bookings WHERE date=? AND status != 'cancelled'", (date_str,) ).fetchall() taken = [(to_min(r["time"]), to_min(r["time"]) + r["minutes"]) for r in rows] slots = [] t = open_m while t + minutes <= close_m: if t >= earliest: overlap = sum(1 for a, b in taken if a < t + minutes and t < b) if overlap < chairs: slots.append(fmt_min(t)) t += step return slots # --------------------------------------------------------------- stats def date_range(qs): """(from, to) ISO days from ?from=&to= or ?days=, in Tangier time.""" today = now_local().date() frm = (qs.get("from") or [""])[0] to = (qs.get("to") or [""])[0] if DATE_RE.match(frm) and DATE_RE.match(to): return (frm, to) if frm <= to else (to, frm) days = max(1, min(3650, int((qs.get("days") or ["30"])[0] or 30))) return (today - dt.timedelta(days=days - 1)).isoformat(), today.isoformat() def stats(frm, to): d0, d1 = dt.date.fromisoformat(frm), dt.date.fromisoformat(to) span = (d1 - d0).days + 1 W = "day BETWEEN ? AND ?" with db() as con: q = lambda sql, *a: con.execute(sql, a).fetchall() one = lambda sql, *a: con.execute(sql, a).fetchone()[0] visits = one(f"SELECT COUNT(*) FROM visits WHERE {W}", frm, to) uniques = one(f"SELECT COUNT(DISTINCT vid) FROM visits WHERE {W}", frm, to) returning = one(f"SELECT COUNT(*) FROM visits WHERE {W} AND is_return=1", frm, to) clicked = one(f"SELECT COUNT(*) FROM visits WHERE {W} AND clicks>0", frm, to) mobile = one(f"SELECT COUNT(*) FROM visits WHERE {W} AND device IN ('Mobile','Tablet')", frm, to) bounced = one(f"SELECT COUNT(*) FROM visits WHERE {W} AND duration<10 AND clicks=0", frm, to) avg_time = one(f"SELECT AVG(duration) FROM visits WHERE {W} AND duration>0", frm, to) or 0 pageviews = one(f"SELECT COUNT(*) FROM events WHERE {W} AND type='pageview'", frm, to) # per-day series (all days, or the last 120 of a long range) series = {r["day"]: r for r in q( f"SELECT day, COUNT(*) v, COUNT(DISTINCT vid) u FROM visits WHERE {W} GROUP BY day", frm, to)} bseries = {r["d"]: r["n"] for r in q( "SELECT substr(created,1,10) d, COUNT(*) n FROM bookings WHERE substr(created,1,10) BETWEEN ? AND ? GROUP BY d", frm, to)} first = max(d0, d1 - dt.timedelta(days=119)) if not series and not bseries: first = max(first, d1 - dt.timedelta(days=29)) daily = [] d = first while d <= d1: k = d.isoformat() r = series.get(k) daily.append({"day": k, "visitors": r["u"] if r else 0, "visits": r["v"] if r else 0, "bookings": bseries.get(k, 0)}) d += dt.timedelta(days=1) hours = [0] * 24 for r in q(f"SELECT hour, COUNT(*) n FROM visits WHERE {W} GROUP BY hour", frm, to): hours[int(r["hour"]) % 24] = r["n"] def counts(sql, *extra): return [{"name": r[0] if r[0] not in (None, "") else "Unknown", "count": r[1]} for r in q(sql, frm, to, *extra)] V = f"FROM visits WHERE {W}" out = { "from": frm, "to": to, "span": span, "visits": visits, "uniques": uniques, "pageviews": pageviews, "returning": returning, "avgTime": round(avg_time), "clickRate": round(100 * clicked / visits) if visits else 0, "mobileRate": round(100 * mobile / visits) if visits else 0, "bounceRate": round(100 * bounced / visits) if visits else 0, "returningRate": round(100 * returning / visits) if visits else 0, "daily": daily, "hours": hours, "referrers": counts(f"SELECT ref, COUNT(*) {V} GROUP BY ref ORDER BY 2 DESC LIMIT 10"), "countries": counts(f"SELECT country, COUNT(*) {V} GROUP BY country ORDER BY 2 DESC LIMIT 10"), "cities": counts(f"SELECT CASE WHEN city IS NULL OR city='' THEN NULL ELSE city || ' (' || country || ')' END, COUNT(*) {V} GROUP BY 1 ORDER BY 2 DESC LIMIT 10"), "isps": counts(f"SELECT isp, COUNT(*) {V} GROUP BY isp ORDER BY 2 DESC LIMIT 10"), "devices": counts(f"SELECT device || ' ยท ' || os, COUNT(*) {V} GROUP BY 1 ORDER BY 2 DESC LIMIT 8"), "browsers": counts(f"SELECT CASE WHEN app!='' THEN app ELSE browser END, COUNT(*) {V} GROUP BY 1 ORDER BY 2 DESC LIMIT 8"), "languages": counts(f"SELECT lang, COUNT(*) {V} GROUP BY lang ORDER BY 2 DESC LIMIT 8"), "sections": counts(f"SELECT name, COUNT(DISTINCT sid) FROM events WHERE {W} AND type='section' GROUP BY name ORDER BY 2 DESC"), "clicks": counts(f"SELECT name, COUNT(*) FROM events WHERE {W} AND type='click' GROUP BY name ORDER BY 2 DESC LIMIT 12"), } funnel_raw = {r[0]: r[1] for r in q( f"SELECT name, COUNT(DISTINCT sid) FROM events WHERE {W} AND type='booking' GROUP BY name", frm, to)} BW = "substr(created,1,10) BETWEEN ? AND ?" b = one(f"SELECT COUNT(*) FROM bookings WHERE {BW}", frm, to) out.update({ "funnel": [{"name": s, "count": funnel_raw.get(s, 0)} for s in ("open", "service", "date", "time", "details", "submitted")], "bookings": b, "byStatus": {r[0]: r[1] for r in q(f"SELECT status, COUNT(*) FROM bookings WHERE {BW} GROUP BY status", frm, to)}, "revenue": one(f"SELECT COALESCE(SUM(price),0) FROM bookings WHERE {BW} AND status IN ('confirmed','done')", frm, to), "topServices": counts(f"SELECT service, COUNT(*) FROM bookings WHERE {BW} AND status!='cancelled' GROUP BY service ORDER BY 2 DESC LIMIT 6"), "upcoming": [dict(r) for r in q( "SELECT * FROM bookings WHERE date>=? AND status IN ('pending','confirmed') ORDER BY date, time LIMIT 8", now_local().date().isoformat())], "conversion": round(100 * b / visits, 1) if visits else 0, }) return out def visit_list(frm, to, search, limit, offset): sql = "SELECT * FROM visits WHERE day BETWEEN ? AND ?" args = [frm, to] if search: like = f"%{search}%" sql += (" AND (ip LIKE ? OR country LIKE ? OR city LIKE ? OR isp LIKE ? OR ref LIKE ? OR ref_url LIKE ?" " OR browser LIKE ? OR app LIKE ? OR os LIKE ? OR device LIKE ? OR lang LIKE ? OR actions LIKE ?)") args += [like] * 12 with db() as con: total = con.execute(sql.replace("SELECT *", "SELECT COUNT(*)", 1), args).fetchone()[0] rows = [dict(r) for r in con.execute(sql + " ORDER BY ts DESC LIMIT ? OFFSET ?", (*args, limit, offset)).fetchall()] return {"total": total, "visits": rows} # ------------------------------------------------------------- validation PHONE_RE = re.compile(r"^[+0-9 ().-]{6,24}$") EMAIL_RE = re.compile(r"^[^@\s]{1,64}@[^@\s]{1,190}\.[^@\s]{2,24}$") TIME_RE = re.compile(r"^\d{2}:\d{2}$") DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") def clean(s, n): return re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", str(s or "")).strip()[:n] def new_ref(): alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" return "EL-" + "".join(secrets.choice(alphabet) for _ in range(5)) # ------------------------------------------------------- team & barbers MIME_TYPES = { ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".webp": "image/webp", ".gif": "image/gif", ".pdf": "application/pdf", ".mp4": "video/mp4", ".webm": "video/webm", ".mov": "video/quicktime", } def barber_session_new(barber_id, hours=BARBER_SESSION_HOURS): tok = secrets.token_urlsafe(32) with db() as con: con.execute("DELETE FROM barber_sessions WHERE expires < ?", (time.time(),)) con.execute( "INSERT INTO barber_sessions (token_hash, barber_id, expires) VALUES (?, ?, ?)", (_th(tok), int(barber_id), time.time() + hours * 3600), ) return tok def barber_session_end(token): if token: with db() as con: con.execute("DELETE FROM barber_sessions WHERE token_hash=?", (_th(token),)) def barber_by_token(token): if not token: return None with db() as con: r = con.execute( """SELECT b.* FROM barber_sessions s JOIN barbers b ON b.id = s.barber_id WHERE s.token_hash=? AND s.expires > ?""", (_th(token), time.time()), ).fetchone() if not r or r["status"] != "active": return None return dict(r) def barber_public(b, for_admin=False): if not b: return None d = dict(b) d.pop("pw_hash", None) d.pop("pw_salt", None) if not for_admin: d.pop("character_notes", None) d.pop("base_salary", None) return d def calc_shift_minutes(cin, cout): if not (TIME_RE.match(str(cin or "")) and TIME_RE.match(str(cout or ""))): return 0 diff = to_min(cout) - to_min(cin) if diff < 0: diff += 24 * 60 return max(1, diff) def services_catalog(content): """Structured list of categories and items for the barber multi-service picker.""" cats = [] vip = content.get("vip") or {} if vip.get("name"): cats.append({ "title": "Signature", "items": [{"name": vip["name"], "min": int(vip.get("min", 60)), "dh": int(vip.get("dh", 0))}], }) pkgs = content.get("packages") or [] if pkgs: cats.append({ "title": "Packages", "items": [{"name": f"{p['tag']} โ€” {p['name']}", "min": int(p["min"]), "dh": int(p["dh"])} for p in pkgs], }) for g in content.get("menu") or []: cats.append({ "title": g.get("title", "Services"), "items": [{"name": it["name"], "min": int(it["min"]), "dh": int(it["dh"])} for it in g.get("items", [])], }) return cats def team_summary(frm, to, filter_barber_id=None): now = now_local() today_str = now.date().isoformat() now_mins = now.hour * 60 + now.minute content = load_content() closed_dates = set((content.get("booking") or {}).get("closedDates") or []) closed_weekdays = {int(h["day"]) for h in (content.get("hours") or []) if h.get("closed")} with db() as con: barbers_raw = [dict(r) for r in con.execute( "SELECT * FROM barbers ORDER BY CASE status WHEN 'active' THEN 0 WHEN 'suspended' THEN 1 ELSE 2 END, name ASC" ).fetchall()] bmap = {b["id"]: b["name"] for b in barbers_raw} works_all = [dict(r) for r in con.execute( "SELECT * FROM barber_works WHERE day BETWEEN ? AND ? ORDER BY day DESC, time DESC, id DESC", (frm, to), ).fetchall()] att_all = [dict(r) for r in con.execute( "SELECT * FROM barber_attendance WHERE day BETWEEN ? AND ? ORDER BY day DESC, check_in DESC, id DESC", (frm, to), ).fetchall()] att_today = [dict(r) for r in con.execute( "SELECT * FROM barber_attendance WHERE day=? ORDER BY id DESC", (today_str,), ).fetchall()] leaves_all = [dict(r) for r in con.execute( "SELECT * FROM leave_requests ORDER BY CASE status WHEN 'pending' THEN 0 ELSE 1 END, date_from DESC, id DESC LIMIT 300" ).fetchall()] products_all = [dict(r) for r in con.execute( "SELECT * FROM product_requests ORDER BY CASE status WHEN 'pending' THEN 0 WHEN 'ordered' THEN 1 ELSE 2 END, id DESC LIMIT 300" ).fetchall()] videos_all = [dict(r) for r in con.execute( "SELECT * FROM training_videos ORDER BY pinned DESC, id DESC" ).fetchall()] # Attach barber_name to works, attendance, leaves, products for w in works_all: w["barber_name"] = bmap.get(w["barber_id"], "Unknown") try: w["services"] = json.loads(w.get("services_json") or "[]") except Exception: w["services"] = [] for a in att_all: a["barber_name"] = bmap.get(a["barber_id"], "Unknown") if a["day"] == today_str and a["check_in"] and not a["check_out"] and a["status"] in ("present", "late", "half-day"): a["live_minutes"] = calc_shift_minutes(a["check_in"], f"{now.hour:02d}:{now.minute:02d}") else: a["live_minutes"] = a["minutes_worked"] for lv in leaves_all: lv["barber_name"] = bmap.get(lv["barber_id"], "Unknown") for pr in products_all: pr["barber_name"] = bmap.get(pr["barber_id"], "Unknown") # Build per-barber statistics d0 = dt.date.fromisoformat(frm) d1 = dt.date.fromisoformat(to) today_date = now.date() barbers_out = [] for b in barbers_raw: bid = b["id"] bp = barber_public(b, for_admin=True) b_works = [w for w in works_all if w["barber_id"] == bid] b_att = [a for a in att_all if a["barber_id"] == bid] b_today = [a for a in att_today if a["barber_id"] == bid] # Today live status open_shift = next((a for a in b_today if a["check_in"] and not a["check_out"] and a["status"] in ("present", "late", "half-day")), None) last_shift = b_today[0] if b_today else None bp["working_now"] = bool(open_shift) bp["today_shift_id"] = open_shift["id"] if open_shift else (last_shift["id"] if last_shift else None) bp["today_check_in"] = open_shift["check_in"] if open_shift else (last_shift["check_in"] if last_shift else "") bp["today_check_out"] = "" if open_shift else (last_shift["check_out"] if last_shift else "") bp["today_status"] = open_shift["status"] if open_shift else (last_shift["status"] if last_shift else "") # Approved full-day leaves for this barber in range leave_days = set() for lv in leaves_all: if lv["barber_id"] == bid and lv["status"] == "approved" and lv["leave_type"] == "full_day": try: ld = max(d0, dt.date.fromisoformat(lv["date_from"])) ld_end = min(d1, dt.date.fromisoformat(lv["date_to"])) while ld <= ld_end: leave_days.add(ld.isoformat()) ld += dt.timedelta(days=1) except ValueError: pass present_days = set() explicit_absent_days = set() late_count = 0 mins_worked = 0 for a in b_att: st = a["status"] if st in ("present", "late", "half-day"): present_days.add(a["day"]) if st == "late": late_count += 1 mins_worked += int(a.get("live_minutes") or a.get("minutes_worked") or 0) elif st == "absent": explicit_absent_days.add(a["day"]) elif st == "leave": leave_days.add(a["day"]) # Also check unlogged absences on past open days since barber account creation/start track_start_str = max(b.get("start_date") or "2000-01-01", (b.get("created") or "")[:10] or "2000-01-01") track_end_str = b.get("end_date") or "9999-12-31" absent_dates = set(explicit_absent_days) if b["status"] == "active": cur = d0 while cur <= min(d1, today_date - dt.timedelta(days=1)): ds = cur.isoformat() if track_start_str <= ds <= track_end_str: if ds not in closed_dates and cur.weekday() not in closed_weekdays: if ds not in present_days and ds not in leave_days: absent_dates.add(ds) cur += dt.timedelta(days=1) std_rev = sum(int(w["standard_price"]) for w in b_works) fin_rev = sum(int(w["final_price"]) for w in b_works) svc_cnt = sum(len(w.get("services") or []) for w in b_works) comm_pct = int(b.get("commission_pct") or 0) bp["stats"] = { "clients": len(b_works), "services_count": svc_cnt, "standard_revenue": std_rev, "revenue": fin_rev, "discounts": max(0, std_rev - fin_rev), "commission_est": round(fin_rev * comm_pct / 100) if comm_pct > 0 else 0, "minutes_worked": mins_worked, "days_present": len(present_days), "days_absent": len(absent_dates), "days_leave": len(leave_days), "late_count": late_count, "absent_dates": sorted(absent_dates, reverse=True), } barbers_out.append(bp) if filter_barber_id: fbid = int(filter_barber_id) works_filtered = [w for w in works_all if w["barber_id"] == fbid] att_filtered = [a for a in att_all if a["barber_id"] == fbid] b_for_totals = [b for b in barbers_out if b["id"] == fbid] else: works_filtered = works_all att_filtered = att_all b_for_totals = barbers_out totals = { "from": frm, "to": to, "revenue": sum(b["stats"]["revenue"] for b in b_for_totals), "standard_revenue": sum(b["stats"]["standard_revenue"] for b in b_for_totals), "discounts": sum(b["stats"]["discounts"] for b in b_for_totals), "clients": sum(b["stats"]["clients"] for b in b_for_totals), "services_count": sum(b["stats"]["services_count"] for b in b_for_totals), "minutes_worked": sum(b["stats"]["minutes_worked"] for b in b_for_totals), "days_present": sum(b["stats"]["days_present"] for b in b_for_totals), "days_absent": sum(b["stats"]["days_absent"] for b in b_for_totals), "days_leave": sum(b["stats"]["days_leave"] for b in b_for_totals), "active_barbers": sum(1 for b in barbers_out if b["status"] == "active"), "working_now": sum(1 for b in barbers_out if b["working_now"]), "total_barbers": len(barbers_out), "pending_products": sum(1 for p in products_all if p["status"] == "pending"), "pending_leaves": sum(1 for l in leaves_all if l["status"] == "pending"), } return { "totals": totals, "barbers": barbers_out, "works": works_filtered[:500], "attendance": att_filtered[:500], "product_requests": products_all, "leave_requests": leaves_all, "videos": videos_all, } # --------------------------------------------------------------- handler class Handler(http.server.SimpleHTTPRequestHandler): server_version = "Elegance" sys_version = "" def __init__(self, *a, **kw): super().__init__(*a, directory=ROOT, **kw) def log_message(self, fmt, *args): if "/api/track" in (self.path or ""): return # keep the console readable super().log_message(fmt, *args) # ---- plumbing def end_headers(self): self.send_header("Cache-Control", "no-store, no-cache, must-revalidate, max-age=0") self.send_header("Pragma", "no-cache") self.send_header("Expires", "0") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("Referrer-Policy", "strict-origin-when-cross-origin") super().end_headers() def ip(self): # behind Apache or nginx set TRUST_PROXY=1: the visitor's address is the # LAST entry of X-Forwarded-For, the one our own proxy appended (earlier # entries come from the visitor and can be forged). Behind Cloudflare # set TRUST_PROXY=cloudflare to use its CF-Connecting-IP header. mode = os.environ.get("TRUST_PROXY") if mode == "cloudflare" and self.headers.get("CF-Connecting-IP"): return self.headers["CF-Connecting-IP"].strip()[:64] if mode in ("1", "cloudflare"): fwd = (self.headers.get("X-Forwarded-For") or "").split(",")[-1].strip() if fwd: return fwd[:64] return self.client_address[0] def secure(self): # the admin cookie is marked Secure once the site is served over HTTPS return self.headers.get("X-Forwarded-Proto", "").lower() == "https" def send_json(self, obj, status=200, headers=None): body = json.dumps(obj, ensure_ascii=False).encode() self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) for k, v in (headers or {}).items(): self.send_header(k, v) self.end_headers() self.wfile.write(body) def read_json(self): n = int(self.headers.get("Content-Length") or 0) if n <= 0 or n > MAX_BODY: return None try: return json.loads(self.rfile.read(n).decode("utf-8")) except (ValueError, UnicodeDecodeError): return None def token(self): for part in (self.headers.get("Cookie") or "").split(";"): k, _, v = part.strip().partition("=") if k == "el_admin": return v return None def barber_token(self): for part in (self.headers.get("Cookie") or "").split(";"): k, _, v = part.strip().partition("=") if k == "el_barber": return v return None def is_admin(self): return session_ok(self.token()) def current_barber(self): return barber_by_token(self.barber_token()) def same_origin(self): # admin writes must come from our own page: a cross-site form cannot # set this header, and the cookie is SameSite=Strict as well return self.headers.get("X-Requested-With") == "elegance" # ---- routing def do_GET(self): for header in ("If-Modified-Since", "If-None-Match"): if header in self.headers: del self.headers[header] url = urlparse(self.path) p = url.path if p.startswith("/api/"): return self.api_get(p, parse_qs(url.query)) # never serve the data folder, the server's own source or dotfiles low = p.lower() if low.startswith("/data") or low.endswith(".py") or "/." in low: return self.send_error(404) try: super().do_GET() except (ConnectionAbortedError, ConnectionResetError, BrokenPipeError): pass # the browser cancelled a photo or seeked a film โ€” not an error def do_HEAD(self): low = urlparse(self.path).path.lower() if low.startswith("/data") or low.endswith(".py") or low.startswith("/api/"): return self.send_error(404) super().do_HEAD() def do_POST(self): p = urlparse(self.path).path try: if p == "/api/track": return self.api_track() if p == "/api/bookings": return self.api_book() if p == "/api/admin/login": return self.api_login() if p == "/api/admin/logout": session_end(self.token()) return self.send_json({"ok": True}, headers={"Set-Cookie": "el_admin=; Path=/; Max-Age=0; HttpOnly; SameSite=Strict"}) if p == "/api/barber/login": return self.api_barber_login() if p == "/api/barber/logout": barber_session_end(self.barber_token()) return self.send_json({"ok": True}, headers={"Set-Cookie": "el_barber=; Path=/; Max-Age=0; HttpOnly; SameSite=Strict"}) if p.startswith("/api/barber/"): b = self.current_barber() if not (b and self.same_origin()): return self.send_json({"error": "unauthorised"}, 401) return self.api_barber_write("POST", p, b) if p == "/api/admin/upload": if not (self.is_admin() and self.same_origin()): self.drain() return self.send_json({"error": "Your admin session has ended. Sign in again, then upload."}, 401) return self.api_admin_upload() if p.startswith("/api/admin/"): if not (self.is_admin() and self.same_origin()): return self.send_json({"error": "unauthorised"}, 401) return self.api_admin_write("POST", p) except (ConnectionAbortedError, ConnectionResetError, BrokenPipeError): return self.send_json({"error": "not found"}, 404) def do_PUT(self): p = urlparse(self.path).path if p.startswith("/api/barber/"): b = self.current_barber() if b and self.same_origin(): return self.api_barber_write("PUT", p, b) return self.send_json({"error": "unauthorised"}, 401) if p.startswith("/api/admin/") and self.is_admin() and self.same_origin(): return self.api_admin_write("PUT", p) self.send_json({"error": "unauthorised"}, 401) def do_PATCH(self): self.do_PUT() def do_DELETE(self): p = urlparse(self.path).path if p.startswith("/api/barber/"): b = self.current_barber() if b and self.same_origin(): return self.api_barber_write("DELETE", p, b) return self.send_json({"error": "unauthorised"}, 401) if p.startswith("/api/admin/") and self.is_admin() and self.same_origin(): return self.api_admin_write("DELETE", p) self.send_json({"error": "unauthorised"}, 401) # ---- public & protected api def api_get(self, p, qs): if p == "/api/content": return self.send_json(load_content()) if p == "/api/slots": content = load_content() svc = all_services(content).get((qs.get("service") or [""])[0]) if not svc: return self.send_json({"slots": []}) return self.send_json({"slots": free_slots(content, (qs.get("date") or [""])[0], svc[0])}) if p.startswith("/api/uploads/"): return self.serve_upload(p[len("/api/uploads/"):]) if p.startswith("/api/barber/"): return self.api_barber_get(p, qs) if p == "/api/admin/me": return self.send_json({"admin": self.is_admin()}) if not p.startswith("/api/admin/") or not self.is_admin(): return self.send_json({"error": "unauthorised"}, 401) if p == "/api/admin/team/summary": frm, to = date_range(qs) bid = (qs.get("barber_id") or [""])[0] return self.send_json(team_summary(frm, to, int(bid) if bid.isdigit() else None)) if p == "/api/admin/team/badges": with db() as con: pp = con.execute("SELECT COUNT(*) FROM product_requests WHERE status='pending'").fetchone()[0] pl = con.execute("SELECT COUNT(*) FROM leave_requests WHERE status='pending'").fetchone()[0] return self.send_json({"pending_products": pp, "pending_leaves": pl, "total": pp + pl}) if p == "/api/admin/bookings": sql, args = "SELECT * FROM bookings WHERE 1=1", [] if qs.get("status"): sql += " AND status=?" args.append(qs["status"][0]) if qs.get("from"): sql += " AND date>=?" args.append(qs["from"][0]) if qs.get("to"): sql += " AND date<=?" args.append(qs["to"][0]) sql += " ORDER BY date DESC, time DESC LIMIT 1000" with db() as con: rows = [dict(r) for r in con.execute(sql, args).fetchall()] return self.send_json({"bookings": rows}) if p == "/api/admin/stats": return self.send_json(stats(*date_range(qs))) if p == "/api/admin/content/versions": with db() as con: rows = [{"id": r["id"], "saved": r["saved"], "size": len(r["value"])} for r in con.execute( "SELECT id, saved, value FROM content_versions ORDER BY id DESC").fetchall()] return self.send_json({"versions": rows}) if p == "/api/admin/visits": frm, to = date_range(qs) search = clean((qs.get("q") or [""])[0], 80) limit = max(1, min(5000, int((qs.get("limit") or ["50"])[0] or 50))) offset = max(0, int((qs.get("offset") or ["0"])[0] or 0)) return self.send_json(visit_list(frm, to, search, limit, offset)) self.send_json({"error": "not found"}, 404) def api_track(self): if not rate_ok("track", self.ip(), 600, 3600): return self.send_json({"ok": False}, 429) body = self.read_json() events = body.get("events") if isinstance(body, dict) else None if not isinstance(events, list): return self.send_json({"ok": False}, 400) now = now_local() rows = [] for e in events[:40]: if not isinstance(e, dict): continue typ = clean(e.get("type"), 20) if typ not in ("pageview", "section", "click", "booking", "time"): continue try: value = float(e.get("value") or 0) except (TypeError, ValueError): value = 0 rows.append(( now.isoformat(timespec="seconds"), now.date().isoformat(), clean(body.get("sid"), 40) or "anon", clean(body.get("vid"), 40) or "anon", typ, clean(e.get("name"), 80), min(value, 86400), clean(e.get("ref"), 120), clean(body.get("device"), 12), clean(body.get("lang"), 12), )) if rows: sid, vid = rows[0][2], rows[0][3] with _lock, db() as con: con.executemany( "INSERT INTO events (ts, day, sid, vid, type, name, value, ref, device, lang) VALUES (?,?,?,?,?,?,?,?,?,?)", rows, ) # one row per visit, with everything known about it for r in rows: typ, name, value = r[4], r[5], r[6] if typ == "pageview": if con.execute("SELECT 1 FROM visits WHERE sid=?", (sid,)).fetchone(): continue ip = self.ip() device, os_, browser, app = parse_ua(self.headers.get("User-Agent")) seen = con.execute("SELECT 1 FROM visits WHERE vid=? LIMIT 1", (vid,)).fetchone() ref_url = clean(body.get("refUrl"), 300) con.execute( "INSERT INTO visits (sid, vid, ts, day, hour, ip, ref, ref_url, device, os, browser, app, lang, screen, client_tz, is_return)" " VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (sid, vid, now.isoformat(timespec="seconds"), now.date().isoformat(), now.hour, ip, r[7] or "direct", ref_url, device, os_, browser, app, r[9], clean(body.get("screen"), 20), clean(body.get("tz"), 60), 1 if seen else 0), ) cached = con.execute("SELECT * FROM ip_geo WHERE ip=?", (ip,)).fetchone() if cached and time.time() - cached["ts"] < GEO_TTL: con.execute("UPDATE visits SET country=?, cc=?, region=?, city=?, tz=?, isp=? WHERE sid=?", (cached["country"], cached["cc"], cached["region"], cached["city"], cached["tz"], cached["isp"], sid)) else: queue_geo(ip) elif typ == "click": con.execute( "UPDATE visits SET clicks=clicks+1, actions=CASE WHEN length(actions)>300 THEN actions" " WHEN actions='' THEN ? ELSE actions || ',' || ? END WHERE sid=?", (name, name, sid)) elif typ == "section": con.execute("UPDATE visits SET sections=sections+1 WHERE sid=?", (sid,)) elif typ == "time": con.execute("UPDATE visits SET duration=MAX(duration, ?) WHERE sid=?", (int(value), sid)) elif typ == "booking" and name == "submitted": con.execute("UPDATE visits SET booked=1 WHERE sid=?", (sid,)) self.send_json({"ok": True}) def api_book(self): if not rate_ok("book", self.ip(), 8, 3600): return self.send_json({"error": "Too many bookings from this connection. Please call us."}, 429) b = self.read_json() if not isinstance(b, dict): return self.send_json({"error": "Invalid request."}, 400) if b.get("website"): # honeypot: people never see this field return self.send_json({"ok": True, "ref": new_ref()}) content = load_content() service = clean(b.get("service"), 160) svc = all_services(content).get(service) date, tm = clean(b.get("date"), 10), clean(b.get("time"), 5) name, phone = clean(b.get("name"), 80), clean(b.get("phone"), 24) email, notes = clean(b.get("email"), 160), clean(b.get("notes"), 600) if not svc: return self.send_json({"error": "Please choose a service."}, 400) if not DATE_RE.match(date) or not TIME_RE.match(tm): return self.send_json({"error": "Please choose a day and a time."}, 400) if len(name) < 2: return self.send_json({"error": "Please tell us your name."}, 400) if not PHONE_RE.match(phone): return self.send_json({"error": "Please enter a valid phone number."}, 400) if email and not EMAIL_RE.match(email): return self.send_json({"error": "Please check your email address."}, 400) with _lock: if tm not in free_slots(content, date, svc[0]): return self.send_json({"error": "That time was just taken. Please choose another.", "taken": True}, 409) ref = new_ref() with db() as con: con.execute( "INSERT INTO bookings (ref, created, service, minutes, price, date, time, name, phone, email, notes) " "VALUES (?,?,?,?,?,?,?,?,?,?,?)", (ref, now_local().isoformat(timespec="seconds"), service, svc[0], svc[1], date, tm, name, phone, email, notes), ) print(f" new booking {ref}: {service} on {date} {tm} for {name} ({phone})", flush=True) self.send_json({"ok": True, "ref": ref, "service": service, "minutes": svc[0], "price": svc[1], "date": date, "time": tm}) def api_login(self): ip = self.ip() now = time.time() fails = [t for t in LOGIN_FAILS.get(ip, []) if now - t < 900] if len(fails) >= 8: return self.send_json({"error": "Too many attempts. Try again in 15 minutes."}, 429) b = self.read_json() or {} if not check_password(str(b.get("password") or "")): fails.append(now) LOGIN_FAILS[ip] = fails time.sleep(0.6) return self.send_json({"error": "Wrong password."}, 401) LOGIN_FAILS.pop(ip, None) tok = session_new(SESSION_HOURS) cookie = f"el_admin={tok}; Path=/; Max-Age={SESSION_HOURS * 3600}; HttpOnly; SameSite=Strict" + ("; Secure" if self.secure() else "") self.send_json({"ok": True}, headers={"Set-Cookie": cookie}) # ---- admin writes def api_admin_write(self, method, p): m = re.match(r"^/api/admin/bookings/(\d+)$", p) if m: bid = int(m.group(1)) if method == "DELETE": with db() as con: con.execute("DELETE FROM bookings WHERE id=?", (bid,)) return self.send_json({"ok": True}) b = self.read_json() or {} fields, args = [], [] if b.get("status") in ("pending", "confirmed", "done", "cancelled", "no-show"): fields.append("status=?") args.append(b["status"]) for k, n in (("date", 10), ("time", 5), ("name", 80), ("phone", 24), ("email", 160), ("notes", 600)): if k in b: v = clean(b[k], n) if k == "date" and not DATE_RE.match(v): continue if k == "time" and not TIME_RE.match(v): continue fields.append(f"{k}=?") args.append(v) if not fields: return self.send_json({"error": "nothing to change"}, 400) with db() as con: con.execute(f"UPDATE bookings SET {', '.join(fields)} WHERE id=?", (*args, bid)) return self.send_json({"ok": True}) if p == "/api/admin/bookings" and method == "POST": # a booking taken by phone or at the desk b = self.read_json() or {} content = load_content() service = clean(b.get("service"), 160) svc = all_services(content).get(service) date, tm = clean(b.get("date"), 10), clean(b.get("time"), 5) if not svc or not DATE_RE.match(date) or not TIME_RE.match(tm) or not clean(b.get("name"), 80): return self.send_json({"error": "Service, date, time and name are required."}, 400) ref = new_ref() with db() as con: con.execute( "INSERT INTO bookings (ref, created, service, minutes, price, date, time, name, phone, email, notes, status) " "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", (ref, now_local().isoformat(timespec="seconds"), service, svc[0], svc[1], date, tm, clean(b.get("name"), 80), clean(b.get("phone"), 24), clean(b.get("email"), 160), clean(b.get("notes"), 600), "confirmed"), ) return self.send_json({"ok": True, "ref": ref}) if p == "/api/admin/content" and method in ("PUT", "POST"): b = self.read_json() if not isinstance(b, dict) or not isinstance(b.get("menu"), list): return self.send_json({"error": "Invalid content."}, 400) save_content(b) return self.send_json({"ok": True}) m = re.match(r"^/api/admin/content/restore/(\d+)$", p) if m and method == "POST": with db() as con: r = con.execute("SELECT value FROM content_versions WHERE id=?", (int(m.group(1)),)).fetchone() if not r: return self.send_json({"error": "That version no longer exists."}, 404) save_content(json.loads(r["value"])) return self.send_json({"ok": True}) if p == "/api/admin/password" and method == "POST": b = self.read_json() or {} if not check_password(str(b.get("current") or "")): return self.send_json({"error": "The current password is wrong."}, 400) new = str(b.get("password") or "") if len(new) < 8: return self.send_json({"error": "Use at least 8 characters."}, 400) set_password(new) return self.send_json({"ok": True}) if p == "/api/admin/stats/reset" and method == "POST": with db() as con: con.execute("DELETE FROM events") con.execute("DELETE FROM visits") return self.send_json({"ok": True}) if p.startswith("/api/admin/team/"): return self.api_admin_team_write(method, p) self.send_json({"error": "not found"}, 404) # ---- uploads (CIN, diplomas, avatars, training videos) def drain(self): """Read and discard a request body we are refusing, so the browser gets our answer instead of a cut connection ("Failed to fetch").""" n = int(self.headers.get("Content-Length") or 0) if 0 < n <= MAX_UPLOAD: while n > 0: chunk = self.rfile.read(min(n, 1 << 20)) if not chunk: break n -= len(chunk) else: self.close_connection = True def api_admin_upload(self): n = int(self.headers.get("Content-Length") or 0) if n <= 0 or n > MAX_UPLOAD: return self.send_json({"error": "File is empty or exceeds 60 MB limit."}, 400) raw_name = unquote(self.headers.get("X-File-Name") or "file.jpg") kind = clean(self.headers.get("X-Upload-Kind") or "doc", 16) _, ext = os.path.splitext(raw_name.lower()) if ext not in MIME_TYPES: self.drain() return self.send_json({"error": "Unsupported file format. Use JPG, PNG, WEBP, PDF or MP4."}, 400) prefix = "doc" if kind in ("cin", "diploma", "doc") else ("vid" if kind == "video" else "img") fname = f"{prefix}_{int(time.time())}_{secrets.token_hex(5)}{ext}" fpath = os.path.join(UPLOADS_DIR, fname) data = self.rfile.read(n) with open(fpath, "wb") as f: f.write(data) return self.send_json({"ok": True, "url": f"/api/uploads/{fname}", "name": fname}) def serve_upload(self, fname): fname = os.path.basename(fname) if not re.match(r"^[a-zA-Z0-9_.-]+$", fname): return self.send_error(404) is_adm = self.is_admin() barber = self.current_barber() if not (is_adm or barber): return self.send_error(401) # Sensitive documents (doc_*) are visible only to Admin or to the owning barber if fname.startswith("doc_") and not is_adm: url_ref = f"/api/uploads/{fname}" if url_ref not in (barber.get("cin_photo"), barber.get("diploma_photo"), barber.get("avatar_photo")): return self.send_error(403) fpath = os.path.join(UPLOADS_DIR, fname) if not os.path.isfile(fpath): return self.send_error(404) _, ext = os.path.splitext(fname.lower()) ctype = MIME_TYPES.get(ext, "application/octet-stream") size = os.path.getsize(fpath) rng = self.headers.get("Range") try: with open(fpath, "rb") as f: if rng and rng.startswith("bytes="): parts = rng[6:].split("-", 1) start = int(parts[0]) if parts[0] else 0 end = int(parts[1]) if len(parts) > 1 and parts[1] else size - 1 end = min(end, size - 1) if start > end or start >= size: self.send_response(416) self.send_header("Content-Range", f"bytes */{size}") self.end_headers() return length = end - start + 1 f.seek(start) chunk = f.read(length) self.send_response(206) self.send_header("Content-Type", ctype) self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Range", f"bytes {start}-{end}/{size}") self.send_header("Content-Length", str(length)) self.end_headers() self.wfile.write(chunk) else: data = f.read() self.send_response(200) self.send_header("Content-Type", ctype) self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Length", str(size)) self.end_headers() self.wfile.write(data) except (ConnectionAbortedError, ConnectionResetError, BrokenPipeError): pass # ---- admin team management def api_admin_team_write(self, method, p): now_iso = now_local().isoformat(timespec="seconds") # Create barber if p == "/api/admin/team/barbers" and method == "POST": b = self.read_json() or {} name = clean(b.get("name"), 80) username = re.sub(r"[^a-z0-9_.-]", "", clean(b.get("username"), 40).lower()) password = str(b.get("password") or "").strip() if len(name) < 2: return self.send_json({"error": "Barber name is required."}, 400) if len(username) < 2: return self.send_json({"error": "Login username is required (letters/numbers)."}, 400) if len(password) < 4: return self.send_json({"error": "Password must be at least 4 characters."}, 400) pw = hash_password(password) level = int(b.get("level") or 1) if level not in (1, 2, 3): level = 1 status = b.get("status") if b.get("status") in ("active", "suspended", "resigned", "terminated") else "active" start_date = clean(b.get("start_date"), 10) if start_date and not DATE_RE.match(start_date): start_date = "" end_date = clean(b.get("end_date"), 10) if end_date and not DATE_RE.match(end_date): end_date = "" try: comm = max(0, min(100, int(b.get("commission_pct") or 0))) except (TypeError, ValueError): comm = 0 try: sal = max(0, int(b.get("base_salary") or 0)) except (TypeError, ValueError): sal = 0 try: with db() as con: cur = con.execute( """INSERT INTO barbers ( created, name, username, pw_salt, pw_hash, cin, cin_photo, diploma_photo, avatar_photo, start_date, end_date, level, phone, email, address, emergency_contact, commission_pct, base_salary, specialty, character_notes, status ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( now_iso, name, username, pw["salt"], pw["hash"], clean(b.get("cin"), 32), clean(b.get("cin_photo"), 200), clean(b.get("diploma_photo"), 200), clean(b.get("avatar_photo"), 200), start_date or now_local().date().isoformat(), end_date, level, clean(b.get("phone"), 24), clean(b.get("email"), 160), clean(b.get("address"), 240), clean(b.get("emergency_contact"), 120), comm, sal, clean(b.get("specialty"), 120), clean(b.get("character_notes"), 2000), status, ), ) bid = cur.lastrowid except sqlite3.IntegrityError: return self.send_json({"error": f"Username '{username}' is already taken."}, 409) return self.send_json({"ok": True, "id": bid}) # Update or delete barber m = re.match(r"^/api/admin/team/barbers/(\d+)$", p) if m: bid = int(m.group(1)) if method == "DELETE": with db() as con: con.execute("DELETE FROM barber_sessions WHERE barber_id=?", (bid,)) con.execute("DELETE FROM barbers WHERE id=?", (bid,)) return self.send_json({"ok": True}) b = self.read_json() or {} fields, args = [], [] if "name" in b and len(clean(b.get("name"), 80)) >= 2: fields.append("name=?") args.append(clean(b["name"], 80)) if "username" in b: u = re.sub(r"[^a-z0-9_.-]", "", clean(b.get("username"), 40).lower()) if len(u) >= 2: fields.append("username=?") args.append(u) if b.get("password"): pw_str = str(b.get("password")).strip() if len(pw_str) < 4: return self.send_json({"error": "Password must be at least 4 characters."}, 400) pw = hash_password(pw_str) fields.extend(["pw_salt=?", "pw_hash=?"]) args.extend([pw["salt"], pw["hash"]]) if "level" in b and int(b.get("level") or 1) in (1, 2, 3): fields.append("level=?") args.append(int(b["level"])) if "status" in b and b.get("status") in ("active", "suspended", "resigned", "terminated"): fields.append("status=?") args.append(b["status"]) for k, lim in ( ("cin", 32), ("cin_photo", 200), ("diploma_photo", 200), ("avatar_photo", 200), ("start_date", 10), ("end_date", 10), ("phone", 24), ("email", 160), ("address", 240), ("emergency_contact", 120), ("specialty", 120), ("character_notes", 2000), ): if k in b: v = clean(b[k], lim) if k in ("start_date", "end_date") and v and not DATE_RE.match(v): continue fields.append(f"{k}=?") args.append(v) for k, mx in (("commission_pct", 100), ("base_salary", 1_000_000)): if k in b: try: v = max(0, min(mx, int(b[k] or 0))) except (TypeError, ValueError): v = 0 fields.append(f"{k}=?") args.append(v) if not fields: return self.send_json({"error": "Nothing to update."}, 400) try: with db() as con: con.execute(f"UPDATE barbers SET {', '.join(fields)} WHERE id=?", (*args, bid)) if b.get("status") and b.get("status") != "active": con.execute("DELETE FROM barber_sessions WHERE barber_id=?", (bid,)) except sqlite3.IntegrityError: return self.send_json({"error": "That username is already used by another barber."}, 409) return self.send_json({"ok": True}) # Admin attendance create / edit / delete if p == "/api/admin/team/attendance" and method == "POST": b = self.read_json() or {} bid = int(b.get("barber_id") or 0) day = clean(b.get("day"), 10) or now_local().date().isoformat() if not bid or not DATE_RE.match(day): return self.send_json({"error": "Barber and valid date are required."}, 400) cin = clean(b.get("check_in"), 5) cout = clean(b.get("check_out"), 5) if cin and not TIME_RE.match(cin): cin = "" if cout and not TIME_RE.match(cout): cout = "" status = b.get("status") if b.get("status") in ("present", "late", "absent", "leave", "half-day") else "present" mins = calc_shift_minutes(cin, cout) if status in ("present", "late", "half-day") else 0 if "minutes_worked" in b and b.get("minutes_worked") not in ("", None): try: mins = max(0, int(b["minutes_worked"])) except (TypeError, ValueError): pass with db() as con: con.execute( "INSERT INTO barber_attendance (barber_id, day, check_in, check_out, minutes_worked, status, notes, created) VALUES (?,?,?,?,?,?,?,?)", (bid, day, cin, cout, mins, status, clean(b.get("notes"), 400), now_iso), ) return self.send_json({"ok": True}) m = re.match(r"^/api/admin/team/attendance/(\d+)$", p) if m: aid = int(m.group(1)) if method == "DELETE": with db() as con: con.execute("DELETE FROM barber_attendance WHERE id=?", (aid,)) return self.send_json({"ok": True}) b = self.read_json() or {} with db() as con: row = con.execute("SELECT * FROM barber_attendance WHERE id=?", (aid,)).fetchone() if not row: return self.send_json({"error": "Record not found"}, 404) day = clean(b.get("day", row["day"]), 10) cin = clean(b.get("check_in", row["check_in"]), 5) cout = clean(b.get("check_out", row["check_out"]), 5) status = b.get("status", row["status"]) if status not in ("present", "late", "absent", "leave", "half-day"): status = row["status"] notes = clean(b.get("notes", row["notes"]), 400) mins = calc_shift_minutes(cin, cout) if (cin and cout and status in ("present", "late", "half-day")) else int(row["minutes_worked"] or 0) if status in ("absent", "leave"): mins = 0 if "minutes_worked" in b and b.get("minutes_worked") not in ("", None): try: mins = max(0, int(b["minutes_worked"])) except (TypeError, ValueError): pass con.execute( "UPDATE barber_attendance SET day=?, check_in=?, check_out=?, minutes_worked=?, status=?, notes=? WHERE id=?", (day, cin, cout, mins, status, notes, aid), ) return self.send_json({"ok": True}) # Admin edit/delete work entry m = re.match(r"^/api/admin/team/works/(\d+)$", p) if m: wid = int(m.group(1)) if method == "DELETE": with db() as con: con.execute("DELETE FROM barber_works WHERE id=?", (wid,)) return self.send_json({"ok": True}) b = self.read_json() or {} fields, args = [], [] if "final_price" in b: try: fields.append("final_price=?") args.append(max(0, int(b["final_price"]))) except (TypeError, ValueError): pass for k, lim in (("time", 5), ("client_name", 80), ("client_notes", 600)): if k in b: fields.append(f"{k}=?") args.append(clean(b[k], lim)) if fields: with db() as con: con.execute(f"UPDATE barber_works SET {', '.join(fields)} WHERE id=?", (*args, wid)) return self.send_json({"ok": True}) # Admin product requests update / delete m = re.match(r"^/api/admin/team/products/(\d+)$", p) if m: pid = int(m.group(1)) if method == "DELETE": with db() as con: con.execute("DELETE FROM product_requests WHERE id=?", (pid,)) return self.send_json({"ok": True}) b = self.read_json() or {} status = b.get("status") if status not in ("pending", "ordered", "restocked", "dismissed"): return self.send_json({"error": "Invalid status"}, 400) with db() as con: con.execute("UPDATE product_requests SET status=?, updated=? WHERE id=?", (status, now_iso, pid)) return self.send_json({"ok": True}) # Admin leave requests approve / reject / delete m = re.match(r"^/api/admin/team/leaves/(\d+)$", p) if m: lid = int(m.group(1)) if method == "DELETE": with db() as con: con.execute("DELETE FROM leave_requests WHERE id=?", (lid,)) return self.send_json({"ok": True}) b = self.read_json() or {} status = b.get("status") if status not in ("pending", "approved", "rejected"): return self.send_json({"error": "Invalid status"}, 400) admin_note = clean(b.get("admin_note"), 300) with db() as con: lv = con.execute("SELECT * FROM leave_requests WHERE id=?", (lid,)).fetchone() if not lv: return self.send_json({"error": "Leave request not found"}, 404) con.execute( "UPDATE leave_requests SET status=?, admin_note=?, updated=? WHERE id=?", (status, admin_note, now_iso, lid), ) # If approved full_day or half_day, log in barber_attendance so the dashboard tracks it if status == "approved" and lv["leave_type"] in ("full_day", "half_day"): try: d_cur = dt.date.fromisoformat(lv["date_from"]) d_end = dt.date.fromisoformat(lv["date_to"] or lv["date_from"]) att_st = "leave" if lv["leave_type"] == "full_day" else "half-day" while d_cur <= d_end: ds = d_cur.isoformat() exists = con.execute( "SELECT id FROM barber_attendance WHERE barber_id=? AND day=? AND status IN ('leave','half-day')", (lv["barber_id"], ds), ).fetchone() if not exists: con.execute( "INSERT INTO barber_attendance (barber_id, day, check_in, check_out, minutes_worked, status, notes, created) VALUES (?,?,?,?,?,?,?,?)", (lv["barber_id"], ds, "", "", 0, att_st, f"Approved leave ({lv['leave_type']})", now_iso), ) d_cur += dt.timedelta(days=1) except ValueError: pass return self.send_json({"ok": True}) # Admin training videos CRUD if p == "/api/admin/team/videos" and method == "POST": b = self.read_json() or {} title = clean(b.get("title"), 140) video_url = clean(b.get("video_url"), 500) if not title or not video_url: return self.send_json({"error": "Title and video file or link are required."}, 400) cat = clean(b.get("category") or "Technique", 40) lvl = int(b.get("target_level") or 0) if lvl not in (0, 1, 2, 3): lvl = 0 pinned = 1 if b.get("pinned") else 0 with db() as con: cur = con.execute( "INSERT INTO training_videos (title, category, target_level, video_url, thumbnail_url, description, pinned, created) VALUES (?,?,?,?,?,?,?,?)", (title, cat, lvl, video_url, clean(b.get("thumbnail_url"), 300), clean(b.get("description"), 1200), pinned, now_iso), ) return self.send_json({"ok": True, "id": cur.lastrowid}) m = re.match(r"^/api/admin/team/videos/(\d+)$", p) if m: vid = int(m.group(1)) if method == "DELETE": with db() as con: con.execute("DELETE FROM training_videos WHERE id=?", (vid,)) return self.send_json({"ok": True}) b = self.read_json() or {} fields, args = [], [] for k, lim in (("title", 140), ("category", 40), ("video_url", 500), ("thumbnail_url", 300), ("description", 1200)): if k in b: fields.append(f"{k}=?") args.append(clean(b[k], lim)) if "target_level" in b: fields.append("target_level=?") args.append(int(b.get("target_level") or 0)) if "pinned" in b: fields.append("pinned=?") args.append(1 if b.get("pinned") else 0) if fields: with db() as con: con.execute(f"UPDATE training_videos SET {', '.join(fields)} WHERE id=?", (*args, vid)) return self.send_json({"ok": True}) return self.send_json({"error": "not found"}, 404) # ---- barber portal api def api_barber_login(self): ip = self.ip() now = time.time() fails = [t for t in LOGIN_FAILS.get(("b", ip), []) if now - t < 900] if len(fails) >= 10: return self.send_json({"error": "Too many attempts. Try again in 15 minutes."}, 429) b = self.read_json() or {} username = re.sub(r"[^a-z0-9_.-]", "", clean(b.get("username"), 40).lower()) password = str(b.get("password") or "").strip() with db() as con: row = con.execute("SELECT * FROM barbers WHERE username=?", (username,)).fetchone() if not row or not hmac.compare_digest(hash_password(password, row["pw_salt"])["hash"], row["pw_hash"]): fails.append(now) LOGIN_FAILS[("b", ip)] = fails time.sleep(0.5) return self.send_json({"error": "Invalid login or password."}, 401) if row["status"] != "active": return self.send_json({"error": "This account is currently inactive. Please contact management."}, 403) LOGIN_FAILS.pop(("b", ip), None) tok = barber_session_new(row["id"], BARBER_SESSION_HOURS) cookie = f"el_barber={tok}; Path=/; Max-Age={BARBER_SESSION_HOURS * 3600}; HttpOnly; SameSite=Strict" + ("; Secure" if self.secure() else "") self.send_json({"ok": True, "barber": barber_public(row)}, headers={"Set-Cookie": cookie}) def api_barber_get(self, p, qs): barber = self.current_barber() if p == "/api/barber/me": if not barber: return self.send_json({"barber": None}) now = now_local() today_str = now.date().isoformat() content = load_content() with db() as con: att_today = [dict(r) for r in con.execute( "SELECT * FROM barber_attendance WHERE barber_id=? AND day=? ORDER BY id DESC", (barber["id"], today_str), ).fetchall()] return self.send_json({ "barber": barber_public(barber), "today": today_str, "now_time": now.strftime("%H:%M"), "attendance_today": att_today, "services_catalog": services_catalog(content), }) if not barber: return self.send_json({"error": "unauthorised"}, 401) if p == "/api/barber/dashboard": frm, to = date_range(qs) now = now_local() today_str = now.date().isoformat() bid = barber["id"] with db() as con: works = [dict(r) for r in con.execute( "SELECT * FROM barber_works WHERE barber_id=? AND day BETWEEN ? AND ? ORDER BY day DESC, time DESC, id DESC", (bid, frm, to), ).fetchall()] att = [dict(r) for r in con.execute( "SELECT * FROM barber_attendance WHERE barber_id=? AND day BETWEEN ? AND ? ORDER BY day DESC, id DESC", (bid, frm, to), ).fetchall()] att_today = [dict(r) for r in con.execute( "SELECT * FROM barber_attendance WHERE barber_id=? AND day=? ORDER BY id DESC", (bid, today_str), ).fetchall()] products = [dict(r) for r in con.execute( "SELECT * FROM product_requests WHERE barber_id=? ORDER BY id DESC LIMIT 60", (bid,), ).fetchall()] leaves = [dict(r) for r in con.execute( "SELECT * FROM leave_requests WHERE barber_id=? ORDER BY id DESC LIMIT 60", (bid,), ).fetchall()] videos = [dict(r) for r in con.execute( "SELECT * FROM training_videos ORDER BY pinned DESC, id DESC" ).fetchall()] for w in works: try: w["services"] = json.loads(w.get("services_json") or "[]") except Exception: w["services"] = [] mins_worked = 0 for a in att: if a["day"] == today_str and a["check_in"] and not a["check_out"] and a["status"] in ("present", "late", "half-day"): a["live_minutes"] = calc_shift_minutes(a["check_in"], now.strftime("%H:%M")) else: a["live_minutes"] = a["minutes_worked"] if a["status"] in ("present", "late", "half-day"): mins_worked += int(a["live_minutes"] or 0) std_rev = sum(int(w["standard_price"]) for w in works) fin_rev = sum(int(w["final_price"]) for w in works) return self.send_json({ "from": frm, "to": to, "today": today_str, "now_time": now.strftime("%H:%M"), "attendance_today": att_today, "works": works, "attendance": att, "product_requests": products, "leave_requests": leaves, "videos": videos, "summary": { "clients_count": len(works), "services_count": sum(len(w.get("services") or []) for w in works), "standard_total": std_rev, "revenue_total": fin_rev, "discount_total": max(0, std_rev - fin_rev), "minutes_worked": mins_worked, "days_present": len({a["day"] for a in att if a["status"] in ("present", "late", "half-day")}), }, }) return self.send_json({"error": "not found"}, 404) def api_barber_write(self, method, p, barber): bid = barber["id"] now = now_local() now_iso = now.isoformat(timespec="seconds") today_str = now.date().isoformat() now_hhmm = now.strftime("%H:%M") # 1. Clock in / Clock out ("bda lkhdma" / "sala lkhdma") if p == "/api/barber/attendance" and method == "POST": b = self.read_json() or {} action = b.get("action") with _lock, db() as con: open_shift = con.execute( "SELECT * FROM barber_attendance WHERE barber_id=? AND day=? AND check_in!='' AND check_out='' AND status IN ('present','late','half-day') ORDER BY id DESC LIMIT 1", (bid, today_str), ).fetchone() if action == "clock_in": if open_shift: return self.send_json({"ok": True, "shift": dict(open_shift)}) con.execute( "INSERT INTO barber_attendance (barber_id, day, check_in, check_out, minutes_worked, status, notes, created) VALUES (?,?,?,?,?,?,?,?)", (bid, today_str, now_hhmm, "", 0, "present", clean(b.get("notes"), 200), now_iso), ) return self.send_json({"ok": True, "status": "clocked_in", "time": now_hhmm}) elif action == "clock_out": if not open_shift: return self.send_json({"error": "No active shift to end."}, 400) mins = calc_shift_minutes(open_shift["check_in"], now_hhmm) con.execute( "UPDATE barber_attendance SET check_out=?, minutes_worked=? WHERE id=?", (now_hhmm, mins, open_shift["id"]), ) return self.send_json({"ok": True, "status": "clocked_out", "time": now_hhmm, "minutes_worked": mins}) return self.send_json({"error": "Invalid attendance action."}, 400) # 2. Record client services ("mani kayji client o kay7asnlo...") if p == "/api/barber/works" and method == "POST": b = self.read_json() or {} day = clean(b.get("day"), 10) or today_str if not DATE_RE.match(day): day = today_str tm = clean(b.get("time"), 5) or now_hhmm if not TIME_RE.match(tm): tm = now_hhmm svcs_in = b.get("services") if not isinstance(svcs_in, list) or not svcs_in: return self.send_json({"error": "Please select at least one service."}, 400) svcs_clean = [] std_price = 0 tot_min = 0 for s in svcs_in[:30]: if not isinstance(s, dict): continue sname = clean(s.get("name"), 140) if not sname: continue sdh = max(0, int(s.get("dh") or 0)) smin = max(5, int(s.get("min") or 15)) svcs_clean.append({"name": sname, "dh": sdh, "min": smin}) std_price += sdh tot_min += smin if not svcs_clean: return self.send_json({"error": "Please select at least one valid service."}, 400) try: fin_price = max(0, int(b.get("final_price") if b.get("final_price") not in ("", None) else std_price)) except (TypeError, ValueError): fin_price = std_price svcs_text = " + ".join(s["name"] for s in svcs_clean) client_name = clean(b.get("client_name"), 80) client_notes = clean(b.get("client_notes"), 600) with db() as con: cur = con.execute( """INSERT INTO barber_works ( barber_id, day, time, client_name, services_json, services_text, total_minutes, standard_price, final_price, client_notes, created ) VALUES (?,?,?,?,?,?,?,?,?,?,?)""", ( bid, day, tm, client_name, json.dumps(svcs_clean, ensure_ascii=False), svcs_text[:400], tot_min, std_price, fin_price, client_notes, now_iso, ), ) return self.send_json({"ok": True, "id": cur.lastrowid}) m = re.match(r"^/api/barber/works/(\d+)$", p) if m and method == "DELETE": wid = int(m.group(1)) with db() as con: con.execute("DELETE FROM barber_works WHERE id=? AND barber_id=?", (wid, bid)) return self.send_json({"ok": True}) # 3. Request missing/low-stock product ("itlab chi produit wala i3lam bi ana produit na9ass...") if p == "/api/barber/products" and method == "POST": b = self.read_json() or {} pname = clean(b.get("product_name"), 120) if len(pname) < 2: return self.send_json({"error": "Please enter the product name."}, 400) try: qty = max(1, min(999, int(b.get("quantity") or 1))) except (TypeError, ValueError): qty = 1 urgency = b.get("urgency") if b.get("urgency") in ("low", "out", "new") else "low" notes = clean(b.get("notes"), 300) with db() as con: cur = con.execute( "INSERT INTO product_requests (barber_id, product_name, quantity, urgency, notes, status, created) VALUES (?,?,?,?,?,?,?)", (bid, pname, qty, urgency, notes, "pending", now_iso), ) return self.send_json({"ok": True, "id": cur.lastrowid}) m = re.match(r"^/api/barber/products/(\d+)$", p) if m and method == "DELETE": pid = int(m.group(1)) with db() as con: con.execute("DELETE FROM product_requests WHERE id=? AND barber_id=? AND status='pending'", (pid, bid)) return self.send_json({"ok": True}) # 4. Request time off / leave ("itlab ra7a nhar wala nass nhar wala sa3 manin 7ta lin...") if p == "/api/barber/leaves" and method == "POST": b = self.read_json() or {} ltype = b.get("leave_type") if ltype not in ("full_day", "half_day", "hours"): return self.send_json({"error": "Please choose leave type (full day, half day, or hours)."}, 400) d_from = clean(b.get("date_from"), 10) if not DATE_RE.match(d_from): return self.send_json({"error": "Please choose a valid date."}, 400) d_to = clean(b.get("date_to"), 10) or d_from if not DATE_RE.match(d_to) or ltype != "full_day": d_to = d_from if d_to < d_from: d_from, d_to = d_to, d_from half_period = "" t_from, t_to = "", "" if ltype == "half_day": half_period = "afternoon" if b.get("half_period") == "afternoon" else "morning" elif ltype == "hours": t_from = clean(b.get("time_from"), 5) t_to = clean(b.get("time_to"), 5) if not (TIME_RE.match(t_from) and TIME_RE.match(t_to)): return self.send_json({"error": "Please specify start time (from) and end time (to)."}, 400) reason = clean(b.get("reason"), 400) with db() as con: cur = con.execute( """INSERT INTO leave_requests ( barber_id, leave_type, date_from, date_to, half_period, time_from, time_to, reason, status, created ) VALUES (?,?,?,?,?,?,?,?,?,?)""", (bid, ltype, d_from, d_to, half_period, t_from, t_to, reason, "pending", now_iso), ) return self.send_json({"ok": True, "id": cur.lastrowid}) m = re.match(r"^/api/barber/leaves/(\d+)$", p) if m and method == "DELETE": lid = int(m.group(1)) with db() as con: con.execute("DELETE FROM leave_requests WHERE id=? AND barber_id=? AND status='pending'", (lid, bid)) return self.send_json({"ok": True}) # 5. Update own profile ("ibadal chi info dialo b7al telephone email address...") if p == "/api/barber/profile" and method in ("POST", "PUT"): b = self.read_json() or {} fields, args = [], [] for k, lim in (("phone", 24), ("email", 160), ("address", 240), ("emergency_contact", 120)): if k in b: v = clean(b[k], lim) if k == "email" and v and not EMAIL_RE.match(v): return self.send_json({"error": "Please enter a valid email address."}, 400) fields.append(f"{k}=?") args.append(v) if b.get("new_password"): cur_pw = str(b.get("current_password") or "").strip() if not hmac.compare_digest(hash_password(cur_pw, barber["pw_salt"])["hash"], barber["pw_hash"]): return self.send_json({"error": "Current password is wrong."}, 400) new_pw = str(b.get("new_password")).strip() if len(new_pw) < 4: return self.send_json({"error": "New password must be at least 4 characters."}, 400) pw = hash_password(new_pw) fields.extend(["pw_salt=?", "pw_hash=?"]) args.extend([pw["salt"], pw["hash"]]) if not fields: return self.send_json({"error": "Nothing to update."}, 400) with db() as con: con.execute(f"UPDATE barbers SET {', '.join(fields)} WHERE id=?", (*args, bid)) updated = con.execute("SELECT * FROM barbers WHERE id=?", (bid,)).fetchone() return self.send_json({"ok": True, "barber": barber_public(updated)}) return self.send_json({"error": "not found"}, 404) class Server(socketserver.ThreadingTCPServer): allow_reuse_address = True # restart on the same port without waiting daemon_threads = True # photos and films load in parallel def backup(dest, keep=14): """Copy the live database with SQLite's own backup (safe while the site runs) to DEST/elegance-YYYY-MM-DD_HHMM.db, and prune old copies.""" os.makedirs(dest, exist_ok=True) name = os.path.join(dest, "elegance-" + now_local().strftime("%Y-%m-%d_%H%M") + ".db") src = sqlite3.connect(DB_PATH) out = sqlite3.connect(name) with out: src.backup(out) out.close() src.close() copies = sorted(f for f in os.listdir(dest) if f.startswith("elegance-") and f.endswith(".db")) for old in copies[:-keep]: os.remove(os.path.join(dest, old)) print(f"backup written: {name} ({len(copies[-keep:])} kept)") if __name__ == "__main__": if len(sys.argv) > 2 and sys.argv[1] == "--backup": init_db() backup(sys.argv[2], int(sys.argv[3]) if len(sys.argv) > 3 else 14) sys.exit(0) if len(sys.argv) > 2 and sys.argv[1] == "--set-password": init_db() migrate_files() set_password(sys.argv[2]) print("admin password updated") sys.exit(0) port = int(sys.argv[1]) if len(sys.argv) > 1 else 4173 init_db() migrate_files() ensure_admin() threading.Thread(target=geo_worker, daemon=True).start() host = os.environ.get("ELEGANCE_HOST", "127.0.0.1") with Server((host, port), Handler) as httpd: print(f"serving http://localhost:{port} (admin: http://localhost:{port}/admin.html)", flush=True) httpd.serve_forever()