وصفات جاهزة

// Recipe 1. Loan file onboarding: ID, proof of address, payslip, then one verdict.
import { extract, postJson } from "./common.mjs";

const reference = "loan-0001";
const files = [
  { file: "cin-test.jpg", mime: "image/jpeg", docType: "national_id", stepKey: "photo_id" },
  { file: "bill-test.pdf", mime: "application/pdf", docType: "utility_bill", stepKey: "proof_of_address" },
  { file: "payslip-test.pdf", mime: "application/pdf", docType: "payslip" },
];

const reads = [];
for (const f of files) reads.push(await extract({ ...f, reference })); // identity first

const unread = reads.filter((r) => r.reader_unavailable);
if (unread.length) throw new Error("A file was not read. Retry later; no verdict was asked for.");

const fields = Object.assign({}, ...reads.map((r) => r.fields).reverse()); // first file wins
const verdict = await postJson("/v1/kyc/verify", {
  reference,
  environment: "sandbox",
  subject: `${fields.first_name} ${fields.last_name}`,
  kind: "individual",
  values: { ...fields, country: "MA" },
  documents: reads.flatMap((r) => r.documents), // unchanged
});

let decision;
if (!verdict.passed) decision = { route: "refuse_or_fix", reasons: verdict.critical_failures.map((c) => `${c.id}: ${c.detail}`) };
else if (verdict.flags.length) decision = { route: "human_review", reasons: verdict.flags.map((c) => c.id) };
else decision = { route: "auto_continue", reasons: [] };

console.log(decision, verdict.case_id);

# Recipe 1. Loan file onboarding: ID, proof of address, payslip, then one verdict.
import os
import requests

BASE = "https://app.sahlfinancial.com/api"
HEADERS = {"Authorization": f"Bearer {os.environ['SAHL_API_KEY']}"}
REFERENCE = "loan-0001"

FILES = [  # identity first
    ("cin-test.jpg", "image/jpeg", "national_id", "photo_id"),
    ("bill-test.pdf", "application/pdf", "utility_bill", "proof_of_address"),
    ("payslip-test.pdf", "application/pdf", "payslip", None),
]


def extract(path, mime, doc_type, step_key):
    data = {"doc_type": doc_type, "reference": REFERENCE, "environment": "sandbox", "kind": "individual"}
    if step_key:
        data["step_key"] = step_key
    with open(path, "rb") as f:
        res = requests.post(f"{BASE}/v1/kyc/extract", headers=HEADERS,
                            files=[("files", (path, f, mime))], data=data, timeout=120)
    res.raise_for_status()
    return res.json()


reads = [extract(*f) for f in FILES]
if any(r["reader_unavailable"] for r in reads):
    raise SystemExit("A file was not read. Retry later; no verdict was asked for.")

fields = {}
for r in reads:  # first file wins, as in the API's own merge
    for k, v in r["fields"].items():
        fields.setdefault(k, v)

res = requests.post(
    f"{BASE}/v1/kyc/verify", headers=HEADERS, timeout=60,
    json={
        "reference": REFERENCE,
        "environment": "sandbox",
        "subject": f"{fields['first_name']} {fields['last_name']}",
        "kind": "individual",
        "values": {**fields, "country": "MA"},
        "documents": [d for r in reads for d in r["documents"]],  # unchanged
    },
)
res.raise_for_status()
verdict = res.json()

if not verdict["passed"]:
    decision = ("refuse_or_fix", [f"{c['id']}: {c['detail']}" for c in verdict["critical_failures"]])
elif verdict["flags"]:
    decision = ("human_review", [c["id"] for c in verdict["flags"]])
else:
    decision = ("auto_continue", [])
print(decision, verdict["case_id"])

// Recipe 2. KYC for an account opening: application form + photo ID + your own duplicate check.
import { extract, postJson } from "./common.mjs";

