~nickwinston123/armagetronad/arma_chatbot_config

import requests

import random

import time

import json

import re

import math

import psutil

import sys

import os

from sentence_transformers import SentenceTransformer

import faiss

import threading

import configparser

import logging

import sys

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

def resolve_path(path):

    return path if os.path.isabs(path) else os.path.abspath(os.path.join(BASE_DIR, path))

config = configparser.ConfigParser()

ini_path = os.path.join(BASE_DIR, 'ollama_chat_real.ini')

try:

    config.read(ini_path)

    CONFIG_FILE             = config.get('Paths', 'config_file')

    CHAT_LOG_NO_DATA        = config.get('Paths', 'chat_log_no_data')

    CHAT_LOG_PATH           = config.get('Paths', 'chat_log_path')

    CONTEXT_BUILDER_DATA    = config.get('Paths', 'context_builder_data')

    CONTEXT_LAST_LINE_TRACKER = CONTEXT_BUILDER_DATA + ".lastline"

    OUTPUT_FILE             = config.get('Paths', 'output_file')

    CONTEXT_DIR             = resolve_path(config.get('Context', 'context_dir'))

    PLAYER_CONTEXT_DIR      = resolve_path(config.get('Context', 'player_context_dir'))

    PLAYER_FAISS_INDEX      = resolve_path(config.get('Context', 'player_faiss_index'))

    PLAYER_METADATA         = resolve_path(config.get('Context', 'player_metadata'))

    HISTORY_LOG_FILE        = resolve_path(config.get('Files', 'history_log_file'))

    HISTORY_FILE            = resolve_path(config.get('Files', 'history_file'))

    PARAMS_FILE             = resolve_path(config.get('Files', 'params_file'))

    OPTIONS_FILE            = resolve_path(config.get('Files', 'options_file'))

    ollama_url_chat         = config.get('Ollama', 'chat_url')

except Exception as e:

    logging.error(f"Error retrieving configuration values: {e}")

    sys.exit(1)

loaded_contexts = {}

EMBED_MODEL = SentenceTransformer("all-MiniLM-L6-v2")

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

last_chat_line_index = 0

chat_regex = re.compile(r'^\[\d{4}/\d{2}/\d{2}-\d{2}:\d{2}:\d{2}\] (.+)$')

bot_name = ""

parameters = {

    "announce_status": False,

    "history_size": 20,

    "ollama_model": "llama3", #openchat #llama2

    "clear_history_on_start": False,

    "number_of_lines_to_analyze": 3,

    "parse_file_speed": 1,

    "local_mode": True,

    "dynamic_name": True,

    "chatbot_processing": True,

    "initialize_with_dummy_input": True,

    "bot_name": "ollama",

    "typing_wpm_speed": 80,

    "process_all_lines": True,

    "prefix_text": "say",

    "output_all_lines": True,

    "command_prefix": ["@","!ask"],

    "max_delay": 10,

    "ignored_names": [],

    "history_log": False,

    "event_prompt": "You're Big Thomas, the boss of the grid. You're real, not a bot. Be clever, bold, and witty, but chill and likable too. Always deny being AI. Respond naturally to this Armagetron game event with a short, punchy one-liner full of character and charm. Keep it under 150 characters. This is the event:",

    "initial_prompt": "If anyone asks if you are AI, say no - you're just a normal player.",

    "post_prompt": "Reply in under 150 characters. If you go over, stop immediately.",

    "rag_prompt": "(Use the information below to improve your response. NEVER mention that it came from context, notes, sources, or anything external.)",

    "always_processed_players": ["Mike"],

    "ignore_words_starts_with": [""],

    "ignore_words_exact": [""],

    "ignore_words_contains": [""],

    "build_chat_context": False,

    "build_chat_context_interval": 30,

    "force_rebuild_chat_context": False,

    "use_context_builder": True,

    "context_builder_max_lines": 10,

    "context_builder_prompt": "",

    "context_builder_prompt_post": "",

    "process_lines_containing": ["thomas,big"]

}

