importieren Anfragen
import Zeit
importieren JSON
aus datetime import datetime, Zeitzone
aus typing import Dict, Optional, Tupel
import smtplib
aus email.mime.text import MIMEText
aus email.mime.multipart import MIMEMultipart
aus dotenv import load_dotenv
import os
aus datetime import timedelta

# Zu Demonstrationszwecken deaktivieren wir SSL-Warnungen
# Verwenden Sie in der Produktion geeignete SSL-Zertifikate!
import urllib3
urllib3.disable_warnings(urllib3.exceptions.UnsichereAnfrageWarnung)

print(„✓ Alle Pakete erfolgreich geladen!“)

# Load environment variables from .env file
load_dotenv()

# Load API credentials from environment variables
HOSTNAME = os.getenv("HOSTNAME")
API_VERSION_INFO_URL = f"https://{HOSTNAME}/api/supportedVersions"

# Fetch available API versions

# Check if API_VERSION is set in environment variables
API_VERSION = os.getenv("API_VERSION")

if API_VERSION:
    # Use the version from environment variable
    print(f"✓ Using API version from environment variable: {API_VERSION}")

else:
    # Fetch API versions from the endpoint
    print(f"\n🔍 Fetching available API versions from {API_VERSION_INFO_URL}...")
    response = requests.get(API_VERSION_INFO_URL, verify=False, timeout=10)
    versions_data = response.json()
    # Extract version strings from the response
    available_versions = [f"v{v['majorVersion']}.{v['minorVersion']}" for v in versions_data]

    if available_versions:
        # Sort versions to get the latest
        API_VERSION = sorted(available_versions, key=lambda v: [int(x) for x in v.lstrip('v').split('.')])[-1]
        print(f"✓ Available API versions: {', '.join(available_versions)}")
        print(f"✓ Using latest version: {API_VERSION}")
    else:
        # Fallback to default version
        API_VERSION = "v6.0"
        print(f"⚠️  No versions found in response, using default: {API_VERSION}")

from dotenv import load_dotenv
import os
from datetime import timedelta

# Load environment variables from .env file
load_dotenv()

API_BASE_URL = f"https://{HOSTNAME}/api/{API_VERSION}"
API_CLIENT_ID = os.getenv("API_CLIENT_ID")
API_CLIENT_SECRET = os.getenv("API_CLIENT_SECRET")

print("✓ Environment variables loaded")
print(f"  - Client ID: {API_CLIENT_ID}")
print(f"  - Client Secret: {'*' * len(API_CLIENT_SECRET) if API_CLIENT_SECRET else 'Not set'}")
print(f"  - Base URL: {API_BASE_URL}")

def get_oauth_token(client_id: str, client_secret: str) -> Optional[Dict]:
    """
    Fetch OAuth2 token using client credentials flow.
    
    Args:
        client_id: OAuth2 client ID
        client_secret: OAuth2 client secret
        
    Returns:
        Dictionary containing token information or None if request fails
    """
    token_url = f"https://{HOSTNAME}/auth/connect/token"
    
    # Prepare the request data for client credentials grant
    data = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret
    }
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    try:
        response = requests.post(token_url, data=data, headers=headers, verify=False, timeout=10)
        response.raise_for_status()
        token_data = response.json()
        return token_data
    except requests.exceptions.RequestException as e:
        print(f"❌ Error fetching OAuth token: {e}")
        return None


def refresh_token() -> bool:
    """
    Refresh the OAuth2 token and update global HEADERS.
    
    Returns:
        True if token refresh was successful, False otherwise
    """
    global API_TOKEN, HEADERS
    
    token_data = get_oauth_token(API_CLIENT_ID, API_CLIENT_SECRET)
    
    if token_data:
        API_TOKEN = token_data.get("access_token", "")
        token_type = token_data.get("token_type", "Bearer")
        
        HEADERS = {
            "Authorization": f"{token_type} {API_TOKEN}",
            "Accept": "application/json"
        }

        return True
    else:
        print("❌ Failed to refresh token")
        return False

# Fetch the token
token_response = get_oauth_token(API_CLIENT_ID, API_CLIENT_SECRET)