const reference = "acct-0001";
const form = {                       // what the client typed in your application
  first_name: "Test", last_name: "Client", date_of_birth: "1988-04-12", citizenship: "MA",
  street1: "10 Rue Exemple", city: "Casablanca", province: "Casablanca-Settat", postal_code: "20000", country: "MA",
  phone: "+212600000000", email: "test.client@example.com",
  occupation: "Analyst", employer_name: "Test Employer SARL", annual_income: "84000",
  net_liquid_assets: "20000", total_net_worth: "60000", source_of_funds: "Employment income",
  objective: "Balanced", horizon: "5-10 years", investment_knowledge: "Good", investment_experience: "< 5 years",
  account_type: "Individual", third_party: "no", pep: "no",
};

const id = await extract({ file: "cin-test.jpg", mime: "image/jpeg", docType: "national_id", stepKey: "photo_id", reference });
if (id.reader_unavailable) throw new Error("The ID was not read. Retry later.");

// The ID you read is the source of truth for the identity fields.
const values = { ...form, ...pick(id.fields, ["first_name", "last_name", "date_of_birth", "id_type", "id_number", "id_expiry"]) };

const duplicate = await isDuplicateInMyDatabase(values); // your own lookup
const verdict = await postJson("/v1/kyc/verify", {
  reference, environment: "sandbox", subject: `${values.first_name} ${values.last_name}`, kind: "individual",
  values,
  documents: id.documents,
  extra_checks: [{
    id: "internal:duplicate_client", label: "No duplicate client in our database", severity: "critical",
    passed: !duplicate, detail: duplicate ? "matches an existing client" : "",
  }],
});

console.log("passed:", verdict.passed, "| completeness:", verdict.completeness.percent + "%");
console.log("critical:", verdict.critical_failures.map((c) => c.id));
console.log("flags:", verdict.flags.map((c) => c.id));
console.log("refused switches:", verdict.policy.overrides_refused);
console.log("store case_id:", verdict.case_id);

function pick(obj, keys) { return Object.fromEntries(keys.filter((k) => obj[k]).map((k) => [k, obj[k]])); }
async function isDuplicateInMyDatabase() { return false; }

# Recipe 2. KYC for an account opening: application form + photo ID + your own duplicate check.
import os
import requests

BASE = "https://app.sahlfinancial.com/api"
HEADERS = {"Authorization": f"Bearer {os.environ['SAHL_API_KEY']}"}
REFERENCE = "acct-0001"

form = {  # what the client typed in your application
    "first_name": "Test", "last_name": "Client", "date_of_birth": "1988-04-12", "citizenship": "MA",
    "street1": "10 Rue Exemple", "city": "Casablanca", "province": "Casablanca-Settat", "postal_code": "20000", "country": "MA",
    "phone": "+212600000000", "email": "test.client@example.com",
    "occupation": "Analyst", "employer_name": "Test Employer SARL", "annual_income": "84000",
    "net_liquid_assets": "20000", "total_net_worth": "60000", "source_of_funds": "Employment income",
    "objective": "Balanced", "horizon": "5-10 years", "investment_knowledge": "Good", "investment_experience": "< 5 years",
    "account_type": "Individual", "third_party": "no", "pep": "no",
}


def is_duplicate_in_my_database(values):  # your own lookup
    return False


with open("cin-test.jpg", "rb") as f:
    res = requests.post(
        f"{BASE}/v1/kyc/extract", headers=HEADERS, timeout=120,
        files=[("files", ("cin-test.jpg", f, "image/jpeg"))],
        data={"doc_type": "national_id", "step_key": "photo_id", "reference": REFERENCE,
              "environment": "sandbox", "kind": "individual"},
    )
res.raise_for_status()
id_read = res.json()
if id_read["reader_unavailable"]:
    raise SystemExit("The ID was not read. Retry later.")

# The ID you read is the source of truth for the identity fields.
identity_keys = ["first_name", "last_name", "date_of_birth", "id_type", "id_number", "id_expiry"]
values = {**form, **{k: id_read["fields"][k] for k in identity_keys if id_read["fields"].get(k)}}

