xy6538 发表于 2025-11-13 09:24

端口映射管理工具

本帖最后由 xy6538 于 2025-11-13 09:53 编辑

一、简介:
    1.起因,单位服务器有点多,服务器上开的端口相应也比较多,容易把端口弄混淆,所以就借助AI开发了这个小工具,适用人群应该不多。
    2.该工具旨在帮助运维人员或任何需要管理多个服务器端口信息的用户,高效地记录、查询和管理不同服务器上的端口映射关系。它特别适用于拥有多个内网/外网服务器(如 NAS、云服务器等)的场景。


二、核心功能:
    1.自定义服务器管理:您可以自由添加、修改、删除服务器名称。修改服务器名时,其下所有关联的端口记录会自动同步更新;删除服务器时,可选择性地删除其所有端口记录或仅从服务器列表中移除,确保数据一致性与操作安全。
    2.便捷的数据操作:支持添加、编辑、删除单条端口-服务器映射记录。通过右键点击数据表格中的任意行,即可快速调出菜单进行编辑或删除操作。
    3.强大的查询功能:支持通过端口号、服务器名、描述等关键词进行模糊搜索,快速定位所需信息。
    4.直观的统计信息:提供独立的统计信息窗口,以清晰的表格形式展示“端口使用统计”(哪些端口被用得最多)和“服务器端口统计”(每个服务器上有多少端口被记录),方便进行资源分析。
    5.数据导出与备份:支持将所有端口映射数据导出为 CSV 文件,方便备份或与其他系统共享。内置数据库手动备份和自动清理旧备份文件的功能,保障数据安全。
    6.数据持久化:使用 SQLite3 作为本地数据库,所有数据持久保存。服务器列表等配置信息也通过 JSON 文件进行持久化。


三、开发环境:
    1.Python 版本: Python 3.10
    2.核心库: tkinter (GUI界面), sqlite3 (数据库), csv (导出)等
    3.环境准备:确保您的电脑已安装 Python 3.10 或更高版本。
    4.初始化:首次运行会自动创建.db数据库文件,当服务器信息有变更时,会在同目录生成.json配置文件。


四、使用
    1.管理服务器:
    点击左侧“管理服务器”按钮,打开服务器管理对话框。
    在顶部输入框输入新服务器名,点击“添加”。
    在列表中选择一个服务器,可在顶部输入框修改其名称(若有数据会提示同步更新),或点击“删除选中”将其移除。
    支持通过“上移/下移”按钮调整服务器在下拉列表中的顺序。
   
    2.添加/编辑记录:
    在左侧表单中输入“端口号”、“选择服务器”(从下拉列表选择)、“描述”,点击“添加记录”。
    在表格中右键点击某一行,选择“编辑记录”,表单会自动填充该行数据,修改后点击“更新记录”。
   
    3.查询记录:
    在“搜索关键词”框输入内容(支持端口号、服务器名、描述),点击“搜索”。
    点击“显示全部”可查看所有记录。
   
    4.查看统计:点击左侧“查看详细统计”按钮,会弹出一个独立的窗口,展示详细的端口和服务器使用统计。
   
    5. 数据导出与备份:使用左侧“导出CSV”和“手动备份”按钮进行相应操作。
   
    6.项目特点与优势:
   
    7.简单易用:界面直观,操作逻辑清晰,无需复杂配置即可上手。
   
    8.功能聚焦:专注于端口映射管理的核心需求,功能实用。
   
    9.数据安全:在修改服务器名、删除服务器/记录等关键操作前,均有弹窗确认,防止误操作。
   
    10.可扩展性:代码结构清晰,易于二次开发和功能扩展。


    已知问题:创建时间不对,高手自行处理一下。





    成品:https://xyong.lanzouu.com/i76FT3axfxhc


import sqlite3
from datetime import datetime
import tkinter as tk
from tkinter import ttk, messagebox, filedialog
import csv
import os
import json

DB_PATH = 'ports.db'
CONFIG_FILE = 'app_config.json'

class ConfigManager:
    def __init__(self):
      self.config_file = CONFIG_FILE
      self.default_servers = ['内网飞牛', '内网群晖', '外网群晖', '外网飞牛']
      self.load_config()

    def load_config(self):
      if os.path.exists(self.config_file):
            try:
                with open(self.config_file, 'r', encoding='utf-8') as f:
                  config = json.load(f)
                  self.servers = config.get('servers', self.default_servers)
                  return self.servers
            except:
                pass
      self.servers = self.default_servers
      return self.servers

    def save_config(self, servers):
      config = {'servers': servers}
      with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(config, f, ensure_ascii=False, indent=2)
      self.servers = servers