if token_response:
    # Extract token details
    API_TOKEN = token_response.get("access_token", "")
    token_type = token_response.get("token_type", "Bearer")
    expires_in = token_response.get("expires_in", 0)
    
    # Update the headers with the new token
    HEADERS = {
        "Authorization": f"{token_type} {API_TOKEN}",
        "Accept": "application/json"
    }
    
    # Calculate expiry time
    current_time = datetime.now(timezone.utc)
    expiry_time = current_time + timedelta(seconds=expires_in)
    
    # Display token information
    print("📋 Token Details:")
    print(f"  Token Type: {token_type}")
    print(f"  Expires In: {expires_in} seconds ({expires_in / 3600:.2f} hours)")
    print(f"  Current Time (UTC): {current_time.strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"  Expiry Time (UTC): {expiry_time.strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"  Token (first 50 chars): {API_TOKEN[:50]}...")
    print(f"\n✓ Authorization headers updated")
else:
    print("⚠️  Failed to retrieve token. Please check your credentials.")

def get_last_job() -> Optional[Dict]:
    """
    Fetch the most recent job from the printer.
    
    Returns:
        Dictionary containing the last job's details or None if request fails
    """
    url = f"{API_BASE_URL}/jobs/last"
    try:
        response = requests.get(url, headers=HEADERS, verify=False, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"❌ Error fetching last job: {e}")
        return None
    
def get_user_message(jobId : str, limit: int = 20) -> Optional[Dict]:
    """
    Fetch the current user message displayed on the printer.
    
    Returns:
        Dictionary containing the user message or None if request fails
    """
    url = f"{API_BASE_URL}/software/usermessages"
    params = {"jobId": jobId, "limit":limit}
    try:
        response = requests.get(url, headers=HEADERS, params=params, verify=False, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"❌ Error fetching user message: {e}")
        return None

def get_all_jobs(limit: int = 10, from_date: Optional[str] = None, to_date: Optional[str] = None) -> Optional[list]:
    """
    Fetch a list of recent jobs.
    
    Args:
        limit: Maximum number of jobs to retrieve
        from_date: Filter jobs starting from this date (ISO 8601 format, e.g., '2024-01-01T00:00:00Z')
        to_date: Filter jobs up to this date (ISO 8601 format, e.g., '2024-12-31T23:59:59Z')
        
    Returns:
        List of job summaries or None if request fails
    """
    url = f"{API_BASE_URL}/jobs"
    params = {"take": limit}
    
    # Add optional date filters if provided
    if from_date:
        params["from"] = from_date
    if to_date:
        params["to"] = to_date
    
    try:
        response = requests.get(url, headers=HEADERS, params=params, verify=False, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"❌ Error fetching jobs list: {e}")
        return None


print("✓ API functions defined")

import pandas as pd
from datetime import datetime, timedelta
from ui import create_job_browser_ui

def fetch_jobs(from_date, to_date):
    """
    Fetch jobs within the specified date range and display them in a table.
    
    Args:
        from_date: Start date for filtering jobs
        to_date: End date for filtering jobs
        
    Returns:
        Pandas DataFrame with job information
    """
    # Convert dates to ISO 8601 format with time
    from_datetime = f"{from_date}T00:00:00.000Z" if from_date else None
    to_datetime = f"{to_date}T23:59:59.999Z" if to_date else None
    
    # Fetch jobs from API
    refresh_token()
    jobs = get_all_jobs(limit=500, from_date=from_datetime, to_date=to_datetime)
    
    if not jobs:
        return pd.DataFrame(columns=['id', 'result', 'timeStarted', 'timeEnded', 'task', 'material'])
    
    # Extract relevant fields
    job_data = []
    for job in jobs:
        job_data.append({
            'id': job.get('id', 'N/A'),
            'result': job.get('result', 'N/A'),
            'timeStarted': job.get('timeStarted', 'N/A'),
            'timeEnded': job.get('timeEnded', 'N/A'),
            'task': job.get('task', 'N/A'),
            'material': job.get('material', 'N/A')
        })
    
    # Create DataFrame
    df = pd.DataFrame(job_data)
    return df

# Launch the interface
demo = create_job_browser_ui(fetch_jobs)
demo.launch(share=False, inline=True)

def check_last_job_status(last_job_id: Optional[str] = None) -> Optional[Dict]:
    """
    Check the status of the last job.
    
    Args:
        last_job_id: Optional job ID to compare against. If provided and matches
                     the current last job, returns None (no change detected)
    
    Returns:
        Dictionary containing:
        - 'result': Job result status (Pending, Successful, Failed, UserCanceled)
        - 'jobId': The job ID
        - 'task': The task name
        - 'timeStarted': When the job started
        - 'timeEnded': When the job ended (None if still running)
        - 'userMessages': List of user messages (only for failed jobs)
        
        Returns None if last_job_id matches the current last job ID
    """
    refresh_token()
    # Fetch the last job
    last_job = get_last_job()
    
    if not last_job:
        return {
            'result': 'Error',
            'jobId': None,
            'task': None,
            'timeStarted': None,
            'timeEnded': None,
            'userMessages': None,
            'error': 'Failed to fetch last job'
        }
    
    job_id = last_job.get('id', 'Unknown')
    
    # If last_job_id is provided and matches current job, return None
    if last_job_id is not None and job_id == last_job_id:
        return None
    
    result = last_job.get('result', 'Unknown')
    task = last_job.get('task', 'N/A')
    time_started = last_job.get('timeStarted', 'N/A')
    time_ended = last_job.get('timeEnded', 'N/A')
    
    # Prepare base response
    response = {
        'result': result,
        'jobId': job_id,
        'task': task,
        'timeStarted': time_started,
        'timeEnded': time_ended,
        'userMessages': None
    }
    
    # If job failed or was canceled, fetch user messages
    if result in ['Failed', 'UserCanceled']:
        user_messages = get_user_message(job_id, limit=20)
        response['userMessages'] = user_messages if user_messages else []
    
    return response


print("✓ Last job status check function defined")

import pandas as pd
from ui import create_job_status_ui

# Track the last job ID
last_job_id_tracker = None

def fetch_job_status():
    """
    Fetch the last job status and format it for display.
    Returns a tuple of (job_info_df, user_messages_df)
    """
    global last_job_id_tracker
    
    result = check_last_job_status(last_job_id_tracker)
    
    if result is None:
        # No new job detected
        return (
            pd.DataFrame([{"Info": f"No new job detected or job status unchanged. Tracked ID: {last_job_id_tracker}"}]),
            pd.DataFrame()
        )
    
    # Update tracker
    last_job_id_tracker = result.get('jobId')
    
    # Create job info DataFrame
    job_info = {
        'Field': ['Job ID', 'Task', 'Result', 'Time Started', 'Time Ended'],
        'Value': [
            result.get('jobId', 'N/A'),
            result.get('task', 'N/A'),
            result.get('result', 'N/A'),
            result.get('timeStarted', 'N/A'),
            result.get('timeEnded', 'N/A')
        ]
    }
    job_info_df = pd.DataFrame(job_info)
    
    # Create user messages DataFrame
    user_messages = result.get('userMessages')
    if user_messages:
        messages_data = []
        for msg in user_messages:
            messages_data.append({
                'Severity': msg.get('severity', 'N/A'),
                'timeRaised': msg.get('timeRaised', 'N/A'),
                'message': msg.get('message', 'N/A'),
                'additionalInfo': msg.get('additionalInfo', 'N/A')
            })
        user_messages_df = pd.DataFrame(messages_data)
    else:
        user_messages_df = pd.DataFrame([{"Info": "No user messages available"}])
    
    return job_info_df, user_messages_df

def clear_last_job_id_tracker():
    global last_job_id_tracker 
    last_job_id_tracker = None
    return [],[]
    

# Create Gradio interface for job status
job_status_demo = create_job_status_ui(fetch_job_status, clear_last_job_id_tracker)

# Launch the interface
job_status_demo.launch(share=False, inline=True)

Die API-Struktur verstehen¶

1. AlleJobs abrufen¶

2. SpezifischeJobdetails abrufen¶

3. Benutzernachrichten für einenAuftrag abrufen¶

Einrichten derUmgebung¶

Konfiguration¶

Einstellung derVariablen¶

API-Anmeldedaten laden¶

Abrufen desZugriffstokens¶

Schritt 1: Verbindung zur API herstellen¶

Schritt 2: Testen derVerbindung¶

Schritt 3:Logik zur Überwachung von Aufträgen¶

Anhang:¶

Anzeigen der EOSCONNECT Core -Dokumentation mitdem OpenAPIEditor¶

SelbstsignierteZertifikate verstehen¶