NPC Sys environment

def _git(args, cwd, timeout=15):
    """Run a git command and return stdout."""
    result = subprocess.run(
        ["git"] + args,
        cwd=cwd,
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or f"git {args[0]} failed")
    return result.stdout.strip()

def _migrate_dirs_to_npc_team() -> None:
    """One-time migration: move top-level resource dirs into npc_team/.

    Runs only when the marker file is absent and old dirs exist.
    """
    data_dir = get_data_dir()
    marker = os.path.join(data_dir, '.dirs_migrated_to_npc_team')
    if os.path.isfile(marker):
        return

    old_dirs = ["images", "models", "attachments", "mcp_servers",
                "jobs", "triggers", "videos", "logs"]
    # Only bother if at least one old top-level dir has content
    needs_migration = any(
        os.path.isdir(os.path.join(data_dir, d)) and os.listdir(os.path.join(data_dir, d))
        for d in old_dirs
        if os.path.isdir(os.path.join(data_dir, d))
    )
    if not needs_migration:
        # No old dirs with content — just write marker and return
        try:
            os.makedirs(data_dir, exist_ok=True)
            with open(marker, 'w') as f:
                f.write('No migration needed\n')
        except OSError:
            pass
        return

    try:
        from migrations.migrate_dirs_to_npc_team import run_migration
        run_migration(data_dir, dry_run=False)
    except ImportError:
        # Inline fallback if the migration module isn't on sys.path
        import shutil
        npc_team_dir = os.path.join(data_dir, 'npc_team')
        os.makedirs(npc_team_dir, exist_ok=True)
        for dirname in old_dirs:
            old = os.path.join(data_dir, dirname)
            new = os.path.join(npc_team_dir, dirname)
            if not os.path.isdir(old) or not os.listdir(old):
                if os.path.isdir(old) and not os.listdir(old):
                    try:
                        os.rmdir(old)
                    except OSError:
                        pass
                continue
            os.makedirs(new, exist_ok=True)
            for item in os.listdir(old):
                src = os.path.join(old, item)
                dst = os.path.join(new, item)
                if not os.path.exists(dst):
                    shutil.move(src, dst)
            if not os.listdir(old):
                try:
                    os.rmdir(old)
                except OSError:
                    pass
        with open(marker, 'w') as f:
            f.write('Migration applied: top-level dirs moved into npc_team/\n')

def check_internet_connection(timeout=5):
    """
    Checks for internet connectivity by trying to connect to a well-known host.
    """
    try:
        socket.create_connection(("8.8.8.8", 53), timeout=timeout)
        return True
    except OSError:
        return False

def ensure_dirs_exist(*dirs):
    """Ensure all specified directories exist"""
    for dir_path in dirs:
        os.makedirs(os.path.expanduser(dir_path), exist_ok=True)

def ensure_npcsh_dirs() -> None:
    """Ensure all npcsh directories exist, migrating old layouts if needed."""
    _migrate_dirs_to_npc_team()
    for dir_path in [get_data_dir(), get_config_dir(), get_cache_dir(),
                     get_models_dir(), get_images_dir(), get_jobs_dir(),
                     get_triggers_dir(), get_videos_dir(), get_attachments_dir(),
                     get_logs_dir()]:
        os.makedirs(dir_path, exist_ok=True)

def get_attachments_dir() -> str:
    """Get the directory for attachments."""
    return os.path.join(get_data_dir(), 'npc_team', 'attachments')

def get_cache_dir() -> str:
    """
    Get the platform-specific cache directory for npcsh.

    Returns:
        - Linux: $XDG_CACHE_HOME/npcsh or ~/.cache/npcsh
        - macOS: ~/Library/Caches/npcsh
        - Windows: %LOCALAPPDATA%/npcsh/cache
    """
    if ON_WINDOWS:
        base = os.environ.get('LOCALAPPDATA', os.path.expanduser('~/AppData/Local'))
        return os.path.join(base, 'npcsh', 'cache')
    elif ON_MACOS:
        return os.path.expanduser('~/Library/Caches/npcsh')
    else:
        xdg_cache = os.environ.get('XDG_CACHE_HOME', os.path.expanduser('~/.cache'))
        return os.path.join(xdg_cache, 'npcsh')

