Skip to content

Integrasi Python

Contoh kode berikut menunjukkan cara menggunakan API Rivalistic dengan Python. Anda dapat menggunakan kode ini sebagai titik awal untuk integrasi Anda sendiri.

Prasyarat

  • Python 3.6+
  • Library requests
  • API key atau kredensial akun Rivalistic

Instalasi Dependensi

bash
pip install requests pillow

Autentikasi

python
import requests

# Autentikasi dengan email dan kata sandi
def get_token(email, password):
    url = "https://api.rivalistic.com/v1/auth/login"
    payload = {
        "email": email,
        "password": password
    }
    response = requests.post(url, json=payload)
    response.raise_for_status()
    data = response.json()
    return data["access_token"]

# Contoh penggunaan
try:
    token = get_token("email@anda.com", "kata_sandi_anda")
    print(f"Token: {token}")
except requests.exceptions.HTTPError as e:
    print(f"Error autentikasi: {e}")

Autentikasi dengan API Key

python
# Menggunakan API key
api_key = "riv__1234567890abcdef"
headers = {"X-API-Key": api_key}

# Contoh permintaan dengan API key
response = requests.get(
    "https://api.rivalistic.com/v1/databases",
    headers=headers
)

Mengelola Database

Membuat Database Baru

python
def create_database(token, name, description=None):
    url = "https://api.rivalistic.com/v1/databases"
    headers = {"Authorization": f"Bearer {token}"}
    payload = {
        "name": name,
        "description": description
    }
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

# Contoh penggunaan
database = create_database(token, "Database Karyawan", "Wajah karyawan perusahaan")
database_id = database["id"]
print(f"Database dibuat dengan ID: {database_id}")

Mendapatkan Daftar Database

python
def list_databases(token):
    url = "https://api.rivalistic.com/v1/databases"
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()["databases"]

# Contoh penggunaan
databases = list_databases(token)
for db in databases:
    print(f"ID: {db['id']}, Nama: {db['metadata']['name']}, Jumlah Gambar: {db['metadata']['image_count']}")

Mengunggah Gambar

python
import base64
from PIL import Image
import io

def upload_image(token, database_id, image_path, metadata=None):
    url = f"https://api.rivalistic.com/v1/databases/{database_id}/images"
    headers = {"Authorization": f"Bearer {token}"}
    
    # Baca dan encode gambar ke base64
    with open(image_path, "rb") as image_file:
        encoded_image = base64.b64encode(image_file.read()).decode('utf-8')
    
    # Siapkan payload
    payload = {
        "image": encoded_image,
        "metadata": metadata or {}
    }
    
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

# Contoh penggunaan
metadata = {
    "name": "John Doe",
    "description": "Marketing Manager",
    "custom_data": {
        "employee_id": "EMP123",
        "department": "Marketing",
        "start_date": "2024-01-15"
    }
}

try:
    result = upload_image(token, database_id, "path/to/image.jpg", metadata)
    print(f"Gambar diunggah dengan ID: {result['image_id']}")
    
    # Cek apakah pengindeksan ulang diperlukan
    if result["reindex_required"]:
        print("Pengindeksan ulang diperlukan setelah mengunggah gambar.")
except requests.exceptions.HTTPError as e:
    print(f"Error mengunggah gambar: {e}")

Pencarian Wajah

Pencarian Wajah Tunggal

python
def search_face(token, database_id, image_path, confidence_threshold=0.4):
    url = "https://api.rivalistic.com/v1/face-search"
    headers = {"Authorization": f"Bearer {token}"}
    
    # Baca dan encode gambar ke base64
    with open(image_path, "rb") as image_file:
        encoded_image = base64.b64encode(image_file.read()).decode('utf-8')
    
    # Siapkan payload
    payload = {
        "image": encoded_image,
        "database_id": database_id,
        "confidence_threshold": confidence_threshold
    }
    
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

# Contoh penggunaan
try:
    result = search_face(token, database_id, "path/to/search_image.jpg", 0.5)
    
    if result["match_found"]:
        print(f"Kecocokan ditemukan dengan skor kepercayaan: {result['confidence']}")
        print(f"Nama: {result['metadata'].get('name', 'Tidak diketahui')}")
        print(f"Metadata tambahan: {result['metadata'].get('custom_data', {})}")
    else:
        print("Tidak ditemukan kecocokan.")
    
    print(f"Waktu pemrosesan: {result['processing_time']} detik")
except requests.exceptions.HTTPError as e:
    print(f"Error pencarian wajah: {e}")

