Merge branch 'dev' of https://github.com/shelllet/winui into dev

assets/script/http_server.py

@@ -0,0 +1,97 @@
+from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
+from urllib.parse import urlparse, parse_qs
+from signal import signal, SIGINT
+from sys import exit
+from json import loads, dumps
+from argparse import ArgumentParser
+
+class HttpHandler(BaseHTTPRequestHandler):
+    protocol_version = 'HTTP/1.1'
+    error_content_type = 'text/plain'
+    error_message_format = "Error %(code)d: %(message)s"
+
+    def do_GET(self):
+        path, args = self.parse_url()
+
+        if path == '/hello' and 'name' in args:
+            name = args['name'][0]
+
+            self.write_response(200, "text/plain", f"Hello, {name}!")
+        else:
+            self.send_error(404, 'Not found')
+
+    def do_POST(self):
+        path, _ = self.parse_url()
+        body = self.read_body()
+
+        if path == '/echo' and body and self.headers['Content-Type'] == "application/json":
+            json_body = self.parse_json(body)
+
+            if json_body:
+                self.write_response(202, "application/json",
+                                    dumps({"message": "Accepted", "request": json_body}, indent=2))
+            else:
+                self.send_error(400, 'Invalid json received')
+        elif path == '/echo' and body:
+            self.write_response(202, "text/plain", f"Accepted: {body}")
+        else:
+            self.send_error(404, 'Not found')
+
+    def parse_url(self):
+        url_components = urlparse(self.path)
+        return url_components.path, parse_qs(url_components.query)
+
+    def parse_json(self, content):
+        try:
+            return loads(content)
+        except Exception:
+            return None
+
+    def read_body(self):
+        try:
+            content_length = int(self.headers['Content-Length'])
+            return self.rfile.read(content_length).decode('utf-8')
+        except Exception:
+            return None
+
+    def write_response(self, status_code, content_type, content):
+        response = content.encode('utf-8')
+
+        self.send_response(status_code)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(response)))
+        self.end_headers()
+        self.wfile.write(response)
+
+    def version_string(self):
+        return "Tiny Http Server"
+
+    def log_error(self, format, *args):
+        pass
+
+
+def start_server(host, port):
+    server_address = (host, port)
+    httpd = ThreadingHTTPServer(server_address, HttpHandler)
+    print(f"Server started on {host}:{port}")
+    httpd.serve_forever()
+
+
+def shutdown_handler(signum, frame):
+    print('Shutting down server')
+    exit(0)
+
+
+def main():
+    signal(SIGINT, shutdown_handler)
+    parser = ArgumentParser(description='Start a tiny HTTP/1.1 server')
+    parser.add_argument('--host', type=str, action='store',
+                        default='127.0.0.1', help='Server host (default: 127.0.0.1)')
+    parser.add_argument('--port', type=int, action='store',
+                        default=8000, help='Server port (default: 8000)')
+    args = parser.parse_args()
+    start_server(args.host, args.port)
+
+
+if __name__ == "__main__":
+    main()

assets/script/tcp_client.py

@@ -0,0 +1,19 @@
+#coding:utf-8
+import sys
+from socket import *
+serverHost = 'localhost'
+serverPort = 6000
+#发送至服务端的默认文本
+message = ['Hello network world']
+#建立一个tcp/ip套接字对象
+sockobj = socket(AF_INET, SOCK_STREAM)
+#连接至服务器及端口
+sockobj.connect((serverHost, serverPort))
+for line in message:
+    #经过套按字发送line至服务端
+    sockobj.send(line.encode())
+    #从服务端接收到的数据，上限为1k
+    data = sockobj.recv(1024)
+    print('Client received:', repr(data))
+#关闭套接字
+sockobj.close( )

assets/script/tcp_server.py

@@ -0,0 +1,27 @@
+#coding:utf-8
+from socket import *
+#''代表服务器为 localhost
+myHost = ''
+#在一个非保留端口号上进行监听
+myPort = 6000
+#设置一个TCP socket对象
+sockobj = socket(AF_INET, SOCK_STREAM)
+#绑定端口号
+sockobj.bind((myHost, myPort))
+#监听，允许5个连结
+sockobj.listen(5)
+#直到进程结束时才结束循环
+while True:
+    #等待客户端连接
+    connection, address = sockobj.accept( )
+    #连接是一个新的socket
+    print('Server connected by', address)
+    while True:
+        #读取客户端套接字的下一行
+        data = connection.recv(1024)
+        #如果没有数量的话，那么跳出循环
+        if not data: break
+        #发送一个回复至客户端
+        connection.send('Echo from server => '.encode() + data)
+    #当socket关闭时eof
+    connection.close( )

assets/script/udp_client.py

@@ -0,0 +1,16 @@
+import socket
+
+s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+addr = ("127.0.0.1", 6000)
+
+
+s.sendto('hello server!'.encode(), addr)
+
+while True:
+	response, addr = s.recvfrom(1024)
+	print(response.decode())
+	if response.decode() == "exit":
+		print("Session is over from the server %s:%s\n" % addr)
+		break
+
+s.close()

assets/script/udp_send.py

@@ -0,0 +1,13 @@
+import socket
+import time
+
+s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+addr = ("127.0.0.1", 6000)
+
+
+
+while True:
+	s.sendto('hello server!'.encode(), addr)
+	time.sleep(1)
+
+s.close()

assets/script/udp_server.py

@@ -0,0 +1,13 @@
+import socket
+
+s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+s.bind(("127.0.0.1", 6001))
+print("UDP bound on port 6001...")
+
+while True:
+	data, addr = s.recvfrom(1024)
+	print("Receive from ", addr, data)
+	if data == b"exit":
+		s.sendto(b"Good bye!\n", addr)
+		continue
+	s.sendto(b"Hello client, i'm a boy!" , addr)