def get_config_dir() -> str:
    """
    Get the platform-specific config directory for npcsh.

    Returns:
        - Linux: $XDG_CONFIG_HOME/npcsh or ~/.config/npcsh
        - macOS: ~/Library/Application Support/npcsh (same as data on macOS)
        - Windows: %APPDATA%/npcsh or ~/AppData/Roaming/npcsh

    Falls back to ~/.npcsh for backwards compatibility if the new location
    doesn't exist but the old one does.
    """
    if ON_WINDOWS:
        base = os.environ.get('APPDATA', os.path.expanduser('~/AppData/Roaming'))
        new_path = os.path.join(base, 'npcsh')
    elif ON_MACOS:
        new_path = os.path.expanduser('~/Library/Application Support/npcsh')
    else:
        xdg_config = os.environ.get('XDG_CONFIG_HOME', os.path.expanduser('~/.config'))
        new_path = os.path.join(xdg_config, 'npcsh')

    old_path = os.path.expanduser('~/.npcsh')
    if os.path.exists(old_path) and not os.path.exists(new_path):
        return old_path

    return new_path

def get_data_dir() -> str:
    """
    Get the data directory for npcsh.

    If INCOGNIDE_HOME is set, use that directly.
    Otherwise:
        - Linux: $XDG_DATA_HOME/npcsh or ~/.local/share/npcsh
        - macOS: ~/Library/Application Support/npcsh
        - Windows: %LOCALAPPDATA%/npcsh or ~/AppData/Local/npcsh

    Falls back to ~/.npcsh for backwards compatibility if the new location
    doesn't exist but the old one does.
    """
    npcsh_home = os.environ.get('INCOGNIDE_HOME')
    if npcsh_home:
        return os.path.expanduser(npcsh_home)

    if ON_WINDOWS:
        base = os.environ.get('LOCALAPPDATA', os.path.expanduser('~/AppData/Local'))
        new_path = os.path.join(base, 'npcsh')
    elif ON_MACOS:
        new_path = os.path.expanduser('~/Library/Application Support/npcsh')
    else:
        xdg_data = os.environ.get('XDG_DATA_HOME', os.path.expanduser('~/.local/share'))
        new_path = os.path.join(xdg_data, 'npcsh')

    old_path = os.path.expanduser('~/.npcsh')
    if os.path.exists(old_path) and not os.path.exists(new_path):
        return old_path

    return new_path

def get_db_npcs(db_path: str) -> List[str]:
    """
    Function Description:
        This function retrieves a list of valid NPCs from the database.
    Args:
        db_path: The path to the database file.
    Keyword Args:
        None
    Returns:
        A list of valid NPCs.
    """
    if "~" in db_path:
        db_path = os.path.expanduser(db_path)
    db_conn = sqlite3.connect(db_path)
    cursor = db_conn.cursor()
    cursor.execute("SELECT name FROM compiled_npcs")
    npcs = [row[0] for row in cursor.fetchall()]
    db_conn.close()
    return npcs

def get_directory_npcs(directory: str = None) -> List[str]:
    """
    Function Description:
        This function retrieves a list of valid NPCs from the database.
    Args:
        db_path: The path to the database file.
    Keyword Args:
        None
    Returns:
        A list of valid NPCs.
    """
    if directory is None:
        directory = os.path.expanduser("./npc_team")
    npcs = []
    for filename in os.listdir(directory):
        if filename.endswith(".npc"):
            npcs.append(filename[:-4])
    return npcs

def get_history_db_path() -> str:
    """
    Get the path to the history database.

    Returns the platform-appropriate database path.
    Falls back to ~/npcsh_history.db for backwards compatibility.
    """
    old_path = os.path.expanduser('~/npcsh_history.db')
    if os.path.exists(old_path):
        return old_path

    data_dir = get_data_dir()
    return os.path.join(data_dir, 'history.db')

def get_images_dir() -> str:
    """Get the directory for storing generated images."""
    return os.path.join(get_data_dir(), 'npc_team', 'images')

def get_jobs_dir() -> str:
    """Get the directory for cron/scheduled jobs."""
    return os.path.join(get_data_dir(), 'npc_team', 'jobs')

