[Python] 纯文本查看 复制代码
#!/usr/bin/env python3
import csv
import ctypes
import ctypes.wintypes
import socket
import threading
import time
import tkinter as tk
from concurrent.futures import ThreadPoolExecutor, as_completed
from tkinter import messagebox, ttk
netapi32 = ctypes.windll.netapi32
shell32 = ctypes.windll.shell32
STYPE_DISKTREE = 0
STYPE_PRINTQ = 1
MAX_PREFERRED_LENGTH = 0xFFFFFFFF
class SHARE_INFO_1(ctypes.Structure):
_fields_ = [
("shi1_netname", ctypes.c_wchar_p),
("shi1_type", ctypes.c_uint32),
("shi1_remark", ctypes.c_wchar_p),
]
def enumerate_shares(ip):
server = f"\\\\{ip}"
buf = ctypes.POINTER(ctypes.c_ubyte)()
entries_read = ctypes.c_uint32(0)
total_entries = ctypes.c_uint32(0)
ret = netapi32.NetShareEnum(
ctypes.c_wchar_p(server),
1,
ctypes.byref(buf),
MAX_PREFERRED_LENGTH,
ctypes.byref(entries_read),
ctypes.byref(total_entries),
None,
)
if ret != 0 or not buf:
return []
shares = []
try:
arr = ctypes.cast(buf, ctypes.POINTER(SHARE_INFO_1))
for i in range(entries_read.value):
si = arr[i]
if si.shi1_type == STYPE_DISKTREE or si.shi1_type == STYPE_PRINTQ:
if si.shi1_netname:
shares.append(si.shi1_netname)
finally:
netapi32.NetApiBufferFree(buf)
return shares
def open_in_explorer(path):
shell32.ShellExecuteW(None, "open", path, None, None, 1)
def copy_to_clipboard(root, text):
try:
root.clipboard_clear()
root.clipboard_append(text)
except Exception:
pass
def probe_port_445(ip, timeout=0.2):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(timeout)
result = s.connect_ex((ip, 445))
s.close()
return result == 0
except Exception:
return False
def reverse_dns(ip):
try:
hostname = socket.gethostbyaddr(ip)[0]
return hostname.rstrip(".")
except Exception:
return "N/A"
def scan_single_ip(ip, cancel_event):
if cancel_event.is_set():
return None
if not probe_port_445(ip):
return {"ip": ip, "type": "progress"}
if cancel_event.is_set():
return None
hostname = reverse_dns(ip)
if cancel_event.is_set():
return None
shares = enumerate_shares(ip)
result = {"ip": ip, "hostname": hostname, "type": "progress"}
if shares:
result["shares"] = shares
return result
class LanFindApp:
def __init__(self, root):
self.root = root
self.root.title("LAN Share Scanner — 跨网段共享扫描器")
self.root.geometry("860x580")
self.root.minsize(600, 400)
self.scanning = False
self.cancel_event = threading.Event()
self.entries = []
self.found_ips = set()
self.total = 254
self.scanned = 0
self._build_ui()
self._poll_events()
def _build_ui(self):
top_frame = ttk.Frame(self.root, padding=(8, 5))
top_frame.pack(fill=tk.X)
ttk.Label(top_frame, text="IP 网段前缀:").pack(side=tk.LEFT)
self.ip_var = tk.StringVar(value="192.168.1")
self.ip_entry = ttk.Entry(top_frame, textvariable=self.ip_var, width=18)
self.ip_entry.pack(side=tk.LEFT, padx=(4, 8))
self.ip_entry.bind("<Return>", lambda e: self.start_scan())
self.scan_btn = ttk.Button(top_frame, text="🔍 扫描", command=self.start_scan)
self.scan_btn.pack(side=tk.LEFT, padx=2)
self.stop_btn = ttk.Button(top_frame, text="⏹ 停止", command=self.stop_scan, state=tk.DISABLED)
self.stop_btn.pack(side=tk.LEFT, padx=2)
self.export_btn = ttk.Button(top_frame, text="📄 导出CSV", command=self.export_csv, state=tk.DISABLED)
self.export_btn.pack(side=tk.LEFT, padx=2)
progress_frame = ttk.Frame(self.root, padding=(8, 2))
progress_frame.pack(fill=tk.X)
self.progress = ttk.Progressbar(progress_frame, mode="determinate", maximum=self.total)
self.progress.pack(fill=tk.X)
tree_frame = ttk.Frame(self.root, padding=(8, 2))
tree_frame.pack(fill=tk.BOTH, expand=True)
columns = ("ip", "hostname", "share", "path")
self.tree = ttk.Treeview(
tree_frame,
columns=columns,
show="headings",
selectmode="browse",
)
self.tree.heading("ip", text="IP 地址")
self.tree.heading("hostname", text="主机名")
self.tree.heading("share", text="共享名")
self.tree.heading("path", text="路径")
self.tree.column("ip", width=140)
self.tree.column("hostname", width=160)
self.tree.column("share", width=120)
self.tree.column("path", width=320)
vsb = ttk.Scrollbar(tree_frame, orient=tk.VERTICAL, command=self.tree.yview)
hsb = ttk.Scrollbar(tree_frame, orient=tk.HORIZONTAL, command=self.tree.xview)
self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
self.tree.grid(row=0, column=0, sticky="nsew")
vsb.grid(row=0, column=1, sticky="ns")
hsb.grid(row=1, column=0, sticky="ew")
tree_frame.rowconfigure(0, weight=1)
tree_frame.columnconfigure(0, weight=1)
self.tree.bind("<Double-1>", self._on_double_click)
self.tree.bind("<Button-3>", self._on_right_click)
self.status_var = tk.StringVar(value="输入网段前缀后点击扫描")
status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W, padding=(6, 2))
status_bar.pack(fill=tk.X, side=tk.BOTTOM)
self.ctx_menu = tk.Menu(self.root, tearoff=0)
self.ctx_menu.add_command(label="📋 复制路径", command=self._ctx_copy_path)
self.ctx_menu.add_command(label="📋 复制 IP", command=self._ctx_copy_ip)
self.ctx_menu.add_command(label="📂 打开共享", command=self._ctx_open_share)
def start_scan(self):
if self.scanning:
return
self.scanning = True
self.cancel_event.clear()
self.entries.clear()
self.found_ips.clear()
self.scanned = 0
for item in self.tree.get_children():
self.tree.delete(item)
self.scan_btn.configure(state=tk.DISABLED)
self.stop_btn.configure(state=tk.NORMAL)
self.export_btn.configure(state=tk.DISABLED)
self.progress["value"] = 0
self.status_var.set("正在扫描...")
prefix = self.ip_var.get().strip()
ips = [f"{prefix}.{i}" for i in range(1, 255)]
thread = threading.Thread(target=self._scan_thread, args=(ips,), daemon=True)
thread.start()
def stop_scan(self):
if self.scanning:
self.cancel_event.set()
self.status_var.set("正在停止...")
def _scan_thread(self, ips):
max_workers = 64
completed = 0
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(scan_single_ip, ip, self.cancel_event): ip
for ip in ips
}
for future in as_completed(futures):
if self.cancel_event.is_set():
break
completed += 1
self.scanned = completed
self.root.after(0, self._update_progress, completed, len(ips))
try:
result = future.result()
except Exception:
continue
if result is None:
continue
ip = result["ip"]
if result.get("shares"):
hostname = result.get("hostname", "N/A")
for share_name in result["shares"]:
path = f"\\\\{ip}\\{share_name}"
self.found_ips.add(ip)
entry = {
"ip": ip,
"hostname": hostname,
"share_name": share_name,
"path": path,
}
self.entries.append(entry)
self.root.after(0, self._insert_entry, entry)
self.root.after(0, self._scan_complete)
def _update_progress(self, scanned, total):
self.progress["value"] = scanned
self.progress["maximum"] = total
def _insert_entry(self, entry):
self.tree.insert("", tk.END, values=(
entry["ip"],
entry["hostname"],
entry["share_name"],
entry["path"],
))
def _scan_complete(self):
self.scanning = False
self.scan_btn.configure(state=tk.NORMAL)
self.stop_btn.configure(state=tk.DISABLED)
self.export_btn.configure(state=tk.NORMAL if self.entries else tk.DISABLED)
self.progress["value"] = self.total
msg = f"扫描完成 — {len(self.found_ips)} 台主机, {len(self.entries)} 个共享"
self.status_var.set(msg)
def export_csv(self):
if not self.entries:
return
ts = int(time.time())
filename = f"lanfind_scan_{ts}.csv"
try:
with open(filename, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.writer(f)
writer.writerow(["IP地址", "主机名", "共享名", "路径"])
for e in self.entries:
writer.writerow([e["ip"], e["hostname"], e["share_name"], e["path"]])
self.status_var.set(f"已导出: {filename}")
except Exception as ex:
messagebox.showerror("导出失败", str(ex))
def _get_selected(self):
sel = self.tree.selection()
if not sel:
return None
item = sel[0]
values = self.tree.item(item, "values")
if not values or len(values) < 4:
return None
return {"ip": values[0], "hostname": values[1], "share_name": values[2], "path": values[3]}
def _on_double_click(self, event):
entry = self._get_selected()
if entry:
open_in_explorer(entry["path"])
def _on_right_click(self, event):
item = self.tree.identify_row(event.y)
if item:
self.tree.selection_set(item)
self.ctx_menu.post(event.x_root, event.y_root)
def _ctx_copy_path(self):
entry = self._get_selected()
if entry:
copy_to_clipboard(self.root,entry["path"])
self.status_var.set("路径已复制到剪贴板")
def _ctx_copy_ip(self):
entry = self._get_selected()
if entry:
copy_to_clipboard(self.root,entry["ip"])
self.status_var.set("IP 已复制到剪贴板")
def _ctx_open_share(self):
entry = self._get_selected()
if entry:
open_in_explorer(entry["path"])
def _poll_events(self):
self.root.after(100, self._poll_events)
def main():
root = tk.Tk()
app = LanFindApp(root)
root.mainloop()
if __name__ == "__main__":
main()
[复制链接]
发表于 2026-7-2 09:36
|
发表于 2026-7-2 09:47