duplicate = is_duplicate_in_my_database(values)
res = requests.post(
    f"{BASE}/v1/kyc/verify", headers=HEADERS, timeout=60,
    json={
        "reference": REFERENCE, "environment": "sandbox",
        "subject": f"{values['first_name']} {values['last_name']}", "kind": "individual",
        "values": values,
        "documents": id_read["documents"],
        "extra_checks": [{
            "id": "internal:duplicate_client", "label": "No duplicate client in our database",
            "severity": "critical", "passed": not duplicate,
            "detail": "matches an existing client" if duplicate else "",
        }],
    },
)
res.raise_for_status()
verdict = res.json()

print("passed:", verdict["passed"], "| completeness:", f"{verdict['completeness']['percent']}%")
print("critical:", [c["id"] for c in verdict["critical_failures"]])
print("flags:", [c["id"] for c in verdict["flags"]])
print("refused switches:", verdict["policy"]["overrides_refused"])
print("store case_id:", verdict["case_id"])

// Recipe 3. Does this payslip support the income the client declared?
// The API reads the payslip. The rules below are yours: the API sets no income rule.
import { extract, postJson } from "./common.mjs";

const reference = "inc-0001";
const applicant = { first_name: "Test", last_name: "Client", employer_name: "Test Employer SARL", annual_income: 84000 };

const read = await extract({ file: "payslip-test.pdf", mime: "application/pdf", docType: "payslip", reference });
if (read.reader_unavailable) throw new Error("The payslip was not read. Retry later.");
const doc = read.documents[0];
const f = doc.fields;

const key = (s) => String(s ?? "").normalize("NFKD").replace(/[^a-z ]/gi, "").toLowerCase().split(/\s+/).filter(Boolean).sort().join(" ");
const findings = [];

if (key(f.document_holder_name || `${f.first_name} ${f.last_name}`) !== key(`${applicant.first_name} ${applicant.last_name}`)) {
  findings.push("holder_does_not_match_applicant");
}
if (key(f.employer_name) !== key(applicant.employer_name)) findings.push("employer_does_not_match");

// Payslip recency is not checked by the API (only address documents are). Your rule: 90 days.
const ageDays = f.document_date ? (Date.now() - Date.parse(f.document_date)) / 86_400_000 : Infinity;
if (!(ageDays <= 90)) findings.push(f.document_date ? "payslip_older_than_90_days" : "payslip_date_not_read");

// Income: only when the payslip states it. Otherwise the API has nothing to compare.
if (f.annual_income) {
  const stated = Number(f.annual_income);
  if (Math.abs(stated - applicant.annual_income) / applicant.annual_income > 0.1) findings.push("income_differs_by_more_than_10_percent");
} else {
  findings.push("income_not_stated_on_payslip");
}

// File-level signals the API already computed.
for (const c of read.checks) if (!c.passed && c.severity !== "info") findings.push(c.id);

// Ask for the capacity band the declared income gives, without any document check.
const { assessment } = await postJson("/v1/kyc/assess", {
  reference, environment: "sandbox", kind: "individual", require_documents: false,
  values: { first_name: applicant.first_name, last_name: applicant.last_name, annual_income: String(applicant.annual_income) },
  documents: [],
});

console.log({ findings, capacity: assessment.capacity });

# Recipe 3. Does this payslip support the income the client declared?
# The API reads the payslip. The rules below are yours: the API sets no income rule.
import os
import re
import unicodedata
from datetime import date

import requests

BASE = "https://app.sahlfinancial.com/api"
HEADERS = {"Authorization": f"Bearer {os.environ['SAHL_API_KEY']}"}
REFERENCE = "inc-0001"
applicant = {"first_name": "Test", "last_name": "Client", "employer_name": "Test Employer SARL", "annual_income": 84000}