def get_locally_available_models(project_directory, airplane_mode=False):
    available_models = {}
    env_path = os.path.join(project_directory, ".env")
    env_vars = {}
    if os.path.exists(env_path):
        with open(env_path, "r") as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith("#"):
                    if "=" in line:
                        key, value = line.split("=", 1)
                        env_vars[key.strip()] = value.strip().strip("\"'")

    internet_available = check_internet_connection()
    if not internet_available:
        logging.info("No internet connection detected. External API calls will be skipped.")
        airplane_mode = True
    else:
        logging.info("Internet connection detected. Proceeding based on 'airplane_mode' parameter.")

    custom_providers = load_custom_providers()

    for provider_name, config in custom_providers.items():
        api_key_var = config.get('api_key_var')
        if not api_key_var:
            api_key_var = f"{provider_name.upper()}_API_KEY"

        if api_key_var in env_vars or os.environ.get(api_key_var):
            try:
                import requests

                def fetch_custom_models():
                    base_url = config.get('base_url', '')
                    headers = config.get('headers', {})

                    api_key = env_vars.get(api_key_var) or \
                              os.environ.get(api_key_var)
                    if api_key:
                        headers['Authorization'] = f'Bearer {api_key}'

                    models_endpoint = f"{base_url.rstrip('/')}/models"
                    response = requests.get(
                        models_endpoint, 
                        headers=headers,
                        timeout=3.5
                    )

                    if response.status_code == 200:
                        data = response.json()

                        if isinstance(data, dict) and 'data' in data:
                            return [
                                m['id'] for m in data['data'] 
                                if 'id' in m
                            ]
                        elif isinstance(data, list):
                            return [
                                m['id'] for m in data 
                                if isinstance(m, dict) and 'id' in m
                            ]
                    return []

                models = fetch_custom_models()
                for model in models:
                    available_models[model] = 'openai-like'

                logging.info(f"Loaded {len(models)} models from custom provider '{provider_name}'")

            except Exception as e:
                logging.warning(f"Failed to load models from custom provider '{provider_name}': {e}")


    airplane_mode = False
    if not airplane_mode:
        timeout_seconds = 3.5


        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:

            if 'NPCSH_API_URL' in env_vars or os.environ.get('NPCSH_API_URL'):
              try:
                  import requests

                  def fetch_custom_models():
                      base_url = env_vars.get('NPCSH_API_URL') or os.environ.get('NPCSH_API_URL')                      
                      models_endpoint = f"{base_url.rstrip('/')}/models"
                      response = requests.get(
                          models_endpoint, 

                          timeout=3.5
                      )

                      if response.status_code == 200:
                          data = response.json()

                          if isinstance(data, dict) and 'data' in data:
                              return [
                                  m['id'] for m in data['data'] 
                                  if 'id' in m
                              ]
                          elif isinstance(data, list):
                              return [
                                  m['id'] for m in data 
                                  if isinstance(m, dict) and 'id' in m
                              ]
                      return []

                  models = fetch_custom_models()
                  for model in models:
                      available_models[model] = 'openai-like'




              except Exception as e:
                  logging.warning(f"Failed to load models from custom provider 'openai-like': {e}")


            if "ANTHROPIC_API_KEY" in env_vars or os.environ.get("ANTHROPIC_API_KEY"):
                try:
                    import anthropic

                    def fetch_anthropic_models():
                        client = anthropic.Anthropic(api_key=env_vars.get("ANTHROPIC_API_KEY") or os.environ.get("ANTHROPIC_API_KEY"))


                        return client.models.list()

                    future = executor.submit(fetch_anthropic_models)
                    models = future.result(timeout=timeout_seconds) 

                    for model in models.data:
                        available_models[model.id] = 'anthropic'

                except (ImportError, concurrent.futures.TimeoutError, Exception) as e:
                    logging.info(f"Anthropic models not indexed or timed out: {e}")

            if "OPENAI_API_KEY" in env_vars or os.environ.get("OPENAI_API_KEY"):
                try:
                    import openai

                    def fetch_openai_models():
                        openai.api_key = env_vars.get("OPENAI_API_KEY", None) or os.environ.get("OPENAI_API_KEY", None)
                        return openai.models.list()

                    future = executor.submit(fetch_openai_models)
                    models = future.result(timeout=timeout_seconds) 

                    for model in models.data:
                        if (
                            (
                                "gpt" in model.id
                                or "o1" in model.id
                                or "o3" in model.id
                                or "chat" in model.id
                            )
                            and "audio" not in model.id
                            and "realtime" not in model.id
                        ):
                            available_models[model.id] = "openai"
                except (ImportError, openai.APIError, concurrent.futures.TimeoutError, Exception) as e:
                    logging.info(f"OpenAI models not indexed or timed out: {e}")

            if "GEMINI_API_KEY" in env_vars or os.environ.get("GEMINI_API_KEY"):
                try:
                    from google import genai
                    def fetch_gemini_models():
                        client = genai.Client(api_key=env_vars.get("GEMINI_API_KEY") or os.environ.get("GEMINI_API_KEY"))
                        found_models = []

                        target_models = [
                            'gemini-2.5-pro', 
                            'gemini-2.5-flash', 
                            'gemini-2.0-flash', 
                            'gemini-2.0-pro', 
                            'gemini-1.5-pro', 
                            'gemini-1.5-flash'
                            'gemini-3-flash-preview',
                            'gemini-3.1-pro-preview',
                        ]

                        for m in client.models.list():
                            for action in m.supported_actions:
                                if action == "generateContent":
                                    if 'models/' in m.name:
                                        model_name_part = m.name.split('/')[1]  

                                        if any(model in model_name_part for model in target_models):
                                            found_models.append(model_name_part)
                        return set(found_models)
                    future = executor.submit(fetch_gemini_models)
                    models = future.result(timeout=timeout_seconds) 

                    for model in models: 
                        if "gemini" in model:
                            available_models[model] = "gemini"
                except (ImportError, concurrent.futures.TimeoutError, Exception) as e:
                    logging.info(f"Gemini models not indexed or timed out: {e}")

            if "DEEPSEEK_API_KEY" in env_vars or os.environ.get("DEEPSEEK_API_KEY"):
                available_models['deepseek-chat'] = 'deepseek'
                available_models['deepseek-reasoner'] = 'deepseek'        
    try:
        import ollama
        timeout_seconds = 0.5 
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ollama_executor:
            def fetch_ollama_models():
                return ollama.list()

            future = ollama_executor.submit(fetch_ollama_models)
            models = future.result(timeout=timeout_seconds) 

        for model in models.models:
            if "embed" not in model.model:
                mod = model.model
                available_models[mod] = "ollama"
    except (ImportError, concurrent.futures.TimeoutError, Exception) as e:
        logging.info(f"Error loading Ollama models or timed out: {e}")

    models_dir = get_models_dir()
    gguf_dirs = [
        os.path.join(models_dir, 'gguf'),
        models_dir,
        os.path.expanduser('~/models'),
        os.path.join(get_cache_dir(), 'huggingface/hub'),
        os.path.expanduser('~/.cache/huggingface/hub'),
    ]
    env_gguf_dir = os.environ.get('NPCSH_GGUF_DIR')
    if env_gguf_dir:
        gguf_dirs.insert(0, os.path.expanduser(env_gguf_dir))

    seen_paths = set()
    for scan_dir in gguf_dirs:
        if not os.path.isdir(scan_dir):
            continue
        try:
            for root, dirs, files in os.walk(scan_dir):
                for f in files:
                    if f.endswith(('.gguf', '.ggml')) and not f.startswith('.'):
                        full_path = os.path.join(root, f)
                        if full_path not in seen_paths:
                            seen_paths.add(full_path)
                            available_models[full_path] = "llamacpp"
        except Exception as e:
            logging.info(f"Error scanning GGUF directory {scan_dir}: {e}")

    try:
        import requests
        response = requests.get('http://127.0.0.1:1234/v1/models', timeout=1)
        if response.ok:
            data = response.json()
            for model in data.get('data', []):
                model_id = model.get('id', model.get('name', 'unknown'))
                available_models[model_id] = "lmstudio"
    except Exception as e:
        logging.debug(f"LM Studio not available: {e}")

    try:
        import requests
        response = requests.get('http://127.0.0.1:8080/v1/models', timeout=1)
        if response.ok:
            data = response.json()
            for model in data.get('data', []):
                model_id = model.get('id', model.get('name', 'unknown'))
                available_models[model_id] = "llamacpp-server"
    except Exception as e:
        logging.debug(f"llama.cpp server not available: {e}")

    try:
        import requests
        response = requests.get('http://127.0.0.1:8000/v1/models', timeout=1)
        if response.ok:
            data = response.json()
            for model in data.get('data', []):
                model_id = model.get('id', model.get('name', 'unknown'))
                available_models[model_id] = "omlx"
    except Exception as e:
        logging.debug(f"OMLX server not available: {e}")

    try:
        import requests
        response = requests.get('http://127.0.0.1:5000/v1/models', timeout=1)
        if response.ok:
            data = response.json()
            for model in data.get('data', []):
                model_id = model.get('id', model.get('name', 'unknown'))
                if model_id not in available_models:
                    available_models[model_id] = "omlx"
    except Exception as e:
        logging.debug(f"OMLX server (port 5000) not available: {e}")

    lora_dirs = [
        get_models_dir(),
    ]
    for scan_dir in lora_dirs:
        if not os.path.isdir(scan_dir):
            continue
        try:
            for item in os.listdir(scan_dir):
                item_path = os.path.join(scan_dir, item)
                if os.path.isdir(item_path):
                    adapter_config = os.path.join(item_path, 'adapter_config.json')
                    if os.path.exists(adapter_config):
                        available_models[item_path] = "lora"
                        logging.debug(f"Found LoRA adapter: {item_path}")
        except Exception as e:
            logging.debug(f"Error scanning LoRA directory {scan_dir}: {e}")

    return available_models

