Nativní integrace AI do Heliosu: možná architektura a best practices

Helios API: Co máme k dispozici

Helios Orange/Red nabízí několik způsobů integrace:

1. REST API (Helios Orange 7+)

Dostupné operace:

• ✅ Čtení entit (faktury, objednávky, sklady...)
• ✅ Vytváření záznamů
• ✅ Aktualizace
• ⚠️ Komplexní business logika (omezená)

Příklad:

GET /api/v1/invoices/issued?date_from=2025-01-01
Authorization: Bearer {api_token}

Response:
{
  "items": [
    {
      "id": 12345,
      "number": "2025001",
      "date_issued": "2025-01-15",
      "date_due": "2025-02-14",
      "customer_id": 678,
      "total": 125000,
      "currency": "CZK",
      "items": [...]
    }
  ],
  "total_count": 156
}

2. SOAP API (starší verze)

Charakteristika:

• XML-based
• Více low-level operací
• Komplexnější, ale silnější
• Podpora transakcí

Příklad:

<soapenv:Envelope>
  <soapenv:Body>
    <CreateInvoice>
      <Invoice>
        <Number>2025001</Number>
        <CustomerId>678</CustomerId>
        <Items>...</Items>
      </Invoice>
    </CreateInvoice>
  </soapenv:Body>
</soapenv:Envelope>

3. Direct DB Access (ne doporučeno)

Možné, ale:

• ❌ Obchází business logiku Heliosu
• ❌ Riziko data corruption
• ❌ Problémy s upgrady
• ⚠️ Použít POUZE read-only!

Use case: Čtení dat pro analytiku, když API nedostačuje.

4. File Export/Import

Helios podporuje:

• CSV exporty
• XML formáty
• Excel (přes XLSX)

Use case:

• Bulk operace
• Offline zpracování
• Legacy integrace

---

Doporučená architektura

High-level diagram

┌──────────────┐
│   HELIOS     │
│   ORANGE     │
└──────┬───────┘
       │ REST/SOAP API
       │
┌──────▼──────────────────────────┐
│   MIDDLEWARE LAYER               │
│   (Python Flask/FastAPI)         │
│   ├─ API adapter (Helios ↔ AI)  │
│   ├─ Data transformation         │
│   ├─ Caching                     │
│   └─ Error handling              │
└──────┬──────────────────────────┘
       │
┌──────▼──────────────────────────┐
│   AI/ML SERVICES                 │
│   ├─ Invoice OCR                 │
│   ├─ Demand forecasting          │
│   ├─ Classification              │
│   ├─ Anomaly detection           │
│   └─ Recommendation engine       │
└──────┬──────────────────────────┘
       │
┌──────▼──────────────────────────┐
│   DATA LAYER                     │
│   ├─ PostgreSQL (analytics)      │
│   ├─ Redis (cache)               │
│   └─ S3/Minio (documents)        │
└──────────────────────────────────┘

Proč middleware?

Bez middleware (direct API):

# ❌ Špatný přístup
ai_result = predict_demand(product_id)
helios.api.create_purchase_order(ai_result)  # Fragile!

S middleware:

# ✅ Správný přístup
ai_result = predict_demand(product_id)
middleware.create_purchase_order_with_validation(ai_result)
# → Validace, retry logic, logging, error handling

Benefity:

• 🛡️ Abstrakce Helios API (snadnější změny)
• 🔄 Retry logic a resilience
• 📊 Centrální logging a monitoring
• 🚀 Caching pro performance
• 🔒 Dodatečná validace a security

---

Implementace: Konkrétní příklady

Use Case #1: AI Fakturace

Workflow:

1. Email s fakturou dorazí
2. AI extrahuje data (OCR)
3. Validace a kontrola
4. Vytvoření záznamu v Heliosu

Middleware API:

from fastapi import FastAPI, UploadFile
import helios_client
import ocr_engine

app = FastAPI()

