call-to-text/setup_directus.py

import sys
import time
import httpx
from config import config


def wait_for_directus(url: str, timeout: int = 60):
    start = time.time()
    while time.time() - start < timeout:
        try:
            resp = httpx.get(f"{url}/server/ping", timeout=5)
            if resp.status_code == 200:
                print("Directus is ready.")
                return True
        except Exception:
            pass
        print("Waiting for Directus...")
        time.sleep(3)
    print("Directus did not become ready.")
    return False


def login_and_get_token(url: str, email: str, password: str) -> str:
    resp = httpx.post(
        f"{url}/auth/login",
        json={"email": email, "password": password},
    )
    resp.raise_for_status()
    return resp.json()["data"]["access_token"]


def create_collection(client: httpx.Client, collection: str, meta: dict):
    resp = client.post("/collections", json={
        "collection": collection,
        "meta": {
            "singleton": False,
            "archive_field": None,
            "archive_app_filter": True,
            "hidden": False,
            "sort": None,
            **meta,
        },
        "schema": {},
    })
    if resp.status_code not in (200, 201):
        if resp.status_code == 409:
            print(f"  Collection '{collection}' already exists.")
            return
        print(f"  Failed to create collection '{collection}': {resp.text}")
        return
    print(f"  Created collection '{collection}'.")


def create_field(client: httpx.Client, collection: str, field: str, field_type: str, meta: dict = None):
    payload = {
        "field": field,
        "type": field_type,
        "meta": meta or {},
        "schema": {},
    }
    if field_type == "json":
        payload["schema"] = {"default": None}
    elif field_type in ("integer", "bigInteger"):
        payload["schema"] = {}
    elif field_type == "text":
        payload["schema"] = {}
    else:
        payload["schema"] = {}

    resp = client.post(f"/fields/{collection}", json=payload)
    if resp.status_code not in (200, 201):
        if resp.status_code == 409:
            print(f"  Field '{collection}.{field}' already exists.")
            return
        print(f"  Failed to create field '{collection}.{field}': {resp.text}")
        return
    print(f"  Created field '{collection}.{field}' ({field_type}).")


def setup():
    url = config.DIRECTUS_URL.rstrip("/")

    if not wait_for_directus(url):
        sys.exit(1)

    token = login_and_get_token(url, "admin@example.com", "admin-password-123")
    client = httpx.Client(
        base_url=url,
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        timeout=30,
    )

    collections = {
        "calls": {"note": "Call recordings with transcripts"},
        "calls_74951284933": {"note": "Calls involving +74951284933"},
        "contacts": {"note": "Phonebook for caller/callee lookup"},
    }

    for col_name, col_meta in collections.items():
        create_collection(client, col_name, col_meta)

    calls_fields = [
        ("call_id", "string", {"interface": "input", "unique": True}),
        ("datetime", "dateTime", {"interface": "datetime"}),
        ("direction", "string", {"interface": "input"}),
        ("caller_number", "string", {"interface": "input"}),
        ("caller_name", "string", {"interface": "input"}),
        ("caller_company", "string", {"interface": "input"}),
        ("callee_number", "string", {"interface": "input"}),
        ("callee_name", "string", {"interface": "input"}),
        ("callee_company", "string", {"interface": "input"}),
        ("employee", "string", {"interface": "input"}),
        ("duration_sec", "integer", {"interface": "input"}),
        ("audio_url", "string", {"interface": "input"}),
        ("transcript", "text", {"interface": "input-multiline"}),
        ("segments", "json", {"interface": "input-code", "options": {"language": "json"}}),
        ("source_email_id", "string", {"interface": "input"}),
        ("raw_metadata", "json", {"interface": "input-code", "options": {"language": "json"}}),
        ("status", "string", {"interface": "input", "default_value": "pending"}),
    ]

    for col in ["calls", "calls_74951284933"]:
        for field_name, field_type, field_meta in calls_fields:
            create_field(client, col, field_name, field_type, field_meta)

    contacts_fields = [
        ("phone_number", "string", {"interface": "input", "unique": True}),
        ("name", "string", {"interface": "input"}),
        ("company", "string", {"interface": "input"}),
        ("notes", "text", {"interface": "input-multiline"}),
    ]

    for field_name, field_type, field_meta in contacts_fields:
        create_field(client, "contacts", field_name, field_type, field_meta)

    print("\nDirectus setup complete! Collections 'calls', 'calls_74951284933', and 'contacts' are ready.")
    client.close()


if __name__ == "__main__":
    setup()