Error Handling Methods

import traceback

def identify_error():
    try:
        # Execute structured processing
        result = workflows.run(custom_dataset_function)
    except Exception as e:
        print(f"Error Type: {type(e).__name__}")
        print(f"Error Message: {str(e)}")
        print(f"Detailed Traceback:")
        traceback.print_exc()

import os

def check_file_exists(file_path):
    if not os.path.exists(file_path):
        print(f"File not found: {file_path}")
        # Suggest alternative file paths
        alternatives = [
            file_path.replace('.csv', '.xlsx'),
            os.path.join('data', os.path.basename(file_path))
        ]
        for alt in alternatives:
            if os.path.exists(alt):
                print(f"Alternative file found: {alt}")
                return alt
        return None
    return file_path

# Set directory permissions
chmod 755 data/
chmod 755 data/structured/
chmod 755 data/logs/

# Set file permissions
chmod 644 data/invoice/invoice.json
chmod 644 data/tasksupport/*.json

import json

def validate_json_file(file_path):
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
        print(f"✅ {file_path} is valid JSON")
        return data
    except json.JSONDecodeError as e:
        print(f"❌ JSON format error in {file_path}:")
        print(f"   Line {e.lineno}, Column {e.colno}: {e.msg}")
        return None
    except FileNotFoundError:
        print(f"❌ File not found: {file_path}")
        return None

def validate_against_schema(data, schema_path):
    try:
        with open(schema_path, 'r') as f:
            schema = json.load(f)

        # Check required fields
        if 'required' in schema:
            for field in schema['required']:
                if field not in data:
                    print(f"❌ Missing required field: {field}")
                    return False

        print("✅ Schema validation passed")
        return True
    except Exception as e:
        print(f"❌ Schema validation error: {e}")
        return False

from rdetoolkit.exceptions import StructuredError

def dataset_with_error_handling(srcpaths, resource_paths):
    try:
        # File reading process
        config = read_config_file("config.json")
    except FileNotFoundError as e:
        # Set RDE error information
        raise StructuredError(
            "Configuration file not found", 
            ecode=3, 
            eobj=e
        ) from e
    except json.JSONDecodeError as e:
        raise StructuredError(
            "Configuration file format is incorrect", 
            ecode=4, 
            eobj=e
        ) from e

    # Normal processing
    return process_data(config)

from rdetoolkit.errors import catch_exception_with_message

@catch_exception_with_message(
    error_message="An unexpected error occurred", 
    error_code=100, 
    verbose=False
)
def dataset_with_decorator(srcpaths, resource_paths):
    # Processing logic
    return process_data()

import logging

def setup_detailed_logging():
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler('debug.log'),
            logging.StreamHandler()
        ]
    )

    logger = logging.getLogger(__name__)
    return logger

def debug_processing(srcpaths, resource_paths):
    logger = setup_detailed_logging()

    logger.info("Starting structured processing")
    logger.debug(f"Input paths: {srcpaths}")
    logger.debug(f"Output paths: {resource_paths}")

    try:
        # Execute processing
        result = your_processing_logic()
        logger.info("Processing completed successfully")
        return result
    except Exception as e:
        logger.error(f"Error occurred during processing: {e}")
        logger.debug("Detailed traceback:", exc_info=True)
        raise

def check_error_file():
    error_file = "job.failed"
    if os.path.exists(error_file):
        with open(error_file, 'r') as f:
            content = f.read()
        print(f"Error information:\n{content}")
        return False
    else:
        print("✅ No error file exists (normal completion)")
        return True

# Check latest log entries
tail -n 20 data/logs/rdesys.log

# Search for error messages
grep -i "error" data/logs/rdesys.log

# Search for warning messages
grep -i "warning" data/logs/rdesys.log