chat_server_02.py
をテンプレートにして作成
[
トップ
] [
新規
|
一覧
|
検索
|
最終更新
|
ヘルプ
|
ログイン
]
開始行:
[[TESLA2]]
#code(Python){{
import socket
import threading
import queue
import tkinter as tk
from tkinter import scrolledtext
from datetime import datetime
import ipget
HOST = ''
PORT = 9998
clients = []
clients_lock = threading.Lock()
# GUIへメッセージを渡すためのQueue
message_queue = queue.Queue()
def log_message(message):
"""GUIに表示するメッセージをQueueへ入れる"""
now = datetime.now().strftime("%H:%M:%S")
message_queue.put(f"[{now}] {message}")
def remove_connection(con, address):
"""クライアントとの接続を切る"""
global clients
log_message(f"[切断] {address}")
try:
con.close()
except Exception:
pass
with clients_lock:
clients = [
c for c in clients
if c[0] != con
]
def server_start():
"""サーバをスタートする"""
global clients
sock = socket.socket(socket.AF_INET, socket.SOCK_STRE...
sock.setsockopt(
socket.SOL_SOCKET,
socket.SO_REUSEADDR,
1
)
try:
sock.bind((HOST, PORT))
sock.listen(10)
log_message(f"サーバ開始 PORT={PORT}")
except Exception as e:
log_message(f"server start error: {e}")
return
while True:
try:
con, address = sock.accept()
log_message(f"[接続] {address}")
with clients_lock:
clients.append((con, address))
handle_thread = threading.Thread(
target=handler,
args=(con, address),
daemon=True
)
handle_thread.start()
except Exception as e:
log_message(
f"accept error: {e}"
)
def handler(con, address):
"""クライアントからデータを受信する"""
try:
fx = con.makefile(
'r',
encoding='utf-8'
)
except ConnectionResetError:
remove_connection(
con,
address
)
return
except Exception as e:
log_message(
f"makefile error: {e}"
)
remove_connection(
con,
address
)
return
while True:
try:
line = fx.readline()
# readline() が "" の場合は
# 相手が接続を閉じた
if line == "":
log_message(
f"connection closed: {address}"
)
remove_connection(
con,
address
)
return
# 改行だけなら無視
if line == "\n":
continue
# GUIへ表示
log_message(
f"[受信] {address} - {line.rstrip()}"
)
# 他のクライアントへ送信
with clients_lock:
current_clients = list(clients)
for client_con, client_address in current_cli...
if client_con != con:
try:
client_con.sendall(
line.encode("utf-8")
)
log_message(
f"[転送] → {client_address} : "
f"{line.rstrip()}"
)
except Exception as e:
log_message(
f"[送信エラー] "
f"{client_address}: {e}"
)
except ConnectionResetError:
log_message(
f"connection reset: {address}"
)
remove_connection(
con,
address
)
return
except Exception as e:
log_message(
f"Exception: {e}, "
f"address={address}"
)
remove_connection(
con,
address
)
return
class ServerGUI:
def __init__(self, root):
self.root = root
root.title(
"Python Communication Server"
)
root.geometry("900x600")
# -------------------------
# タイトル
# -------------------------
title_label = tk.Label(
root,
text="TCP Communication Server",
font=("Helvetica", 18)
)
title_label.pack(
pady=10
)
# -------------------------
# サーバ情報
# -------------------------
self.info_label = tk.Label(
root,
text=f"PORT : {PORT}",
font=("Helvetica", 12)
)
self.info_label.pack()
# -------------------------
# 通信ログ表示
# -------------------------
self.text_area = scrolledtext.ScrolledText(
root,
width=100,
height=28,
font=("Courier", 11)
)
self.text_area.pack(
padx=10,
pady=10,
fill=tk.BOTH,
expand=True
)
# -------------------------
# ボタン
# -------------------------
button_frame = tk.Frame(root)
button_frame.pack(
pady=5
)
clear_button = tk.Button(
button_frame,
text="Clear",
width=15,
command=self.clear_log
)
clear_button.pack(
side=tk.LEFT,
padx=5
)
quit_button = tk.Button(
button_frame,
text="Quit",
width=15,
command=self.quit_program
)
quit_button.pack(
side=tk.LEFT,
padx=5
)
# Queueを定期的にチェック
self.check_queue()
def clear_log(self):
self.text_area.delete(
"1.0",
tk.END
)
def quit_program(self):
self.root.destroy()
def check_queue(self):
try:
while True:
message = message_queue.get_nowait()
self.text_area.insert(
tk.END,
message + "\n"
)
# 常に最新行を表示
self.text_area.see(
tk.END
)
except queue.Empty:
pass
# 100ms後に再チェック
self.root.after(
100,
self.check_queue
)
def main():
# ----------------------------
# GUI開始
# ----------------------------
root = tk.Tk()
gui = ServerGUI(root)
# ----------------------------
# IPアドレス表示
# ----------------------------
try:
ipx = ipget.ipget()
ip_address = ipx.ipaddr(
'eth0'
)
log_message(
f"eth0: {ip_address}"
)
except Exception as e:
log_message(
f"IP address error: {e}"
)
# ----------------------------
# TCP Server開始
# ----------------------------
server_thread = threading.Thread(
target=server_start,
daemon=True
)
server_thread.start()
# ----------------------------
# GUIメインループ
# ----------------------------
root.mainloop()
if __name__ == "__main__":
main()
}}
----
#counter
終了行:
[[TESLA2]]
#code(Python){{
import socket
import threading
import queue
import tkinter as tk
from tkinter import scrolledtext
from datetime import datetime
import ipget
HOST = ''
PORT = 9998
clients = []
clients_lock = threading.Lock()
# GUIへメッセージを渡すためのQueue
message_queue = queue.Queue()
def log_message(message):
"""GUIに表示するメッセージをQueueへ入れる"""
now = datetime.now().strftime("%H:%M:%S")
message_queue.put(f"[{now}] {message}")
def remove_connection(con, address):
"""クライアントとの接続を切る"""
global clients
log_message(f"[切断] {address}")
try:
con.close()
except Exception:
pass
with clients_lock:
clients = [
c for c in clients
if c[0] != con
]
def server_start():
"""サーバをスタートする"""
global clients
sock = socket.socket(socket.AF_INET, socket.SOCK_STRE...
sock.setsockopt(
socket.SOL_SOCKET,
socket.SO_REUSEADDR,
1
)
try:
sock.bind((HOST, PORT))
sock.listen(10)
log_message(f"サーバ開始 PORT={PORT}")
except Exception as e:
log_message(f"server start error: {e}")
return
while True:
try:
con, address = sock.accept()
log_message(f"[接続] {address}")
with clients_lock:
clients.append((con, address))
handle_thread = threading.Thread(
target=handler,
args=(con, address),
daemon=True
)
handle_thread.start()
except Exception as e:
log_message(
f"accept error: {e}"
)
def handler(con, address):
"""クライアントからデータを受信する"""
try:
fx = con.makefile(
'r',
encoding='utf-8'
)
except ConnectionResetError:
remove_connection(
con,
address
)
return
except Exception as e:
log_message(
f"makefile error: {e}"
)
remove_connection(
con,
address
)
return
while True:
try:
line = fx.readline()
# readline() が "" の場合は
# 相手が接続を閉じた
if line == "":
log_message(
f"connection closed: {address}"
)
remove_connection(
con,
address
)
return
# 改行だけなら無視
if line == "\n":
continue
# GUIへ表示
log_message(
f"[受信] {address} - {line.rstrip()}"
)
# 他のクライアントへ送信
with clients_lock:
current_clients = list(clients)
for client_con, client_address in current_cli...
if client_con != con:
try:
client_con.sendall(
line.encode("utf-8")
)
log_message(
f"[転送] → {client_address} : "
f"{line.rstrip()}"
)
except Exception as e:
log_message(
f"[送信エラー] "
f"{client_address}: {e}"
)
except ConnectionResetError:
log_message(
f"connection reset: {address}"
)
remove_connection(
con,
address
)
return
except Exception as e:
log_message(
f"Exception: {e}, "
f"address={address}"
)
remove_connection(
con,
address
)
return
class ServerGUI:
def __init__(self, root):
self.root = root
root.title(
"Python Communication Server"
)
root.geometry("900x600")
# -------------------------
# タイトル
# -------------------------
title_label = tk.Label(
root,
text="TCP Communication Server",
font=("Helvetica", 18)
)
title_label.pack(
pady=10
)
# -------------------------
# サーバ情報
# -------------------------
self.info_label = tk.Label(
root,
text=f"PORT : {PORT}",
font=("Helvetica", 12)
)
self.info_label.pack()
# -------------------------
# 通信ログ表示
# -------------------------
self.text_area = scrolledtext.ScrolledText(
root,
width=100,
height=28,
font=("Courier", 11)
)
self.text_area.pack(
padx=10,
pady=10,
fill=tk.BOTH,
expand=True
)
# -------------------------
# ボタン
# -------------------------
button_frame = tk.Frame(root)
button_frame.pack(
pady=5
)
clear_button = tk.Button(
button_frame,
text="Clear",
width=15,
command=self.clear_log
)
clear_button.pack(
side=tk.LEFT,
padx=5
)
quit_button = tk.Button(
button_frame,
text="Quit",
width=15,
command=self.quit_program
)
quit_button.pack(
side=tk.LEFT,
padx=5
)
# Queueを定期的にチェック
self.check_queue()
def clear_log(self):
self.text_area.delete(
"1.0",
tk.END
)
def quit_program(self):
self.root.destroy()
def check_queue(self):
try:
while True:
message = message_queue.get_nowait()
self.text_area.insert(
tk.END,
message + "\n"
)
# 常に最新行を表示
self.text_area.see(
tk.END
)
except queue.Empty:
pass
# 100ms後に再チェック
self.root.after(
100,
self.check_queue
)
def main():
# ----------------------------
# GUI開始
# ----------------------------
root = tk.Tk()
gui = ServerGUI(root)
# ----------------------------
# IPアドレス表示
# ----------------------------
try:
ipx = ipget.ipget()
ip_address = ipx.ipaddr(
'eth0'
)
log_message(
f"eth0: {ip_address}"
)
except Exception as e:
log_message(
f"IP address error: {e}"
)
# ----------------------------
# TCP Server開始
# ----------------------------
server_thread = threading.Thread(
target=server_start,
daemon=True
)
server_thread.start()
# ----------------------------
# GUIメインループ
# ----------------------------
root.mainloop()
if __name__ == "__main__":
main()
}}
----
#counter
ページ名:
![[PukiWiki]](image/pukiwiki.png "[PukiWiki]")
