406 lines
15 KiB
Python
406 lines
15 KiB
Python
from rich import print as pr
|
||
from rich.layout import Layout
|
||
from rich.panel import Panel
|
||
import termcharts
|
||
import ollama
|
||
import tkinter as tk
|
||
from tkinter import filedialog
|
||
import re
|
||
import heapq
|
||
import time
|
||
import sys
|
||
|
||
# Officially advanced to the C-series tier!
|
||
cc_version = "ITA C-1-i"
|
||
|
||
# P7MJ's ITA (In the Air) chat mood analyzer, as a YAY project between 3/2 and 3/3 and finished on 3/3 (A-2-i). No longer so yay...
|
||
# Version B-1-i Refining (aka B-2-i Testing) completed on 3/5
|
||
# Working on B-1-i 3/6 and 3/8. Procrastinating between. A-series, get ready to be lost in time, but i still have some copies
|
||
# B-2-i Testing testing on 3/8. Prelim testing completed.
|
||
# B-2-ii Final working on 3/8. Removing all TEST comments and running more test cases
|
||
# B-2-iii Final working on 3/9 Morning. Added a safeguard in case the LLM hallucinates giving the super_llm_rate.
|
||
# B-2-iv Final working on 3/9 Morning. Fixed and added static colors (then removed them)
|
||
# Presenting is 3/9
|
||
# C-1-i Patched 6/14/2026. Upgraded regex parser for flexible AM/PM 12-hour exporter logs & added crash safeguards.
|
||
|
||
rating_data = []
|
||
total_messages = 0
|
||
messages_rated = 0
|
||
posanslist = []
|
||
neganslist = []
|
||
chat_summary = ""
|
||
chat_message_list = ""
|
||
|
||
# Find indices of 5 ratings with largest value
|
||
def get_n_largest_indices(data, n):
|
||
largest_items = heapq.nlargest(n, enumerate(data), key=lambda x: x[1])
|
||
indices = [index for index, value in largest_items]
|
||
return indices
|
||
|
||
# Make single or more messages readable
|
||
def readable_format(inputs):
|
||
processed = "<log>\n"
|
||
for i in range(len(inputs)):
|
||
processed += f"{inputs[i][0]}: {inputs[i][1]}\n"
|
||
processed += "</log>"
|
||
return processed
|
||
|
||
# One readable message
|
||
def readable_one_message(single_input_list):
|
||
return f"{single_input_list[0]}: {single_input_list[1]}"
|
||
|
||
# Find indices of 5 ratings with smallest value
|
||
def get_n_smallest_indices(data, n):
|
||
smallest_items = heapq.nsmallest(n, enumerate(data), key=lambda x: x[1])
|
||
indices = [index for index, value in smallest_items]
|
||
return indices
|
||
|
||
# Processes the raw chat log into a readable list
|
||
def parse_chat_log(file_path):
|
||
# FIXED FOR C-1-i: Matches flexible [M/D/YYYY H:MM AM/PM] style formats seamlessly
|
||
pattern = re.compile(r"\[\d{1,2}/\d{1,2}/\d{4} \d{1,2}:\d{2}\s*(?:AM|PM)?\]\s+(.*)")
|
||
chat_data = []
|
||
current_user = None
|
||
current_message = []
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith("===") or line.startswith("Exported"):
|
||
continue
|
||
match = pattern.match(line)
|
||
if match:
|
||
if current_user:
|
||
chat_data.append([current_user, "\n".join(current_message)])
|
||
current_user = match.group(1)
|
||
current_message = []
|
||
else:
|
||
current_message.append(line)
|
||
if current_user:
|
||
chat_data.append([current_user, "\n".join(current_message)])
|
||
return chat_data
|
||
|
||
def split_thy_list_for_thy_batch_function(inputses):
|
||
processing = inputses.copy()
|
||
batches = []
|
||
if len(inputses) > 5:
|
||
while True:
|
||
if len(processing) > 5:
|
||
batches.append(processing[:5])
|
||
del processing[:5]
|
||
else:
|
||
batches.append(processing)
|
||
break
|
||
return batches
|
||
else:
|
||
batches.append(inputses)
|
||
return batches
|
||
|
||
# Filters and processes the values, replacing text answers and 0.00s
|
||
def float_filter(value):
|
||
try:
|
||
float(value)
|
||
if float(value) == 0.00:
|
||
return -0.10
|
||
else:
|
||
return float(value)
|
||
except ValueError:
|
||
return -0.5
|
||
|
||
def process_llm_list(input_string):
|
||
stripped = input_string.strip()
|
||
splitted = stripped.split(", ")
|
||
for i in range(len(splitted)):
|
||
try:
|
||
splitted[i] = float(splitted[i])
|
||
except:
|
||
print(f"E {splitted[i]} not_float")
|
||
splitted[i] = -2
|
||
return splitted
|
||
|
||
# Uses the LLM to rate messages
|
||
def llm_rate(user_input, length):
|
||
global messages_rated
|
||
response = ollama.chat(model='llama3.2', messages=[
|
||
{
|
||
'role': 'system',
|
||
'content': (
|
||
"Sentiment Analyzer: Output ONLY a float between -1.0 and 1.0."
|
||
"Chat message format is user: message. Analyze single message."
|
||
"Scoring: Hostile/Insulting (-0.8), Dismissive/Sarcastic (-0.5),"
|
||
"Apathetic/Short (-0.2), Neutral/Functional (0.0), Positive/Link (0.1-1.0)."
|
||
"No prose."
|
||
)
|
||
},
|
||
{'role': 'user', 'content': readable_one_message(user_input)}
|
||
])
|
||
|
||
messages_rated += 1
|
||
print(f"\r{messages_rated}/{length} messages rated by LLM.", end = "")
|
||
rating_data.append(response['message']['content'])
|
||
|
||
# Rate all the messages at once
|
||
def super_llm_rate(inputs):
|
||
global chat_message_list
|
||
while True:
|
||
print("LLM All-at-once Rating Mode rating...", end = " ", flush=True)
|
||
response = ollama.chat(model='llama3.2', messages=[
|
||
{
|
||
'role': 'system',
|
||
'content': (
|
||
"Sentiment Analyzer: Output ONLY a list of floats between -1.0 and 1.0, like '0.0, 0.5, -0.5'."
|
||
"Chat messages are between <log> and </log>. Analyze the chat records and give rating to each message in order."
|
||
"Scoring: Hostile/Insulting (-0.8), Dismissive/Sarcastic (-0.5),"
|
||
"Apathetic/Short (-0.2), Neutral/Functional (0.0), Positive/Link (0.1-1.0)."
|
||
f"There are {len(chat_message_list)} messages in total. Make sure you have the same amount of ratings."
|
||
"Exactly a comma followed by a space between ratings. No prose. Double check amount."
|
||
)
|
||
},
|
||
{'role': 'user', 'content': readable_format(inputs)}
|
||
])
|
||
finished = process_llm_list(response['message']['content'])
|
||
# FIXED FOR C-1-i: Corrected global variable reference target
|
||
if len(finished) == len(chat_message_list):
|
||
for i in range(len(finished)):
|
||
rating_data.append(finished[i])
|
||
break
|
||
else:
|
||
print("LLM Hallucinated!")
|
||
|
||
# Accepts a list of unprocessed parsed entries
|
||
def batch_llm_rate(inputs):
|
||
global messages_rated
|
||
print("Creating batches...", end = " ", flush=True)
|
||
try:
|
||
batcheses = split_thy_list_for_thy_batch_function(inputs)
|
||
print("Success!")
|
||
except:
|
||
print("Fail!")
|
||
sys.exit(0)
|
||
|
||
print("Rating...", end = "", flush=True)
|
||
for i in range(len(batcheses)):
|
||
that_one_batch = batcheses[i]
|
||
response = ollama.chat(model='llama3.2', messages=[
|
||
{
|
||
'role': 'system',
|
||
'content': (
|
||
"Sentiment Analyzer: Output ONLY a list of floats between -1.0 and 1.0, like '0.0, 0.5, -0.5'."
|
||
"Chat messages are between <log> and </log>. Analyze the chat records and give rating to each message in order."
|
||
"Scoring: Hostile/Insulting (-0.8), Dismissive/Sarcastic (-0.5),"
|
||
"Apathetic/Short (-0.2), Neutral/Functional (0.0), Positive/Link (0.1-1.0)."
|
||
f"There are {len(that_one_batch)} messages in total. Make sure you have {len(that_one_batch)} ratings."
|
||
"Exactly a comma followed by a space between ratings. No prose. Double check amount."
|
||
)
|
||
},
|
||
{'role': 'user', 'content': readable_format(that_one_batch)}
|
||
])
|
||
finished_list = process_llm_list(response['message']['content'])
|
||
for k in range(len(finished_list)):
|
||
rating_data.append(finished_list[k])
|
||
print("\r" + f"{len(rating_data)}/{len(inputs)} messages rated", end = "")
|
||
|
||
# Summarizes logs
|
||
def chat_summarizer(lists):
|
||
global chat_summary
|
||
print("Summarizing chat...", end = " ", flush=True)
|
||
response = ollama.chat(model='llama3.2', messages=[
|
||
{
|
||
'role': 'system',
|
||
'content': (
|
||
'Chat Summarizer Tool: Sumarize given chat between <log> and </log>. Return brief summary and lists of topics discussed in order. Explain chat flow. Hypothesize user personalities. Use as few lines as possible.'
|
||
)
|
||
},
|
||
{'role': 'user', 'content': readable_format(lists)}
|
||
])
|
||
chat_summary = response['message']['content']
|
||
print("Success!")
|
||
|
||
# Start ITA Agent
|
||
def start_ita_agent(mood, summary):
|
||
messages = [
|
||
{
|
||
'role': 'system',
|
||
'content': (f"Sentiment Analysis Tool. Respond to the user's questions on analyzed chat."
|
||
f"Analyzed data was: mood = {mood} with 0 being average."
|
||
f"summary: {summary}. Respond in short and concise sentences."
|
||
)
|
||
}
|
||
]
|
||
|
||
print("Type 'exit' or 'quit' to end the session.\n")
|
||
|
||
while True:
|
||
user_input = input("You > ")
|
||
if user_input.lower() in ['exit', 'quit']:
|
||
break
|
||
|
||
messages.append({'role': 'user', 'content': user_input})
|
||
print("ITA Analysis Agent > ", end="", flush=True)
|
||
|
||
full_response = ""
|
||
stream = ollama.chat(model='llama3.2', messages=messages, stream=True)
|
||
|
||
for chunk in stream:
|
||
content = chunk['message']['content']
|
||
print(content, end="", flush=True)
|
||
full_response += content
|
||
|
||
messages.append({'role': 'assistant', 'content': full_response})
|
||
print()
|
||
|
||
# Counts answers for statistics chart
|
||
def countanswers(lst):
|
||
goodanswers = 0
|
||
badanswers = 0
|
||
neutralanswers = 0
|
||
for z in range(len(lst)):
|
||
if lst[z] > -0.3 and lst[z] < 0.3:
|
||
neutralanswers += 1
|
||
elif lst[z] > 0.3:
|
||
goodanswers += 1
|
||
elif lst[z] < -0.3:
|
||
badanswers += 1
|
||
return [goodanswers, badanswers, neutralanswers]
|
||
|
||
|
||
# MAIN
|
||
if __name__ == "__main__":
|
||
while True:
|
||
print(f"""
|
||
{"=" * 80}
|
||
ITA (In the Air) Mood Interpretor | Version: {cc_version} - P7MJ
|
||
{"=" * 80}
|
||
|
||
Choose a chat log to evaluate
|
||
[1] goodlog.txt
|
||
[2] badlog.txt
|
||
[3] testlog.txt
|
||
[4] specify""")
|
||
log_to_choose = input(" > ")
|
||
if log_to_choose == "1":
|
||
log_to_choose = "goodlog.txt"
|
||
break
|
||
elif log_to_choose == "2":
|
||
log_to_choose = "badlog.txt"
|
||
break
|
||
elif log_to_choose == "3":
|
||
log_to_choose = "testlog.txt"
|
||
break
|
||
elif log_to_choose == "4":
|
||
log_to_choose = input("Text file name (w/extension) > ")
|
||
break
|
||
else:
|
||
print("Invalid Option!")
|
||
|
||
start_time = time.perf_counter()
|
||
|
||
print("Parsing chat message file...", end = " ")
|
||
try:
|
||
chat_message_list = parse_chat_log(f"{log_to_choose}")
|
||
except Exception as e:
|
||
input(f"Error opening file! {e}. Press enter to exit...")
|
||
sys.exit(0)
|
||
|
||
# FIXED FOR C-1-i: Clean layout exit fallback if the file exists but has no matched lines
|
||
if len(chat_message_list) == 0:
|
||
input("\nError: No messages parsed! Check your timestamp formats. Press enter to exit...")
|
||
sys.exit(0)
|
||
|
||
print("Complete.")
|
||
|
||
while True:
|
||
rating_mode = input("\nChoose rating mode:\n[1] Individual message rating\n[2] Batches of 5 rating\n[3] All-at-once rating\n > ")
|
||
if rating_mode in ["1", "2", "3"]:
|
||
break
|
||
else:
|
||
print("Not an integer or within range!")
|
||
|
||
if rating_mode == "1":
|
||
for m in range(len(chat_message_list)):
|
||
prompt = str(chat_message_list[m][1])
|
||
llm_rate(chat_message_list[m], len(chat_message_list))
|
||
elif rating_mode == "2":
|
||
batch_llm_rate(chat_message_list)
|
||
elif rating_mode == "3":
|
||
super_llm_rate(chat_message_list)
|
||
|
||
# Filters and processes rating data
|
||
for i in range(len(rating_data)):
|
||
rating_data[i] = float_filter(rating_data[i])
|
||
|
||
# FIXED FOR C-1-i: Division-by-zero protection guard
|
||
if len(rating_data) > 0:
|
||
sum_of_numbers = 0
|
||
for k in range(len(rating_data)):
|
||
sum_of_numbers += rating_data[k]
|
||
sum_of_numbers /= len(rating_data)
|
||
else:
|
||
sum_of_numbers = 0.0
|
||
|
||
list_of_best_answers = get_n_largest_indices(rating_data, 5)
|
||
list_of_worst_answers = get_n_smallest_indices(rating_data, 5)
|
||
|
||
# FIXED FOR C-1-i: Added safe index length checks to cleanly populate panels if under 5 items exist
|
||
for j in range(5):
|
||
if j < len(list_of_best_answers) and list_of_best_answers[j] < len(chat_message_list):
|
||
posanslist.append(f"[{j+1}] {chat_message_list[list_of_best_answers[j]][0]}: {chat_message_list[list_of_best_answers[j]][1]}")
|
||
else:
|
||
posanslist.append(f"[{j+1}] N/A: No data available")
|
||
|
||
for k in range(5):
|
||
if k < len(list_of_worst_answers) and list_of_worst_answers[k] < len(chat_message_list):
|
||
neganslist.append(f"[{k+1}] {chat_message_list[list_of_worst_answers[k]][0]}: {chat_message_list[list_of_worst_answers[k]][1]}")
|
||
else:
|
||
neganslist.append(f"[{k+1}] N/A: No data available")
|
||
|
||
stastisticresults = countanswers(rating_data)
|
||
print()
|
||
|
||
chat_summarizer(chat_message_list)
|
||
|
||
end_time = time.perf_counter()
|
||
elapsed_time = end_time - start_time
|
||
|
||
data = {"Positive": stastisticresults[0], "Negative": stastisticresults[1], "Neutral": stastisticresults[2]}
|
||
chart1 = termcharts.bar(data, title="Chart of message emotions", rich=True)
|
||
|
||
layout = Layout()
|
||
layout.split_column(
|
||
Layout(name="upper", size=20),
|
||
Layout(name="lower")
|
||
)
|
||
|
||
layout["upper"].split_row(
|
||
Layout(name="uleft", ratio=1),
|
||
Layout(name="uright", ratio=3),
|
||
)
|
||
|
||
layout["uright"].split_row(
|
||
Layout(Panel(f"STATISTICS\n\nRating Average: {round(sum_of_numbers, 3)}\nPositive Ratings: {stastisticresults[0]}\nNegative Ratings: {stastisticresults[1]}\nNeutral Ratings: {stastisticresults[2]}\nTotal messages analyzed: {len(rating_data)}\nTotal time elapsed: {elapsed_time:.4f} seconds")),
|
||
Layout(Panel(fr"""
|
||
___ _________ ________
|
||
|\ \|\___ ___\\ __ \
|
||
\ \ \|___ \ \_\ \ \|\ \
|
||
\ \ \ \ \ \ \ \ __ \
|
||
\ \ \ \ \ \ \ \ \ \ \
|
||
\ \__\ \ \__\ \ \__\ \__\
|
||
\|__| \|__| \|__|\|__|
|
||
|
||
©️2026 P7MJ
|
||
All Rights Reserved
|
||
|
||
In the Air, {cc_version} by P7MJ
|
||
Know the mood - and what to say next.
|
||
"""))
|
||
)
|
||
|
||
layout["lower"].split_row(
|
||
Layout(Panel(f"OUTSTANDING MESSAGES\n\nMOST POSITIVE ANSWERS\n{posanslist[0][:200]}\n{posanslist[1][:200]}\n{posanslist[2][:200]}\n{posanslist[3][:200]}\n{posanslist[4][:200]}\n\n\nMOST NEGATIVE ANSWERS\n{neganslist[0][:200]}\n{neganslist[1][:200]}\n{neganslist[2][:200]}\n{neganslist[3][:200]}\n{neganslist[4][:200]}")),
|
||
Layout(Panel(f"CHAT SUMMARY\n\n{chat_summary}")),
|
||
)
|
||
|
||
layout["uleft"].update(Panel(chart1))
|
||
|
||
pr(layout)
|
||
input("PRESS ENTER TO LAUNCH ITA CHAT AGENT")
|
||
start_ita_agent(round(sum_of_numbers, 3), chat_summary) |