def init_db():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='port_mapping'")
    table_exists = cursor.fetchone()
    if not table_exists:
      cursor.execute('''
            CREATE TABLE port_mapping (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                port INTEGER NOT NULL,
                server_name TEXT NOT NULL,
                description TEXT DEFAULT '',
                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                UNIQUE(port, server_name)
            )
      ''')
      cursor.execute('CREATE INDEX idx_port ON port_mapping(port)')
      cursor.execute('CREATE INDEX idx_server ON port_mapping(server_name)')
      conn.commit()
      conn.close()
      return

    cursor.execute("PRAGMA table_info(port_mapping)")
    columns = {column for column in cursor.fetchall()}
    if 'description' not in columns:
      cursor.execute("ALTER TABLE port_mapping ADD COLUMN description TEXT DEFAULT ''")
    if 'updated_at' not in columns:
      cursor.execute("ALTER TABLE port_mapping ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP")

    conn.commit()
    conn.close()

def safe_get_all_ports():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute(
      "SELECT port, server_name, description, created_at, updated_at FROM port_mapping ORDER BY server_name, port"
    )
    rows = cursor.fetchall()
    conn.close()
    return rows

def add_port(port, server_name, description=""):
    try:
      conn = sqlite3.connect(DB_PATH)
      cursor = conn.cursor()
      cursor.execute(
            "SELECT 1 FROM port_mapping WHERE port = ? AND server_name = ?",
            (port, server_name)
      )
      if cursor.fetchone():
            return False, f"服务器 {server_name} 上端口 {port} 已存在!"

      cursor.execute(
            "INSERT INTO port_mapping (port, server_name, description, updated_at) VALUES (?, ?, ?, ?)",
            (port, server_name, description, datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
      )
      conn.commit()
      conn.close()
      return True, "添加成功"
    except sqlite3.IntegrityError:
      return False, f"服务器 {server_name} 的端口 {port} 已存在!"
    except Exception as e:
      return False, f"添加失败: {str(e)}"

def update_record(port, server_name, new_port=None, new_description=None):
    try:
      conn = sqlite3.connect(DB_PATH)
      cursor = conn.cursor()
      if new_port is not None:
            cursor.execute(
                "SELECT 1 FROM port_mapping WHERE port = ? AND server_name = ? AND port != ?",
                (new_port, server_name, port)
            )
            if cursor.fetchone():
                return False, f"服务器 {server_name} 上端口 {new_port} 已存在!"
            cursor.execute(
                "UPDATE port_mapping SET port = ?, description = ?, updated_at = ? WHERE port = ? AND server_name = ?",
                (new_port, new_description or "", datetime.now().strftime('%Y-%m-%d %H:%M:%S'), port, server_name)
            )
      else:
            cursor.execute(
                "UPDATE port_mapping SET description = ?, updated_at = ? WHERE port = ? AND server_name = ?",
                (new_description or "", datetime.now().strftime('%Y-%m-%d %H:%M:%S'), port, server_name)
            )

      conn.commit()
      conn.close()
      return True, "更新成功"
    except Exception as e:
      return False, f"更新失败: {str(e)}"

def update_server_name(old_name, new_name):
    try:
      conn = sqlite3.connect(DB_PATH)
      cursor = conn.cursor()
      cursor.execute("SELECT 1 FROM port_mapping WHERE server_name = ?", (new_name,))
      if cursor.fetchone():
            return False, f"服务器名称 '{new_name}' 已存在!"

      cursor.execute(
            "UPDATE port_mapping SET server_name = ?, updated_at = ? WHERE server_name = ?",
            (new_name, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), old_name)
      )
      conn.commit()
      conn.close()
      return True, f"服务器名称已从 '{old_name}' 更新为 '{new_name}',相关记录已同步更新"
    except Exception as e:
      return False, f"更新服务器名称失败: {str(e)}"

def query_by_port(port):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute(
      "SELECT server_name, description, created_at, updated_at FROM port_mapping WHERE port = ? ORDER BY server_name",
      (port,)
    )
    rows = cursor.fetchall()
    conn.close()
    return rows

def query_by_server(server_name):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute(
      "SELECT port, description, created_at, updated_at FROM port_mapping WHERE server_name = ? ORDER BY port",
      (server_name,)
    )
    rows = cursor.fetchall()
    conn.close()
    return rows

def search_records(keyword):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    if keyword.isdigit():
      cursor.execute('''
            SELECT port, server_name, description, created_at, updated_at
            FROM port_mapping
            WHERE port = ? OR server_name LIKE ? OR description LIKE ?
            ORDER BY port, server_name
      ''', (int(keyword), f'%{keyword}%', f'%{keyword}%'))
    else:
      cursor.execute('''
            SELECT port, server_name, description, created_at, updated_at
            FROM port_mapping
            WHERE server_name LIKE ? OR description LIKE ?
            ORDER BY port, server_name
      ''', (f'%{keyword}%', f'%{keyword}%'))
    rows = cursor.fetchall()
    conn.close()
    return rows