Pencarian Wajah Batch

python
def batch_search_faces(token, database_id, image_paths, confidence_threshold=0.4):
    url = "https://api.rivalistic.com/v1/face-search-batch"
    headers = {"Authorization": f"Bearer {token}"}
    
    # Siapkan gambar
    images = []
    for i, path in enumerate(image_paths):
        with open(path, "rb") as image_file:
            encoded_image = base64.b64encode(image_file.read()).decode('utf-8')
        images.append({
            "image": encoded_image,
            "reference_id": f"img_{i}"
        })
    
    # Siapkan payload
    payload = {
        "images": images,
        "database_id": database_id,
        "confidence_threshold": confidence_threshold
    }
    
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

# Contoh penggunaan
image_paths = ["path/to/image1.jpg", "path/to/image2.jpg", "path/to/image3.jpg"]

try:
    results = batch_search_faces(token, database_id, image_paths, 0.5)
    
    print(f"Total gambar diproses: {results['total_processed']}")
    print(f"Waktu pemrosesan total: {results['processing_time']} detik")
    
    for result in results["results"]:
        ref_id = result["reference_id"]
        if result["match_found"]:
            print(f"{ref_id}: Kecocokan ditemukan dengan skor {result['confidence']}")
            print(f" Nama: {result['metadata'].get('name', 'Tidak diketahui')}")
        else:
            print(f"{ref_id}: Tidak ditemukan kecocokan.")
except requests.exceptions.HTTPError as e:
    print(f"Error pencarian batch: {e}")

Pengindeksan Ulang Database

python
def reindex_database(token, database_id, force=False):
    url = f"https://api.rivalistic.com/v1/databases/{database_id}/reindex"
    headers = {"Authorization": f"Bearer {token}"}
    payload = {"force": force}
    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

def check_reindex_status(token, database_id):
    url = f"https://api.rivalistic.com/v1/databases/{database_id}/reindex/status"
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    return response.json()

# Contoh penggunaan
try:
    # Mulai pengindeksan
    reindex = reindex_database(token, database_id)
    job_id = reindex["job_id"]
    print(f"Pengindeksan dimulai dengan job ID: {job_id}")
    print(f"Estimasi waktu: {reindex['estimated_time']} detik")
    
    # Cek status (dalam aplikasi nyata, gunakan loop dengan delay)
    status = check_reindex_status(token, database_id)
    print(f"Status: {status['status']}")
    print(f"Kemajuan: {status['progress']*100:.2f}%")
except requests.exceptions.HTTPError as e:
    print(f"Error pengindeksan: {e}")

Aplikasi Lengkap

python
import requests
import base64
import time
import argparse
import os

class RivalisticClient:
    def __init__(self, api_key=None, email=None, password=None):
        self.base_url = "https://api.rivalistic.com/v1"
        self.token = None
        self.api_key = api_key
        
        if api_key:
            self.headers = {"X-API-Key": api_key}
        elif email and password:
            self.token = self.get_token(email, password)
            self.headers = {"Authorization": f"Bearer {self.token}"}
        else:
            raise ValueError("Berikan API key atau kredensial email/kata sandi")
    
    def get_token(self, email, password):
        url = f"{self.base_url}/auth/login"
        payload = {"email": email, "password": password}
        response = requests.post(url, json=payload)
        response.raise_for_status()
        return response.json()["access_token"]
    
    def list_databases(self):
        url = f"{self.base_url}/databases"
        response = requests.get(url, headers=self.headers)
        response.raise_for_status()
        return response.json()["databases"]
    
    def create_database(self, name, description=None):
        url = f"{self.base_url}/databases"
        payload = {"name": name, "description": description}
        response = requests.post(url, json=payload, headers=self.headers)
        response.raise_for_status()
        return response.json()
    
    def upload_image(self, database_id, image_path, metadata=None):
        url = f"{self.base_url}/databases/{database_id}/images"
        with open(image_path, "rb") as image_file:
            encoded_image = base64.b64encode(image_file.read()).decode('utf-8')
        
        payload = {
            "image": encoded_image,
            "metadata": metadata or {}
        }
        
        response = requests.post(url, json=payload, headers=self.headers)
        response.raise_for_status()
        return response.json()
    
    def search_face(self, database_id, image_path, confidence_threshold=0.4):
        url = f"{self.base_url}/face-search"
        with open(image_path, "rb") as image_file:
            encoded_image = base64.b64encode(image_file.read()).decode('utf-8')
        
        payload = {
            "image": encoded_image,
            "database_id": database_id,
            "confidence_threshold": confidence_threshold
        }
        
        response = requests.post(url, json=payload, headers=self.headers)
        response.raise_for_status()
        return response.json()
    
    def reindex_database(self, database_id, force=False):
        url = f"{self.base_url}/databases/{database_id}/reindex"
        payload = {"force": force}
        response = requests.post(url, json=payload, headers=self.headers)
        response.raise_for_status()
        return response.json()