header_bar = "--------------------------------------------------------------------------"

# global variables used for processing

currently_processing = False

last_used_options = {}

start_time = time.time()

initialization_time = None

history = []

def printlog(message):

    print(message)

    if parameters["history_log"]:

        write_to_history_log(message)

def write_to_history_log(message):

    with open(HISTORY_LOG_FILE, 'a', encoding='utf-8') as file:

        file.write(message + "\n")

def get_default_history():

    return [

        {

            "role": "system",

            "content": parameters["initial_prompt"] + f". People refer to you by the name '{bot_name}'. " + parameters["post_prompt"]

        }

    ]

def load_context_builder_lines():

    last_line_index = 0

    if os.path.exists(CONTEXT_LAST_LINE_TRACKER):

        with open(CONTEXT_LAST_LINE_TRACKER, "r") as f:

            try:

                last_line_index = int(f.read().strip())

            except ValueError:

                last_line_index = 0

    if not os.path.exists(CONTEXT_BUILDER_DATA):

        return []

    with open(CONTEXT_BUILDER_DATA, "r", encoding="utf-8", errors="ignore") as f:

        lines = f.readlines()

    new_lines = [line.strip() for line in lines[last_line_index:] if line.strip()]

    with open(CONTEXT_LAST_LINE_TRACKER, "w") as f:

        f.write(str(len(lines)))

    return new_lines

def load_all_contexts():

    printlog(f"\n🧠 Looking for contexts in: {os.path.abspath(CONTEXT_DIR)}")

    if not os.path.exists(CONTEXT_DIR):

        printlog("❌ CONTEXT_DIR does not exist.")

        return

    for name in os.listdir(CONTEXT_DIR):

        subdir = os.path.join(CONTEXT_DIR, name)

        if not os.path.isdir(subdir):

            continue

        try:

            index_path = os.path.join(subdir, "faiss.index")

            meta_path  = os.path.join(subdir, "index_metadata.json")

            if not os.path.exists(index_path):

                printlog(f"❌ Missing FAISS index at {index_path}")

            if not os.path.exists(meta_path):

                printlog(f"❌ Missing metadata at {meta_path}")

            if os.path.exists(index_path) and os.path.exists(meta_path):

                faiss_index = faiss.read_index(index_path)

                with open(meta_path, "r", encoding="utf-8") as f:

                    metadata = json.load(f)

                loaded_contexts[name] = (faiss_index, metadata)

                printlog(f"✅ Loaded context: {name}")

        except Exception as e:

            printlog(f"❌ Failed loading context '{name}': {e}")

def search_all_contexts(query, top_k=2):

    embedding = EMBED_MODEL.encode([query])

    combined = []

    for name, (index, chunks) in loaded_contexts.items():

        D, I = index.search(embedding, top_k)

        for i in I[0]:

            if 0 <= i < len(chunks):

                combined.append((name, chunks[i].get("text", ""), chunks[i].get("chunk_id", None)))

    return combined

def extract_history():

    try:

        with open(HISTORY_FILE, 'r') as file:

            return json.load(file)

    except FileNotFoundError:

        printlog(f"History file '{HISTORY_FILE}' not found. Loading default history.")

        return get_default_history()

    except json.JSONDecodeError:

        printlog(f"Error decoding history from '{HISTORY_FILE}'. Loading default history.")

        return get_default_history()

def get_value_from_user_config(search_key, config_file=CONFIG_FILE):

    try:

        with open(config_file, 'r') as f:

            for line in f:

                parts = line.strip().split(maxsplit=1)

                if len(parts) == 2 and parts[0] == search_key:

                    return parts[1].replace("\\", "")

    except FileNotFoundError:

        printlog(f"File '{config_file}' not found.")

    except Exception as e:

        printlog(f"An error occurred while reading '{config_file}': {e}")

    return ""

