tigrlvovichx228/khuinia-kakaeto

import socket import threading import json import os import sys import time from datetime import datetime class SimpleMessenger: def init(self): self.my_port = 5000 self.target_ip = "127.0.0.1" self.target_port = 5000 self.socket = None self.connected = False self.listening = False self.running = True self.messages = [] def clear(self): """Очистка экрана""" os.system('cls' if os.name == 'nt' else 'clear') def print_header(self): """Печать заголовка""" print("\n" + "█"*50) print(" ОФФЛАЙН МЕССЕНДЖЕР v1.0") print("█"*50) def print_status(self): """Печать статуса""" status = "ПОДКЛЮЧЕН" if self.connected else "НЕ ПОДКЛЮЧЕН" color = "\033[92m" if self.connected else "\033[91m" print(f"\nСтатус: {color}{status}\033[0m") if self.connected: print(f"Соединение: {self.target_ip}:{self.target_port}") print("-"*50) def show_menu(self): """Показать главное меню""" self.clear() self.print_header() self.print_status() print("\n1. 📡 Запустить сервер (режим ожидания)") print("2. 🔌 Подключиться к серверу") print("3. 💬 Отправить сообщение") print("4. 📁 Отправить файл") print("5. ❌ Отключиться") print("6. ⚙️ Настройки") print("7. 🚪 Выход") print("\n" + "-"*50) def start_server(self): """Запуск сервера""" if self.listening: print("⚠️ Сервер уже запущен!") return try: server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(('0.0.0.0', self.my_port)) server.listen(1) print(f"\n✅ Сервер запущен на порту {self.my_port}") print("⏳ Ожидание подключения...") # Принимаем подключение в отдельном потоке def accept(): try: self.socket, addr = server.accept() self.connected = True self.target_ip, self.target_port = addr print(f"\n🎉 Подключено к {self.target_ip}:{self.target_port}") # Запускаем прием сообщений receive_thread = threading.Thread(target=self.receive_messages, daemon=True) receive_thread.start() server.close() except Exception as e: print(f"\n❌ Ошибка: {e}") threading.Thread(target=accept, daemon=True).start() self.listening = True except Exception as e: print(f"\n❌ Ошибка запуска сервера: {e}") def connect(self): """Подключение к серверу""" if self.connected: print("⚠️ Уже подключены!") return try: print(f"\n🔄 Подключение к {self.target_ip}:{self.target_port}...") self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.connect((self.target_ip, self.target_port)) self.connected = True print(f"✅ Подключено успешно!") # Запускаем прием сообщений receive_thread = threading.Thread(target=self.receive_messages, daemon=True) receive_thread.start() except Exception as e: print(f"\n❌ Ошибка подключения: {e}") def send_message(self): """Отправка сообщения""" if not self.connected: print("\n❌ Нет подключения!") return print("\n📝 Введите сообщение (пустая строка для отмены):") message = input("➤ ").strip() if not message: return data = {"type": "msg", "content": message, "time": datetime.now().strftime("%H:%M:%S") } try: self.socket.send((json.dumps(data) + "\n").encode()) print(f"\n💬 [{datetime.now().strftime('%H:%M:%S')}] Вы: {message}") self.messages.append(("Вы", message, datetime.now().strftime("%H:%M:%S"))) except Exception as e: print(f"\n❌ Ошибка отправки: {e}") self.disconnect() def send_file(self): """Отправка файла""" if not self.connected: print("\n❌ Нет подключения!") return path = input("\n📂 Введите путь к файлу: ").strip() if not os.path.exists(path): print("❌ Файл не найден!") return try: name = os.path.basename(path) size = os.path.getsize(path) print(f"📤 Отправка файла {name} ({size} байт)...") # Отправляем информацию о файле info = { "type": "file", "name": name, "size": size } self.socket.send((json.dumps(info) + "\n").encode()) # Отправляем содержимое файла with open(path, "rb") as f: sent = 0 while sent < size: chunk = f.read(4096) if not chunk: break self.socket.send(chunk) sent += len(chunk) print(f"✅ Файл {name} отправлен!") self.messages.append(("Вы", f"[Файл] {name}", datetime.now().strftime("%H:%M:%S"))) except Exception as e: print(f"❌ Ошибка: {e}") def receive_messages(self): """Прием сообщений""" buffer = "" while self.connected: try: data = self.socket.recv(4096).decode() if not data: break buffer += data while "\n" in buffer: line, buffer = buffer.split("\n", 1) if line.strip(): self.process_message(line) except Exception as e: if self.connected: print(f"\n⚠️ Потеряно соединение: {e}") break self.disconnect() def process_message(self, data_str): """Обработка полученного сообщения""" try: data = json.loads(data_str) if data["type"] == "msg": text = f"[{data['time']}] Собеседник: {data['content']}" print(f"\n💬 {text}") self.messages.append(("Собеседник", data['content'], data['time'])) print("\n➤ ", end="", flush=True) elif data["type"] == "file": self.receive_file(data["name"], data["size"]) except Exception as e: print(f"❌ Ошибка обработки: {e}") def receive_file(self, name, size): """Получение файла""" save_path = input(f"\n📥 Сохранить файл '{name}' как (Enter для '{name}'): ").strip() if not save_path: save_path = name try: print(f"📥 Получение файла {name} ({size} байт)...") received = 0 with open(save_path, "wb") as f: while received < size: chunk = self.socket.recv(min(4096, size - received)) if not chunk: break f.write(chunk) received += len(chunk) print(f"✅ Файл сохранен как {save_path}") self.messages.append(("Собеседник", f"[Файл] {name}", datetime.now().strftime("%H:%M:%S"))) print("\n➤ ", end="", flush=True)except Exception as e: print(f"❌ Ошибка сохранения: {e}") def disconnect(self): """Отключение""" if self.socket: try: self.socket.close() except: pass self.socket = None self.connected = False self.listening = False print("\n🔌 Отключено") def settings(self): """Настройки""" self.clear() self.print_header() print("\n⚙️ НАСТРОЙКИ") print("-"*50) print(f"1. Мой порт: {self.my_port}") print(f"2. IP собеседника: {self.target_ip}") print(f"3. Порт собеседника: {self.target_port}") print("4. Показать историю сообщений") print("5. Очистить историю") print("6. Назад") print("-"*50) choice = input("\nВыберите параметр: ") if choice == "1": try: port = int(input("Введите новый порт: ")) self.my_port = port print(f"✅ Порт изменен на {port}") except: print("❌ Неверный порт!") elif choice == "2": ip = input("Введите IP адрес: ") self.target_ip = ip print(f"✅ IP изменен на {ip}") elif choice == "3": try: port = int(input("Введите порт: ")) self.target_port = port print(f"✅ Порт изменен на {port}") except: print("❌ Неверный порт!") elif choice == "4": self.show_history() elif choice == "5": self.messages = [] print("✅ История очищена") input("\nНажмите Enter для продолжения...") def show_history(self): """Показать историю сообщений""" self.clear() self.print_header() print("\n📜 ИСТОРИЯ СООБЩЕНИЙ") print("-"*50) if not self.messages: print("История пуста") else: for sender, text, time in self.messages: print(f"[{time}] {sender}: {text}") print("-"*50) input("\nНажмите Enter для продолжения...") def run(self): """Основной цикл""" while self.running: self.show_menu() choice = input("Выберите действие (1-7): ").strip() if choice == "1": self.start_server() elif choice == "2": self.connect() elif choice == "3": self.send_message() elif choice == "4": self.send_file() elif choice == "5": self.disconnect() elif choice == "6": self.settings() elif choice == "7": print("\n👋 До свидания!") self.running = False else: print("\n❌ Неверный выбор!") if self.running and not self.connected and choice not in ["1", "6", "7"]: input("\nНажмите Enter для продолжения...") self.disconnect()

язык не указан · рейтинг 0 · 👍 0 · ❤️ 0 · 💎 0 · 0 форков · ссылка на SourceCraft

Дата анализа: 29.09.2026 22:56