People Database

#!/usr/bin/env python3
"""Serves people.html and stores entries in ~/Documents/AppData/people.db (SQLite).
Run:  python3 people_server.py   then open http://127.0.0.1:8766

Fields: name, surname, gender, email (unique), phone (optional).
Spreadsheet: GET /api/export.csv, POST /api/import (CSV, opens in Excel/LibreOffice).
"""
import csv
import io
import json
import re
import sqlite3
import unicodedata
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path

HOST, PORT = "127.0.0.1", 8766
HTML = Path(__file__).with_name("people.html")
DB_DIR = Path.home() / "Documents" / "AppData"
DB_PATH = DB_DIR / "people.db"
GENDERS = {g.lower(): g for g in ("Female", "Male", "Non-binary", "Prefer not to say")}
COLUMNS = ("name", "surname", "gender", "email", "phone")
NAME_RE = re.compile(r"^[^\W\d_](?:[^\W\d_]|[ '\-.])*$")   # letters (any script), space ' - .
EMAIL_RE = re.compile(r"^[A-Za-z0-9._%+\-]+@[A-Za-z0-9](?:[A-Za-z0-9\-]*[A-Za-z0-9])?(?:\.[A-Za-z0-9](?:[A-Za-z0-9\-]*[A-Za-z0-9])?)+$")
PHONE_RE = re.compile(r"^\+?[0-9 ()\-]{6,20}$")
MAX_NAME, MAX_EMAIL = 60, 254
MAX_IMPORT_BYTES, MAX_IMPORT_ROWS = 1_000_000, 5000


def connect() -> sqlite3.Connection:
    return sqlite3.connect(DB_PATH)


def init_db() -> None:
    """Create the database if missing; migrate the older layout (age NOT NULL) without losing rows."""
    DB_DIR.mkdir(parents=True, exist_ok=True)
    with connect() as db:
        db.execute("""CREATE TABLE IF NOT EXISTS people (
            email      TEXT PRIMARY KEY COLLATE NOCASE,
            name       TEXT NOT NULL,
            surname    TEXT NOT NULL,
            gender     TEXT NOT NULL,
            phone      TEXT,
            age        INTEGER,            -- legacy, no longer collected
            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)""")
        cols = {r[1]: r for r in db.execute("PRAGMA table_info(people)")}
        if "age" in cols and cols["age"][3]:      # old table: age was NOT NULL
            db.execute("ALTER TABLE people RENAME TO people_old")
            db.execute("""CREATE TABLE people (
                email TEXT PRIMARY KEY COLLATE NOCASE, name TEXT NOT NULL, surname TEXT NOT NULL,
                gender TEXT NOT NULL, phone TEXT, age INTEGER,
                created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)""")
            db.execute("INSERT INTO people SELECT email,name,surname,gender,phone,age,created_at FROM people_old")
            db.execute("DROP TABLE people_old")


def clean(v) -> str:
    """Filter one input: string only, NFC, no control/format chars, collapsed whitespace, trimmed."""
    if v is None:
        return ""
    s = unicodedata.normalize("NFC", str(v))
    s = "".join(" " if ch.isspace() else ch for ch in s if ch.isspace() or unicodedata.category(ch)[0] != "C")
    return re.sub(r" {2,}", " ", s).strip()


def validate(d: dict) -> tuple[dict | None, str]:
    rec = {k: clean(d.get(k)) for k in COLUMNS}
    for k in ("name", "surname"):
        label = k.capitalize()
        if not rec[k]:
            return None, f"{label} is required."
        if len(rec[k]) > MAX_NAME or not NAME_RE.match(rec[k]):
            return None, f"{label} may only contain letters, spaces, hyphens, apostrophes and full stops (max {MAX_NAME})."
    rec["gender"] = GENDERS.get(rec["gender"].lower(), "")
    if not rec["gender"]:
        return None, "Choose a gender option."
    rec["email"] = rec["email"].lower()
    if len(rec["email"]) > MAX_EMAIL or not EMAIL_RE.match(rec["email"]):
        return None, "Enter a valid email address."
    if rec["phone"] and not PHONE_RE.match(rec["phone"]):
        return None, "Phone may only contain digits, spaces, ( ) - and a leading + (6-20 characters)."
    rec["phone"] = rec["phone"] or None
    return rec, ""


def insert(db: sqlite3.Connection, rec: dict) -> None:
    db.execute("INSERT INTO people (email, name, surname, gender, phone) "
               "VALUES (:email, :name, :surname, :gender, :phone)", rec)


def sheet_safe(v: str) -> str:
    """Stop spreadsheet apps from running a cell as a formula (leading ' is stripped again on import)."""
    return "'" + v if v[:1] in ("=", "@", "\t", "\r") else v


def unsheet(v: str) -> str:
    return v[1:] if v[:2] in ("'=", "'@") else v


