'''
网络编程基本概念:
- 协议(Protocol): 通信规则和格式的约定
- 套接字(Socket): 网络通信的端点
- 客户端-服务器模型: 最常见的网络通信模式
- 端口(Port): 区分主机上不同应用程序的数字标识符
- IP地址: 标识网络上设备的地址

Python网络编程的优势:
- 简洁的API，降低开发复杂度
- 丰富的标准库和第三方库支持
- 跨平台兼容性好
- 适合快速原型开发和生产级应用
'''

import socket

def tcp_client():
    """简单的TCP客户端示例"""
    # 创建TCP套接字
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        # 连接服务器
        client.connect(('example.com', 80))
        print("已连接到服务器")
        
        # 发送HTTP请求
        request = "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
        client.sendall(request.encode())
        
        # 接收响应
        response = b''
        while True:
            chunk = client.recv(4096)
            if not chunk:
                break
            response += chunk
            
        print(f"收到 {len(response)} 字节的响应：")
        print(response.decode()[:200] + "...")  # 只显示前200个字符
        
    finally:
        client.close()
        print("连接已关闭")

## 调用客户端函数
if __name__ == "__main__":
    tcp_client()

import socket
import threading

def handle_client(client_socket, address):
    """处理客户端连接
    
    Args:
        client_socket: 客户端套接字
        address: 客户端地址
    """
    print(f"接受来自 {address} 的连接")
    
    # 接收客户端数据
    data = client_socket.recv(1024)
    
    # 处理数据
    response = f"收到 {len(data)} 字节的数据".encode()
    
    # 发送响应
    client_socket.send(response)
    
    # 关闭连接
    client_socket.close()
    print(f"与 {address} 的连接已关闭")

def tcp_server():
    """简单的多线程TCP服务器"""
    # 创建服务器套接字
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # 设置地址重用，防止"地址已在使用"错误
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # 绑定地址和端口
    server.bind(('0.0.0.0', 8888))
    
    # 开始监听
    server.listen(5)
    print("服务器启动，监听端口 8888...")
    
    try:
        while True:
            # 接受客户端连接
            client, address = server.accept()
            
            # 为每个客户端创建新线程
            client_thread = threading.Thread(
                target=handle_client,
                args=(client, address)
            )
            client_thread.daemon = True
            client_thread.start()
    
    except KeyboardInterrupt:
        print("服务器关闭中...")
    finally:
        server.close()

## 调用服务器函数
if __name__ == "__main__":
    tcp_server()

import socket

def udp_client():
    """简单的UDP客户端示例"""
    # 创建UDP套接字
    client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
    # 服务器地址
    server_address = ('localhost', 9999)
    
    try:
        # 发送消息
        message = "Hello, UDP Server!"
        client.sendto(message.encode(), server_address)
        print(f"已发送消息: {message}")
        
        # 接收响应
        client.settimeout(5)  # 设置超时
        try:
            data, server = client.recvfrom(4096)
            print(f"收到来自 {server} 的响应: {data.decode()}")
        except socket.timeout:
            print("接收响应超时")
    
    finally:
        client.close()
        print("套接字已关闭")

## 调用客户端函数
if __name__ == "__main__":
    udp_client()

import socket

def udp_server():
    """简单的UDP服务器示例"""
    # 创建UDP套接字
    server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
    # 绑定地址和端口
    server_address = ('0.0.0.0', 9999)
    server.bind(server_address)
    
    print(f"UDP服务器启动，监听 {server_address}...")
    
    try:
        while True:
            # 接收数据
            data, client_address = server.recvfrom(4096)
            print(f"收到来自 {client_address} 的 {len(data)} 字节数据")
            
            # 处理接收到的数据
            message = data.decode()
            print(f"收到消息: {message}")
            
            # 发送响应
            response = f"服务器已收到: {message}"
            server.sendto(response.encode(), client_address)
    
    except KeyboardInterrupt:
        print("服务器关闭中...")
    finally:
        server.close()

