""" 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