with open("payslip-test.pdf", "rb") as fh:
    res = requests.post(
        f"{BASE}/v1/kyc/extract", headers=HEADERS, timeout=120,
        files=[("files", ("payslip-test.pdf", fh, "application/pdf"))],
        data={"doc_type": "payslip", "reference": REFERENCE, "environment": "sandbox", "kind": "individual"},
    )
res.raise_for_status()
read = res.json()
if read["reader_unavailable"]:
    raise SystemExit("The payslip was not read. Retry later.")
f = read["documents"][0]["fields"]


def key(s):
    s = unicodedata.normalize("NFKD", str(s or ""))
    return " ".join(sorted(re.sub(r"[^a-z ]", "", s.lower()).split()))


findings = []
if key(f.get("document_holder_name") or f"{f.get('first_name')} {f.get('last_name')}") != key(
    f"{applicant['first_name']} {applicant['last_name']}"
):
    findings.append("holder_does_not_match_applicant")
if key(f.get("employer_name")) != key(applicant["employer_name"]):
    findings.append("employer_does_not_match")

# Payslip recency is not checked by the API (only address documents are). Your rule: 90 days.
if f.get("document_date"):
    if (date.today() - date.fromisoformat(f["document_date"])).days > 90:
        findings.append("payslip_older_than_90_days")
else:
    findings.append("payslip_date_not_read")

# Income: only when the payslip states it. Otherwise the API has nothing to compare.
if f.get("annual_income"):
    stated = float(f["annual_income"])
    if abs(stated - applicant["annual_income"]) / applicant["annual_income"] > 0.1:
        findings.append("income_differs_by_more_than_10_percent")
else:
    findings.append("income_not_stated_on_payslip")

# File-level signals the API already computed.
findings += [c["id"] for c in read["checks"] if not c["passed"] and c["severity"] != "info"]

res = requests.post(
    f"{BASE}/v1/kyc/assess", headers=HEADERS, timeout=60,
    json={
        "reference": REFERENCE, "environment": "sandbox", "kind": "individual", "require_documents": False,
        "values": {"first_name": applicant["first_name"], "last_name": applicant["last_name"],
                   "annual_income": str(applicant["annual_income"])},
        "documents": [],
    },
)
res.raise_for_status()
print({"findings": findings, "capacity": res.json()["assessment"]["capacity"]})

// Recipe 4. eID check: start, poll until the client finishes, keep the PDF, then verify.
import { writeFile } from "node:fs/promises";
import { sahl, postJson } from "./common.mjs";

const reference = "eid-0001";
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// 1. Start. The client is emailed now, even in sandbox. Use an address you control.
const started = await postJson("/v1/kyc/eid", {
  reference, first_name: "Test", last_name: "Client", email: "you@your-domain.example",
  country: "CA", language: "en", documents: 1, environment: "sandbox",
});
console.log("eID request", started.key, "started for", started.reference);

// 2. Poll: every 30 s for 10 minutes, then every 5 minutes, for up to 24 hours.
const deadline = Date.now() + 24 * 3600_000;
let result;
for (let n = 0; Date.now() < deadline; n++) {
  result = await (await sahl(`/v1/kyc/eid/${started.key}?environment=sandbox`)).json();
  if (result.complete) break;
  await sleep(n < 20 ? 30_000 : 300_000);
}
if (!result?.complete) throw new Error("Not completed in 24 hours. Start a new request if needed.");

console.log("passed:", result.passed);
for (const c of result.checks.filter((c) => !c.passed)) console.log(" failed:", c.id, c.detail);

// 3. Keep the report within about seven days: the provider then deletes the personal details.
const pdf = Buffer.from(await (await sahl(`/v1/kyc/eid/${started.key}/report`)).arrayBuffer());
await writeFile(`eid-${started.key}.pdf`, pdf);

// 4. Verify with the SAME reference and environment, so a policy eID requirement can be met.
const verdict = await postJson("/v1/kyc/verify", {
  reference, environment: "sandbox", kind: "individual", require_documents: false,
  values: { first_name: "Test", last_name: "Client", country: "CA" }, documents: [],
});
console.log("verify passed:", verdict.passed, verdict.critical_failures.map((c) => c.id));