def export_csv() -> bytes:
    buf = io.StringIO(newline="")
    w = csv.writer(buf)
    w.writerow(COLUMNS)
    with connect() as db:
        for r in db.execute("SELECT name, surname, gender, email, phone FROM people ORDER BY created_at, rowid"):
            w.writerow([sheet_safe(c or "") for c in r])
    return b"\xef\xbb\xbf" + buf.getvalue().encode()      # BOM so Excel reads UTF-8


def import_csv(raw: bytes) -> dict:
    """Row-by-row: valid new rows are saved; duplicates and bad rows are reported, never fatal."""
    try:
        text = raw.decode("utf-8-sig")
    except UnicodeDecodeError:
        raise ValueError("File must be UTF-8 CSV (in Excel: Save As > CSV UTF-8).")
    rdr = csv.reader(io.StringIO(text))
    header = [clean(h).lower() for h in next(rdr, [])]
    missing = [c for c in ("name", "surname", "gender", "email") if c not in header]
    if missing:
        raise ValueError("Missing column(s): " + ", ".join(missing) + ". Expected: " + ", ".join(COLUMNS) + ".")
    idx = {c: header.index(c) for c in COLUMNS if c in header}
    added, dupes, rejected, seen = 0, 0, [], set()
    with connect() as db:
        for n, row in enumerate(rdr, 2):
            if n - 1 > MAX_IMPORT_ROWS:
                raise ValueError(f"Too many rows (max {MAX_IMPORT_ROWS}).")
            if not any(c.strip() for c in row):
                continue
            rec, err = validate({c: unsheet(row[i]) if i < len(row) else "" for c, i in idx.items()})
            if err:
                rejected.append({"row": n, "error": err})
                continue
            if rec["email"] in seen:
                dupes += 1
                continue
            seen.add(rec["email"])
            try:
                insert(db, rec)
                added += 1
            except sqlite3.IntegrityError:
                dupes += 1
    return {"added": added, "duplicates": dupes, "rejected": rejected[:50], "rejected_count": len(rejected)}


class Handler(BaseHTTPRequestHandler):
    def send(self, code: int, body: bytes, ctype: str = "application/json", extra: dict | None = None) -> None:
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("X-Content-Type-Options", "nosniff")
        for k, v in (extra or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)

    def reply(self, code: int, obj) -> None:
        self.send(code, json.dumps(obj).encode())

    def host_ok(self) -> bool:  # blocks DNS-rebinding style access from other sites
        return self.headers.get("Host", "") in {f"{HOST}:{PORT}", f"localhost:{PORT}"}

    def body(self, limit: int) -> bytes:
        n = int(self.headers.get("Content-Length", 0))
        if n < 0 or n > limit:
            raise ValueError("Request too large.")
        return self.rfile.read(n)

    def do_GET(self):
        if not self.host_ok():
            return self.reply(403, {"error": "Forbidden"})
        if self.path in ("/", "/people.html"):
            return self.send(200, HTML.read_bytes(), "text/html; charset=utf-8")
        if self.path == "/api/people":
            with connect() as db:
                rows = db.execute("SELECT name, surname, gender, email, phone FROM people "
                                  "ORDER BY created_at DESC, rowid DESC").fetchall()
            return self.reply(200, [dict(zip(COLUMNS, r)) for r in rows])
        if self.path == "/api/export.csv":
            return self.send(200, export_csv(), "text/csv; charset=utf-8",
                             {"Content-Disposition": 'attachment; filename="people.csv"'})
        self.reply(404, {"error": "Not found"})

    def do_POST(self):
        if not self.host_ok() or self.path not in ("/api/people", "/api/import"):
            return self.reply(404, {"error": "Not found"})
        ctype = self.headers.get("Content-Type", "").split(";")[0]
        try:
            if self.path == "/api/import":
                if ctype != "text/csv":
                    return self.reply(415, {"error": "CSV required"})
                return self.reply(200, import_csv(self.body(MAX_IMPORT_BYTES)))
            if ctype != "application/json":
                return self.reply(415, {"error": "JSON required"})
            data = json.loads(self.body(10_000))
            if not isinstance(data, dict):
                raise ValueError
        except ValueError as e:
            return self.reply(400, {"error": str(e) or "Invalid request."})
        rec, err = validate(data)
        if err:
            return self.reply(400, {"error": err})
        try:
            with connect() as db:
                insert(db, rec)
        except sqlite3.IntegrityError:
            return self.reply(409, {"error": "That email address is already in the database."})
        self.reply(201, {"ok": True})

    def log_message(self, *a):  # keep the terminal quiet
        pass


if __name__ == "__main__":
    init_db()
    print(f"Database: {DB_PATH}\nOpen http://{HOST}:{PORT}  (Ctrl+C to stop)")
    HTTPServer((HOST, PORT), Handler).serve_forever()

