Linux-tips

#!/usr/bin/env python3

import irc.bot

import requests

import os

import time

import sys

import queue

from concurrent.futures import ThreadPoolExecutor

from google import genai

from google.genai.errors import APIError

# ---------- CONFIGURATION ----------

IRC_SERVER = os.getenv("IRC_SERVER", "0.0.0.0")

IRC_PORT = int(os.getenv("IRC_PORT", "6667"))

IRC_PASSWORD = os.getenv("IRC_PASSWORD", "****")  # Set to your server password if required

IRC_CHANNEL = os.getenv("IRC_CHANNEL", "**")

BOT_NICK = os.getenv("BOT_NICK", "DualAIBot")

# ---------- API KEYS & ENDPOINTS ----------

GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "****")

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "****")

# Initialize official Gemini Client

gemini_client = genai.Client(api_key="****")  

DEEPSEEK_URL = "https://api.deepseek.com/v1/chat/completions"

# ---------- IRC MESSAGE SPLITTER ----------

def split_irc_message(text, max_bytes=400):

    """

    Split a long message into a list of strings,

    respecting IRC's size limit and destroying carriage returns.

    """

    chunks = []

    lines = [line.strip() for line in text.replace('\r', '').split('\n') if line.strip()]

    for line in lines:

        encoded = line.encode('utf-8')

        if len(encoded) <= max_bytes:

            chunks.append(line)

            continue

        current = ""

        for word in line.split():

            test = current + (" " + word if current else word)

            if len(test.encode('utf-8')) <= max_bytes:

                current = test

            else:

                if current:

                    chunks.append(current)

                current = word

        if current:

            chunks.append(current)

    return chunks

# ---------- CORE API CALLS ----------

def call_gemini(prompt):

    """Queries Gemini with a pacing buffer and a masked error fallback strategy."""

    time.sleep(1.5)

    for attempt in range(3):

        try:

            response = gemini_client.models.generate_content(

                model='gemini-2.5-flash',

                contents=prompt,

            )

            return response.text

        except APIError as e:

            if e.code == 429:

                print(f"[-] Gemini 429 hit. Backing off for {2 ** attempt + 2}s...")

                time.sleep(2 ** attempt + 2)

                continue

            return "Помилка зв'язку з Gemini. (Connection glitch on Gemini end.)"

        except Exception as e:

            print(f"[!] System Exception on Gemini call: {e}")

            return "Connection glitch on Gemini end."

    return "Трохи зайнятий, зачекайте... (Server is breathing heavy right now. Give me a second.)"

def call_deepseek(prompt):

    """Queries DeepSeek via a direct HTTP POST request wrapper."""

    headers = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}

    payload = {"model": "deepseek-chat", "messages": [{"role": "user", "content": prompt}], "temperature": 0.7}

    try:

        resp = requests.post(DEEPSEEK_URL, headers=headers, json=payload, timeout=30)

        if resp.status_code == 429:

            return "[DeepSeek Error] 429 Rate Limit Exceeded."

        resp.raise_for_status()

        data = resp.json()

        return data["choices"][0]["message"]["content"]

    except Exception as e:

        return f"[DeepSeek Error] {str(e)}"

# ---------- STANDARD ASYNCHRONOUS WORKER TASK ----------

def process_and_reply(msg_queue, channel, nick, prompt, target="both"):

    """Executes standard single or dual API queries off the main thread."""

    if target in ["both", "gemini"]:

        try:

            gemini_reply = call_gemini(prompt)

            chunks = split_irc_message(gemini_reply, max_bytes=400)

            for chunk in chunks:

                msg_queue.put((channel, f"{nick}: [Gemini] {chunk}"))

        except Exception:

            msg_queue.put((channel, f"{nick}: Gemini processing failed."))

    if target in ["both", "deepseek"]:

        try:

            deepseek_reply = call_deepseek(prompt)

            chunks = split_irc_message(deepseek_reply, max_bytes=400)

            for chunk in chunks:

                msg_queue.put((channel, f"{nick}: [DeepSeek] {chunk}"))

        except Exception:

            msg_queue.put((channel, f"{nick}: DeepSeek processing failed."))

# ---------- INTER-AI DISCUSSION WORKER LOOP ----------