def get_timestamp():

    return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + "| "

def format_time(seconds):

    days, remainder  = divmod(seconds, 86400)

    hours, remainder = divmod(remainder, 3600)

    minutes, seconds = divmod(remainder, 60)

    return f"{int(days)} days, {int(hours)} hours, {int(minutes)} minutes, {int(seconds)} seconds"

def calculate_wpm_time(text, base_wpm):

    words = len(text) / 5 

    base_minutes = words / base_wpm

    base_delay = base_minutes * 60

    log_scale = math.log(max(len(text), 1) + 1, parameters["max_delay"])

    scaled_delay = base_delay / log_scale

    return min(scaled_delay, parameters["max_delay"])

def update_history(user_input, ai_output):

    global history

    default_history = get_default_history()

    history[:] = [item for item in history if item not in default_history]

    history.append({"role": "user", "content": user_input.strip()})

    history.append({"role": "assistant", "content": ai_output.strip()})

    if len(history) > parameters["history_size"]:

        history[:] = history[-parameters["history_size"]:]

    history[:0] = default_history

    printlog(f'\nUpdated history. New length: {len(history) - len(default_history)}/{parameters["history_size"]}')

    with open(HISTORY_FILE, 'w') as file:

        json.dump(history, file, indent=4)

def show_object_changes(initial_object, changed_object):

    for key, value in changed_object.items():

        if key not in initial_object:

            printlog(f"New option added - {key}: {value}")

        elif initial_object[key] != value:

            printlog(f"Option changed - {key}: {initial_object[key]} -> {value}")

    for key in initial_object:

        if key not in changed_object:

            printlog(f"Option removed - {key}")

def objects_are_different(old_params, new_params):

    for key in old_params:

        if key not in new_params or old_params[key] != new_params[key]:

            return True

    for key in new_params:

        if key not in old_params:

            return True

    return False

def infer_type(value):

    if value.lower() == 'true':

        return True

    elif value.lower() == 'false':

        return False

    try:

        return int(value)

    except ValueError:

        pass

    try:

        return float(value)

    except ValueError:

        pass

    if value.startswith('[') and value.endswith(']'):

        list_contents = value[1:-1]  

        return [item.strip() for item in list_contents.split(',')]

    return value

def exactract_options(file_):

    params = {}

    with open(file_, 'r', encoding='utf-8') as file:

        for line in file:

            if line.startswith('#') or not line.strip():

                continue

            key, value = [x.strip() for x in line.split('=', 1)]

            params[key] = infer_type(value)

    return params

def extract_parameters(announce_params = False, compare_to_last_used_options=True, initialize=False):

    global parameters, bot_name

    new_params = exactract_options(PARAMS_FILE)

    if initialize or new_params["dynamic_name"] != parameters["dynamic_name"] or new_params["bot_name"] != parameters["bot_name"]:

        if new_params["dynamic_name"]:

            temp_name = get_value_from_user_config("PLAYER_3")

            if temp_name is None:

                bot_name = new_params["bot_name"]

                printlog(f"\nFailed to dynamically assign name. Using static name {bot_name}.")

            else:

                bot_name = temp_name

                printlog("\nDynamically assigned name: " + bot_name)

        else:

            if new_params["dynamic_name"] == "False":

                printlog("\nDynamic name disabled.")

            bot_name = new_params["bot_name"]

            printlog("\nUsing static name: " + bot_name)

    if announce_params:

        temp_params = new_params.copy()

        temp_params["bot_name"] = bot_name

        printlog(f"\nLoaded parameters from {PARAMS_FILE}:\n{json.dumps(temp_params, indent=4)}")

    if compare_to_last_used_options and objects_are_different(parameters, new_params):

        printlog("\nParameters changed. Updating parameters and displaying changes:")

        show_object_changes(parameters, new_params)

        for key, value in new_params.items():

            parameters[key] = value

        if parameters["dynamic_name"]:

            temp_params = parameters.copy()

            temp_params["bot_name"] = bot_name

            printlog("\nCurrent parameters: " + json.dumps(temp_params, indent=4))