<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>People Database</title>
<style>
  :root { --bg:#fff; --fg:#1a1a1a; --mut:#666; --line:#ccc; --acc:#2557d6; --err:#c62828; --ok:#2e7d32; }
  @media (prefers-color-scheme: dark) { :root { --bg:#181a1f; --fg:#e8e8e8; --mut:#999; --line:#3a3d45; --acc:#6d95ff; --err:#ff8a80; --ok:#81c784; } }
  * { box-sizing:border-box; }
  body { margin:0; padding:16px; background:var(--bg); color:var(--fg); font:15px system-ui,sans-serif; }
  main { max-width:640px; margin:0 auto; display:grid; gap:16px; }
  h1 { font-size:20px; margin:0; }
  form { display:grid; grid-template-columns:1fr 1fr; gap:12px; }
  .wide { grid-column:1 / -1; }
  label { display:grid; gap:4px; color:var(--mut); font-size:13px; }
  input,select { font:inherit; padding:8px; color:var(--fg); background:transparent; border:1px solid var(--line); border-radius:6px; width:100%; }
  button { font:inherit; padding:10px; border:0; border-radius:6px; background:var(--acc); color:#fff; cursor:pointer; }
  .bar { display:flex; gap:8px; flex-wrap:wrap; }
  .btn { display:inline-block; padding:8px 12px; border:1px solid var(--line); border-radius:6px; color:var(--fg); text-decoration:none; cursor:pointer; font-size:13px; }
  #msg { min-height:1.2em; font-size:13px; }
  .err { color:var(--err); } .ok { color:var(--ok); }
  .tbl { overflow-x:auto; }
  table { border-collapse:collapse; width:100%; font-size:13px; }
  th,td { text-align:left; padding:6px 8px; border-bottom:1px solid var(--line); white-space:nowrap; }
  th { color:var(--mut); font-weight:600; }
  @media (max-width:480px) { form { grid-template-columns:1fr; } }
</style>
</head>
<body>
<main>
  <h1>People Database</h1>
  <div class="bar"><a class="btn" href="/api/export.csv" download>Export to spreadsheet (CSV)</a>
    <label class="btn">Import from spreadsheet <input id="imp" type="file" accept=".csv,text/csv" hidden></label></div>
  <form id="f">
    <label>Name <input name="name" required maxlength="60" autocomplete="given-name"></label>
    <label>Surname <input name="surname" required maxlength="60" autocomplete="family-name"></label>
    <label class="wide">Gender
      <select name="gender" required>
        <option value="">Select…</option><option>Female</option><option>Male</option>
        <option>Non-binary</option><option>Prefer not to say</option>
      </select>
    </label>
    <label class="wide">Email address (must be unique) <input name="email" type="email" required maxlength="254" autocomplete="email"></label>
    <label class="wide">Phone number (optional) <input name="phone" type="tel" pattern="\+?[0-9 ()\-]{6,20}" autocomplete="tel"></label>
    <button class="wide" type="submit">Add to database</button>
  </form>
  <div id="msg" role="status"></div>
  <div class="tbl">
    <table>
      <thead><tr><th>Name</th><th>Surname</th><th>Gender</th><th>Email</th><th>Phone</th></tr></thead>
      <tbody id="rows"></tbody>
    </table>
  </div>
</main>
<script>
const $ = (id) => document.getElementById(id);
const say = (text, cls) => { $("msg").textContent = text; $("msg").className = cls || ""; };

async function load() {
  const res = await fetch("/api/people");
  const rows = await res.json();
  $("rows").replaceChildren(...rows.map((r) => {
    const tr = document.createElement("tr");
    for (const k of ["name", "surname", "gender", "email", "phone"]) {
      const td = document.createElement("td"); td.textContent = r[k] ?? "—"; tr.append(td);
    }
    return tr;
  }));
}

$("f").addEventListener("submit", async (e) => {
  e.preventDefault();
  const data = Object.fromEntries(new FormData(e.target));
  try {
    const res = await fetch("/api/people", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(data) });
    const out = await res.json();
    if (!res.ok) return say(out.error, "err");
    e.target.reset(); say("Saved.", "ok"); load();
  } catch { say("Can't reach the server. Is people_server.py running?", "err"); }
});

$("imp").addEventListener("change", async (e) => {
  const f = e.target.files[0]; e.target.value = "";
  if (!f) return;
  try {
    const res = await fetch("/api/import", { method: "POST", headers: { "Content-Type": "text/csv" }, body: await f.arrayBuffer() });
    const out = await res.json();
    if (!res.ok) return say(out.error, "err");
    const bad = out.rejected.map((r) => `row ${r.row}: ${r.error}`).join(" | ");
    say(`Imported ${out.added}, duplicates skipped ${out.duplicates}, rejected ${out.rejected_count}.` + (bad ? " " + bad : ""), out.rejected_count ? "err" : "ok");
    load();
  } catch { say("Import failed. Is people_server.py running?", "err"); }
});

load().catch(() => say("Can't reach the server. Start it with: python3 ~/Ccode/people_server.py", "err"));
</script>
</body>
</html>