homelab-docs/import-cocktaildb-to-bar-assistant.py

#!/usr/bin/env python3
"""
Import all cocktails from TheCocktailDB into Bar Assistant

Requirements:
    pip install requests

Usage:
    python import-cocktaildb-to-bar-assistant.py
"""

import requests
import time
import json
from typing import Dict, List

# Configuration
BAR_ASSISTANT_URL = "https://cocktails.nianticbooks.com"
BAR_ASSISTANT_API = f"{BAR_ASSISTANT_URL}/bar/api"
COCKTAILDB_API = "https://www.thecocktaildb.com/api/json/v1/1"

# You'll need to get these from Bar Assistant
# Login to Bar Assistant, go to Settings -> API Tokens, create a token
BAR_ASSISTANT_TOKEN = "3|dqtZPWZoKdU59fFRLkGrL83I9HKm9EqmLCgO4kkI96199a82"  # Set this to your API token
BAR_ID = 1  # Usually 1 for your first bar


def get_all_cocktails_from_cocktaildb() -> List[Dict]:
    """Fetch all cocktails from TheCocktailDB"""
    print("Fetching all cocktails from TheCocktailDB...")
    all_cocktails = []

    # Get all cocktails by first letter (a-z, 0-9)
    letters = 'abcdefghijklmnopqrstuvwxyz0123456789'

    for letter in letters:
        print(f"  Fetching cocktails starting with '{letter}'...")
        url = f"{COCKTAILDB_API}/search.php?f={letter}"

        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()
            data = response.json()

            if data.get('drinks'):
                cocktails = data['drinks']
                all_cocktails.extend(cocktails)
                print(f"    Found {len(cocktails)} cocktails")

            time.sleep(0.5)  # Be nice to the API

        except Exception as e:
            print(f"    Error fetching '{letter}': {e}")
            continue

    print(f"\n[OK] Total cocktails found: {len(all_cocktails)}")
    return all_cocktails


def convert_cocktaildb_to_bar_assistant(cocktail: Dict) -> Dict:
    """Convert TheCocktailDB format to Bar Assistant format"""

    # Extract ingredients
    ingredients = []
    for i in range(1, 16):  # TheCocktailDB supports up to 15 ingredients
        ingredient_key = f"strIngredient{i}"
        measure_key = f"strMeasure{i}"

        ingredient_name = cocktail.get(ingredient_key)
        measure = cocktail.get(measure_key)

        if ingredient_name and ingredient_name.strip():
            ingredients.append({
                "name": ingredient_name.strip(),
                "amount": measure.strip() if measure else "",
                "units": "",
                "optional": False
            })

    # Build the cocktail data
    bar_assistant_cocktail = {
        "name": cocktail.get('strDrink', 'Unknown'),
        "instructions": cocktail.get('strInstructions', ''),
        "description": f"Imported from TheCocktailDB. Category: {cocktail.get('strCategory', 'N/A')}",
        "source": f"TheCocktailDB (ID: {cocktail.get('idDrink')})",
        "garnish": cocktail.get('strGarnish', ''),
        "glass": cocktail.get('strGlass', ''),
        "method": cocktail.get('strCategory', ''),
        "tags": [
            cocktail.get('strCategory', ''),
            cocktail.get('strAlcoholic', ''),
            cocktail.get('strIBA', '') if cocktail.get('strIBA') else None
        ],
        "ingredients": ingredients,
        "images": [
            {"url": cocktail.get('strDrinkThumb'), "copyright": "TheCocktailDB"}
        ] if cocktail.get('strDrinkThumb') else []
    }

    # Clean up None values from tags
    bar_assistant_cocktail['tags'] = [tag for tag in bar_assistant_cocktail['tags'] if tag]

    return bar_assistant_cocktail


def import_to_bar_assistant(cocktail_data: Dict, headers: Dict) -> bool:
    """Import a cocktail into Bar Assistant via API"""
    url = f"{BAR_ASSISTANT_API}/cocktails"

    try:
        response = requests.post(url, json=cocktail_data, headers=headers, timeout=30)
        response.raise_for_status()
        return True
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 409:
            print(f"    [!] Already exists: {cocktail_data['name']}")
        else:
            print(f"    [X] HTTP Error {e.response.status_code}: {cocktail_data['name']}")
        return False
    except Exception as e:
        print(f"    [X] Error importing {cocktail_data['name']}: {e}")
        return False


def main():
    """Main execution"""
    print("="*60)
    print("TheCocktailDB -> Bar Assistant Bulk Import")
    print("="*60)

    # Check if token is set
    if not BAR_ASSISTANT_TOKEN:
        print("\n[ERROR] ERROR: BAR_ASSISTANT_TOKEN not set!")
        print("\nTo get your API token:")
        print("1. Login to Bar Assistant at https://cocktails.nianticbooks.com")
        print("2. Go to Settings (or Profile)")
        print("3. Find 'Personal Access Tokens' or 'API Tokens'")
        print("4. Create a new token")
        print("5. Copy the token and set it in this script")
        print("\nThen edit this script and set:")
        print("  BAR_ASSISTANT_TOKEN = 'your-token-here'")
        return

    # Set up headers
    headers = {
        "Authorization": f"Bearer {BAR_ASSISTANT_TOKEN}",
        "Content-Type": "application/json",
        "Accept": "application/json",
        "Bar-Assistant-Bar-Id": str(BAR_ID)
    }

    # Step 1: Fetch all cocktails from TheCocktailDB
    cocktails = get_all_cocktails_from_cocktaildb()

    if not cocktails:
        print("\n[ERROR] No cocktails found from TheCocktailDB")
        return

    # Step 2: Convert and import
    print(f"\nImporting {len(cocktails)} cocktails to Bar Assistant...")
    print("This may take a while...\n")

    success_count = 0
    skip_count = 0
    error_count = 0

    for idx, cocktail in enumerate(cocktails, 1):
        cocktail_name = cocktail.get('strDrink', 'Unknown')
        print(f"[{idx}/{len(cocktails)}] {cocktail_name}...")

        # Convert format
        bar_assistant_data = convert_cocktaildb_to_bar_assistant(cocktail)

        # Import
        if import_to_bar_assistant(bar_assistant_data, headers):
            success_count += 1
            print(f"    [OK] Imported successfully")
        else:
            skip_count += 1

        # Small delay to avoid overwhelming the server
        time.sleep(0.5)

    # Summary
    print("\n" + "="*60)
    print("Import Complete!")
    print("="*60)
    print(f"[OK] Successfully imported: {success_count}")
    print(f"[!] Skipped (duplicates):  {skip_count}")
    print(f"[X] Errors:                {error_count}")
    print(f"\nTotal processed: {len(cocktails)}")
    print(f"\nView your cocktails at: {BAR_ASSISTANT_URL}")


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n\n[!] Import cancelled by user")
    except Exception as e:
        print(f"\n\n[ERROR] Error: {e}")