@app.post("/api/invoices/process")
async def process_invoice(file: UploadFile):
    # 1. OCR extraction
    pdf_bytes = await file.read()
    extracted_data = ocr_engine.extract_invoice(pdf_bytes)

    # 2. AI validation
    validation_result = ai_validator.check(extracted_data)
    if validation_result.confidence < 0.90:
        return {
            "status": "review_needed",
            "data": extracted_data,
            "issues": validation_result.issues
        }

    # 3. Find supplier in Helios
    supplier = helios_client.find_supplier_by_ico(
        extracted_data.ico
    )
    if not supplier:
        return {
            "status": "error",
            "message": "Supplier not found in Helios"
        }

    # 4. AI classification (středisko, účet)
    classification = ai_classifier.classify(
        description=extracted_data.description,
        supplier=supplier.name,
        amount=extracted_data.total
    )

    # 5. Create invoice in Helios
    try:
        invoice = helios_client.create_invoice({
            "number": extracted_data.invoice_number,
            "supplier_id": supplier.id,
            "date_issued": extracted_data.date_issued,
            "date_due": extracted_data.date_due,
            "total": extracted_data.total,
            "vat": extracted_data.vat,
            "account": classification.account,
            "cost_center": classification.cost_center,
            "items": extracted_data.items
        })

        return {
            "status": "success",
            "helios_invoice_id": invoice.id,
            "confidence": validation_result.confidence
        }

    except helios_client.HeliosAPIError as e:
        # Retry logic, logging, etc.
        return {
            "status": "error",
            "message": str(e)
        }

Helios API klient:

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class HeliosClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.api_key = api_key

        # Setup session with retry logic
        self.session = requests.Session()
        retry = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

    def _headers(self):
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def find_supplier_by_ico(self, ico):
        """Find supplier by IČO"""
        response = self.session.get(
            f"{self.base_url}/api/v1/suppliers",
            headers=self._headers(),
            params={"ico": ico}
        )
        response.raise_for_status()
        data = response.json()
        return data["items"][0] if data["items"] else None

    def create_invoice(self, invoice_data):
        """Create incoming invoice"""
        response = self.session.post(
            f"{self.base_url}/api/v1/invoices/received",
            headers=self._headers(),
            json=invoice_data
        )
        response.raise_for_status()
        return response.json()

Use Case #2: Predikce poptávky → Auto-objednávky

Workflow:

1. Každé ráno: AI predikuje poptávku na 8 týdnů
2. Výpočet optimálních objednávek
3. Vytvoření nákupních žádanek v Heliosu
4. Notifikace nákupčímu

Scheduled job:

import schedule
import time
from datetime import datetime, timedelta

def daily_demand_forecast():
    print(f"[{datetime.now()}] Starting demand forecast...")

    # 1. Load products from Helios
    products = helios_client.get_all_products()

    # 2. For each product, predict demand
    predictions = []
    for product in products:
        # Get historical sales
        sales_history = helios_client.get_sales_history(
            product_id=product.id,
            months=24
        )

        # AI prediction
        forecast = ai_forecaster.predict(
            product_id=product.id,
            history=sales_history,
            horizon_weeks=8
        )

        # Current stock level
        stock = helios_client.get_stock_level(product.id)

        # Calculate order requirement
        order_qty = calculate_order_quantity(
            forecast=forecast,
            current_stock=stock,
            lead_time=product.lead_time_days,
            safety_stock=product.safety_stock
        )

        if order_qty > 0:
            predictions.append({
                "product": product,
                "forecast": forecast,
                "order_qty": order_qty,
                "urgency": calculate_urgency(stock, forecast)
            })

    # 3. Create purchase requisitions in Helios
    for pred in predictions:
        if pred["urgency"] == "HIGH":
            # Auto-create for high urgency
            create_purchase_requisition(pred)
        else:
            # Just notify for normal items
            notify_purchaser(pred)

    print(f"Processed {len(products)} products, {len(predictions)} orders needed")

def create_purchase_requisition(prediction):
    """Create nákupní žádanka in Helios"""
    supplier = helios_client.get_preferred_supplier(
        prediction["product"].id
    )

    requisition = helios_client.create_purchase_requisition({
        "product_id": prediction["product"].id,
        "quantity": prediction["order_qty"],
        "supplier_id": supplier.id,
        "requested_delivery": datetime.now() + timedelta(
            days=prediction["product"].lead_time_days
        ),
        "note": f"AI recommendation - Predicted demand: {prediction['forecast'].total_qty} units",
        "urgency": prediction["urgency"]
    })

    # Log to analytics DB
    analytics_db.log_prediction(prediction, requisition.id)

# Schedule daily at 6 AM
schedule.every().day.at("06:00").do(daily_demand_forecast)

# Run
while True:
    schedule.run_pending()
    time.sleep(60)

---

Deployment: Možnosti

Option 1: On-premise (na serveru zákazníka)

Architecture:

┌──────────────────────────────────┐
│  Zákaznická síť                  │
│                                  │
│  ┌─────────┐    ┌─────────────┐ │
│  │ Helios  │←───│ AI Middleware│ │
│  │ Server  │    │   (Docker)   │ │
│  └─────────┘    └──────┬───────┘ │
│                        │         │
│                 ┌──────▼───────┐ │
│                 │   PostgreSQL │ │
│                 │   + Redis    │ │
│                 └──────────────┘ │
└──────────────────────────────────┘

Pros:

• ✅ Maximální kontrola nad daty
• ✅ Žádné cloud závislosti
• ✅ Rychlá komunikace s Heliosem (LAN)

Cons:

• ❌ Vyžaduje hardware zákazníka
• ❌ Údržba a monitoring na straně zákazníka
• ❌ Složitější updates

Deployment (Docker):

# docker-compose.yml
version: '3.8'

services:
  middleware:
    image: ai-middleware:latest
    ports:
      - "8000:8000"
    environment:
      - HELIOS_API_URL=http://helios-server:8080
      - HELIOS_API_KEY=${HELIOS_API_KEY}
      - DATABASE_URL=postgresql://user:pass@postgres:5432/ai_db
      - REDIS_URL=redis://redis:6379
    depends_on:
      - postgres
      - redis
    restart: always

  postgres:
    image: postgres:15
    volumes:
      - postgres_data:/var/lib/postgresql/data
    environment:
      - POSTGRES_PASSWORD=${DB_PASSWORD}

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  redis_data:

Option 2: Cloud hybrid

Architecture:

┌──────────────────────┐      ┌────────────────────┐
│  Zákaznická síť      │      │     AWS/Azure      │
│                      │      │                    │
│  ┌─────────┐         │      │  ┌──────────────┐ │
│  │ Helios  │◄────────┼──────┼─►│ AI Services  │ │
│  │ Server  │ VPN/API │      │  │  (Managed)   │ │
│  └─────────┘         │      │  └──────────────┘ │
└──────────────────────┘      └────────────────────┘

Pros:

• ✅ Silnější AI compute (GPU)
• ✅ Snadnější scaling
• ✅ Managed services (nižší údržba)

Cons:

• ❌ Data v cloudu (GDPR concerns)
• ❌ Latence (API přes internet)
• ❌ Vyšší náklady (cloud fees)

Best for:

• Compute-intensive AI (computer vision, NLP)
• Burst workloads (měsíční predikce)

Option 3: SaaS (plně managed)

Architecture:

┌──────────────────────┐
│  Zákaznická síť      │
│                      │
│  ┌─────────┐         │
│  │ Helios  │◄────────┼───┐
│  │ Server  │         │   │ API (přes internet)
│  └─────────┘         │   │
└──────────────────────┘   │
                           │
┌──────────────────────────▼──────┐
│      SaaS Platforma              │
│  (Multi-tenant AI middleware)    │
│                                  │
│  Tenant 1 │ Tenant 2 │ Tenant N  │
└──────────────────────────────────┘

Pros:

• ✅ Zero maintenance
• ✅ Rychlý onboarding
• ✅ Nízká vstupní investice

Cons:

• ❌ Méně flexibility
• ❌ Vendor lock-in
• ❌ Možné GDPR komplikace

---

Security best practices

1. API Key management

❌ Špatně:

# Hardcoded v kódu
HELIOS_API_KEY = "abc123xyz456"

✅ Správně:

# Environment variables
import os
HELIOS_API_KEY = os.getenv("HELIOS_API_KEY")

# Nebo secrets management (Azure Key Vault, AWS Secrets Manager)
from azure.keyvault.secrets import SecretClient
secret = secret_client.get_secret("helios-api-key")

2. Least privilege

Helios API přístup:

• AI middleware by měl mít POUZE nutná oprávnění
• Read-only tam, kde stačí
• Separate API keys pro různé služby

# Příklad: Dva API keys
HELIOS_READ_KEY:  # Pro čtení (predikce, analýzy)
  permissions: [read_products, read_sales, read_stock]

HELIOS_WRITE_KEY:  # Pro zápis (vytváření žádanek)
  permissions: [create_purchase_requisition]

3. Data encryption

At rest:

# Database: PostgreSQL s encryption
postgresql://user:pass@host/db?sslmode=require

# S3: Server-side encryption
boto3.client('s3').put_object(
    Bucket='invoices',
    Key='invoice.pdf',
    Body=pdf_bytes,
    ServerSideEncryption='AES256'
)

In transit:

# HTTPS only
helios_api_url = "https://helios.company.cz/api"  # ✅
# NOT http:// ❌

4. Audit logging

import logging