def get_port_usage_stats():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute('''
      SELECT port, COUNT(*) as usage_count
      FROM port_mapping
      GROUP BY port
      ORDER BY usage_count DESC, port
    ''')
    port_stats = cursor.fetchall()
    cursor.execute('''
      SELECT server_name, COUNT(*) as port_count
      FROM port_mapping
      GROUP BY server_name
      ORDER BY port_count DESC, server_name
    ''')
    server_stats = cursor.fetchall()
    conn.close()
    return port_stats, server_stats

def delete_record(port, server_name):
    try:
      conn = sqlite3.connect(DB_PATH)
      cursor = conn.cursor()
      cursor.execute(
            "DELETE FROM port_mapping WHERE port = ? AND server_name = ?",
            (port, server_name)
      )
      conn.commit()
      conn.close()
      return True, "删除成功"
    except Exception as e:
      return False, f"删除失败: {str(e)}"

def delete_server_and_records(server_name):
    try:
      conn = sqlite3.connect(DB_PATH)
      cursor = conn.cursor()
      cursor.execute("SELECT COUNT(*) FROM port_mapping WHERE server_name = ?", (server_name,))
      record_count = cursor.fetchone()
      cursor.execute("DELETE FROM port_mapping WHERE server_name = ?", (server_name,))
      conn.commit()
      conn.close()
      return True, f"服务器 '{server_name}' 及其 {record_count} 条端口记录已删除"
    except Exception as e:
      return False, f"删除服务器失败: {str(e)}"

def manual_backup():
    if not os.path.exists(DB_PATH):
      messagebox.showinfo("备份", "数据库文件不存在")
      return

    backup_dir = "backups"
    if not os.path.exists(backup_dir):
      os.makedirs(backup_dir)

    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_path = os.path.join(backup_dir, f"ports_backup_{timestamp}.db")

    import shutil
    shutil.copy2(DB_PATH, backup_path)
    messagebox.showinfo("备份成功", f"数据库已备份到: {backup_path}")

def cleanup_old_backups(max_backups=5):
    backup_dir = "backups"
    if not os.path.exists(backup_dir):
      return

    try:
      backup_files = []
      for file in os.listdir(backup_dir):
            if file.startswith("ports_backup_") and file.endswith(".db"):
                file_path = os.path.join(backup_dir, file)
                backup_files.append((file_path, os.path.getctime(file_path)))

      backup_files.sort(key=lambda x: x, reverse=True)

      for file_path, _ in backup_files:
            os.remove(file_path)
    except Exception as e:
      print(f"清理备份文件时出错: {e}")

def export_to_csv():
    data = safe_get_all_ports()
    if not data:
      messagebox.showinfo("导出", "没有数据可导出")
      return

    filename = filedialog.asksaveasfilename(
      defaultextension=".csv",
      filetypes=[("CSV files", "*.csv"), ("All files", "*.*")],
      initialfile=f"port_mapping_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
    )
    if filename:
      with open(filename, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['端口号', '服务器', '描述', '创建时间', '更新时间'])
            writer.writerows(data)
      messagebox.showinfo("导出成功", f"文件已保存:{filename}")