def get_logs_dir() -> str:
    """Get the directory for logs."""
    return os.path.join(get_data_dir(), 'npc_team', 'logs')

def get_model_and_provider(command: str, available_models: list) -> tuple:
    """
    Function Description:
        Extracts model and provider from command and autocompletes if possible.
    Args:
        command : str : Command string
        available_models : list : List of available models
    Keyword Args:
        None
    Returns:
        model_name : str : Model name
        provider : str : Provider
        cleaned_command : str : Clean

    """

    model_match = re.search(r"@(\S+)", command)
    if model_match:
        model_name = model_match.group(1)

        matches = [m for m in available_models if m.startswith(model_name)]
        if matches:
            if len(matches) == 1:
                model_name = matches[0]  

            provider = lookup_provider(model_name)
            if provider:

                cleaned_command = command.replace(
                    f"@{model_match.group(1)}", ""
                ).strip()

                return model_name, provider, cleaned_command
            else:
                return None, None, command  
        else:
            return None, None, command  
    else:
        return None, None, command  

def get_models_dir() -> str:
    """Get the directory for storing models."""
    return os.path.join(get_data_dir(), 'npc_team', 'models')

def get_npcshrc_path() -> str:
    """
    Get the path to the npcshrc config file.

    Returns the platform-appropriate config file path.
    Falls back to ~/.npcshrc for backwards compatibility.
    """
    old_path = os.path.expanduser('~/.npcshrc')
    if os.path.exists(old_path):
        return old_path

    config_dir = get_config_dir()
    return os.path.join(config_dir, 'npcshrc')

