#!/usr/bin/env python3
"""Dawilo CSV appointment bridge and durable webhook inbox. Python 3.11+, no packages.
Run --help and read README.md before enabling --apply. No vendor database writes.
"""
import argparse
import csv
import datetime as dt
import hashlib
import hmac
import json
import os
from pathlib import Path
import re
import sqlite3
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from http.server import BaseHTTPRequestHandler, HTTPServer

MAX_BYTES = 10 * 1024 * 1024
EVENTS = {"appointment.created": "practice_id", "prescription.dispensed": "pharmacy_id", "referral.completed": "lab_id"}

class BridgeError(Exception):
    pass

def encoded(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")

def fingerprint(value):
    return hashlib.sha256(encoded(value)).hexdigest()

def valid_id(value):
    try:
        return str(uuid.UUID(value)) == value.lower()
    except (ValueError, TypeError, AttributeError):
        return False

def validate_config(config):
    url = urllib.parse.urlsplit(config.get("base_url", ""))
    if url.scheme != "https" or not url.hostname or url.username or url.password or url.query or url.fragment or url.path not in ("", "/"):
        raise BridgeError("base_url must be the HTTPS Dawilo origin, without a path or credentials")
    if config.get("environment") not in ("sandbox", "live") or not valid_id(config.get("practice_id")):
        raise BridgeError("Set environment and the connected public practice_id")
    if not isinstance(config.get("practitioners"), dict) or not config["practitioners"]:
        raise BridgeError("Map each vendor doctor ID to a Dawilo practitioner ID")
    return config

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        return None

class Client:
    def __init__(self, config, key):
        validate_config(config)
        if not re.fullmatch(r"dwi_" + config["environment"] + r"_[a-f0-9]{64}", key):
            raise BridgeError("DAWILO_API_KEY must match the configured environment")
        self.config, self.key = config, key
        self.opener = urllib.request.build_opener(NoRedirect())

    def call(self, path, payload=None, request_id=None):
        headers = {"Authorization": "Bearer " + self.key, "Accept": "application/json"}
        if request_id:
            headers["Idempotency-Key"] = request_id
        if payload is not None:
            headers["Content-Type"] = "application/json"
        request = urllib.request.Request(self.config["base_url"].rstrip("/") + "/api/partner/v1/" + path,
                                         data=None if payload is None else encoded(payload), headers=headers)
        try:
            with self.opener.open(request, timeout=25) as response:
                raw = response.read(1024 * 1024 + 1)
                if len(raw) > 1024 * 1024:
                    raise BridgeError("Oversized API response; retry the exact same input")
                result = json.loads(raw)
        except urllib.error.HTTPError as error:
            # Never print remote responses: they can contain patient data.
            retry = error.headers.get("Retry-After", "60")
            raise BridgeError(f"HTTP {error.code}; " + (f"retry the same input after {retry} seconds" if error.code == 429 else "queued row retained; review authorization, validation or conflict before retry")) from None
        except (urllib.error.URLError, TimeoutError, OSError, ValueError):
            raise BridgeError("Delivery uncertain; queued row retained. Retry the exact same input") from None
        if not isinstance(result, dict) or result.get("meta", {}).get("environment") != self.config["environment"]:
            raise BridgeError("Environment mismatch in response; queued row retained")
        return result

    def create(self, payload, request_id):
        return self.call("appointments", payload, request_id)

class State:
    def __init__(self, path, binding):
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        if path.is_symlink():
            raise BridgeError("State must be a regular private file")
        fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o600)
        os.close(fd)
        if os.name != "nt":
            os.chmod(path, 0o600)
        self.db = sqlite3.connect(path, timeout=5)
        self.db.row_factory = sqlite3.Row
        self.db.execute("PRAGMA synchronous=FULL")
        self.db.executescript("""
          CREATE TABLE IF NOT EXISTS settings (id TEXT PRIMARY KEY, value TEXT NOT NULL);
          CREATE TABLE IF NOT EXISTS imports (external_id TEXT PRIMARY KEY, payload_hash TEXT NOT NULL,
            request_id TEXT NOT NULL UNIQUE, state TEXT NOT NULL, result TEXT);
          CREATE TABLE IF NOT EXISTS inbox (event_id TEXT PRIMARY KEY, payload_hash TEXT NOT NULL,
            payload TEXT NOT NULL, received_at TEXT NOT NULL, processed_at TEXT);
        """)
        with self.db:
            row = self.db.execute("SELECT value FROM settings WHERE id='binding'").fetchone()
            value = fingerprint(binding)
            if row and row[0] != value:
                raise BridgeError("State belongs to another server, key or practice. Restore its original configuration; do not create a new state to retry existing imports")
            self.db.execute("INSERT OR IGNORE INTO settings VALUES ('binding', ?)", (value,))

    def import_one(self, external_id, payload, client):
        digest = fingerprint(payload)
        # An exclusive write transaction also serializes concurrent bridge processes.
        # The pending receipt commits before the network request. A retry reuses it.
        with self.db:
            self.db.execute("BEGIN IMMEDIATE")
            row = self.db.execute("SELECT * FROM imports WHERE external_id=?", (external_id,)).fetchone()
            if row and row["payload_hash"] != digest:
                raise BridgeError("Existing source row changed; no overwrite or duplicate created. Reconcile using the revision-based appointment API")
            if row and row["state"] == "done":
                return "already imported"
            if not row:
                self.db.execute("INSERT INTO imports VALUES (?, ?, ?, 'pending', NULL)", (external_id, digest, "bridge-" + str(uuid.uuid4())))
            row = self.db.execute("SELECT * FROM imports WHERE external_id=?", (external_id,)).fetchone()
        result = client.create(payload, row["request_id"])
        data = result.get("data", {})
        if not valid_id(data.get("id")) or not re.fullmatch(r"[a-f0-9]{64}", str(data.get("revision", ""))) or data.get("practice_id") != payload["practice_id"]:
            raise BridgeError("Invalid receipt; pending import retained for an exact retry")
        metadata = {k: data[k] for k in ("id", "revision", "practice_id")}
        with self.db:
            self.db.execute("UPDATE imports SET state='done',result=? WHERE external_id=? AND request_id=?", (encoded(metadata).decode(), external_id, row["request_id"]))
        return "imported"

    def receive(self, event):
        digest = fingerprint(event)
        with self.db:
            row = self.db.execute("SELECT payload_hash FROM inbox WHERE event_id=?", (event["id"],)).fetchone()
            if row and row[0] != digest:
                raise BridgeError("Event ID reused with different content")
            self.db.execute("INSERT OR IGNORE INTO inbox VALUES (?, ?, ?, ?, NULL)", (event["id"], digest, encoded(event).decode(), dt.datetime.now(dt.timezone.utc).isoformat()))