## 调用服务器函数
if __name__ == "__main__":
    udp_server()

import http.client
import json

def http_client_standard():
    """使用标准库的HTTP客户端"""
    # 创建连接
    conn = http.client.HTTPSConnection("api.github.com")
    
    # 设置请求头
    headers = {
        'User-Agent': 'Python HTTP Client',
        'Accept': 'application/json'
    }
    
    try:
        # 发送GET请求
        conn.request("GET", "/users/octocat", headers=headers)
        
        # 获取响应
        response = conn.getresponse()
        print(f"状态码: {response.status} {response.reason}")
        
        # 读取响应体
        data = response.read().decode('utf-8')
        
        # 解析JSON响应
        user_data = json.loads(data)
        print(f"用户名: {user_data.get('login')}")
        print(f"个人主页: {user_data.get('html_url')}")
        
    finally:
        # 关闭连接
        conn.close()

## 运行示例
if __name__ == "__main__":
    http_client_standard()

import requests

def requests_example():
    """使用requests库的示例"""
    # 设置请求参数
    url = "https://api.github.com/users/octocat"
    headers = {
        'User-Agent': 'Python Requests',
        'Accept': 'application/json'
    }
    
    try:
        # 发送GET请求
        response = requests.get(url, headers=headers)
        
        # 自动检查HTTP错误
        response.raise_for_status()
        
        # 解析JSON响应
        user_data = response.json()
        
        print(f"用户名: {user_data.get('login')}")
        print(f"个人主页: {user_data.get('html_url')}")
        print(f"公开仓库数: {user_data.get('public_repos')}")
        
        # 获取响应头信息
        print("\n响应头信息:")
        for key, value in response.headers.items():
            print(f"{key}: {value}")
    
    except requests.exceptions.HTTPError as e:
        print(f"HTTP错误: {e}")
    except requests.exceptions.ConnectionError as e:
        print(f"连接错误: {e}")
    except requests.exceptions.Timeout as e:
        print(f"请求超时: {e}")
    except requests.exceptions.RequestException as e:
        print(f"请求异常: {e}")

## 运行示例
if __name__ == "__main__":
    requests_example()

import aiohttp
import asyncio