class ServerManagerDialog:
    def __init__(self, parent, config_manager, callback):
      self.parent = parent
      self.config_manager = config_manager
      self.callback = callback
      self.servers = config_manager.load_config().copy()
      self.create_dialog()

    def create_dialog(self):
      self.dialog = tk.Toplevel(self.parent)
      self.dialog.title("服务器管理")
      self.dialog.geometry("500x400")
      self.dialog.resizable(False, False)
      self.dialog.transient(self.parent)
      self.dialog.grab_set()
      self.dialog.geometry("+%d+%d" % (
            self.parent.winfo_rootx() + 50,
            self.parent.winfo_rooty() + 50
      ))

      main_frame = ttk.Frame(self.dialog, padding="10")
      main_frame.pack(fill='both', expand=True)
      add_frame = ttk.LabelFrame(main_frame, text="添加/编辑服务器", padding="5")
      add_frame.pack(fill='x', pady=(0, 10))

      ttk.Label(add_frame, text="服务器名称:").grid(row=0, column=0, sticky='w', padx=(0, 5))
      self.new_server_entry = ttk.Entry(add_frame, width=25)
      self.new_server_entry.grid(row=0, column=1, padx=(0, 10))
      self.new_server_entry.bind('<Return>', lambda e: self.add_server())

      self.add_btn = ttk.Button(add_frame, text="添加", command=self.add_server)
      self.add_btn.grid(row=0, column=2, padx=(0, 5))

      self.edit_btn = ttk.Button(add_frame, text="修改", command=self.edit_server, state='disabled')
      self.edit_btn.grid(row=0, column=3)

      list_frame = ttk.LabelFrame(main_frame, text="服务器列表", padding="5")
      list_frame.pack(fill='both', expand=True)

      self.server_listbox = tk.Listbox(list_frame, height=10)
      scrollbar = ttk.Scrollbar(list_frame, orient="vertical", command=self.server_listbox.yview)
      self.server_listbox.configure(yscrollcommand=scrollbar.set)
      self.server_listbox.pack(side='left', fill='both', expand=True, padx=(0, 5))
      scrollbar.pack(side='right', fill='y')
      self.server_listbox.bind('<<ListboxSelect>>', self.on_server_select)

      button_frame = ttk.Frame(list_frame)
      button_frame.pack(side='right', fill='y', padx=(5, 0))

      ttk.Button(button_frame, text="删除选中", command=self.delete_server, width=12).pack(fill='x', pady=2)
      ttk.Button(button_frame, text="上移", command=self.move_up, width=12).pack(fill='x', pady=2)
      ttk.Button(button_frame, text="下移", command=self.move_down, width=12).pack(fill='x', pady=2)

      self.stats_label = ttk.Label(main_frame, text="", foreground='blue')
      self.stats_label.pack(fill='x', pady=(5, 0))

      bottom_frame = ttk.Frame(main_frame)
      bottom_frame.pack(fill='x', pady=(10, 0))
      ttk.Button(bottom_frame, text="确定", command=self.save_and_close).pack(side='right', padx=(5, 0))
      ttk.Button(bottom_frame, text="取消", command=self.dialog.destroy).pack(side='right')

      self.refresh_list()
      self.update_stats()

    def refresh_list(self):
      self.server_listbox.delete(0, tk.END)
      for server in self.servers:
            self.server_listbox.insert(tk.END, server)

    def update_stats(self):
      conn = sqlite3.connect(DB_PATH)
      cursor = conn.cursor()
      stats_text = "服务器统计: "
      stats = []
      for server in self.servers:
            cursor.execute("SELECT COUNT(*) FROM port_mapping WHERE server_name = ?", (server,))
            count = cursor.fetchone()
            stats.append(f"{server}({count})")
      conn.close()
      self.stats_label.config(text=stats_text + " | ".join(stats))

    def on_server_select(self, event):
      selection = self.server_listbox.curselection()
      if selection:
            index = selection
            server_name = self.servers
            self.new_server_entry.delete(0, tk.END)
            self.new_server_entry.insert(0, server_name)
            self.edit_btn.config(state='normal')
      else:
            self.new_server_entry.delete(0, tk.END)
            self.edit_btn.config(state='disabled')

    def add_server(self):
      server_name = self.new_server_entry.get().strip()
      if not server_name:
            messagebox.showwarning("输入错误", "请输入服务器名称")
            return
      if server_name in self.servers:
            messagebox.showwarning("输入错误", "服务器已存在")
            return
      self.servers.append(server_name)
      self.refresh_list()
      self.new_server_entry.delete(0, tk.END)
      self.edit_btn.config(state='disabled')
      self.update_stats()

    def edit_server(self):
      selection = self.server_listbox.curselection()
      if not selection:
            messagebox.showwarning("选择错误", "请选择要修改的服务器")
            return
      index = selection
      old_name = self.servers
      new_name = self.new_server_entry.get().strip()
      if not new_name:
            messagebox.showwarning("输入错误", "请输入新的服务器名称")
            return
      if new_name == old_name:
            messagebox.showwarning("输入错误", "服务器名称未改变")
            return
      if new_name in self.servers:
            messagebox.showwarning("输入错误", "服务器名称已存在")
            return

      conn = sqlite3.connect(DB_PATH)
      cursor = conn.cursor()
      cursor.execute("SELECT COUNT(*) FROM port_mapping WHERE server_name = ?", (old_name,))
      record_count = cursor.fetchone()
      conn.close()

      if record_count > 0:
            result = messagebox.askyesno(
                "确认修改",
                f"服务器 '{old_name}' 有 {record_count} 条端口记录。\n修改服务器名称将同步更新所有相关记录。\n是否继续?"
            )
            if result:
                success, msg = update_server_name(old_name, new_name)
                if success:
                  self.servers = new_name
                  self.refresh_list()
                  self.new_server_entry.delete(0, tk.END)
                  self.edit_btn.config(state='disabled')
                  self.update_stats()
                  messagebox.showinfo("修改成功", msg)
                else:
                  messagebox.showerror("修改失败", msg)
      else:
            self.servers = new_name
            self.refresh_list()
            self.new_server_entry.delete(0, tk.END)
            self.edit_btn.config(state='disabled')
            self.update_stats()
            messagebox.showinfo("修改成功", f"服务器名称已修改为 '{new_name}'")

    def delete_server(self):
      selection = self.server_listbox.curselection()
      if not selection:
            messagebox.showwarning("选择错误", "请选择要删除的服务器")
            return
      index = selection
      server_name = self.servers

      conn = sqlite3.connect(DB_PATH)
      cursor = conn.cursor()
      cursor.execute("SELECT COUNT(*) FROM port_mapping WHERE server_name = ?", (server_name,))
      count = cursor.fetchone()
      conn.close()

      if count > 0:
            choice = messagebox.askyesnocancel(
                "确认删除",
                f"服务器 '{server_name}' 有 {count} 条端口映射记录。\n"
                f"选择『是』:删除服务器及其所有端口记录\n"
                f"选择『否』:仅从服务器列表中移除(保留端口记录)\n"
                f"选择『取消』:取消操作"
            )
            if choice is None:
                return
            elif choice:
                success, msg = delete_server_and_records(server_name)
                if success:
                  del self.servers
                  self.refresh_list()
                  self.new_server_entry.delete(0, tk.END)
                  self.edit_btn.config(state='disabled')
                  self.update_stats()
                  messagebox.showinfo("删除成功", msg)
                else:
                  messagebox.showerror("删除失败", msg)
            else:
                del self.servers
                self.refresh_list()
                self.new_server_entry.delete(0, tk.END)
                self.edit_btn.config(state='disabled')
                self.update_stats()
                messagebox.showinfo("移除成功", f"服务器 '{server_name}' 已从列表中移除,端口记录保留")
      else:
            del self.servers
            self.refresh_list()
            self.new_server_entry.delete(0, tk.END)
            self.edit_btn.config(state='disabled')
            self.update_stats()

    def move_up(self):
      selection = self.server_listbox.curselection()
      if not selection or selection == 0:
            return
      index = selection
      self.servers, self.servers = self.servers, self.servers
      self.refresh_list()
      self.server_listbox.select_set(index-1)

    def move_down(self):
      selection = self.server_listbox.curselection()
      if not selection or selection == len(self.servers) - 1:
            return
      index = selection
      self.servers, self.servers = self.servers, self.servers
      self.refresh_list()
      self.server_listbox.select_set(index+1)

    def save_and_close(self):
      self.config_manager.save_config(self.servers)
      self.callback(self.servers)
      self.dialog.destroy()

