ОБНОВЕНО · 03.10.2026

Ключове и карти за достъп: заявка, оценка на риска и одобрение от човек

Строим на локален AI сървър от класа NVIDIA GB10 регистър за ключове и карти за достъп: заявката се записва, малка услуга ѝ дава оценка от 0 до 100, а n8n решава кой трябва да я одобри. Оценката е съвет към човек, не присъда за човек, и всичко остава на машината.

⏱ 1,5–2 ч Напреднали GX10 NVIDIA GB10 · ARM64 n8n · PostgreSQL · FastAPI · Docker

Docker · n8n · PostgreSQL🔒 локално FastAPI + asyncpg (скорът)🔒 локално SMTP за известията🌐 глобален

🔄

ОБНОВЕНО · 03.10.2026 — какво

Урокът е преработен. Махнахме: примерната фирма с „40 обекта“ и „120 служители“, твърдението за наредба и срок за съхранение (не са сверени и зависят от обекта), автоматичния отказ при висок риск, остарелия възел Cron (сега Schedule Trigger), паролата във връзката към базата, услугата, слушаща на 0.0.0.0, и „работния процес като JSON“ — той беше непълен и не можеше да се внесе. Поправихме: часът се взема в местно време, а не в UTC; изгледът с просрочените ключове не повтаря напомнянията; липсващата връзка към базата в кода. Добавихме: защита на журнала срещу промяна, Wait възел с ограничение във времето, отделна услуга за скора във вътрешната мрежа на Docker, QR кодове само с код на ключ и кутия за защита на личните данни. Работният процес в n8n описваме по възли — внасяй го на ръка и го тествай.

⚠️

Какво не сме пускали сами

Нямахме машина от класа GB10. Нито един SQL, контейнер или работен процес от този урок не е пускан. Единствено функцията за скора е изпълнена с примерни стойности на обикновен компютър. Затова няма етикет „ТЕСТВАНО“. Особено непроверени: имената на полетата в изхода на Wait възела и работата на asyncpg и uvicorn на ARM64.

01Какво ще научиш

Как да опишеш ключове, зони и заявки в PostgreSQL и да направиш журнала „само за добавяне“.
Как да сметнеш оценка на риска от пет фактора и защо часът трябва да е в местно време.
Как n8n да пусне заявката по път според оценката и зоната, без автоматичен отказ.
Как да спреш работния процес, докато човек отговори (възел Wait), и какво става, ако не отговори.
Как да пращаш напомняния за невърнати ключове, без да се повтарят.
Как да пазиш малко лични данни: псевдоним вместо име и QR код само с код на ключ.

02Преди да започнеш

Работещ стек с n8n и Postgres в Docker Compose — по урока за n8n.
Основни познания по SQL и Python.
Достъп до пощенски сървър (SMTP) за известията. Това е единственото нещо, което може да е извън машината — 🌐 глобален, ако ползваш външен пощенски доставчик.
Само измислени данни. Не вкарвай реални ключове, хора или обекти, докато не мине правната проверка.

⚖️

Кой е взел кой ключ и кога е лична информация

Журналът за ключове и карти е обработване на лични данни на служители. Основание, информиране на хората, срок на съхранение, кой чете журнала и правата на засегнатите зависят от организацията и обекта. Преди да използваш системата с реални ключове и хора, провери с юрист или с длъжностното лице по защита на данните. Този урок е технически и не е правна консултация.

03Стъпки

Какво ще стои къде

Защо на части? Така всяка част се проверява отделно: базата пази истината, малката услуга смята, n8n организира хората.

Част	За какво е	Достъп
PostgreSQL, база `keys`	Зони, ключове, заявки, журнал	вътрешна
scorer (FastAPI)	Смята оценката и пътя на заявката	вътрешна, `http://scorer:8194`
n8n	Приема заявки, изчаква одобрение, праща напомняния	`127.0.0.1:5678` (както в урока за n8n)

Базата: таблици, журнал и изглед