def read_rows(path, config):
    if Path(path).stat().st_size > MAX_BYTES:
        raise BridgeError("Split the export into files smaller than 10 MiB")
    columns = config.get("columns", {})
    required = ["external_id", "practitioner_id", "name", "phone", "birthDate", "date", "time"]
    if any(not isinstance(columns.get(k), str) for k in required):
        raise BridgeError("Map every required column, including birthDate")
    delimiter = config.get("delimiter", ";")
    if len(delimiter) != 1:
        raise BridgeError("delimiter must be one character")
    rows, seen = [], set()
    with open(path, newline="", encoding=config.get("encoding", "utf-8-sig")) as source:
        reader = csv.DictReader(source, delimiter=delimiter)
        if not reader.fieldnames or len(set(reader.fieldnames)) != len(reader.fieldnames) or any(v not in reader.fieldnames for v in columns.values()):
            raise BridgeError("Missing or duplicate CSV headers; check the mapping, delimiter and encoding")
        for index, row in enumerate(reader, start=2):
            try:
                if index > 10001 or None in row or any(v is None for v in row.values()):
                    raise ValueError()
                values = {k: row[v].strip() for k, v in columns.items()}
                external_id = values.pop("external_id")
                if not external_id or len(external_id) > 120 or external_id in seen:
                    raise ValueError()
                seen.add(external_id)
                doctor = config["practitioners"].get(values["practitioner_id"])
                if not doctor or not re.fullmatch(r"[-a-z0-9]{3,60}", doctor):
                    raise ValueError()
                date = dt.datetime.strptime(values["date"], config.get("date_format", "%Y-%m-%d")).date()
                birth = dt.datetime.strptime(values["birthDate"], config.get("birth_date_format", "%Y-%m-%d")).date()
                clock = dt.datetime.strptime(values["time"], "%H:%M").time()
                phone = re.sub(r"[\s()-]", "", values["phone"])
                phone = "+213" + phone[1:] if phone.startswith("0") else phone
                now = dt.datetime.now(dt.timezone(dt.timedelta(hours=1)))
                if not re.fullmatch(r"\+213[567]\d{8}", phone) or not 2 <= len(values["name"]) <= 80 or birth > now.date() or birth.year < now.year - 130:
                    raise ValueError()
                # Import is for new future bookings, never a historical chart migration.
                if dt.datetime.combine(date, clock, now.tzinfo) <= now or date > now.date() + dt.timedelta(days=90):
                    raise ValueError()
                service = values.get("service", "consultation")
                service = config.get("services", {}).get(service, service)
                if service not in ("consultation", "controle"):
                    raise ValueError()
                payload = dict(practice_id=config["practice_id"], practitioner_id=doctor, name=values["name"], phone=phone,
                               birthDate=birth.isoformat(), date=date.isoformat(), time=clock.strftime("%H:%M"), service=service)
                rows.append((external_id, payload))
            except (ValueError, KeyError, TypeError):
                raise BridgeError(f"CSV row {index}: invalid date, birth date, phone, doctor, service or duplicate source ID; no rows were sent") from None
    return rows