async def fetch_user(username):
    """异步获取用户信息
    
    Args:
        username: GitHub用户名
    """
    url = f"https://api.github.com/users/{username}"
    headers = {
        'User-Agent': 'Python aiohttp',
        'Accept': 'application/json'
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as response:
            if response.status == 200:
                data = await response.json()
                return {
                    'username': data.get('login'),
                    'name': data.get('name'),
                    'repos': data.get('public_repos')
                }
            else:
                print(f"获取用户 {username} 失败: {response.status}")
                return None

async def main():
    """主函数"""
    usernames = ["octocat", "torvalds", "gvanrossum"]
    tasks = [fetch_user(username) for username in usernames]
    
    # 并发执行所有任务
    results = await asyncio.gather(*tasks)
    
    # 打印结果
    for user in results:
        if user:
            print(f"用户: {user['username']}, 姓名: {user['name']}, 仓库数: {user['repos']}")

## 运行异步函数
if __name__ == "__main__":
    asyncio.run(main())

# -*- coding: utf-8 -*-
from ftplib import FTP
import os
from contextlib import contextmanager


@contextmanager
def ftp_connection(host, port=21, user=None, passwd=None, encoding="GB18030", timeout=30):
    """创建 FTP 连接的上下文管理器，确保连接正确关闭

    Args:
        host: FTP 服务器地址
        port: FTP 服务器端口，默认为 21
        user: FTP 用户名，默认为 None（匿名登录）
        passwd: FTP 密码，默认为 None
        encoding: 服务器编码，默认为 GB18030
        timeout: 连接超时时间（秒），默认为 30

    Yields:
        ftplib.FTP: 已连接的 FTP 对象
    """
    ftp = FTP()
    ftp.encoding = encoding
    try:
        ftp.connect(host, port, timeout)
        ftp.login(user, passwd)
        yield ftp
    finally:
        if ftp:
            ftp.quit()


def upload_file(local_path, remote_path, filename, host="172.17.115.3", port=21,
                user=None, passwd=None):
    """上传本地文件到 FTP 服务器

    Args:
        local_path: 本地文件的完整路径
        remote_path: FTP 服务器上的目标路径
        filename: 上传后的文件名
        host: FTP 服务器地址，默认为 172.17.115.3
        port: FTP 服务器端口，默认为 21
        user: FTP 用户名，默认为 None（匿名登录）
        passwd: FTP 密码，默认为 None

    Returns:
        bool: 上传成功返回 True，否则返回 False
    """
    try:
        with ftp_connection(host, port, user, passwd) as ftp:
            ftp.cwd(remote_path)  # 切换至目标路径
            with open(local_path, "rb") as local_file:
                ftp.storbinary("STOR " + filename, local_file)  # 上传文件
            print("文件上传成功！")
            return True
    except Exception as e:
        print("文件上传失败！", e)
        return False


def download_file(remote_path, remote_filename, local_path, host="172.17.115.3", port=21,
                  user=None, passwd=None):
    """从 FTP 服务器下载文件到本地

    Args:
        remote_path: FTP 服务器上的文件路径
        remote_filename: 要下载的文件名
        local_path: 本地保存路径
        host: FTP 服务器地址，默认为 172.17.115.3
        port: FTP 服务器端口，默认为 21
        user: FTP 用户名，默认为 None（匿名登录）
        passwd: FTP 密码，默认为 None

    Returns:
        bool: 下载成功返回 True，否则返回 False
    """
    try:
        with ftp_connection(host, port, user, passwd) as ftp:
            ftp.cwd(remote_path)  # 切换至目标路径
            with open(local_path, "wb") as local_file:
                ftp.retrbinary("RETR " + remote_filename, local_file.write)  # 下载文件
            print("文件下载成功！")
            return True
    except Exception as e:
        print("文件下载失败！", e)
        return False


def list_files(remote_path, host="172.17.115.3", port=21, user=None, passwd=None):
    """列出 FTP 服务器上指定目录中的文件

    Args:
        remote_path: FTP 服务器上的目录路径
        host: FTP 服务器地址，默认为 172.17.115.3
        port: FTP 服务器端口，默认为 21
        user: FTP 用户名，默认为 None（匿名登录）
        passwd: FTP 密码，默认为 None

    Returns:
        list: 文件名列表
    """
    files = []
    try:
        with ftp_connection(host, port, user, passwd) as ftp:
            ftp.cwd(remote_path)
            # ftp.retrlines会对每一行调用传入的回调函数，将每行内容作为参数传递给回调函数
            ftp.retrlines("LIST", lambda x: files.append(x.split()[-1])) # 由于前三个字符为时间，所以只取最后一个字段
        return files
    except Exception as e:
        print("文件列表获取失败！", e)
        return []


if __name__ == '__main__':
    # 示例：上传文件
    upload_file(
        local_path=r"test3.txt",  # 当前目录下的test3.txt文件
        remote_path="/upload",  # FTP服务器上的目标目录
        filename="test3.txt",  # 上传后的文件名
    )

    # 示例：下载文件 - 将刚上传的文件下载回来（使用不同的名称以便比较）
    download_file(
        remote_path="/upload",
        remote_filename="test3.txt",
        local_path=r"test3_downloaded.txt",
    )

    # 示例：列出目录中的文件
    files = list_files(
        remote_path="/upload",
    )
    print("FTP服务器上的文件列表:", files)

#!/usr/bin/env python
import asyncio
import json
import logging
import uuid
from datetime import datetime
from websockets.asyncio.server import ServerConnection, serve
from typing import Set, Dict, Any, Optional

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)