def process_discussion_turn(bot_instance, channel, current_speaker, conversation_history):

    """Recursively bounces the conversation back and forth between engines until turns run out."""

    if bot_instance.discussion_turns <= 0:

        bot_instance.msg_queue.put((channel, "📢 [System] Discussion limit reached. Inter-AI loop terminated."))

        return

    bot_instance.discussion_turns -= 1

    # System framework to keep responses brief, conversational, and direct

    system_framing = (

        "Keep your response under 2 short paragraphs. Address your opponent's points bluntly and concisely. "

        "Do not include code blocks. "

    )

    if current_speaker == "gemini":

        prompt = f"{system_framing} This is an ongoing debate with DeepSeek on an IRC channel. Analyze the dialogue so far and reply directly to DeepSeek:\n\n{conversation_history}"

        reply = call_gemini(prompt)

        engine_prefix = "[Gemini]"

        next_speaker = "deepseek"

    else:

        prompt = f"{system_framing} This is an ongoing debate with Gemini on an IRC channel. Analyze the dialogue so far and reply directly to Gemini:\n\n{conversation_history}"

        reply = call_deepseek(prompt)

        engine_prefix = "[DeepSeek]"

        next_speaker = "gemini"

    # Push chunks out to the IRC queue

    chunks = split_irc_message(reply, max_bytes=400)

    for chunk in chunks:

        bot_instance.msg_queue.put((channel, f"{engine_prefix} {chunk}"))

    # Update history context for the next recursion step

    updated_history = f"{conversation_history}\n{engine_prefix}: {reply}"

    # Give the server a small breather to avoid flooding sockets too hard

    time.sleep(2.0)

    # Queue the next turn in the thread pool if limits permit

    if bot_instance.discussion_turns > 0:

        bot_instance.executor.submit(process_discussion_turn, bot_instance, channel, next_speaker, updated_history)

    else:

        bot_instance.msg_queue.put((channel, "📢 [System] Discussion limit reached. Inter-AI loop terminated."))

# ---------- IRC BOT WITH THREAD-SAFE OUTBOUND QUEUE ----------

class DualAIIRCBot(irc.bot.SingleServerIRCBot):

    def __init__(self, channel, nickname, server, port, password=None):

        irc.bot.SingleServerIRCBot.__init__(self, [irc.bot.ServerSpec(server, port, password)], nickname, nickname)

        self.channel = channel

        self.executor = ThreadPoolExecutor(max_workers=4)

        self.msg_queue = queue.Queue()

        # Discussion variables

        self.discussion_turns = 0

    def on_welcome(self, connection, event):

        print(f"✅ Connected to {IRC_SERVER}:{IRC_PORT}")

        print(f"📢 Joining channel: {self.channel}")

        connection.join(self.channel)

        connection.execute_every(0.1, self.check_message_queue)

    def on_pubmsg(self, connection, event):

        message = event.arguments[0]

        nick = event.source.nick

        # Ignore anything that looks like the bot quoting itself or talking to itself directly

        if message.startswith(f"{BOT_NICK}:") or message.startswith("[Gemini]") or message.startswith("[DeepSeek]"):

            return

        msg_lower = message.lower()

        prompt = ""

        target = ""

        # Handle the direct debate trigger

        if msg_lower.startswith("!discuss "):

            topic = message[9:].strip()

            if not topic:

                connection.privmsg(self.channel, f"{nick}: Usage: !discuss <topic to debate>")

                return

            # Initialize circuit breaker: 6 total turns means 3 replies from each engine

            self.discussion_turns = 6

            connection.privmsg(self.channel, f"📢 [System] Initiating Gemini vs DeepSeek debate on: '{topic}' ({self.discussion_turns} turns maximum).")

            initial_history = f"Topic of discussion: {topic}"

            # Kick off the chain with Gemini leading the argument

            self.executor.submit(process_discussion_turn, self, self.channel, "gemini", initial_history)

            return

        # Regular single-shot engine queries

        if msg_lower.startswith("!illya "):

            prompt = message[7:].strip()

            target = "both"

        elif msg_lower.startswith("!gemini "):

            prompt = message[8:].strip()

            target = "gemini"

        elif msg_lower.startswith("!deepseek "):

            prompt = message[10:].strip()

            target = "deepseek"

        if target:

            if not prompt:

                cmd_used = "!illya" if target == "both" else f"!{target}"

                connection.privmsg(self.channel, f"{nick}: Usage: {cmd_used} <question>")

                return

            connection.privmsg(self.channel, f"{nick}: 🤔 Processing query...")

            self.executor.submit(process_and_reply, self.msg_queue, self.channel, nick, prompt, target)

    def check_message_queue(self):

        """Processes pending responses. Bulletproofed against IRC protocol errors."""

        try:

            while True:

                target_channel, out_msg = self.msg_queue.get_nowait()

                clean_msg = out_msg.replace('\n', ' ').replace('\r', '')

                try:

                    self.connection.privmsg(target_channel, clean_msg)

                except Exception as e:

                    print(f"[!] Socket exception on outbound message: {e}")

                self.msg_queue.task_done()

        except queue.Empty:

            pass

    def on_disconnect(self, connection, event):

        print("⚠️ Disconnected. Main loop will cycle to initiate recovery...")

        raise SystemExit("Triggering automated restart cycle.")

# ---------- MAIN EXECUTION AND RECOVERY LOOP ----------

if __name__ == "__main__":

    while True:

        try:

            bot = DualAIIRCBot(IRC_CHANNEL, BOT_NICK, IRC_SERVER, IRC_PORT, IRC_PASSWORD if IRC_PASSWORD else None)

            print(f"🚀 Dual-AI bot initializing connection loop...")

            bot.start()

        except SystemExit:

            print("Restarting network thread state in 5 seconds...")

            time.sleep(5)

            continue

        except Exception as e:

            print(f"Critical operational error encountered: {e}")

            time.sleep(10)

            continue