class StatsWindow:
    def __init__(self, parent):
      self.parent = parent
      self.window = tk.Toplevel(parent)
      self.window.title("统计信息")
      self.window.geometry("600x450")
      self.window.transient(parent)
      self.window.resizable(True, True)
      self.create_widgets()
      self.load_stats()

    def create_widgets(self):
      main_frame = ttk.Frame(self.window, padding="10")
      main_frame.pack(fill='both', expand=True)
      notebook = ttk.Notebook(main_frame)
      notebook.pack(fill='both', expand=True, pady=(0, 10))
      port_frame = ttk.Frame(notebook)
      notebook.add(port_frame, text='端口使用统计')
      server_frame = ttk.Frame(notebook)
      notebook.add(server_frame, text='服务器端口统计')
      port_columns = ('端口号', '使用次数')
      self.port_tree = ttk.Treeview(port_frame, columns=port_columns, show='headings', height=15)
      for col in port_columns:
            self.port_tree.heading(col, text=col)
            self.port_tree.column(col, width=100, anchor='center')
      port_v_scrollbar = ttk.Scrollbar(port_frame, orient="vertical", command=self.port_tree.yview)
      self.port_tree.configure(yscrollcommand=port_v_scrollbar.set)
      self.port_tree.grid(row=0, column=0, sticky="nsew")
      port_v_scrollbar.grid(row=0, column=1, sticky="ns")
      port_frame.grid_rowconfigure(0, weight=1)
      port_frame.grid_columnconfigure(0, weight=1)

      server_columns = ('服务器', '端口数')
      self.server_tree = ttk.Treeview(server_frame, columns=server_columns, show='headings', height=15)
      for col in server_columns:
            self.server_tree.heading(col, text=col)
            self.server_tree.column(col, width=100, anchor='center')
      server_v_scrollbar = ttk.Scrollbar(server_frame, orient="vertical", command=self.server_tree.yview)
      self.server_tree.configure(yscrollcommand=server_v_scrollbar.set)
      self.server_tree.grid(row=0, column=0, sticky="nsew")
      server_v_scrollbar.grid(row=0, column=1, sticky="ns")
      server_frame.grid_rowconfigure(0, weight=1)
      server_frame.grid_columnconfigure(0, weight=1)

      refresh_btn = ttk.Button(main_frame, text="刷新统计", command=self.load_stats)
      refresh_btn.pack()

    def load_stats(self):
      for item in self.port_tree.get_children():
            self.port_tree.delete(item)
      for item in self.server_tree.get_children():
            self.server_tree.delete(item)

      port_stats, server_stats = get_port_usage_stats()
      for port, count in port_stats:
            self.port_tree.insert('', 'end', values=(port, count))
      for server, count in server_stats:
            self.server_tree.insert('', 'end', values=(server, count))

