Automating Content Ingestion and Processing

import requests
import time
import json
import os
from datetime import datetime
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# --- Configuration ---
CLIENT_ID = os.getenv("GLOO_CLIENT_ID", "YOUR_CLIENT_ID")
CLIENT_SECRET = os.getenv("GLOO_CLIENT_SECRET", "YOUR_CLIENT_SECRET")
TOKEN_URL = "https://platform.ai.gloo.com/oauth2/token"
API_URL = "https://platform.ai.gloo.com/ingestion/v1/real_time_upload"
PUBLISHER_ID = "b6ada63e-f1ce-4a16-80da-9864f58c5bd7"  # Replace with your publisher ID

# --- State Management ---
access_token_info = {}

def get_access_token():
    """Retrieves a new access token."""
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    data = {"grant_type": "client_credentials", "scope": "api/access"}
    response = requests.post(TOKEN_URL, headers=headers, data=data, auth=(CLIENT_ID, CLIENT_SECRET))
    response.raise_for_status()
    token_data = response.json()
    token_data['expires_at'] = int(time.time()) + token_data['expires_in']
    return token_data

def is_token_expired(token_info):
    """Checks if the token is expired or close to expiring."""
    if not token_info or 'expires_at' not in token_info:
        return True
    return time.time() > (token_info['expires_at'] - 60)

def upload_content(content_data):
    """Uploads content to the Realtime API."""
    global access_token_info
    if is_token_expired(access_token_info):
        print("Token is expired or missing. Fetching a new one...")
        access_token_info = get_access_token()

    headers = {
        "Authorization": f"Bearer {access_token_info['access_token']}",
        "Content-Type": "application/json"
    }

    response = requests.post(API_URL, headers=headers, json=content_data)
    response.raise_for_status()
    return response.json()

def process_file(file_path):
    """Processes a single file and uploads its content."""
    try:
        with open(file_path, 'r', encoding='utf-8') as file:
            content = file.read()
        
        # Extract metadata from filename and content
        filename = os.path.basename(file_path)
        title = filename.replace('.txt', '').replace('.md', '').replace('_', ' ').title()
        
        content_data = {
            "content": content,
            "publisherId": PUBLISHER_ID,
            "item_title": title,
            "author": ["Automated Ingestion"],
            "publication_date": datetime.now().strftime("%Y-%m-%d"),
            "type": "Article",
            "pub_type": "technical",
            "item_tags": ["automated", "ingestion"],
            "evergreen": True,
            "drm": ["aspen", "kallm"]
        }
        
        result = upload_content(content_data)
        print(f"✅ Successfully uploaded: {title}")
        print(f"   Response: {result['message']}")
        return True
        
    except Exception as e:
        print(f"❌ Failed to process {file_path}: {e}")
        return False

class ContentHandler(FileSystemEventHandler):
    """Handles file system events for content monitoring."""
    
    def on_created(self, event):
        if not event.is_directory and event.src_path.endswith(('.txt', '.md')):
            print(f"📄 New file detected: {event.src_path}")
            time.sleep(1)  # Allow file write to complete
            process_file(event.src_path)

def start_file_watcher(watch_directory):
    """Starts monitoring a directory for new content files."""
    if not os.path.exists(watch_directory):
        os.makedirs(watch_directory)
        print(f"Created watch directory: {watch_directory}")
    
    event_handler = ContentHandler()
    observer = Observer()
    observer.schedule(event_handler, watch_directory, recursive=True)
    
    print(f"🔍 Monitoring directory: {watch_directory}")
    print("   Supported file types: .txt, .md")
    print("   Press Ctrl+C to stop")
    
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        print("\n👋 Stopping file monitor...")
    observer.join()

def batch_process_directory(directory_path):
    """Processes all supported files in a directory."""
    if not os.path.exists(directory_path):
        print(f"Directory does not exist: {directory_path}")
        return
    
    processed = 0
    failed = 0
    
    for filename in os.listdir(directory_path):
        if filename.endswith(('.txt', '.md')):
            file_path = os.path.join(directory_path, filename)
            if process_file(file_path):
                processed += 1
            else:
                failed += 1
            
            # Rate limiting - avoid overwhelming the API
            time.sleep(1)
    
    print(f"\n📊 Batch processing complete:")
    print(f"   ✅ Processed: {processed} files")
    print(f"   ❌ Failed: {failed} files")

# --- Example Usage ---
if __name__ == "__main__":
    import sys
    
    if len(sys.argv) < 2:
        print("Usage:")
        print("  python main.py watch <directory>     # Monitor directory for new files")
        print("  python main.py batch <directory>     # Process all files in directory")
        print("  python main.py single <file_path>    # Process single file")
        sys.exit(1)
    
    command = sys.argv[1]
    
    if command == "watch" and len(sys.argv) > 2:
        start_file_watcher(sys.argv[2])
    elif command == "batch" and len(sys.argv) > 2:
        batch_process_directory(sys.argv[2])
    elif command == "single" and len(sys.argv) > 2:
        process_file(sys.argv[2])
    else:
        print("Invalid command or missing arguments")