Сложи схемата във файл schema.sql. Журналът е защитен с тригер: UPDATE и DELETE дават грешка. Защо? Журналът е доказателство какво се е случило; ако може да се пренаписва, не доказва нищо. Допълнително махни тези права от ролите на приложенията; администраторът на базата пак може да го промени, затова правиш и копия.

sql · schema.sql

-- база „keys“ (PostgreSQL)
CREATE TABLE IF NOT EXISTS zones (
    zone_code   text PRIMARY KEY,
    site_code   text NOT NULL,
    zone_name   text,
    sensitivity smallint NOT NULL DEFAULT 1 CHECK (sensitivity BETWEEN 0 AND 3)
);  -- 0 PUBLIC, 1 STANDARD, 2 RESTRICTED, 3 CRITICAL

CREATE TABLE IF NOT EXISTS staff (
    employee_ref text PRIMARY KEY,   -- псевдоним, не име
    role         text NOT NULL DEFAULT 'staff'
);

CREATE TABLE IF NOT EXISTS keys (
    id        serial PRIMARY KEY,
    key_code  text NOT NULL UNIQUE,  -- стойността в QR кода
    key_type  text NOT NULL CHECK (key_type IN ('physical', 'card')),
    zone_code text NOT NULL REFERENCES zones(zone_code),
    active    boolean NOT NULL DEFAULT true
);