def audit_log(action, user, resource, result):
    logger.info({
        "timestamp": datetime.now().isoformat(),
        "action": action,
        "user": user,
        "resource": resource,
        "result": result,
        "ip": request.remote_addr
    })

# Usage:
audit_log(
    action="create_invoice",
    user="ai_middleware",
    resource=f"invoice #{invoice.id}",
    result="success"
)

---

Monitoring & Observability

Metriky k sledování

1. API Performance:

# Prometheus metrics
from prometheus_client import Histogram, Counter

api_latency = Histogram(
    'helios_api_latency_seconds',
    'Helios API call latency'
)

api_errors = Counter(
    'helios_api_errors_total',
    'Helios API errors',
    ['endpoint', 'status_code']
)

@api_latency.time()
def call_helios_api():
    try:
        response = helios_client.get(...)
        return response
    except Exception as e:
        api_errors.labels(
            endpoint='/api/v1/products',
            status_code=500
        ).inc()
        raise

2. AI Model Performance:

# Monitoring AI accuracy
def track_prediction_accuracy(prediction, actual):
    error = abs(prediction - actual) / actual

    metrics_db.insert({
        "timestamp": datetime.now(),
        "model": "demand_forecast",
        "prediction": prediction,
        "actual": actual,
        "mape": error * 100  # Mean Absolute Percentage Error
    })

3. Business metrics:

-- Dashboard query: Denní přehled
SELECT
  DATE(created_at) as date,
  COUNT(*) as invoices_processed,
  SUM(CASE WHEN status='auto_approved' THEN 1 ELSE 0 END) as auto_approved,
  AVG(confidence_score) as avg_confidence,
  SUM(CASE WHEN status='error' THEN 1 ELSE 0 END) as errors
FROM ai_invoice_processing
WHERE created_at >= CURRENT_DATE - INTERVAL '7 days'
GROUP BY DATE(created_at)
ORDER BY date DESC;

Alerting

# Slack alerts pro kritické eventy
def alert_on_helios_api_failure():
    if api_errors_last_hour > 10:
        slack.send_message(
            channel="#ops",
            text=f"⚠️  Helios API errors: {api_errors_last_hour} in last hour"
        )

def alert_on_low_accuracy():
    if daily_mape > 15:  # >15% error
        slack.send_message(
            channel="#ai-team",
            text=f"⚠️  AI forecast accuracy dropped to {daily_mape}%"
        )

---

Cost optimization

1. Caching

from functools import lru_cache
import redis

redis_client = redis.Redis(host='localhost', port=6379)

def get_product_cached(product_id):
    # Try cache first
    cache_key = f"product:{product_id}"
    cached = redis_client.get(cache_key)

    if cached:
        return json.loads(cached)

    # Cache miss → fetch from Helios
    product = helios_client.get_product(product_id)

    # Store in cache (TTL 1 hour)
    redis_client.setex(
        cache_key,
        3600,
        json.dumps(product)
    )

    return product

2. Batch operations

# ❌ Slow: Individual API calls
for product_id in product_ids:
    product = helios_client.get_product(product_id)  # 200 API calls!

# ✅ Fast: Batch API call
products = helios_client.get_products_batch(product_ids)  # 1 API call

3. Asynchronní zpracování

from celery import Celery

celery_app = Celery('tasks', broker='redis://localhost:6379')

@celery_app.task
def process_invoice_async(invoice_pdf_bytes):
    # Long-running OCR + AI processing
    # Runs in background worker
    extracted_data = ocr_engine.extract(invoice_pdf_bytes)
    validated = ai_validator.validate(extracted_data)
    helios_client.create_invoice(validated)

# Usage: Non-blocking
process_invoice_async.delay(pdf_bytes)
# Returns immediately, processed by worker

---

Závěr: Checklist pro úspěšnou integraci

Před startem:

• Audit Helios API (jaká verze, jaké endpointy)
• Definice use cases (co chceme automatizovat)
• Security requirements (on-prem vs. cloud)
• Budget a timeline

Development:

• Middleware architektura (ne direct API!)
• Robust error handling
• Logging a monitoring od začátku
• Unit + integration testy

Deployment:

• Staging environment (test na kopii Heliosu)
• Postupný rollout (piloty)
• Fallback plán (co když AI selže)
• Dokumentace pro OPS tým

Production:

• 24/7 monitoring
• Regular AI model retraining
• Performance tuning
• User feedback loop

Chcete pomoc s architekturou? Nabízíme bezplatnou konzultaci a design review vaší integrace s Heliosem.