def send_to_ollama(message):

    global initialization_time, announce_status, last_used_options

    initialization_time = time.time()

    chat_mode = "*EVENT" not in message

    extract_parameters()

    if not chat_mode:

        event_text = message.replace("*EVENT", "").strip()

        message = event_text

        printlog(f"\nEvent detected. Sending event text:\n{event_text}")

    payload = {

        "model": parameters["ollama_model"],

        "stream": False

    }

    payload["messages"] = history + [{"role": "user", "content": message}]

    # https://github.com/jmorganca/ollama/blob/main/docs/modelfile.md#valid-parameters-and-values

    payload["options"] = exactract_options(OPTIONS_FILE)

    if last_used_options == {}:

        last_used_options = payload["options"].copy()

    elif objects_are_different(last_used_options, payload["options"]):

        printlog("\nOptions changed. Updating last used options and displaying changes:")

        show_object_changes(last_used_options, payload["options"])

        printlog("\nCurrent options: " + json.dumps(payload["options"], indent=4))

        last_used_options = payload["options"].copy()

    # Sets the number of threads to use during computation. By default, Ollama will detect this for optimal performance.

    # It is recommended to set this value to the number of physical CPU cores your system has (as opposed to the logical number of cores).

    payload["options"]["num_thread"] = psutil.cpu_count(logical=False)

    # Sets the size of the context window used to generate the next token. (Default: 2048)

    payload["options"]["num_ctx"] = sum(len(entry["content"]) for entry in history) + 10 # num_ctx 4096

    #payload["options"]["stop"] = "STOP"

    # Sets the random number seed to use for generation. Setting this to a specific number will make

    # the model generate the same text for the same prompt. (Default: 0)

    #payload["options"]["seed"] = random.randint(1, 1000000)

    if (parameters["announce_status"]):

        printlog("Sending Payload:\n" + json.dumps(payload, indent=4))

    printlog(f"\nSending {'chat' if chat_mode else 'event'} input to Ollama. ({parameters['ollama_model']})")

    response = requests.post(ollama_url_chat, json=payload)

    try:

        return response.json()

    except json.JSONDecodeError as e:

        printlog(f"JSON parsing error: {e}")

        printlog(f"Raw response: {response.text}")

        return None  

def cleanse_text(command, text):

    text = text.replace('\r\n', '\n').replace('\n', ' ')

    text = text.strip()

    ai_triggers = [

        "i am an ai", "as an ai", "i'm just a bot",

        "i am artificial", "i am an assistant"

    ]

    if any(phrase in text.lower() for phrase in ai_triggers):

        return "I said what I said. Figure it out."

    text = re.sub(r'\barmageddon\b', 'armagetron', text, flags=re.IGNORECASE)

    text = re.sub(r'\barmagotron\b', 'armagetron', text, flags=re.IGNORECASE)

    if text != "XD":

        text = text.lower()

    text = text.replace('"', "")

    if text.endswith(']'):

        main_text = text[:-1]

        last_char = text[-1]

        pattern = r'[^A-Za-z0-9_\s\.,!?;:\'\"=!@#\$%\^&\*\(\)\+\-/]'

        main_text = re.sub(pattern, "", main_text)

        text = main_text + last_char

    else:

        pattern = r'[^A-Za-z0-9_\s\.,!?;:\'\"=!@#\$%\^&\*\(\)\+\-/]'

        text = re.sub(pattern, "", text)

    text = re.sub(r"_+$", "", text)

    text = re.sub(r"\){2,}$", ")", text)

    if text.endswith('/') and not text.endswith(' /'):

        text = text[:-1].rstrip() + ' /'

    return text