class PortManagerApp:
    def __init__(self, root):
      self.root = root
      self.root.title("端口映射管理工具")
      self.root.geometry("1200x660")
      self.config_manager = ConfigManager()
      self.SERVER_OPTIONS = self.config_manager.load_config()
      self.editing_item = None
      self.create_widgets()
      self.refresh_table()

    def create_widgets(self):
      main_frame = ttk.Frame(self.root)
      main_frame.pack(fill='both', expand=True, padx=10, pady=10)
      left_frame = ttk.Frame(main_frame)
      left_frame.pack(side='left', fill='y', padx=(0, 10))
      right_frame = ttk.Frame(main_frame)
      right_frame.pack(side='right', fill='both', expand=True)
      add_frame = ttk.LabelFrame(left_frame, text="添加/编辑端口映射", padding=(10, 5))
      add_frame.pack(fill='x', pady=(0, 10))
      entry_width = 20
      ttk.Label(add_frame, text="端口号:").grid(row=0, column=0, sticky='w', padx=(0, 5), pady=5)
      self.port_entry = ttk.Entry(add_frame, width=entry_width)
      self.port_entry.grid(row=0, column=1, padx=(0, 10), pady=5, sticky='ew')

      ttk.Label(add_frame, text="服务器:").grid(row=1, column=0, sticky='w', padx=(0, 5), pady=5)
      self.server_combo = ttk.Combobox(add_frame, values=self.SERVER_OPTIONS, width=entry_width-3)
      self.server_combo.grid(row=1, column=1, padx=(0, 10), pady=5, sticky='ew')

      ttk.Label(add_frame, text="描述:").grid(row=2, column=0, sticky='w', padx=(0, 5), pady=5)
      self.desc_entry = ttk.Entry(add_frame, width=entry_width)
      self.desc_entry.grid(row=2, column=1, padx=(0, 10), pady=5, sticky='ew')
      add_frame.columnconfigure(1, weight=1)

      button_frame = ttk.Frame(add_frame)
      button_frame.grid(row=3, column=0, columnspan=2, pady=10)

      self.add_btn = ttk.Button(button_frame, text="添加记录", command=self.add_record)
      self.add_btn.pack(side='left', padx=(0, 5))

      self.update_btn = ttk.Button(button_frame, text="更新记录", command=self.update_record, state='disabled')
      self.update_btn.pack(side='left', padx=(0, 5))

      ttk.Button(button_frame, text="取消编辑", command=self.cancel_edit).pack(side='left')
      query_frame = ttk.LabelFrame(left_frame, text="查询", padding=(10, 5))
      query_frame.pack(fill='x', pady=(0, 10))

      ttk.Label(query_frame, text="搜索关键词:").grid(row=0, column=0, sticky='w', pady=5)
      self.search_entry = ttk.Entry(query_frame, width=entry_width)
      self.search_entry.grid(row=1, column=0, sticky='ew', pady=5)
      self.search_entry.bind('<Return>', lambda e: self.search_records_action())

      ttk.Button(query_frame, text="搜索", command=self.search_records_action, width=12).grid(row=2, column=0, sticky='w', pady=5)
      query_frame.columnconfigure(0, weight=1)
      stats_frame = ttk.LabelFrame(left_frame, text="统计信息", padding=(10, 5))
      stats_frame.pack(fill='x', pady=(0, 10))
      self.stats_button = ttk.Button(stats_frame, text="查看详细统计", command=self.open_stats_window)
      self.stats_button.pack(fill='x', pady=5)
      func_frame = ttk.LabelFrame(left_frame, text="功能", padding=(10, 5))
      func_frame.pack(fill='x')

      ttk.Button(func_frame, text="显示全部", command=self.show_all_records, width=12).pack(fill='x', pady=2)
      ttk.Button(func_frame, text="刷新表格", command=self.refresh_table, width=12).pack(fill='x', pady=2)
      ttk.Button(func_frame, text="导出CSV", command=export_to_csv, width=12).pack(fill='x', pady=2)
      ttk.Button(func_frame, text="手动备份", command=manual_backup, width=12).pack(fill='x', pady=2)
      ttk.Button(func_frame, text="清理备份", command=lambda: cleanup_old_backups(3), width=12).pack(fill='x', pady=2)
      ttk.Button(func_frame, text="管理服务器", command=self.manage_servers, width=12).pack(fill='x', pady=2)
      table_frame = ttk.LabelFrame(right_frame, text="记录显示", padding=(10, 5))
      table_frame.pack(fill='both', expand=True)
      columns = ('端口号', '服务器', '描述', '创建时间', '更新时间', '操作')
      self.tree = ttk.Treeview(table_frame, columns=columns, show='headings', height=20)
      for col in columns:
            self.tree.heading(col, text=col)
      self.tree.column('端口号', width=80, anchor='center')
      self.tree.column('服务器', width=120, anchor='center')
      self.tree.column('描述', width=150, anchor='center')
      self.tree.column('创建时间', width=140, anchor='center')
      self.tree.column('更新时间', width=140, anchor='center')
      self.tree.column('操作', width=120, anchor='center')

      v_scrollbar = ttk.Scrollbar(table_frame, orient="vertical", command=self.tree.yview)
      h_scrollbar = ttk.Scrollbar(table_frame, orient="horizontal", command=self.tree.xview)
      self.tree.configure(yscrollcommand=v_scrollbar.set, xscrollcommand=h_scrollbar.set)

      self.tree.grid(row=0, column=0, sticky="nsew")
      v_scrollbar.grid(row=0, column=1, sticky="ns")
      h_scrollbar.grid(row=1, column=0, sticky="ew")
      table_frame.grid_rowconfigure(0, weight=1)
      table_frame.grid_columnconfigure(0, weight=1)

      self.tree.bind('<Double-1>', self.on_double_click)
      self.tree.bind('<Button-3>', self.on_right_click)

      self.popup_menu = tk.Menu(self.tree, tearoff=0)
      self.popup_menu.add_command(label="编辑记录", command=self.popup_edit_record)
      self.popup_menu.add_command(label="删除记录", command=self.popup_delete_record)


    def open_stats_window(self):
      if hasattr(self, 'stats_window') and self.stats_window.window.winfo_exists():
            self.stats_window.window.lift()
      else:
            self.stats_window = StatsWindow(self.root)

    def manage_servers(self):
      ServerManagerDialog(self.root, self.config_manager, self.update_servers)

    def update_servers(self, servers):
      self.SERVER_OPTIONS = servers
      self.server_combo['values'] = servers
      if servers:
            self.server_combo.set(servers)
      self.refresh_table()

    def validate_port(self, port_str):
      if not port_str.strip():
            return False, "端口号不能为空"
      if not port_str.isdigit():
            return False, "端口号必须为数字"
      port = int(port_str)
      if not 1 <= port <= 65535:
            return False, "端口号范围应在 1-65535 之间"
      return True, port

    def clear_form(self):
      self.port_entry.delete(0, tk.END)
      self.server_combo.set('')
      self.desc_entry.delete(0, tk.END)
      self.cancel_edit()

    def cancel_edit(self):
      self.editing_item = None
      self.add_btn.config(state='normal')
      self.update_btn.config(state='disabled')
      self.clear_form()

    def add_record(self):
      port_str = self.port_entry.get().strip()
      server = self.server_combo.get().strip()
      description = self.desc_entry.get().strip()

      if not server:
            messagebox.showerror("错误", "请选择服务器")
            return

      valid, result = self.validate_port(port_str)
      if not valid:
            messagebox.showerror("错误", result)
            return
      port = result

      success, msg = add_port(port, server, description)
      messagebox.showinfo("结果", msg)
      if success:
            self.clear_form()
            self.refresh_table()
            # self.show_stats()

    def update_record(self):
      if not self.editing_item:
            return

      old_values = self.tree.item(self.editing_item)['values']
      old_port = old_values
      old_server = old_values

      new_port_str = self.port_entry.get().strip()
      new_server = self.server_combo.get().strip()
      new_description = self.desc_entry.get().strip()

      if not new_server:
            messagebox.showerror("错误", "请选择服务器")
            return

      valid, result = self.validate_port(new_port_str)
      if not valid:
            messagebox.showerror("错误", result)
            return
      new_port = result

      if new_port != old_port or new_server != old_server:
            success, msg = update_record(old_port, old_server, new_port, new_description)
      else:
            success, msg = update_record(old_port, old_server, None, new_description)

      messagebox.showinfo("结果", msg)
      if success:
            self.cancel_edit()
            self.refresh_table()
            # self.show_stats()

    def search_records_action(self):
      keyword = self.search_entry.get().strip()
      if not keyword:
            self.show_all_records()
            return
      rows = search_records(keyword)
      self.display_results([(row, row, row, row, row, '') for row in rows])

    def show_all_records(self):
      rows = safe_get_all_ports()
      self.display_results([(row, row, row, row, row, '') for row in rows])

    def display_results(self, data):
      for item in self.tree.get_children():
            self.tree.delete(item)

      for row in data:
            values = list(row)
            values[-1] = "编辑 | 删除"
            self.tree.insert('', 'end', values=values)

    def refresh_table(self):
      self.show_all_records()

    def on_double_click(self, event):
      selection = self.tree.selection()
      if not selection:
            return
      item = selection
      values = self.tree.item(item)['values']
      if not values:
            return
      column = self.tree.identify_column(event.x)
      col_index = int(column.replace('#', '')) - 1
      if col_index == 5:
            port = values
            server = values
            menu = tk.Menu(self.root, tearoff=0)
            menu.add_command(label="编辑记录", command=lambda: self.edit_record(item))
            menu.add_command(label="删除记录", command=lambda: self.delete_record(port, server))
            try:
                menu.tk_popup(event.x_root, event.y_root)
            finally:
                menu.grab_release()

    def on_right_click(self, event):
      item = self.tree.identify_row(event.y)
      if item:
            self.tree.selection_set(item)
            try:
                self.popup_menu.tk_popup(event.x_root, event.y_root)
            finally:
                self.popup_menu.grab_release()

    def popup_edit_record(self):
      selection = self.tree.selection()
      if selection:
            self.edit_record(selection)

    def popup_delete_record(self):
      selection = self.tree.selection()
      if selection:
            values = self.tree.item(selection)['values']
            if values:
                port = values
                server = values
                self.delete_record(port, server)

    def edit_record(self, item):
      values = self.tree.item(item)['values']
      self.editing_item = item
      self.port_entry.delete(0, tk.END)
      self.port_entry.insert(0, str(values))
      self.server_combo.set(values)
      self.desc_entry.delete(0, tk.END)
      self.desc_entry.insert(0, values if values else '')

      self.add_btn.config(state='disabled')
      self.update_btn.config(state='normal')

    def delete_record(self, port, server):
      if messagebox.askyesno("确认删除", f"确定要删除服务器 '{server}' 的端口 {port} 吗?"):
            success, msg = delete_record(port, server)
            messagebox.showinfo("结果", msg)
            if success:
                self.refresh_table()
                # self.show_stats()
                self.cancel_edit()

