python3 verify.py person-a.jpg person-b.jpg
python3 represent.py face.jpg
# Optional explicit local export of sensitive embeddings (new file only):
python3 represent.py face.jpg embeddings.json
python3 compare.py vectors.json

"""Shared standard-library client. No automatic inference retries or payload logs."""
import base64
import json
import mimetypes
import math
import os
import re
import sys
import uuid
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener


class ExampleError(Exception):
    def __init__(self, code, **details):
        super().__init__(code)
        self.details = details


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def config(env=None):
    env = os.environ if env is None else env
    key = env.get("DEEPFACE_API_KEY", "").strip()
    if not key:
        raise ExampleError("missing_api_key")
    base = env.get("DEEPFACE_API_BASE_URL", "https://api.deepface.dev").rstrip("/")
    try:
        url = urlsplit(base)
        valid = url.hostname and (url.scheme == "https" or (url.scheme == "http" and url.hostname in ("localhost", "127.0.0.1", "::1")))
        if not valid or url.username or url.password or url.query or url.fragment:
            raise ValueError()
    except ValueError:
        raise ExampleError("invalid_api_base") from None
    return {"base": base, "key": key, "model": env.get("DEEPFACE_MODEL")}


def get_model(settings):
    try:
        with build_opener(NoRedirect()).open(settings["base"] + "/capabilities", timeout=15) as response:
            capabilities = json.load(response)
        model = settings["model"] or capabilities["default_model"]
        if model not in capabilities["allowed_models"]:
            raise ValueError()
        return model
    except Exception:
        raise ExampleError("capabilities_or_model_unavailable") from None


def image(filename):
    media_type = mimetypes.guess_type(filename)[0]
    if media_type not in ("image/jpeg", "image/png", "image/webp"):
        raise ExampleError("provide_local_jpeg_png_or_webp")
    return "data:" + media_type + ";base64," + base64.b64encode(Path(filename).read_bytes()).decode("ascii")


def finite(value):
    return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)


def integer(value):
    return isinstance(value, int) and not isinstance(value, bool)


def named(value):
    return isinstance(value, str) and bool(value.strip())


def valid_result(endpoint, data):
    if endpoint == "/verify":
        return isinstance(data, dict) and isinstance(data.get("verified"), bool) and finite(data.get("distance")) and finite(data.get("threshold"))
    if endpoint == "/represent":
        return isinstance(data, list) and bool(data) and all(isinstance(face, dict) and named(face.get("model_name")) and isinstance(face.get("embedding"), list) and bool(face["embedding"]) and all(finite(value) for value in face["embedding"]) for face in data)
    if endpoint != "/compare" or not isinstance(data, dict):
        return False
    threshold, source = data.get("threshold"), data.get("source")

    def valid_item(item):
        return isinstance(item, dict) and integer(item.get("index")) and item["index"] >= 0 and finite(item.get("distance")) and ("match" not in item if threshold is None else isinstance(item.get("match"), bool))

    return named(data.get("model_name")) and data.get("metric") in ("cosine", "euclidean", "euclidean_l2") and (threshold is None or finite(threshold)) and isinstance(source, dict) and source.get("type") in ("image", "vector") and integer(source.get("vector_length")) and source["vector_length"] > 0 and isinstance(data.get("results"), list) and bool(data["results"]) and all(valid_item(item) for item in data["results"]) and ("best_match" not in data or valid_item(data["best_match"]))


def compute(endpoint, payload, settings):
    request_id = str(uuid.uuid4())
    request = Request(settings["base"] + endpoint, data=json.dumps(payload, allow_nan=False).encode(), headers={"Authorization": "Bearer " + settings["key"], "Content-Type": "application/json", "x-request-id": request_id}, method="POST")
    try:
        try:
            response = build_opener(NoRedirect()).open(request, timeout=120)
        except HTTPError as error:
            response = error
        with response:
            status = response.code
            echoed = response.headers.get("x-request-id", "")
            if re.fullmatch(r"[a-zA-Z0-9_-]{1,128}", echoed):
                request_id = echoed
            data = json.load(response)
    except Exception:
        raise ExampleError("compute_outcome_unknown", request_id=request_id, retryable=False) from None
    if status == 202:
        raise ExampleError("compute_outcome_unknown", request_id=request_id, retryable=False)
    no_face = isinstance(data, dict) and data.get("outcome") == "no_face_detected"
    if status < 200 or status >= 300 or no_face:
        code = data.get("error") if isinstance(data, dict) else None
        code = code if isinstance(code, str) and re.fullmatch(r"[a-z0-9_]{1,80}", code) else "no_face_detected" if no_face else "http_" + str(status)
        raise ExampleError(code, status=status, request_id=request_id, retryable=False)
    valid = valid_result(endpoint, data)
    if not valid:
        raise ExampleError("compute_outcome_unknown", request_id=request_id, retryable=False)
    return {"request_id": request_id, "data": data}


def summarize(result):
    data = result["data"]
    summary = {"request_id": result["request_id"], "outcome": "completed"}
    if isinstance(data, list):
        return dict(summary, faces=len(data), dimensions=[len(face["embedding"]) for face in data])
    if isinstance(data.get("verified"), bool):
        return dict(summary, verified=data["verified"], **{key: data[key] for key in ("distance", "threshold") if isinstance(data.get(key), (int, float)) and not isinstance(data[key], bool) and math.isfinite(data[key])})
    summary.update(comparisons=len(data["results"]), match_evaluation="not_requested" if data.get("threshold") is None else "threshold_applied")
    if data.get("threshold") is not None:
        summary.update(threshold=data["threshold"], matches=sum(item.get("match") is True for item in data["results"]))
    return summary


def save_result(filename, data):
    with os.fdopen(os.open(filename, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as output:
        json.dump(data, output, allow_nan=False)


def run(main):
    try:
        main()
    except ExampleError as error:
        print(json.dumps(dict(error=str(error), **error.details)), file=sys.stderr)
        sys.exit(1)
    except Exception:
        print('{"error":"local_input_or_output_error"}', file=sys.stderr)
        sys.exit(1)

import json
import sys
from client import config, get_model, image, compute, summarize, run


def main():
    settings = config()
    payload = {"img1": image(sys.argv[1]), "img2": image(sys.argv[2]), "model_name": get_model(settings)}
    print(json.dumps(summarize(compute("/verify", payload, settings))))


if __name__ == "__main__":
    run(main)

import json
import sys
from client import config, get_model, image, compute, summarize, save_result, run


def main():
    settings = config()
    result = compute("/represent", {"img": image(sys.argv[1]), "model_name": get_model(settings)}, settings)
    print(json.dumps(summarize(result)))
    if len(sys.argv) > 2:
        save_result(sys.argv[2], result["data"])


if __name__ == "__main__":
    run(main)

import json
import sys
from pathlib import Path
from client import config, get_model, compute, summarize, run


def main():
    settings = config()
    source = json.loads(Path(sys.argv[1]).read_text())
    fields = ("source_vector", "target_vector", "target_vectors", "distance_metric", "threshold")
    payload = {key: source[key] for key in fields if key in source}
    payload["model_name"] = get_model(settings)
    print(json.dumps(summarize(compute("/compare", payload, settings))))


if __name__ == "__main__":
    run(main)