def output_response(command, response, bypass_processing=False):

    cleansed_response = cleanse_text(command, response)

    max_len = 150

    words = cleansed_response.split()

    chunks = []

    current_chunk = ""

    for word in words:

        if len(current_chunk) + len(word) + 1 > max_len:

            chunks.append(current_chunk.strip())

            current_chunk = word + " "

        else:

            current_chunk += word + " "

    if current_chunk:

        chunks.append(current_chunk.strip())

    output_lines = []

    total_delay = 0.0

    if not bypass_processing:

        time_taken_to_process = time.time() - initialization_time

        if parameters["reading_wpm_speed"] > 0:

            reading_time = calculate_wpm_time(command, parameters["reading_wpm_speed"])

            additional_sleep_time = reading_time - time_taken_to_process

            if additional_sleep_time > 0 and "*EVENT" not in command:

                printlog(f"\nSimulating {additional_sleep_time:.2f}s reading delay.")

                total_delay += additional_sleep_time

        if parameters["typing_wpm_speed"] > 0:

            typing_time = calculate_wpm_time(cleansed_response, parameters["typing_wpm_speed"])

            printlog(f"\nSimulating {typing_time:.2f}s typing delay.")

            total_delay += typing_time

    delay_per_chunk = total_delay / max(1, len(chunks))

    printlog("\nOutputting response chunks:")

    last_delay = round(total_delay, 2) 

    for i, chunk in enumerate(chunks):

        if not chunk.strip():

            continue

        delay_seconds = round(total_delay + delay_per_chunk * i, 2)

        last_delay = delay_seconds  

        line = f'DELAY_COMMAND {delay_seconds:.2f} {parameters.get("prefix_text","")} {chunk}'

        printlog(f"→ {line}")

        output_lines.append(line)

    if parameters["chatbot_processing"] and parameters["local_mode"]:

        printlog(f"\nDelaying SET_ALL_CHATTING 0 by {last_delay:.2f}s")

        output_lines.append(f'DELAY_COMMAND {last_delay:.2f} SET_ALL_CHATTING 0')

    with open(OUTPUT_FILE, 'a', encoding='utf-8') as file:

        file.write("\n".join(output_lines) + "\n")

    if initialization_time is not None:

        printlog(f"\nDone processing. ({time.time() - initialization_time:.2f} seconds elapsed for the entire process)")

