Python Socket库基础方法与应用详解

socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0)

import socket

def start_web_server(host='', port=8080):
    with socket.socket() as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        s.listen(5)
        print(f"Server running on {port}")

        while True:
            conn, addr = s.accept()
            with conn:
                request = conn.recv(1024).decode()
                # 解析HTTP请求头
                headers = request.split('\n')
                filename = headers[0].split()[1][1:] or 'index.html'

                try:
                    with open(filename, 'rb') as f:
                        content = f.read()
                    response = b'HTTP/1.1 200 OK\n\n' + content
                except FileNotFoundError:
                    response = b'HTTP/1.1 404 Not Found\n\n<h1>404 Error</h1>'

                conn.sendall(response)

# 使用示例
start_web_server()

import socket
import threading

clients = []

def handle_client(client):
    while True:
        try:
            msg = client.recv(1024)
            if not msg:
                break
            for c in clients:
                if c != client:
                    c.sendall(msg)
        except:
            break
    clients.remove(client)
    client.close()

def start_chat_server(port=9000):
    with socket.socket() as s:
        s.bind(('', port))
        s.listen()
        print(f"Chat server started on {port}")

        while True:
            client, addr = s.accept()
            clients.append(client)
            thread = threading.Thread(target=handle_client, args=(client,))
            thread.start()

# 客户端实现需配合使用连接和发送/接收线程

import socket
import hashlib

def send_file(filename, host='127.0.0.1', port=6000):
    with socket.socket() as s:
        s.connect((host, port))
        with open(filename, 'rb') as f:
            file_data = f.read()
            # 计算文件哈希值
            file_hash = hashlib.md5(file_data).hexdigest().encode()
            # 先发送哈希值
            s.sendall(file_hash)
            # 发送文件数据
            s.sendall(file_data)
        print("File sent successfully")

def receive_file(port=6000):
    with socket.socket() as s:
        s.bind(('', port))
        s.listen()
        conn, addr = s.accept()
        with conn:
            # 接收哈希值
            file_hash = conn.recv(32)
            file_data = b''
            while True:
                data = conn.recv(4096)
                if not data:
                    break
                file_data += data
            # 验证完整性
            if hashlib.md5(file_data).hexdigest().encode() == file_hash:
                with open('received_file', 'wb') as f:
                    f.write(file_data)
                print("File received successfully")
            else:
                print("File corrupted")

import socket
from concurrent.futures import ThreadPoolExecutor

def scan_port(host, port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(1)
        try:
            s.connect((host, port))
            print(f"Port {port} is open")
            return True
        except:
            return False

def port_scanner(host='127.0.0.1', start=1, end=1024):
    print(f"Scanning {host}...")
    with ThreadPoolExecutor(max_workers=100) as executor:
        results = executor.map(lambda p: scan_port(host, p), range(start, end+1))
    return sum(results)

# 使用示例
port_scanner(end=100)