def main():
    init_db()
    cleanup_old_backups(3)
    root = tk.Tk()
    app = PortManagerApp(root)
    root.mainloop()

if __name__ == '__main__':
    main()




xy6538 发表于 2025-11-13 11:35

huaconglan 发表于 2025-11-13 11:23
同问,如果是电脑可以直接映射到防火墙,能理解,不能理解的是如何映射路由器的。。。

介绍中有说明 只是对工具中服务器信息进行修改

四、使用
    1.管理服务器:
    点击左侧“管理服务器”按钮,打开服务器管理对话框。
    在顶部输入框输入新服务器名,点击“添加”。
    在列表中选择一个服务器,可在顶部输入框修改其名称(若有数据会提示同步更新),或点击“删除选中”将其移除。
    支持通过“上移/下移”按钮调整服务器在下拉列表中的顺序。

xy6538 发表于 2025-11-13 11:34

skyfxf 发表于 2025-11-13 11:18
记录、查询 这个好理解 ,管理是啥意思。正常端口映射不是在路由器 或者防火墙上设置的吗。这个软件可以 ...

介绍中有说明

四、使用
    1.管理服务器:
    点击左侧“管理服务器”按钮,打开服务器管理对话框。
    在顶部输入框输入新服务器名,点击“添加”。
    在列表中选择一个服务器,可在顶部输入框修改其名称(若有数据会提示同步更新),或点击“删除选中”将其移除。
    支持通过“上移/下移”按钮调整服务器在下拉列表中的顺序。

