demandes d'importation demandes
importation temps
import json
from datetime import datetime, timezone
from typing import Dict, Optional, Tuple
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from dotenv import load_dotenv
import os
from datetime import timedelta

# À des fins de démonstration, nous désactiverons les avertissements SSL
# En production, utilisez des certificats SSL appropriés !
import urllib3
urllib3.disable_warnings(urllib3.exceptions.Avertissement de requête non sécurisée)

imprimer(« ✓ Tous les paquets ont été chargés avec succès ! »)

# 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)

Comprendre lastructure de l'API¶

1. Obtenir tousles emplois¶

2. Obtenirdes détails spécifiques sur le poste¶

3. Obtenir les messages utilisateur pour unetâche¶

Configuration de l'environnement¶

Configuration¶

Définition desvariables¶

Chargementdes informations d'identification API¶

Récupération dujeton d'accès¶

Étape 1 : Connexion àl'API¶

Étape 2 : Test de laconnexion¶

Étape 3 :Logique de surveillance des tâches¶

Annexe :¶

Affichage de la documentation EOSCONNECT Core avecl'éditeur OpenAPI¶

Comprendreles certificats auto-signés¶