def verify_event(raw, signature, secret, environment, resources, clock=None):
    if len(raw) > 65536 or not secret:
        raise BridgeError("Invalid webhook")
    match = re.fullmatch(r"t=(\d{1,12}),v1=([a-f0-9]{64})", signature or "")
    if not match or abs((time.time() if clock is None else clock) - int(match[1])) > 300:
        raise BridgeError("Invalid or expired signature")
    expected = hmac.new(secret.encode(), match[1].encode() + b"." + raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, match[2]):
        raise BridgeError("Invalid signature")
    try:
        event = json.loads(raw)
        resource = EVENTS[event["type"]]
        data = event["data"]
        if event["environment"] != environment or event["api_version"] != "1" or not valid_id(event["id"]) or not valid_id(data["id"]) or data[resource] not in resources:
            raise ValueError()
        if set(data) != {"id", "status", "revision", resource} or set(event) != {"id", "type", "created_at", "api_version", "environment", "data"}:
            raise ValueError()
        if not isinstance(data["status"], str) or not isinstance(data["revision"], (str, int)):
            raise ValueError()
        return event
    except (ValueError, KeyError, TypeError):
        raise BridgeError("Unexpected event, environment or resource") from None


def serve(state, config, secret, port):
    if not secret:
        raise BridgeError("Set DAWILO_WEBHOOK_SECRET to the complete per-key signing secret")
    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass  # Do not log request headers, URLs or patient data.
        def do_POST(self):
            try:
                length = int(self.headers.get("Content-Length", "0"))
                if self.path != "/webhooks/dawilo" or not 0 < length <= 65536 or self.headers.get("Transfer-Encoding"):
                    self.send_error(400); return
                self.connection.settimeout(10)
                raw = self.rfile.read(length)
                if len(raw) != length:
                    self.send_error(400); return
                event = verify_event(raw, self.headers.get("X-Dawilo-Signature"), secret, config["environment"], config.get("webhook_resources", [config["practice_id"]]))
                state.receive(event)  # Commit durable inbox before acknowledging.
                self.send_response(204); self.end_headers()
            except BridgeError:
                self.send_error(400)
            except (sqlite3.Error, OSError):
                self.send_error(503)
            except (ValueError, TimeoutError):
                self.send_error(400)
    print("Webhook inbox listening on loopback; forward HTTPS /webhooks/dawilo here. Database application remains pending until acknowledged.")
    HTTPServer(("127.0.0.1", port), Handler).serve_forever()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=["check", "import", "serve", "events", "ack"])
    parser.add_argument("--config", default="config.json")
    parser.add_argument("--csv")
    parser.add_argument("--state", default="private/dawilo-bridge.sqlite")
    parser.add_argument("--apply", action="store_true", help="send new appointments; otherwise preview only")
    parser.add_argument("--event-id", help="ack only after your local database transaction commits")
    parser.add_argument("--port", type=int, default=8790)
    args = parser.parse_args()
    config = validate_config(json.loads(Path(args.config).read_text(encoding="utf-8-sig")))
    if args.command == "import":
        if not args.csv:
            raise BridgeError("--csv is required")
        rows = read_rows(args.csv, config)  # Validate entire batch before sending anything.
        if not args.apply:
            print(f"{len(rows)} valid appointment rows. Preview only: nothing sent, no database changed. Slot availability is checked by the server when applying.")
            return
    key = os.environ.get("DAWILO_API_KEY", "")
    client = Client(config, key)
    if args.command == "check":
        result = client.call("practice-configuration?" + urllib.parse.urlencode({"practice_id": config["practice_id"]}))
        print(json.dumps(result, ensure_ascii=False, indent=2)); return
    state = State(args.state, {"origin": config["base_url"].rstrip("/"), "environment": config["environment"], "practice": config["practice_id"], "key": hashlib.sha256(key.encode()).hexdigest()})
    if args.command == "import":
        # A stopped batch is resumable. Never delete state or change IDs to fix an error.
        for index, (external_id, payload) in enumerate(rows, 1):
            print(f"row {index}: {state.import_one(external_id, payload, client)}")
    elif args.command == "serve":
        serve(state, config, os.environ.get("DAWILO_WEBHOOK_SECRET", ""), args.port)
    elif args.command == "events":
        for row in state.db.execute("SELECT payload FROM inbox WHERE processed_at IS NULL ORDER BY received_at, event_id"):
            print(row[0])
    elif args.command == "ack":
        if not valid_id(args.event_id):
            raise BridgeError("Provide the committed --event-id")
        with state.db:
            changed = state.db.execute("UPDATE inbox SET processed_at=? WHERE event_id=? AND processed_at IS NULL", (dt.datetime.now(dt.timezone.utc).isoformat(), args.event_id)).rowcount
        print(f"{changed} event acknowledged; event receipt retained")

if __name__ == "__main__":
    try:
        main()
    except (BridgeError, OSError, sqlite3.Error, ValueError, csv.Error) as error:
        print(str(error) if isinstance(error, BridgeError) else "Cannot read configuration/export or persist state. Nothing queued was discarded.", file=sys.stderr)
        sys.exit(1)