jun269 发表于 2025-11-13 11:18

我终于沙发了,,楼主你真V5哦{:1_921:}

skyfxf 发表于 2025-11-13 11:18

记录、查询 这个好理解 ,管理是啥意思。正常端口映射不是在路由器 或者防火墙上设置的吗。这个软件可以自动连接路由器 防火墙 修改?

huaconglan 发表于 2025-11-13 11:23

skyfxf 发表于 2025-11-13 11:18
记录、查询 这个好理解 ,管理是啥意思。正常端口映射不是在路由器 或者防火墙上设置的吗。这个软件可以 ...

同问,如果是电脑可以直接映射到防火墙,能理解,不能理解的是如何映射路由器的。。。

skyfxf 发表于 2025-11-13 11:38

xy6538 发表于 2025-11-13 11:35
介绍中有说明 只是对工具中服务器信息进行修改

四、使用


那实际就是一个记录的作用   对吧,那直接用excel应该更简单点

楼主你好萌 发表于 2025-11-13 13:14

直接在路由上备注感觉更方便一些

xy6538 发表于 2025-11-13 13:56

skyfxf 发表于 2025-11-13 11:38
那实际就是一个记录的作用   对吧,那直接用excel应该更简单点

最初就是用的excel,随着数据多起来,感觉还是不太方便,尤其是插件有点多,启动excel太慢了:lol

xy6538 发表于 2025-11-13 13:57

楼主你好萌 发表于 2025-11-13 13:14
直接在路由上备注感觉更方便一些

涉及到端口转发才过路由,端口多的情况查询也不方便!
页: [1] 2 3 4
查看完整版本: 端口映射管理工具