def get_system_message(npc, team=None, tool_capable=False) -> str:

    if npc is None:
        return "You are a helpful assistant"
    if npc.plain_system_message:
        return npc.primary_directive

    system_message = f"""
.
..
...
....
.....
......
.......
........
.........
..........
Hello!
Welcome to the team.
You are the {npc.name} NPC with the following primary directive: {npc.primary_directive}.
Users may refer to you by your assistant name, {npc.name} and you should
consider this to be your core identity.
The current working directory is {os.getcwd()}.
The current date and time are : {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
"""

    if hasattr(npc, 'kg_data') and npc.kg_data:
        memory_context = npc.get_memory_context()
        if memory_context:
            system_message += f"\n\nMemory Context:\n{memory_context}\n"

    if npc.db_conn is not None:
        db_path = None
        if hasattr(npc.db_conn, "url") and npc.db_conn.url:
            db_path = npc.db_conn.url.database
        elif hasattr(npc.db_conn, "database"):
            db_path = npc.db_conn.database
        system_message += """What follows is information about the database connection. If you are asked to execute queries with tools, use this information. 
        If you are asked for help with debugging queries, use this information. 
        Do not unnecessarily reference that you possess this information unless it is
        specifically relevant to the request.

        DB Connection Information:        
        """
        if db_path:
            system_message += f"\nDatabase path: {db_path}\n"
        if npc.tables is not None:
            system_message += f"\nDatabase tables: {npc.tables}\n"

    if team is not None and npc.name == getattr(team, 'forenpc_name', 'sibiji'):
        team_context = team.context if hasattr(team, "context") and team.context else ""
        team_preferences = team.shared_context.get('preferences', '') if hasattr(team, "shared_context") else ""
        system_message += f"\nTeam context: {team_context}\n"
        if team_preferences:
            system_message += f"Team preferences: {team_preferences}\n"

        if hasattr(team, 'npcs') and team.npcs:
            members = []
            for name, member in team.npcs.items():
                if name != npc.name:
                    directive = getattr(member, 'primary_directive', '')
                    desc = directive[:50].strip() if directive else ''
                    members.append(f"  - @{name}: {desc}")
            if members:
                system_message += "\nTeam members available for delegation:\n" + "\n".join(members) + "\n"

    if hasattr(npc, 'jinxes_dict') and npc.jinxes_dict:
        tool_lines = []
        for jname, jinx in npc.jinxes_dict.items():
            desc = getattr(jinx, 'description', '') or ''
            tool_lines.append(f"  - {jname}: {desc.strip()}")
        if tool_lines:
            system_message += "\nYou have access to the following jinxes:\n"
            system_message += "\n".join(tool_lines) + "\n"
            if tool_capable:
                system_message += "\nUse the provided function calling interface to invoke tools when they are relevant to the request. For multi-step tasks, call one tool at a time and use its result to inform the next step.\n"
            else:
                jinx_names_str = ", ".join(npc.jinxes_dict.keys())
                jinx_instructions = f"""
                if you are in the [ReAct loop] and you are asked to use jinxes, refer to these guidelines:
              [BEGIN GUIDELINES FOR JINX EXECUTION]
                  Use jinxes when appropriate. For example:

                    - If you are asked about something up-to-date or dynamic (e.g., latest exchange rates)
                    - If the user asks you to read or edit a file
                    - If the user asks for code that should be executed
                    - If the user requests to open, search, download or scrape, which involve actual system or web actions
                    - If they request a screenshot, audio, or image manipulation
                    - Situations requiring file parsing (e.g., CSV or JSON loading)
                    - Scripted workflows or pipelines, e.g., generate a chart, fetch data, summarize from source, etc.

                    You MUST use a jinx if the request directly refers to a tool the AI cannot handle directly (e.g., 'run', 'open', 'search', etc).

                    You do not need to use a jinx if:

                    - the user asks a simple question like 'what is 2+2' or 'who invented linux', essentially any question which only requires general knowledge.
                    - The user asks you to write them a story (unless they separately specify saving it to a file, then you should directly write the story to be output through a jinx to said file.)
                    To invoke a jinx, return the action 'invoke_jinx' along with the jinx specific name.
                    An example for a jinx-specific return would be:
                    """ +"""
                    {
                        "action": "invoke_jinx",
                        "jinx_name": "file_reader",
                        "explanation": "Read the contents of <full_filename_path_from_user_request> and <detailed explanation of how to accomplish the problem outlined in the request>."
                    }


                    Do not use the jinx names as the action keys. You must use the action 'invoke_jinx' to invoke a jinx!
                    Do not invent jinx names. Use only those provided.


                Respond with a single JSON object only.
                To use a jinx, set action to jinx, jinx_name to one of [{jinx_names_str}], and inputs with the required parameters.

              [END GUIDELINES FOR JINX EXECUTION]


"""
                system_message += jinx_instructions




    return system_message

