resource_access.py

"""
Resource access utilities for managing local files
"""
import os
import logging
import json
from typing import Dict, List, Optional, Any, Union

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Constants
RESOURCE_FOLDER = os.path.join(os.path.dirname(os.path.abspath(__file__)), "resource")
METADATA_FILE = os.path.join(RESOURCE_FOLDER, "metadata.jsonl")

class ResourceAccess:
    """Manages access to resource files"""
    
    @staticmethod
    def list_resources() -> List[str]:
        """List all files in the resources directory"""
        try:
            return [f for f in os.listdir(RESOURCE_FOLDER) if os.path.isfile(os.path.join(RESOURCE_FOLDER, f))]
        except Exception as e:
            logger.error(f"Error listing resources: {e}")
            return []
    
    @staticmethod
    def get_file_path(filename: str) -> Optional[str]:
        """Get the full path for a specific file"""
        file_path = os.path.join(RESOURCE_FOLDER, filename)
        return file_path if os.path.exists(file_path) else None
    
    @staticmethod
    def get_files_by_extension(extension: str) -> List[str]:
        """Get a list of files with a specific extension"""
        if not extension.startswith('.'):
            extension = f".{extension}"
        
        try:
            return [f for f in os.listdir(RESOURCE_FOLDER) 
                   if os.path.isfile(os.path.join(RESOURCE_FOLDER, f)) 
                   and f.lower().endswith(extension.lower())]
        except Exception as e:
            logger.error(f"Error getting files by extension {extension}: {e}")
            return []
    
    @staticmethod
    def load_metadata() -> Dict[str, Dict]:
        """Load metadata from the metadata.jsonl file"""
        metadata = {}
        
        try:
            with open(METADATA_FILE, 'r', encoding='utf-8') as f:
                for line in f:
                    data = json.loads(line.strip())
                    task_id = data.get('task_id')
                    if task_id:
                        metadata[task_id] = data
            logger.info(f"Loaded {len(metadata)} entries from metadata")
        except Exception as e:
            logger.error(f"Error loading metadata: {e}")
        
        return metadata
    
    @staticmethod
    def find_tasks_with_file(filename: str) -> List[Dict]:
        """Find tasks that reference a specific file"""
        tasks = []
        
        try:
            metadata = ResourceAccess.load_metadata()
            for task_id, data in metadata.items():
                if 'resources' in data:
                    for resource in data['resources']:
                        if resource.get('file_name') == filename:
                            tasks.append(data)
                            break
        except Exception as e:
            logger.error(f"Error finding tasks with file {filename}: {e}")
        
        return tasks
    
    @staticmethod
    def file_exists(filename: str) -> bool:
        """Check if a file exists in the resources directory"""
        return os.path.exists(os.path.join(RESOURCE_FOLDER, filename))
    
    @staticmethod
    def get_metadata_for_task(task_id: str) -> Optional[Dict]:
        """Get metadata for a specific task ID"""
        try:
            metadata = ResourceAccess.load_metadata()
            return metadata.get(task_id)
        except Exception as e:
            logger.error(f"Error getting metadata for task {task_id}: {e}")
            return None