Spaces:
Sleeping
Sleeping
File size: 3,591 Bytes
922f271 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 |
"""
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
|