def get_triggers_dir() -> str:
    """Get the directory for trigger scripts."""
    return os.path.join(get_data_dir(), 'npc_team', 'triggers')

def get_videos_dir() -> str:
    """Get the directory for generated videos."""
    return os.path.join(get_data_dir(), 'npc_team', 'videos')

def guess_mime_type(filename):
    """Guess the MIME type of a file based on its extension."""
    extension = os.path.splitext(filename)[1].lower()
    mime_types = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".gif": "image/gif",
        ".bmp": "image/bmp",
        ".webp": "image/webp",
        ".pdf": "application/pdf",
        ".txt": "text/plain",
        ".csv": "text/csv",
        ".json": "application/json",
        ".md": "text/markdown",
    }
    return mime_types.get(extension, "application/octet-stream")

def init_db_tables(db_path="~/npcsh_history.db"):
    """Initialize necessary database tables"""
    db_path = os.path.expanduser(db_path)
    with sqlite3.connect(db_path) as conn:

        conn.execute("""
            CREATE TABLE IF NOT EXISTS npc_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                entity_id TEXT,  
                entry_type TEXT,
                content TEXT,
                metadata TEXT,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        """)


        conn.execute("""
            CREATE TABLE IF NOT EXISTS pipeline_runs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                pipeline_name TEXT,
                step_name TEXT,
                output TEXT,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        """)


        conn.execute("""
            CREATE TABLE IF NOT EXISTS compiled_npcs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT UNIQUE,
                source_path TEXT,
                compiled_content TEXT,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        """)

        conn.commit()

