termux irc client
A page to document the various attempts at creating an irc client in Termux.
termux IRC client first try
irc_client_advanced.py
import socket
import ssl
import threading
import sys
import time
import queue
from datetime import datetime
class IRCClient:
def __init__(self):
self.server = "irc.rizon.net"
self.port = 6697 # SSL port for Rizon
self.channels = ["#/g/tv", "#/sp/"] # Default channels to join
self.active_channel = self.channels[0] if self.channels else None # Default active channel
self.nickname = "LullSac"
self.realname = "Termux IRC Client"
self.irc = None
self.context = ssl.create_default_context()
self.running = True
self.message_queue = queue.Queue()
self.lock = threading.Lock() # For thread-safe channel list updates
def connect(self):
try:
# Create socket and wrap with SSL
raw_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.irc = self.context.wrap_socket(raw_socket, server_hostname=self.server)
print(f"Connecting to {self.server}:{self.port}...")
self.irc.connect((self.server, self.port))
# Send user and nick information
self.send(f"USER {self.nickname} 0 * :{self.realname}")
self.send(f"NICK {self.nickname}")
return True
except Exception as e:
print(f"Connection error: {e}")
return False
def send(self, message):
try:
self.irc.send(f"{message}\r\n".encode('utf-8'))
except Exception as e:
print(f"Error sending message: {e}")
def get_timestamp(self):
# Return current time in 12-hour format (e.g., 3:45 PM) with blue ANSI color
timestamp = datetime.now().strftime("%I:%M %p")
return f"\033[34m[{timestamp}]\033[0m"
def handle_input(self):
while self.running:
try:
# Show prompt with active channel
with self.lock:
prompt = f"[{self.active_channel}] > " if self.active_channel else "[No channel] > "
sys.stdout.write(prompt)
sys.stdout.flush()
message = input().strip()
if not self.running:
break
if message:
self.message_queue.put(message)
except KeyboardInterrupt:
self.message_queue.put("QUIT :Goodbye")
break
def handle_server(self):
while self.running:
try:
data = self.irc.recv(2048).decode('utf-8')
if not data:
print(f"{self.get_timestamp()} Disconnected from server.")
self.running = False
break
for line in data.strip().split('\r\n'):
if not line:
continue
# Add blue timestamp to all server lines
print(f"{self.get_timestamp()} {line}")
# Handle PING
if line.startswith("PING"):
self.send(f"PONG {line.split()[1]}")
# Join channels after MOTD
if "376" in line or "422" in line:
with self.lock:
for channel in self.channels:
self.send(f"JOIN {channel}")
print(f"{self.get_timestamp()} Joined {channel}")
self.active_channel = self.channels[0] if self.channels else None
# Parse messages for display
if "PRIVMSG" in line:
sender = line[1:line.index('!')]
target = line.split()[2]
msg = line[line.index(':', 1) + 1:]
# Timestamp already added above, just format the message
print(f"{self.get_timestamp()} <{sender}@{target}> {msg}")
# Handle channel join confirmation
if "JOIN" in line and self.nickname in line:
channel = line.split(':', 2)[-1]
with self.lock:
if channel not in self.channels:
self.channels.append(channel)
print(f"{self.get_timestamp()} Added {channel} to active channels.")
if not self.active_channel:
self.active_channel = channel
# Handle parting a channel
if "PART" in line and self.nickname in line:
channel = line.split()[2]
with self.lock:
if channel in self.channels:
self.channels.remove(channel)
print(f"{self.get_timestamp()} Left {channel}")
if self.active_channel == channel:
self.active_channel = self.channels[0] if self.channels else None
print(f"{self.get_timestamp()} Active channel switched to: {self.active_channel or 'None'}")
except Exception as e:
print(f"{self.get_timestamp()} Server error: {e}")
self.running = False
break
def process_commands(self):
while self.running:
try:
message = self.message_queue.get(timeout=1)
if message.lower() == "quit":
self.send("QUIT :Goodbye")
self.running = False
elif message.startswith('/'):
self.handle_command(message)
else:
# Send to active channel
with self.lock:
if self.active_channel:
self.send(f"PRIVMSG {self.active_channel} :{message}")
else:
print(f"{self.get_timestamp()} No active channel. Use /join <channel> or /switch <index>.")
except queue.Empty:
continue
except Exception as e:
print(f"{self.get_timestamp()} Command error: {e}")
def handle_command(self, command):
parts = command.split(maxsplit=2)
cmd = parts[0].lower()
if cmd == "/nick" and len(parts) > 1:
self.nickname = parts[1]
self.send(f"NICK {self.nickname}")
elif cmd == "/join" and len(parts) > 1:
channel = parts[1]
self.send(f"JOIN {channel}")
with self.lock:
if channel not in self.channels:
self.channels.append(channel)
print(f"{self.get_timestamp()} Requested to join {channel}")
elif cmd == "/msg" and len(parts) > 2:
target = parts[1]
msg = parts[2]
self.send(f"PRIVMSG {target} :{msg}")
elif cmd == "/list":
with self.lock:
if self.channels:
print(f"{self.get_timestamp()} Active channels:")
for i, channel in enumerate(self.channels):
mark = "*" if channel == self.active_channel else " "
print(f"{self.get_timestamp()} {i}: {channel}{mark}")
else:
print(f"{self.get_timestamp()} No channels joined.")
elif cmd == "/part" and len(parts) > 1:
channel = parts[1]
self.send(f"PART {channel}")
elif cmd == "/switch" and len(parts) > 1:
try:
index = int(parts[1])
with self.lock:
if 0 <= index < len(self.channels):
self.active_channel = self.channels[index]
print(f"{self.get_timestamp()} Switched to active channel: {self.active_channel}")
else:
print(f"{self.get_timestamp()} Invalid index. Use /list to see channels.")
except ValueError:
print(f"{self.get_timestamp()} Invalid index. Use a number (e.g., /switch 0).")
else:
print(f"{self.get_timestamp()} Unknown command or invalid syntax. Try: /nick, /join, /msg, /list, /part, /switch")
def run(self):
if not self.connect():
return
# Start threads
server_thread = threading.Thread(target=self.handle_server)
input_thread = threading.Thread(target=self.handle_input)
command_thread = threading.Thread(target=self.process_commands)
server_thread.start()
input_thread.start()
command_thread.start()
# Wait for threads to finish
try:
server_thread.join()
input_thread.join()
command_thread.join()
except KeyboardInterrupt:
self.running = False
# Clean up
self.irc.close()
print(f"{self.get_timestamp()} Connection closed.")
def main():
client = IRCClient()
client.run()
if __name__ == "__main__":
main()
The client will:
- Connect to irc.rizon.net as LullSac.
- Attempt to join #/g/tv and #/sp/.
- Set #/g/tv as the active channel.
- Show a prompt like [#/g/tv] > .
- Display all messages with blue timestamps (e.g., [3:45 PM] <LullSac@#/g/tv> Hello).
Use the UI:
- Type messages (e.g., Hello) to send to the active channel (e.g., #/g/tv).
- /list: See channels (e.g., 0: #/g/tv *, 1: #/sp/).
- /switch 1: Switch to #/sp/ (prompt changes to [#/sp/] > ).
- /join #newchannel: Join another channel.
- /part #/sp/: Leave a channel.
- /msg #/g/tv Hi: Send to a specific channel.
- /nick NewName: Change nickname.
- /msg SomeUser Hi: Send a private message.
- quit: Exit the client.