from pathlib import Path from typing import List import yaml from utils.constants import DEFAULT_COMPRESSION_LEVELS def load_papers_config(config_path: str) -> List[dict]: """ Load papers configuration from YAML file. Args: config_path: Path to YAML configuration file Returns: List of paper dicts with normalized fields: path, name, token_limits Raises: FileNotFoundError: If config file doesn't exist ValueError: If config is invalid or required fields missing """ config_file = Path(config_path) if not config_file.exists(): raise FileNotFoundError(f"Config file not found: {config_path}") try: with open(config_file, "r") as f: config = yaml.safe_load(f) except yaml.YAMLError as e: raise ValueError(f"Invalid YAML in {config_path}: {e}") if not config or "papers" not in config: raise ValueError("Config must contain 'papers' key") papers = config["papers"] if not isinstance(papers, list): raise ValueError("'papers' must be a list") normalized_papers = [] for idx, entry in enumerate(papers): try: normalized = _normalize_paper_entry(entry) normalized_papers.append(normalized) except ValueError as e: raise ValueError(f"Invalid paper entry at index {idx}: {e}") return normalized_papers def _normalize_paper_entry(entry: dict) -> dict: """ Normalize and validate a single paper entry. Args: entry: Paper entry dict from YAML Returns: Normalized entry with all required fields and sensible defaults Raises: ValueError: If required fields missing or invalid """ if not isinstance(entry, dict): raise ValueError("Paper entry must be a dictionary") # Validate required field: path if "path" not in entry: raise ValueError("Missing required field: 'path'") path = entry["path"] if not isinstance(path, str): raise ValueError("'path' must be a string") if not Path(path).exists(): raise ValueError(f"Paper path does not exist: {path}") # Optional field: name (default to filename) name = entry.get("name") if name is None: name = Path(path).name # Optional field: token_limits (default to DEFAULT_COMPRESSION_LEVELS) token_limits = entry.get("token_limits") if token_limits is None: token_limits = DEFAULT_COMPRESSION_LEVELS else: if not isinstance(token_limits, list): raise ValueError("'token_limits' must be a list") for limit in token_limits: if not isinstance(limit, int) or limit <= 0: raise ValueError( f"Token limits must be positive integers, got: {limit}" ) # Optional field: token_tolerances (tolerance values for each token_limit) token_tolerances = entry.get("token_tolerances") if token_tolerances is not None: if not isinstance(token_tolerances, list): raise ValueError("'token_tolerances' must be a list") for tolerance in token_tolerances: if not isinstance(tolerance, int) or tolerance < 0: raise ValueError( f"Token tolerances must be non-negative integers, got: {tolerance}" ) # Optional field: context_pdfs (list of reference PDFs) context_pdfs = entry.get("context_pdfs") if context_pdfs is not None: if not isinstance(context_pdfs, list): raise ValueError("'context_pdfs' must be a list") for pdf_path in context_pdfs: if not isinstance(pdf_path, str): raise ValueError("context_pdfs must be a list of strings") if not Path(pdf_path).exists(): raise FileNotFoundError(f"Context PDF not found: {pdf_path}") return { "path": path, "name": name, "token_limits": token_limits, "token_tolerances": token_tolerances, "description": entry.get("description", ""), "context_pdfs": context_pdfs or [], }