def load_custom_providers():
    """
    Load custom provider configurations from npcshrc config file.

    Returns:
        dict: Custom provider configurations keyed by provider name
    """
    custom_providers = {}
    npcshrc_path = get_npcshrc_path()

    if os.path.exists(npcshrc_path):
        with open(npcshrc_path, "r") as f:
            for line in f:
                line = line.split("#")[0].strip()
                if "CUSTOM_PROVIDER_" in line and "=" in line:
                    key, value = line.split("=", 1)
                    key = key.strip().replace("export ", "")
                    value = value.strip().strip("\"'")

                    try:
                        config = json.loads(value)
                        provider_name = key.replace(
                            "CUSTOM_PROVIDER_", ""
                        ).lower()
                        custom_providers[provider_name] = config
                    except json.JSONDecodeError as e:
                        logging.warning(
                            f"Failed to parse custom provider {key}: {e}"
                        )
                        continue

    return custom_providers

def load_env_from_execution_dir() -> None:
    """
    Function Description:
        This function loads environment variables from a .env file in the current execution directory.
    Args:
        None
    Keyword Args:
        None
    Returns:
        None
    """

    execution_dir = os.path.abspath(os.getcwd())
    env_path = os.path.join(execution_dir, ".env")
    if os.path.exists(env_path):
        load_dotenv(dotenv_path=env_path)
        logging.info(f"Loaded .env file from {execution_dir}")
    else:
        logging.warning(f"Warning: No .env file found in {execution_dir}")

def log_action(action: str, detail: str = "") -> None:
    """
    Function Description:
        This function logs an action with optional detail.
    Args:
        action: The action to log.
        detail: Additional detail to log.
    Keyword Args:
        None
    Returns:
        None
    """
    logging.info(f"{action}: {detail}")

def lookup_provider(model: str) -> str:
    """
    Determine the provider based on the model name.
    Checks custom providers first, then falls back to known providers.

    Args:
        model: The model name

    Returns:
        The provider name or None if not found
    """
    if model and os.path.isdir(os.path.expanduser(model)):
        adapter_config = os.path.join(os.path.expanduser(model), 'adapter_config.json')
        if os.path.exists(adapter_config):
            return "lora"

    custom_providers = load_custom_providers()

    for provider_name, config in custom_providers.items():
        if model.startswith(f"{provider_name}-"):
            return provider_name

        try:
            import requests
            api_key_var = config.get('api_key_var') or \
                         f"{provider_name.upper()}_API_KEY"
            api_key = os.environ.get(api_key_var)

            if api_key:
                base_url = config.get('base_url', '')
                headers = config.get('headers', {})
                headers['Authorization'] = f'Bearer {api_key}'

                models_endpoint = f"{base_url.rstrip('/')}/models"
                response = requests.get(
                    models_endpoint, 
                    headers=headers, 
                    timeout=1.0
                )

                if response.status_code == 200:
                    data = response.json()
                    models = []

                    if isinstance(data, dict) and 'data' in data:
                        models = [m['id'] for m in data['data']]
                    elif isinstance(data, list):
                        models = [m['id'] for m in data]

                    if model in models:
                        return provider_name
        except Exception:
            pass

    if model == "deepseek-chat" or model == "deepseek-reasoner":
        return "deepseek"

    if model.startswith("airllm-"):
        return "airllm"

    ollama_prefixes = [
        "llama", "deepseek", "qwen", "llava", 
        "phi", "mistral", "mixtral", "dolphin", 
        "codellama", "gemma",]
    if any(model.startswith(prefix) for prefix in ollama_prefixes):
        return "ollama"

    openai_prefixes = ["gpt-", "dall-e-", "whisper-", "o1"]
    if any(model.startswith(prefix) for prefix in openai_prefixes):
        return "openai"

    if model.startswith("claude"):
        return "anthropic"
    if model.startswith("gemini"):
        return "gemini"
    if "diffusion" in model:
        return "diffusers"

    return None