一个简单的局域网文本传输服务器

image-20250828152804960

import http.server
import socketserver
import urllib.parse
import html
import netifaces

DEFAULT_PORT = 8000
MAX_PORT_ATTEMPTS = 100  # 最大尝试端口数量
TEXT_HISTORY = []  # 存储提交的文本历史

class TextHandler(http.server.SimpleHTTPRequestHandler):
    def do_GET(self):
        # 处理favicon请求
        if self.path == '/favicon.ico':
            self.send_response(204)  # No Content
            self.end_headers()
            return
        
        self.send_response(200)
        self.send_header("Content-type", "text/html; charset=utf-8")
        self.end_headers()
        
        # 构建历史文本显示
        history_html = "<h2>历史文本</h2><ul>"
        if TEXT_HISTORY:
            for i, text in enumerate(TEXT_HISTORY, 1):
                history_html += f"<li>消息 {i}: {html.escape(text)}</li>"
        else:
            history_html += "<li>暂无消息</li>"
        history_html += "</ul>"
        
        page_html = f"""
        <html>
        <head>
            <title>局域网文本传输</title>
            <meta charset="utf-8">
        </head>
        <body>
            <h1>局域网文本传输</h1>
            <form method="POST" action="/">
                <textarea name="text" rows="10" cols="50" placeholder="请输入要传输的文本"></textarea><br>
                <input type="submit" value="发送文本">
            </form>
            {history_html}
        </body>
        </html>
        """
        print(f"处理GET请求,发送页面,历史消息数: {len(TEXT_HISTORY)}")  # 调试输出
        self.wfile.write(page_html.encode('utf-8'))

    def do_POST(self):
        try:
            content_length = int(self.headers.get('Content-Length', 0))
            if content_length > 0:
                post_data = self.rfile.read(content_length).decode('utf-8')
                print(f"接收到的原始POST数据: {post_data}")  # 调试输出
                params = urllib.parse.parse_qs(post_data)
                text = params.get('text', [''])[0]
                if text.strip():  # 仅保存非空文本
                    TEXT_HISTORY.append(text)
                    print(f"已保存文本: {text}")  # 调试输出
                else:
                    text = "未收到有效输入内容"
            else:
                text = "未收到输入内容"
            
            self.send_response(200)
            self.send_header("Content-type", "text/html; charset=utf-8")
            self.end_headers()
            
            # 提交后显示确认页面
            response = f"""
            <html>
            <head>
                <title>文本接收</title>
                <meta charset="utf-8">
            </head>
            <body>
                <h1>文本已提交</h1>
                <p>提交的内容: {html.escape(text) if text else '无内容'}</p>
                <a href="/">返回</a>
            </body>
            </html>
            """
            self.wfile.write(response.encode('utf-8'))
        except Exception as e:
            print(f"处理POST请求时出错: {e}")  # 调试输出
            self.send_response(500)
            self.send_header("Content-type", "text/html; charset=utf-8")
            self.end_headers()
            self.wfile.write(f"<html><body><h1>服务器错误</h1><p>{str(e)}</p></body></html>".encode('utf-8'))

def get_local_ip():
    try:
        # 获取所有网络接口
        interfaces = netifaces.interfaces()
        for iface in interfaces:
            addrs = netifaces.ifaddresses(iface)
            # 检查IPv4地址
            if netifaces.AF_INET in addrs:
                for addr in addrs[netifaces.AF_INET]:
                    ip = addr['addr']
                    # 筛选局域网IP(192.168.x.x)
                    if ip.startswith('192.168.'):
                        return ip
        # 如果未找到192.168.x.x地址,返回默认本地IP
        return '127.0.0.1'
    except Exception:
        return '127.0.0.1'

def run_server():
    port = DEFAULT_PORT
    server_started = False
    
    for attempt in range(MAX_PORT_ATTEMPTS):
        try:
            httpd = socketserver.TCPServer(("", port), TextHandler)
            server_started = True
            break
        except OSError as e:
            if "Address already in use" in str(e):
                print(f"端口 {port} 已被占用,尝试端口 {port + 1}")
                port += 1
            else:
                raise e
    
    if not server_started:
        print(f"错误:无法在端口 {DEFAULT_PORT}{port} 之间找到空闲端口")
        return
    
    local_ip = get_local_ip()
    print(f"服务器运行在 http://0.0.0.0:{port}")
    print(f"局域网访问地址: http://{local_ip}:{port}")
    
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n服务器已停止")
        httpd.server_close()


if __name__ == "__main__":
    run_server()

以下是gemini给的一些扩展方向和应用场景的建议,从简单到复杂排列:

1. 功能增强与体验优化 (Enhancements & UX Improvements)

这些是在现有基础上直接进行的改进,可以让你更好地使用这个工具。

  • 历史记录持久化: 目前 TEXT_HISTORY 是一个内存变量,服务器重启后历史记录就会丢失。你可以将历史记录保存到文件中(如 txt, json, csv),每次服务器启动时加载,实现历史记录的持久化。
  • 增加时间戳: 在每条消息旁边显示提交的时间,这样可以更清楚地知道消息是什么时候发送的。
  • 支持删除/清空历史: 在前端页面上为每条消息添加一个“删除”按钮,或者添加一个“清空所有历史”的按钮。这需要在后端 do_POSTdo_GET 中处理相应的请求。
  • 美化前端页面: 使用更现代的 CSS 框架 (如 Bootstrap, Tailwind CSS) 来美化页面,使其在不同设备上(尤其是移动设备)有更好的显示效果(响应式设计)。
  • 实时刷新: 目前必须手动刷新页面才能看到新的消息。你可以使用 JavaScript 在前端实现定时轮询(每隔几秒向服务器请求一次最新数据)或者更高阶的 WebSocket 技术,实现消息的实时推送,打造一个真正的局域网聊天室。
  • 生成二维码: 在服务器启动时,将局域网访问地址 http://<local_ip>:<port> 生成一个二维码并显示在控制台或网页上。这样,其他人用手机扫一扫就能直接访问,非常方便。

2. 从文本到文件 (From Text to Files)

这是最直接、最实用的扩展方向,将纯文本传输升级为文件传输。

  • 局域网文件共享工具:
    • 在网页上添加一个文件上传的表单 (<input type="file">)。
    • 修改后端的 do_POST 方法来处理 multipart/form-data 类型的请求,接收并保存上传的文件到服务器的特定文件夹。
    • 在主页上列出所有已上传的文件,并提供下载链接。
    • 进阶玩法: 可以实现拖拽上传、显示上传进度条、文件分享有效期、密码保护等功能。这基本上就是一个轻量级的个人局域网“网盘”或“NAS”。

3. 有趣的应用方向 (Fun Applications)

基于这个核心框架,可以构建一些特定场景的好玩应用。

  • 局域网匿名投票/问卷系统:
    • 创建一个发起投票的页面,输入问题和选项。
    • 将投票信息提交到服务器,生成一个唯一的投票链接。
    • 局域网内的其他人访问这个链接进行投票。
    • 服务器实时统计票数,并可以在结果页面上以图表(如柱状图)的形式动态展示。
  • 剪贴板共享工具 (Clipboard Sharing):
    • 核心思想是将文本传输功能聚焦于“剪贴板”。
    • 电脑 -> 手机: 在电脑上复制一段文本,粘贴到网页上提交。然后手机访问这个网页,就能看到最新的文本,直接复制使用。
    • 手机 -> 电脑: 反之亦然。
    • 进阶玩法: 可以结合浏览器扩展(Extension)或者桌面客户端,实现自动同步剪贴板内容,做到真正的“隔空复制粘贴”。
  • 简单的代码/笔记片段分享 (Code/Note Snippet Sharing):
    • 针对程序员或团队协作场景。
    • 在文本输入框中增加对代码高亮的支持(例如使用 highlight.js 这样的前端库)。
    • 提交代码片段后,页面上会以美观的、高亮的形式展示出来,方便其他人查看和复制。
  • 临时消息板/“阅后即焚”:
    • 可以给每条消息设置一个“生命周期”。例如,只显示最近的 10 条消息,或者消息在提交 5 分钟后自动删除。
    • 实现一个“阅后即焚”模式:生成一个一次性访问链接,该链接里的消息在被查看一次后即从服务器删除。
  • 本地物联网 (IoT) 控制面板:
    • 如果你有一些支持网络请求的智能设备(比如某些智能灯泡、开关),你可以用这个服务器作为它们的控制中心。
    • 在网页上创建一些按钮,比如“开灯”、“关灯”。
    • 点击按钮时,前端发送一个 POST 请求到你的 Python 服务器,服务器收到请求后,再向你的智能设备的 API 地址发送相应的控制指令。

如何选择?

  • 新手入门: 建议从 功能增强 开始,比如“增加时间戳”和“历史记录持久化”,这些改动相对简单,能帮助你更好地理解代码。
  • 实用主义: 文件共享 功能是价值最高的扩展,能解决生活中“手机和电脑互传文件”的痛点。
  • 追求好玩: 剪贴板共享匿名投票系统 是非常有趣且实现起来不太复杂的项目。