# Recipe 4. eID check: start, poll until the client finishes, keep the PDF, then verify.
import os
import time

import requests

BASE = "https://app.sahlfinancial.com/api"
HEADERS = {"Authorization": f"Bearer {os.environ['SAHL_API_KEY']}"}
REFERENCE = "eid-0001"

# 1. Start. The client is emailed now, even in sandbox. Use an address you control.
res = requests.post(
    f"{BASE}/v1/kyc/eid", headers=HEADERS, timeout=60,
    json={"reference": REFERENCE, "first_name": "Test", "last_name": "Client",
          "email": "you@your-domain.example", "country": "CA", "language": "en",
          "documents": 1, "environment": "sandbox"},
)
res.raise_for_status()
key = res.json()["key"]
print("eID request", key, "started for", REFERENCE)

# 2. Poll: every 30 s for 10 minutes, then every 5 minutes, for up to 24 hours.
deadline = time.time() + 24 * 3600
result, n = None, 0
while time.time() < deadline:
    r = requests.get(f"{BASE}/v1/kyc/eid/{key}", params={"environment": "sandbox"}, headers=HEADERS, timeout=60)
    r.raise_for_status()
    result = r.json()
    if result["complete"]:
        break
    time.sleep(30 if n < 20 else 300)
    n += 1
if not result or not result["complete"]:
    raise SystemExit("Not completed in 24 hours. Start a new request if needed.")

print("passed:", result["passed"])
for c in result["checks"]:
    if not c["passed"]:
        print(" failed:", c["id"], c["detail"])

# 3. Keep the report within about seven days: the provider then deletes the personal details.
pdf = requests.get(f"{BASE}/v1/kyc/eid/{key}/report", headers=HEADERS, timeout=60)
pdf.raise_for_status()
with open(f"eid-{key}.pdf", "wb") as fh:
    fh.write(pdf.content)

# 4. Verify with the SAME reference and environment, so a policy eID requirement can be met.
res = requests.post(
    f"{BASE}/v1/kyc/verify", headers=HEADERS, timeout=60,
    json={"reference": REFERENCE, "environment": "sandbox", "kind": "individual", "require_documents": False,
          "values": {"first_name": "Test", "last_name": "Client", "country": "CA"}, "documents": []},
)
res.raise_for_status()
verdict = res.json()
print("verify passed:", verdict["passed"], [c["id"] for c in verdict["critical_failures"]])

// Recipe 5. A field was not read. The API has no per-field confidence, so look at what is absent.
import { extract, postJson } from "./common.mjs";

// What the checks need per document type (see "Document types and fields").
const EXPECTED = {
  passport: ["first_name", "last_name", "date_of_birth", "id_number"],
  national_id: ["first_name", "last_name", "id_number"],
  drivers_license: ["first_name", "last_name", "id_number"],
  utility_bill: ["street1", "city", "postal_code", "document_holder_name"],
  bank_statement: ["bank_name", "document_holder_name"],
};
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// Retry only when the file was never read (reader_unavailable). A blank read is not retried: send a better file.
async function readWithRetry(args, tries = 3) {
  for (let i = 0; i < tries; i++) {
    const read = await extract(args);
    if (!read.reader_unavailable) return read;
    await sleep(2000 * 2 ** i); // 2 s, 4 s, 8 s
  }
  return null;
}

function problems(read) {
  const doc = read.documents[0];
  const out = [];
  for (const k of EXPECTED[doc.doc_type] ?? []) if (!doc.fields[k]) out.push({ field: k, ask: "type_it_or_rescan" });
  for (const c of read.checks) {
    if (c.passed || c.severity === "info") continue;
    if (c.id.startsWith("doctype:")) out.push({ check: c.id, ask: "upload_the_right_document", detail: c.detail });
    else if (c.id.startsWith("legible:")) out.push({ check: c.id, ask: "better_scan", detail: c.detail });
    else if (c.id.startsWith("expiry:")) out.push({ check: c.id, ask: "valid_document", detail: c.detail });
    else out.push({ check: c.id, ask: "review", detail: c.detail });
  }
  return out;
}