CREATE TABLE IF NOT EXISTS key_requests (
    id              serial PRIMARY KEY,
    employee_ref    text NOT NULL,
    key_id          int  NOT NULL REFERENCES keys(id),
    reason          text NOT NULL,
    expected_return timestamptz NOT NULL,
    status          text NOT NULL DEFAULT 'PENDING'
        CHECK (status IN ('PENDING','SCORED','APPROVED','REJECTED','ISSUED','RETURNED','LOST')),
    risk_score      smallint,
    risk_class      text,
    issued_at       timestamptz,
    returned_at     timestamptz,
    created_at      timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_kr_employee ON key_requests (employee_ref);
CREATE INDEX IF NOT EXISTS idx_kr_status   ON key_requests (status);

CREATE TABLE IF NOT EXISTS audit_log (
    id         bigserial PRIMARY KEY,
    request_id int REFERENCES key_requests(id),
    event_type text NOT NULL,
    actor      text NOT NULL,        -- 'system' или псевдоним
    details    jsonb,
    event_at   timestamptz NOT NULL DEFAULT now()
);

-- записите само се добавят: UPDATE и DELETE се блокират
CREATE OR REPLACE FUNCTION audit_log_append_only() RETURNS trigger AS $$
BEGIN
    RAISE EXCEPTION 'audit_log is append-only';
END;
$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS audit_log_no_change ON audit_log;
CREATE TRIGGER audit_log_no_change
    BEFORE UPDATE OR DELETE ON audit_log
    FOR EACH ROW EXECUTE FUNCTION audit_log_append_only();

-- издадени ключове с изтекъл срок за връщане
CREATE OR REPLACE VIEW v_overdue_keys AS
SELECT r.id AS request_id, r.employee_ref, k.key_code, z.zone_name, z.sensitivity,
       EXTRACT(EPOCH FROM (now() - r.expected_return)) / 60 AS overdue_min
FROM key_requests r
JOIN keys  k ON k.id = r.key_id
JOIN zones z ON z.zone_code = k.zone_code
WHERE r.status = 'ISSUED' AND r.expected_return < now();

Идентификаторът на служителя е псевдоним (employee_ref), не име. Съответствието „псевдоним — човек“ се пази отделно от тази система.

Скорът: чисти функции, които се тестват без база

Първо логиката без база — така се проверява с обикновен компютър. Оценката е сума от пет фактора; числата са учебни, не стандарт. Заменяй ги със свои, след като поговориш с хората, които отговарят за обекта.

Фактор	Макс.	Правило
Час на деня (местно време)	25	07–19 ч = 0; 19–22 ч = 10; 22–07 ч = 25
Чувствителност на зоната	30	0 / 5 / 15 / 30 за PUBLIC / STANDARD / RESTRICTED / CRITICAL
Роля	15	директор 0 · ръководител на обект 5 · старши 10 · служител и временен 15
Закъснения за 90 дни	20	0 % = 0; под 10 % = 5; под 25 % = 12; от 25 % = 20; без история = 5
Активни ключове у служителя	10	0–1 = 0; 2–3 = 5; 4 и повече = 10

Класове: LOW под 30, MEDIUM 30–69, HIGH от 70. Пътят: LOW → автоматично; MEDIUM и HIGH → ръководител (при HIGH се уведомява и директор); зона CRITICAL → двама различни ръководители, независимо от оценката.

python · scorer/scoring.py

from datetime import datetime
from zoneinfo import ZoneInfo

ROLE_POINTS = {"director": 0, "site_manager": 5, "senior": 10, "staff": 15, "temporary": 15}
ZONE_POINTS = {0: 0, 1: 5, 2: 15, 3: 30}

def time_points(local_time: datetime) -> int:
    h = local_time.hour
    if 7 <= h < 19:
        return 0
    if 19 <= h < 22:
        return 10
    return 25          # 22:00-07:00

def history_points(late: int, total: int) -> int:
    if total == 0:
        return 5       # без история: малък принос
    rate = late / total
    if rate == 0:
        return 0
    if rate < 0.10:
        return 5
    if rate < 0.25:
        return 12
    return 20

def active_points(active: int) -> int:
    return 0 if active <= 1 else 5 if active <= 3 else 10

def score(local_time, sensitivity, role, late, total, active):
    f = {
        "time": time_points(local_time),
        "zone": ZONE_POINTS.get(sensitivity, 5),
        "role": ROLE_POINTS.get(role, 15),
        "history": history_points(late, total),
        "active": active_points(active),
    }
    pts = min(100, sum(f.values()))
    cls = "LOW" if pts < 30 else "MEDIUM" if pts < 70 else "HIGH"
    return pts, cls, f

def route(cls: str, sensitivity: int) -> str:
    if sensitivity >= 3:
        return "TWO_MANAGERS"      # КРИТИЧНА зона: никога автоматично
    return "AUTO" if cls == "LOW" else "MANAGER"

Провери с три ръчно измислени случая (пример: нощ + критична зона + временен служител → HIGH; ден + обикновена зона + старши → LOW). Така научаваш дали правилата правят това, което искаш, още преди базата.

✅

Няма автоматичен отказ

В стария вариант висок скор отхвърляше заявката сам. Тук не. Отказът е решение за човек — оставяме го на човек, а моделът само подрежда реда и подготвя обяснението.

Услугата за скора (FastAPI + asyncpg)

Малката услуга чете заявката от базата, смята оценката и записва резултата и журнала в една транзакция. Адресът на базата идва от променлива на средата — не се пише в кода.

python · scorer/key_risk_scorer.py

import json
import os
from datetime import datetime
from zoneinfo import ZoneInfo

import asyncpg
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from scoring import score, route

app = FastAPI(title="Key risk scorer")
DATABASE_URL = os.environ["DATABASE_URL"]
LOCAL_TZ = ZoneInfo(os.environ.get("TZ", "Europe/Sofia"))


class ScoreIn(BaseModel):
    request_id: int


@app.post("/score")
async def score_request(body: ScoreIn):
    conn = await asyncpg.connect(DATABASE_URL)
    try:
        req = await conn.fetchrow(
            """SELECT r.employee_ref, z.sensitivity, COALESCE(s.role, 'staff') AS role
               FROM key_requests r
               JOIN keys  k ON k.id = r.key_id
               JOIN zones z ON z.zone_code = k.zone_code
               LEFT JOIN staff s ON s.employee_ref = r.employee_ref
               WHERE r.id = $1""", body.request_id)
        if req is None:
            raise HTTPException(status_code=404, detail="request not found")
        hist = await conn.fetchrow(
            """SELECT count(*) FILTER (WHERE returned_at - expected_return > interval '30 minutes') AS late,
                      count(*) AS total
               FROM key_requests
               WHERE employee_ref = $1 AND status = 'RETURNED'
                 AND created_at > now() - interval '90 days'""", req["employee_ref"])
        active = await conn.fetchval(
            "SELECT count(*) FROM key_requests WHERE employee_ref = $1 AND status = 'ISSUED'",
            req["employee_ref"])
        pts, cls, factors = score(datetime.now(LOCAL_TZ), req["sensitivity"], req["role"],
                                  hist["late"], hist["total"], active)
        path = route(cls, req["sensitivity"])
        async with conn.transaction():
            await conn.execute(
                "UPDATE key_requests SET risk_score = $1, risk_class = $2, status = 'SCORED' WHERE id = $3",
                pts, cls, body.request_id)
            await conn.execute(
                "INSERT INTO audit_log (request_id, event_type, actor, details) "
                "VALUES ($1, 'SCORED', 'system', $2::jsonb)",
                body.request_id, json.dumps({"score": pts, "class": cls, "route": path, "factors": factors}))
    finally:
        await conn.close()
    return {"score": pts, "risk_class": cls, "route": path, "factors": factors,
            "notify_director": cls == "HIGH"}

Пускане като услуга в същия Compose

Защо не на машината направо? n8n работи в контейнер; за него „localhost“ е самият контейнер. Ако и скорът е услуга в същия проект, n8n го вика по име и портът не се публикува към мрежата. Версиите на пакетите не са закачени; закачи ги в Dockerfile по текущите издания (към 03.10.2026: FastAPI 0.142.2, uvicorn 0.54.0, asyncpg 0.31.0 — виж PyPI).

dockerfile · scorer/Dockerfile

FROM python:3.13-slim
WORKDIR /app
RUN pip install --no-cache-dir fastapi "uvicorn[standard]" asyncpg
COPY scoring.py key_risk_scorer.py ./
CMD ["uvicorn", "key_risk_scorer:app", "--host", "0.0.0.0", "--port", "8194"]

yaml · добави към compose.yaml (в services)

  scorer:
    build: ./scorer
    restart: unless-stopped
    depends_on:
      postgres:
        condition: service_healthy
    environment:
      DATABASE_URL: postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@postgres:5432/keys
      TZ: ${TZ}
    # без „ports“: достига се само от другите контейнери (n8n) като http://scorer:8194

bash

# 1) отделна база и таблиците (от папката на стека с n8n)
docker compose exec postgres sh -c 'createdb -U "$POSTGRES_USER" keys'
docker compose exec -T postgres sh -c 'psql -U "$POSTGRES_USER" -d keys' < schema.sql

# 2) услугата за скор
mkdir -p scorer   # сложи в нея scoring.py, key_risk_scorer.py и Dockerfile
docker compose up -d --build scorer

# 3) проба от вътрешността на контейнера на n8n (първо вмъкни тестова заявка)
docker compose exec n8n wget -qO- --post-data='{"request_id":1}' \
  --header='Content-Type: application/json' http://scorer:8194/score

Не сме пускали тези команди. Ако нещо не стартира, виж docker compose logs scorer.

Работен процес 1: заявка → оценка → път

Създай в n8n работен процес с възлите по-долу. Описваме го по възли, защото внесеният JSON лесно се чупи при смяна на версията.

№	Възел	Настройки
1	Webhook	POST, път `key-request`, удостоверяване с Header Auth (поле: служител, ключ, причина, очакван час на връщане)
2	Postgres	Execute Query: вмъкване в `key_requests` с `RETURNING id`; стойностите — като параметри на заявката, не залепени в текст
3	HTTP Request	POST `http://scorer:8194/score`, тяло `{"request_id": id}`
4	Switch	по полето `route`: AUTO · MANAGER · TWO_MANAGERS
5 · AUTO	Postgres + Send Email	статус APPROVED, запис в журнала (actor = system), известие до дежурния за издаване
5 · MANAGER	Send Email → Wait	писмо с оценката, разбивката по фактори и два адреса: `{{ $execution.resumeUrl }}?action=approve` и `?action=reject`
5 · TWO_MANAGERS	два пъти Send Email → Wait	два Wait възела един след друг, всеки със свой `Webhook Suffix` и друг ръководител; одобрението на двамата е нужно
6	IF	по `action`; липса на отговор → клон „ескалация към директор“
7	Postgres	запис в журнала: кой, какво, кога

Wait възелът: „Resume“ = On Webhook Call, метод GET (щракване върху връзка), „Limit Wait Time“ = 2 часа, „Ignore Bots“ = включено (иначе предварителният преглед на връзката в пощата може да „отговори“ вместо човек). Адресът за продължаване е уникален за всяко изпълнение (по документацията на n8n); все пак се отнасяй към писмото като към чувствително. След изтичане на времето работният процес продължава без отговор — затова клонът за ескалация е задължителен. ⚠️ Точното име на полето с избраното действие в изхода на Wait възела не сме го проверявали; виж изхода при тест.

Работен процес 2: напомняния за невърнати ключове

Schedule Trigger на всеки 30 минути (остарелият възел Cron вече не е препоръчан път; документацията на n8n описва Schedule Trigger). Работният процес трябва да е публикуван, иначе графикът не върви. Заявката по-долу пропуска ниво, за което вече има запис в журнала, и затова напомнянията не се повтарят.
sql · заявка на възела Postgres
```
SELECT * FROM (
    SELECT request_id, zone_name, sensitivity, round(overdue_min) AS overdue_min,
           CASE WHEN overdue_min >= 90 THEN 3 WHEN overdue_min >= 60 THEN 2 ELSE 1 END AS level
    FROM v_overdue_keys
    WHERE overdue_min >= 30
) t
WHERE NOT EXISTS (
    SELECT 1 FROM audit_log a
    WHERE a.request_id = t.request_id
      AND a.event_type = 'REMINDER_SENT'
      AND (a.details->>'level')::int >= t.level
);
```
Switch по level: 1 — напомняне до служителя (30 мин.), 2 — до служителя и ръководителя (60 мин.), 3 — ескалация до директора (90 мин.). След всяко писмо — запис в журнала със event_type = REMINDER_SENT и details = {"level": N}.
Работен процес 3: сканиране на QR кода

Webhook key-scan (POST, Header Auth) получава key_code и действие issue или return, обновява статуса (ISSUED / RETURNED) и часа, и записва в журнала. QR кодовете съдържат само кода на ключа — никога име.
python · qr.py
```
import qrcode

for code in ["KEY-0001", "KEY-0002", "CARD-0001"]:
    qrcode.make(code).save(f"{code}.png")   # само кодът на ключа, никога име
```
Нужни са pip install "qrcode[pil]" (версия 8.2 по PyPI към 03.10.2026) и Pillow за PNG.
Сигурност — минимумът
- Удостоверяване на двата уеб адреса за входа (Header Auth); не ги отваряй към интернет.
- Услугата за скора няма публикуван порт.
- Паролите са генерирани и стоят в .env с права 600, не в кода, не в урока.
- Журналът е защитен с тригер, правата за промяна са махнати, а копията на базата се правят редовно (виж урока за n8n).
- Пази минимум лични данни и определи срок на съхранение заедно с юриста.

05Какво следва

Серията · GX10

GX10: всички уроци

Индексът на серията: други задачи за локалния AI сървър — работни процеси, документи, търсене и още.

Към индекса → Стълбата

Бърз експеримент

Пробваме една твоя задача пред теб и ти оставяме готов помощник за нея. Излизаш с резултат.

Виж Стълбата

06Източници

n8n: възел Wait — продължаване с извикване на уеб адрес, ограничение във времето, „Ignore Bots“, Webhook Suffix, $execution.resumeUrl.
n8n: Schedule Trigger — интервали, публикуване, часова зона.
n8n: възел Webhook — удостоверяване.
FastAPI · asyncpg 🔒 локално.
PostgreSQL: тригери в PL/pgSQL.
PyPI: fastapi · uvicorn · asyncpg · qrcode — версии.
Регламент (ЕС) 2016/679 (GDPR) — официален текст; не е правна консултация.