# Contoh penggunaan aplikasi lengkap
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Klien Rivalistic API")
    parser.add_argument("--mode", choices=["list", "create", "upload", "search", "reindex"], required=True)
    parser.add_argument("--api-key", help="API key Rivalistic")
    parser.add_argument("--email", help="Email akun Rivalistic")
    parser.add_argument("--password", help="Kata sandi akun Rivalistic")
    parser.add_argument("--database-id", help="ID database")
    parser.add_argument("--name", help="Nama database atau gambar")
    parser.add_argument("--description", help="Deskripsi database atau gambar")
    parser.add_argument("--image-path", help="Path ke file gambar")
    parser.add_argument("--threshold", type=float, default=0.4, help="Ambang batas kepercayaan")
    args = parser.parse_args()
    
    try:
        # Inisialisasi klien
        client = RivalisticClient(
            api_key=args.api_key,
            email=args.email,
            password=args.password
        )
        
        # Proses berdasarkan mode
        if args.mode == "list":
            databases = client.list_databases()
            print(f"Total database: {len(databases)}")
            for db in databases:
                print(f"ID: {db['id']}, Nama: {db['metadata']['name']}, Gambar: {db['metadata']['image_count']}")
        
        elif args.mode == "create":
            if not args.name:
                raise ValueError("Parameter --name diperlukan untuk membuat database")
            result = client.create_database(args.name, args.description)
            print(f"Database dibuat: {result['id']}")
        
        elif args.mode == "upload":
            if not args.database_id or not args.image_path:
                raise ValueError("Parameter --database-id dan --image-path diperlukan untuk mengunggah")
            if not os.path.exists(args.image_path):
                raise ValueError(f"File tidak ditemukan: {args.image_path}")
            
            metadata = {
                "name": args.name,
                "description": args.description
            }
            result = client.upload_image(args.database_id, args.image_path, metadata)
            print(f"Gambar diunggah: {result['image_id']}")
            if result["reindex_required"]:
                print("Pengindeksan ulang diperlukan.")
        
        elif args.mode == "search":
            if not args.database_id or not args.image_path:
                raise ValueError("Parameter --database-id dan --image-path diperlukan untuk pencarian")
            if not os.path.exists(args.image_path):
                raise ValueError(f"File tidak ditemukan: {args.image_path}")
            
            result = client.search_face(args.database_id, args.image_path, args.threshold)
            if result["match_found"]:
                print(f"Kecocokan ditemukan dengan skor: {result['confidence']}")
                print(f"ID gambar yang cocok: {result['matched_image']}")
                print(f"Metadata: {result['metadata']}")
            else:
                print("Tidak ditemukan kecocokan.")
            print(f"Waktu pemrosesan: {result['processing_time']} detik")
        
        elif args.mode == "reindex":
            if not args.database_id:
                raise ValueError("Parameter --database-id diperlukan untuk pengindeksan ulang")
            result = client.reindex_database(args.database_id)
            print(f"Pengindeksan dimulai: {result['job_id']}")
            print(f"Status: {result['status']}")
            print(f"Estimasi waktu: {result['estimated_time']} detik")
    
    except requests.exceptions.HTTPError as e:
        print(f"Error API: {e}")
    except Exception as e:
        print(f"Error: {e}")

Anda dapat menggunakan aplikasi lengkap ini dengan perintah berikut:

bash
# Daftar semua database
python rivalistic_client.py --mode list --api-key YOUR_API_KEY

# Buat database baru
python rivalistic_client.py --mode create --api-key YOUR_API_KEY --name "Database Karyawan" --description "Wajah karyawan perusahaan"

# Unggah gambar
python rivalistic_client.py --mode upload --api-key YOUR_API_KEY --database-id db_12345 --image-path /path/to/image.jpg --name "John Doe" --description "Marketing Manager"

# Cari wajah
python rivalistic_client.py --mode search --api-key YOUR_API_KEY --database-id db_12345 --image-path /path/to/search.jpg --threshold 0.5

# Indeks ulang database
python rivalistic_client.py --mode reindex --api-key YOUR_API_KEY --database-id db_12345

Dokumentasi API Rivalistic