const reference = "fix-0001";
const read = await readWithRetry({ file: "cin-test.jpg", mime: "image/jpeg", docType: "national_id", stepKey: "photo_id", reference });
if (!read) {
  console.log("Reader not available after 3 tries. Queue the file and tell the client it is being processed.");
} else {
  const todo = problems(read);
  console.log(todo.length ? { needs_attention: todo } : "all expected fields read");
  // What the person types overrides what was read. Send it in `values`; the entries go back unchanged.
  const typed = { id_number: "BK123456" }; // from your form, only for the fields in `todo`
  const verdict = await postJson("/v1/kyc/verify", {
    reference, environment: "sandbox", kind: "individual",
    values: { ...read.fields, ...typed }, documents: read.documents,
  });
  console.log("passed:", verdict.passed);
}

# Recipe 5. A field was not read. The API has no per-field confidence, so look at what is absent.
import os
import time

import requests

BASE = "https://app.sahlfinancial.com/api"
HEADERS = {"Authorization": f"Bearer {os.environ['SAHL_API_KEY']}"}

# What the checks need per document type (see "Document types and fields").
EXPECTED = {
    "passport": ["first_name", "last_name", "date_of_birth", "id_number"],
    "national_id": ["first_name", "last_name", "id_number"],
    "drivers_license": ["first_name", "last_name", "id_number"],
    "utility_bill": ["street1", "city", "postal_code", "document_holder_name"],
    "bank_statement": ["bank_name", "document_holder_name"],
}


def read_with_retry(path, mime, doc_type, step_key, reference, tries=3):
    """Retry only when the file was never read (reader_unavailable). A blank read is not retried."""
    for i in range(tries):
        with open(path, "rb") as f:
            res = requests.post(
                f"{BASE}/v1/kyc/extract", headers=HEADERS, timeout=120,
                files=[("files", (path, f, mime))],
                data={"doc_type": doc_type, "step_key": step_key, "reference": reference,
                      "environment": "sandbox", "kind": "individual"},
            )
        res.raise_for_status()
        read = res.json()
        if not read["reader_unavailable"]:
            return read
        time.sleep(2 * 2 ** i)  # 2 s, 4 s, 8 s
    return None


def problems(read):
    doc = read["documents"][0]
    out = [{"field": k, "ask": "type_it_or_rescan"} for k in EXPECTED.get(doc["doc_type"], []) if not doc["fields"].get(k)]
    for c in read["checks"]:
        if c["passed"] or c["severity"] == "info":
            continue
        if c["id"].startswith("doctype:"):
            ask = "upload_the_right_document"
        elif c["id"].startswith("legible:"):
            ask = "better_scan"
        elif c["id"].startswith("expiry:"):
            ask = "valid_document"
        else:
            ask = "review"
        out.append({"check": c["id"], "ask": ask, "detail": c["detail"]})
    return out


REFERENCE = "fix-0001"
read = read_with_retry("cin-test.jpg", "image/jpeg", "national_id", "photo_id", REFERENCE)
if read is None:
    print("Reader not available after 3 tries. Queue the file and tell the client it is being processed.")
else:
    todo = problems(read)
    print({"needs_attention": todo} if todo else "all expected fields read")
    typed = {"id_number": "BK123456"}  # from your form, only for the fields in `todo`
    res = requests.post(
        f"{BASE}/v1/kyc/verify", headers=HEADERS, timeout=60,
        json={"reference": REFERENCE, "environment": "sandbox", "kind": "individual",
              "values": {**read["fields"], **typed}, "documents": read["documents"]},
    )
    res.raise_for_status()
    print("passed:", res.json()["passed"])