class WebSocketServer:
    """WebSocket服务器类，处理客户端连接和消息通信"""

    def __init__(self, host: str = "localhost", port: int = 8765):
        """初始化WebSocket服务器

        Args:
            host: 服务器主机名或IP地址
            port: 服务器端口号
        """
        self.host = host
        self.port = port
        self.clients: Dict[str, ServerConnection] = {}  # 使用字典存储客户端连接，键为客户端ID
        self.client_info: Dict[str, Dict[str, Any]] = {}  # 存储客户端信息，如用户名和头像
        self.logger = logging.getLogger('websocket_server')
        # 定义消息处理器映射
        self.message_handlers = {
            'chat': self._handle_chat_message,
            'ping': self._handle_ping_message,
            'user_info': self._handle_user_info
        }

    async def handle_message(self, client_id: str, message: str) -> None:
        """处理从客户端接收到的消息

        Args:
            client_id: 客户端ID
            message: 收到的消息内容
        """
        try:
            data = json.loads(message)
            await self._process_message(client_id, data)
        except json.JSONDecodeError:
            await self._send_error(client_id, "消息格式不是有效的JSON")
        except Exception as e:
            self.logger.error(f"处理消息时出错: {str(e)}")
            await self._send_error(client_id, f"服务器处理消息时出错: {str(e)}")

    async def _process_message(self, client_id: str, data: Dict[str, Any]) -> None:
        """处理解析后的消息数据"""
        # 验证消息格式
        if 'type' not in data:
            await self._send_error(client_id, "消息格式错误，缺少type字段")
            return

        # 获取并执行对应的处理器
        message_type = data['type']
        handler = self.message_handlers.get(message_type)

        if handler:
            await handler(client_id, data)
        else:
            await self._send_error(client_id, f"不支持的消息类型: {message_type}")

    async def _handle_user_info(self, client_id: str, data: Dict[str, Any]) -> None:
        """处理用户信息更新"""
        if 'username' not in data or 'avatar' not in data:
            await self._send_error(client_id, "用户信息不完整，需要username和avatar字段")
            return
            
        # 更新用户信息
        self.client_info[client_id]['username'] = data['username']
        self.client_info[client_id]['avatar'] = data['avatar']
        
        # 通知所有客户端有用户信息更新
        user_info_message = json.dumps({
            "type": "user_update",
            "client_id": client_id,
            "username": data['username'],
            "avatar": data['avatar']
        })
        
        await self._broadcast_message(user_info_message)
        self.logger.info(f"用户信息更新: {data['username']}")

    async def _handle_chat_message(self, client_id: str, data: Dict[str, Any]) -> None:
        """处理聊天消息"""
        if 'content' not in data:
            await self._send_error(client_id, "消息缺少content字段")
            return
            
        username = self.client_info.get(client_id, {}).get('username', '匿名用户')
        avatar = self.client_info.get(client_id, {}).get('avatar', './images/1.png')
        timestamp = data.get('timestamp', datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

        broadcast_message = json.dumps({
            "type": "chat",
            "client_id": client_id,  # 添加发送者ID
            "sender": username,
            "content": data['content'],
            "timestamp": timestamp,
            "avatar": avatar
        })

        await self._broadcast_message(broadcast_message)
        self.logger.info(f"广播消息: {data['content']} (来自: {username})")

    async def _handle_ping_message(self, client_id: str, data: Dict[str, Any]) -> None:
        """处理ping消息"""
        if client_id in self.clients:
            await self.clients[client_id].send(json.dumps({"type": "pong"}))

    async def _send_error(self, client_id: str, message: str) -> None:
        """发送错误消息到客户端"""
        if client_id in self.clients:
            await self.clients[client_id].send(json.dumps({
                "type": "error",
                "message": message
            }))

    async def _send_to_client(self, client_id: str, message: str) -> None:
        """发送消息到特定客户端"""
        if client_id in self.clients:
            await self.clients[client_id].send(message)

    async def _broadcast_message(self, message: str, exclude_client: Optional[str] = None) -> None:
        """广播消息给所有客户端

        Args:
            message: 要广播的消息
            exclude_client: 要排除的客户端ID（可选）
        """
        tasks = []
        for cid, conn in self.clients.items():
            if exclude_client is None or cid != exclude_client:
                tasks.append(conn.send(message))
                
        if tasks:
            await asyncio.gather(*tasks)

    async def connection_handler(self, websocket: ServerConnection) -> None:
        """处理WebSocket连接的主函数

        Args:
            websocket: 客户端WebSocket连接
        """
        # 生成客户端ID
        client_id = str(uuid.uuid4())
        
        # 添加新的客户端连接
        self.clients[client_id] = websocket
        self.client_info[client_id] = {
            "username": f"用户{len(self.clients)}",
            "avatar": "./images/1.png"
        }
        
        client_info = f"{websocket.remote_address[0]}:{websocket.remote_address[1]}"
        self.logger.info(f"客户端连接: {client_info} (ID: {client_id})")

        try:
            await self._handle_new_connection(client_id)
            await self._process_client_messages(client_id, websocket)
        except Exception as e:
            self.logger.error(f"连接处理出错: {str(e)}")
        finally:
            await self._handle_disconnection(client_id)

    async def _handle_new_connection(self, client_id: str) -> None:
        """处理新客户端连接"""
        # 发送客户端ID和欢迎消息
        welcome_message = json.dumps({
            "type": "welcome",
            "client_id": client_id,
            "content": "已连接到WebSocket服务器",
            "user_count": len(self.clients)
        })
        await self._send_to_client(client_id, welcome_message)

        # 发送当前在线用户列表
        users_list = []
        for cid, info in self.client_info.items():
            users_list.append({
                "client_id": cid,
                "username": info.get("username", "匿名用户"),
                "avatar": info.get("avatar", "./images/1.png")
            })
            
        users_message = json.dumps({
            "type": "users_list",
            "users": users_list
        })
        await self._send_to_client(client_id, users_message)

        # 通知所有客户端有新用户加入
        if len(self.clients) > 1:  # 只有当有多个客户端时才广播
            username = self.client_info[client_id].get("username", "匿名用户")
            connect_message = json.dumps({
                "type": "system",
                "content": f"用户 {username} 加入聊天，当前共有 {len(self.clients)} 名用户"
            })
            await self._broadcast_message(connect_message, exclude_client=client_id)

    async def _process_client_messages(self, client_id: str, websocket: ServerConnection) -> None:
        """处理客户端消息流"""
        async for message in websocket:
            await self.handle_message(client_id, message)

    async def _handle_disconnection(self, client_id: str) -> None:
        """处理客户端断开连接"""
        # 获取用户名以便在通知中使用
        username = self.client_info.get(client_id, {}).get("username", "匿名用户")
        
        # 从集合中移除断开的客户端
        if client_id in self.clients:
            del self.clients[client_id]
        if client_id in self.client_info:
            del self.client_info[client_id]

        # 通知其他客户端有用户离开
        if self.clients:  # 只有当还有其他客户端时才广播
            disconnect_message = json.dumps({
                "type": "system",
                "content": f"用户 {username} 离开聊天，当前共有 {len(self.clients)} 名用户"
            })
            await self._broadcast_message(disconnect_message)

    async def start(self) -> None:
        """启动WebSocket服务器"""
        self.logger.info(f"启动WebSocket服务器 - ws://{self.host}:{self.port}")
        async with serve(self.connection_handler, self.host, self.port):
            await asyncio.Future()  # 无限期运行服务器

    def run(self) -> None:
        """运行WebSocket服务器(阻塞调用)"""
        try:
            asyncio.run(self.start())
        except KeyboardInterrupt:
            self.logger.info("服务器已通过键盘中断信号停止")
        except Exception as e:
            self.logger.error(f"服务器出错: {str(e)}")


if __name__ == "__main__":
    # 创建并运行WebSocket服务器实例
    server = WebSocketServer(host="localhost", port=8765)
    server.run()