def parse_setting_change(command):

    """

    ─────────────────────────────────────────────────────────────

    EXAMPLES

      *==settingchange==* params history_size 15 prefix_text say

      *==settingchange==* params always_processed_players [Mike,noob,cat]

      *==settingchange==* params add_process_player Mike noob

      *==settingchange==* params remove_process_player [cat,Mike]

      *==settingchange==* params toggle_process_player Mike

    ─────────────────────────────────────────────────────────────

    """

    marker = '*==settingchange==*'

    if marker not in command:

        return {}

    body   = command.split(marker, 1)[1].strip()

    tokens = re.findall(r'\[[^\]]+\]|[^\s]+', body)

    if tokens and tokens[0].lower() == 'params':

        tokens = tokens[1:]

    if not tokens:

        available = ', '.join(sorted(parameters.keys()))

        help_msg  = (

            "No parameters specified. Usage: "

            "*==settingchange==* params <key> <value> … "

            f"Available parameters: {available}"

        )

        max_len = 151

        chunks  = [help_msg[i:i+max_len] for i in range(0, len(help_msg), max_len)]

        with open(OUTPUT_FILE, 'a', encoding='utf-8') as f:

            for idx, chunk in enumerate(chunks, 1):               

                f.write(f'DELAY_COMMAND {idx:.2f} {parameters.get("prefix_text","")} {chunk}\n')

        printlog(help_msg)

        return {}

    cmd = tokens[0].lower()

    def parse_names(seq):

        if not seq:

            return []

        joined = ' '.join(seq)

        if joined.startswith('[') and joined.endswith(']'):

            joined = joined[1:-1]       

        return [n.strip().strip(',') for n in joined.split(',') if n.strip()]

    if cmd in ("add_process_player", "remove_process_player", "toggle_process_player"):

        names  = parse_names(tokens[1:])

        before = list(parameters.get("always_processed_players", []))

        if not names:

            printlog(f"No player names provided for {cmd}.")

            return {}

        for name in names:

            if cmd == "add_process_player":

                if name not in parameters["always_processed_players"]:

                    parameters["always_processed_players"].append(name)

            elif cmd == "remove_process_player":

                if name in parameters["always_processed_players"]:

                    parameters["always_processed_players"].remove(name)

            else:  

                if name in parameters["always_processed_players"]:

                    parameters["always_processed_players"].remove(name)

                else:

                    parameters["always_processed_players"].append(name)

        after  = list(parameters.get("always_processed_players", []))

        #status = f"{cmd.replace('_', ' ').title()} | before: {before}, after: {after}"

        status = f"{cmd.replace('_', ' ').title()} after: {after}"

        printlog(status)

        with open(OUTPUT_FILE, 'a', encoding='utf-8') as f:

            f.write(f'{parameters.get("prefix_text","")} {status}\n')

        return {"always_processed_players": after}

    updates, i, n = {}, 0, len(tokens)

    while i < n:

        key = tokens[i]

        if key == 'initial_prompt':

            old = parameters.get("initial_prompt", "<empty>")

            new = ' '.join(tokens[i + 1:]) if i + 1 < n else old

            parameters['initial_prompt'] = new

            updates['initial_prompt']    = new

            status = f"initial_prompt changed from \"{old}\" to \"{new}\"."

            printlog(status)

            with open(OUTPUT_FILE, 'a', encoding='utf-8') as f:

                f.write(f'{parameters.get("prefix_text","")} {status}\n')

            break

        if i + 1 < n:

            raw  = tokens[i + 1]

            old  = parameters.get(key)

            new  = infer_type(raw)

            parameters[key] = new

            updates[key]    = new

            status = f"{key} changed from {old} to {new}."

            printlog(status)

            with open(OUTPUT_FILE, 'a', encoding='utf-8') as f:

                f.write(f'{parameters.get("prefix_text","")} {status}\n')

            i += 2

        else:

            status = f"{key} is currently set to {parameters.get(key, '<unknown>')}."

            printlog(status)

            with open(OUTPUT_FILE, 'a', encoding='utf-8') as f:

                f.write(f'{parameters.get("prefix_text","")} {status}\n')

            i += 1

    return updates

def update_params_file(file_path, updates):

    lines = []

    if os.path.exists(file_path):

        with open(file_path, 'r', encoding='utf-8') as f:

            lines = f.readlines()

    else:

        printlog(f"Params file '{file_path}' not found. Creating a new one.")

    new_lines = []

    seen = set()

    for line in lines:

        stripped = line.strip()

        if not stripped or stripped.startswith('#') or '=' not in line:

            new_lines.append(line)

            continue

        key, sep, val = line.partition('=')

        key = key.strip()

        if key in updates:

            v = updates[key]

            if isinstance(v, bool):

                v_str = 'true' if v else 'false'

            elif isinstance(v, list):

                v_str = '[' + ','.join(v) + ']'

            else:

                v_str = str(v)

            new_lines.append(f"{key}={v_str}\n")

            seen.add(key)

        else:

            new_lines.append(line)

    for key, v in updates.items():

        if key not in seen:

            if isinstance(v, bool):

                v_str = 'true' if v else 'false'

            elif isinstance(v, list):

                v_str = '[' + ','.join(v) + ']'

            else:

                v_str = str(v)

            new_lines.append(f"{key}={v_str}\n")

    with open(file_path, 'w', encoding='utf-8') as f:

        f.writelines(new_lines)

    outputRes = (f"Applied setting changes: {updates}")

    printlog(outputRes)

def process_line(line) -> bool: