utils.py

import os
import re
import ast
import sys
import yaml
import json
import logging
import pandas as pd
from typing import Dict
from pathlib import Path

def load_system_prompt(system_prompt_file):
    if system_prompt_file and os.path.exists(system_prompt_file):
        with open(system_prompt_file, "r", encoding="utf-8") as f:
            text = f.read()
        match = re.search(r"```\n?(.*?)```", text, re.DOTALL)
        if match:
            return match.group(1).strip()
        return text.strip()
    else:
        raise FileNotFoundError(f"System prompt file not found: {system_prompt_file}")

def load_config(config_path: str = None) -> dict:
    if config_path is None:
        config_path = Path(__file__).resolve().parent.parent / "configs/config.yaml"

    if not Path(config_path).exists():
        raise FileNotFoundError(
            f"Config file not found: {config_path}\n"
        )

    with open(config_path, "r", encoding="utf-8") as f:
        cfg = yaml.safe_load(f) or {}

    if not isinstance(cfg, dict):
        raise ValueError(
            f"Config file must contain a YAML mapping (dict), got {type(cfg).__name__}. "
        )

    # Normalize None-valued sections to empty dicts so chained .get() never
    # fails with AttributeError (e.g. `llm:` with no sub-keys → None).
    # Recurse into nested dicts so `paths:\n  vector_db:` also gets normalized.
    def _normalize_nulls(d):
        for key in list(d.keys()):
            if d[key] is None:
                d[key] = {}
            elif isinstance(d[key], dict):
                _normalize_nulls(d[key])
    _normalize_nulls(cfg)

    return cfg

def parse_llm_json(raw_text):
    if not raw_text or not raw_text.strip():
        return None, "Empty response"
    text = raw_text.strip()
    # Strip markdown fences
    fence_match = re.search(r"```(?:json)?\s*\n?(.*?)```", text, re.DOTALL)
    if fence_match:
        text = fence_match.group(1).strip()
 
    # Attempt 1: Standard JSON parsing (double quotes)
    try:
        return json.loads(text), None
    except json.JSONDecodeError:
        pass
 
    # Attempt 2: Extract JSON object from surrounding text
    brace_match = re.search(r"\{.*\}", text, re.DOTALL)
    if brace_match:
        extracted = brace_match.group(0)
        try:
            return json.loads(extracted), None
        except json.JSONDecodeError:
            pass
 
        # Attempt 3: Python dict syntax (single quotes) — common with Llama models.
        # ast.literal_eval safely parses Python literals including single-quoted dicts.
        # This handles: {'commodity': 'Coffee', 'intent': 'Crop Nutrition'}
        # and also apostrophes: {'reasoning': "The farmer's crop..."}
        try:
            result = ast.literal_eval(extracted)
            if isinstance(result, dict):
                return result, None
        except (ValueError, SyntaxError):
            pass
 
    return None, f"Could not parse JSON from: {text[:200]}"

def setup_logging(output_dir, log_file):
    log_path = os.path.join(output_dir, log_file)
    # 1. Create a custom module-level logger
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.INFO)  # Capture everything at the logger level

    # 2. Create a handler to write logs to a file
    file_handler = logging.FileHandler(log_path, mode="a")
    file_handler.setLevel(logging.INFO)  # File only gets errors and criticals

    # 3. Create a handler to output logs to the console
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)  # Console gets Info, Warning, Error, Critical

    # 4. Define a reusable string format for your log outputs
    formatter = logging.Formatter('%(asctime)s | %(name)s | %(levelname)s | %(message)s')
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)

    # 5. Attach the configured handlers to your custom logger
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger

def validate_classification(parsed, classification_fields:Dict):
    errors = []
    if parsed is None:
        return False, ["Parsed result is None"]
    
    for field in classification_fields.keys():
        if field not in parsed:
            errors.append(f"Missing field: {field}")
            continue
        if parsed.get(field) not in classification_fields[field]:
            errors.append(f"Invalid {field}: {parsed.get(field)}")
    return len(errors) == 0, errors

def load_checkpoint(cfg, logger):
    ckpt_path = os.path.join(cfg.output_dir, cfg.checkpoint_file)
    if os.path.exists(ckpt_path):
        df = pd.read_parquet(ckpt_path)
        logger.info(f"RESUMING from checkpoint: {len(df)} queries already classified")
        return df, len(df)
    return None, 0


def save_checkpoint(df, cfg, logger):
    ckpt_path = os.path.join(cfg.output_dir, cfg.checkpoint_file)
    df.to_parquet(ckpt_path, index=False)
    logger.info(f"Checkpoint saved: {len(df)} queries")

def load_queries(cfg, logger):
    # --- Load input data ---
    logger.info(f"Loading data from {cfg.input_path}")
    df_input = pd.read_csv(cfg.input_path) if cfg.input_path.split('.')[-1] == 'csv' else pd.read_parquet(cfg.input_path)

    size = cfg.test_size if cfg.test_size>=0 else df_input.shape[0]
    if cfg.shuffle:
        df_input = df_input.sample(n=size, random_state=42).reset_index(drop=True)

    queries = df_input[cfg.query_column].tolist()
    
    if cfg.classification_type==2:
        assert cfg.response_column is not None, 'Adoption phase analysis requires LLM responses. Specify response column.'
        responses = df_input[cfg.response_column].tolist()
        queries = ['\n\n'.join([f"## Query\n\n{query}", f"### Response\n\n{response}"]) for query, response in zip(queries, responses)]

    return df_input, queries