[Python] 纯文本查看 复制代码
import hashlib
import json
import os
import shutil
import socket
import sys
import threading
import time
import uuid
from datetime import datetime
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import quote, unquote, urlparse
from urllib.request import Request, urlopen
from PyQt5.QtCore import QObject, QSize, Qt, QUrl, pyqtSignal
from PyQt5.QtGui import QColor, QDrag
from PyQt5.QtWidgets import (
QApplication,
QCheckBox,
QDialog,
QFileDialog,
QFrame,
QHBoxLayout,
QLabel,
QListWidget,
QListWidgetItem,
QMainWindow,
QMessageBox,
QPlainTextEdit,
QPushButton,
QVBoxLayout,
QWidget,
)
UDP_DISCOVERY_PORT = 45678
UDP_BUFFER_SIZE = 65535
HELLO_INTERVAL_SEC = 3
PEER_TIMEOUT_SEC = 12
HTTP_TIMEOUT_SEC = 10
def now_ts() -> float:
return time.time()
def fmt_time(ts: float) -> str:
return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
def fmt_size(size: int) -> str:
units = ["B", "KB", "MB", "GB", "TB"]
n = float(size)
for unit in units:
if n < 1024 or unit == units[-1]:
if unit == "B":
return f"{int(n)} {unit}"
return f"{n:.2f} {unit}"
n /= 1024.0
return f"{size} B"
def safe_name(name: str) -> str:
cleaned = "".join(c if c.isalnum() or c in "._- ()[]{}" else "_" for c in name)
cleaned = cleaned.strip()
return cleaned or "file"
def detect_local_ip() -> str:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
sock.connect(("8.8.8.8", 80))
return sock.getsockname()[0]
except OSError:
return "127.0.0.1"
finally:
sock.close()
def find_free_port() -> int:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("0.0.0.0", 0))
port = sock.getsockname()[1]
sock.close()
return port
def sha1_file(path: Path) -> str:
h = hashlib.sha1()
with path.open("rb") as f:
while True:
chunk = f.read(1024 * 1024)
if not chunk:
break
h.update(chunk)
return h.hexdigest()
class ShareCore(QObject):
files_changed = pyqtSignal()
stats_changed = pyqtSignal(dict)
log_event = pyqtSignal(str)
def __init__(self):
super().__init__()
self.lock = threading.RLock()
self.stop_event = threading.Event()
self.node_id = uuid.uuid4().hex[:12]
self.node_name = socket.gethostname()
self.local_ip = detect_local_ip()
self.http_port = find_free_port()
self.base_dir = Path.home() / ".lan_soft_share"
self.shared_dir = self.base_dir / "shared"
self.mirror_dir = self.base_dir / "mirror"
self.auto_sync_default_dir = self.base_dir / "auto_sync_downloads"
self.settings_file = self.base_dir / "settings.json"
self.shared_dir.mkdir(parents=True, exist_ok=True)
self.mirror_dir.mkdir(parents=True, exist_ok=True)
self.auto_sync_default_dir.mkdir(parents=True, exist_ok=True)
self.auto_sync_enabled = False
self.auto_sync_dir = self.auto_sync_default_dir
self.cleanup_shared_on_exit = True
self._load_settings()
self.auto_sync_dir.mkdir(parents=True, exist_ok=True)
self.files = {}
self.peers = {}
self.local_file_paths = {}
self.downloading = set()
self.udp_socket = None
self.http_server = None
self.http_thread = None
self.threads = []
def start(self):
self._start_http()
self._start_udp()
self._spawn(self._udp_recv_loop, "udp-recv")
self._spawn(self._hello_loop, "hello")
self._spawn(self._peer_gc_loop, "peer-gc")
self._emit_stats()
self.log(
"已启动，拖拽文件到窗口即可共享。"
f" 自动接收={'开启' if self.auto_sync_enabled else '关闭'}。"
)
if self.auto_sync_enabled:
self._trigger_auto_sync_backfill()
def stop(self):
with self.lock:
local_file_ids = [
fid for fid, meta in self.files.items() if meta.get("is_local")
]
cleanup_on_exit = bool(self.cleanup_shared_on_exit)
self._broadcast_files_removed(local_file_ids)
self.stop_event.set()
if self.udp_socket:
try:
self.udp_socket.close()
except OSError:
pass
if self.http_server:
try:
self.http_server.shutdown()
except OSError:
pass
try:
self.http_server.server_close()
except OSError:
pass
for t in self.threads:
t.join(timeout=1.2)
if self.http_thread:
self.http_thread.join(timeout=1.2)
if cleanup_on_exit:
self._cleanup_shared_dir()
def log(self, text: str):
self.log_event.emit(f"[{datetime.now().strftime('%H:%M:%S')}] {text}")
def _load_settings(self):
if not self.settings_file.exists():
return
try:
data = json.loads(self.settings_file.read_text(encoding="utf-8"))
self.auto_sync_enabled = bool(data.get("auto_sync_enabled", False))
auto_dir = str(data.get("auto_sync_dir", "")).strip()
if auto_dir:
self.auto_sync_dir = Path(auto_dir).expanduser()
self.cleanup_shared_on_exit = bool(data.get("cleanup_shared_on_exit", True))
except Exception:
# Fall back to defaults if settings are malformed.
self.auto_sync_enabled = False
self.auto_sync_dir = self.auto_sync_default_dir
self.cleanup_shared_on_exit = True
def _save_settings(self):
try:
payload = {
"auto_sync_enabled": self.auto_sync_enabled,
"auto_sync_dir": str(self.auto_sync_dir),
"cleanup_shared_on_exit": self.cleanup_shared_on_exit,
}
self.settings_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
except Exception as e:
self.log(f"保存设置失败: {e}")
def set_auto_sync_enabled(self, enabled: bool):
enabled = bool(enabled)
with self.lock:
self.auto_sync_enabled = enabled
self._save_settings()
self._emit_stats()
self.log(f"自动接收已{'开启' if enabled else '关闭'}")
if enabled:
self._trigger_auto_sync_backfill()
def set_auto_sync_dir(self, folder: str) -> bool:
target = Path(folder).expanduser()
try:
target.mkdir(parents=True, exist_ok=True)
except Exception as e:
self.log(f"无法使用该目录: {e}")
return False
with self.lock:
self.auto_sync_dir = target
self._save_settings()
self._emit_stats()
self.log(f"自动接收目录: {target}")
if self.auto_sync_enabled:
self._trigger_auto_sync_backfill()
return True
def set_cleanup_shared_on_exit(self, enabled: bool):
enabled = bool(enabled)
with self.lock:
self.cleanup_shared_on_exit = enabled
self._save_settings()
self._emit_stats()
self.log(f"退出自动清理共享副本已{'开启' if enabled else '关闭'}")
def _trigger_auto_sync_backfill(self):
with self.lock:
if not self.auto_sync_enabled:
return
candidates = []
for fid, meta in self.files.items():
if meta.get("is_local"):
continue
local_path = meta.get("local_path", "")
if local_path and os.path.exists(local_path):
continue
if meta.get("status") in ("remote", "error"):
candidates.append(fid)
for fid in candidates:
self._spawn(
lambda file_id=fid: self._download_remote(
file_id, target_dir=self.auto_sync_dir, reason="auto-receive"
),
f"auto-{fid[:8]}",
)
def get_stats(self) -> dict:
with self.lock:
return {
"node_name": self.node_name,
"node_id": self.node_id,
"local_ip": self.local_ip,
"http_port": self.http_port,
"peer_count": len(self.peers),
"file_count": len(self.files),
"auto_sync_enabled": self.auto_sync_enabled,
"auto_sync_dir": str(self.auto_sync_dir),
"cleanup_shared_on_exit": self.cleanup_shared_on_exit,
}
def get_files_snapshot(self):
with self.lock:
items = list(self.files.values())
return sorted(items, key=lambda x: x.get("added_at", 0), reverse=True)
def share_paths(self, paths):
changed = False
for p in paths:
path = Path(p)
if not path.exists() or not path.is_file():
continue
if self._share_single(path):
changed = True
if changed:
self.files_changed.emit()
self._emit_stats()
def _share_single(self, path: Path) -> bool:
size = path.stat().st_size
sha1 = sha1_file(path)
file_id = f"{sha1}_{size}"
with self.lock:
if file_id in self.files and self.files[file_id].get("is_local"):
self.log(f"已共享过: {path.name}")
return False
target_name = f"{file_id}_{safe_name(path.name)}"
target_path = self.shared_dir / target_name
if not target_path.exists():
shutil.copy2(str(path), str(target_path))
meta = {
"file_id": file_id,
"name": path.name,
"size": size,
"sha1": sha1,
"added_at": now_ts(),
"owner_id": self.node_id,
"owner_name": self.node_name,
"owner_host": self.local_ip,
"owner_port": self.http_port,
"status": "ready",
"is_local": True,
"local_path": str(target_path),
"sources": [{"host": self.local_ip, "port": self.http_port}],
}
with self.lock:
self.files[file_id] = meta
self.local_file_paths[file_id] = str(target_path)
self._broadcast(
{
"type": "NEW_FILE",
"node_id": self.node_id,
"node_name": self.node_name,
"http_port": self.http_port,
"meta": {
"file_id": file_id,
"name": path.name,
"size": size,
"sha1": sha1,
"added_at": meta["added_at"],
"owner_id": self.node_id,
"owner_name": self.node_name,
},
}
)
self.log(f"已共享: {path.name} ({fmt_size(size)})")
return True
def _start_http(self):
core = self
class Handler(BaseHTTPRequestHandler):
def log_message(self, fmt, *args):
return
def _send_json(self, data, status=HTTPStatus.OK):
payload = json.dumps(data, ensure_ascii=False).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == "/index":
self._send_json(core._local_index())
return
if parsed.path.startswith("/file/"):
file_id = unquote(parsed.path[len("/file/") :])
with core.lock:
fpath = core.local_file_paths.get(file_id)
if not fpath or not os.path.exists(fpath):
self._send_json({"error": "not_found"}, status=HTTPStatus.NOT_FOUND)
return
size = os.path.getsize(fpath)
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "application/octet-stream")
self.send_header("Content-Length", str(size))
self.end_headers()
with open(fpath, "rb") as f:
while True:
chunk = f.read(1024 * 1024)
if not chunk:
break
self.wfile.write(chunk)
return
self._send_json({"error": "unknown_endpoint"}, status=HTTPStatus.NOT_FOUND)
self.http_server = ThreadingHTTPServer(("0.0.0.0", self.http_port), Handler)
self.http_thread = threading.Thread(target=self.http_server.serve_forever, daemon=True)
self.http_thread.start()
def _local_index(self):
with self.lock:
out = []
for f in self.files.values():
if not f.get("is_local"):
continue
out.append(
{
"file_id": f["file_id"],
"name": f["name"],
"size": f["size"],
"sha1": f["sha1"],
"added_at": f["added_at"],
"owner_id": self.node_id,
"owner_name": self.node_name,
}
)
return out
def _start_udp(self):
self.udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
self.udp_socket.bind(("", UDP_DISCOVERY_PORT))
self.udp_socket.settimeout(1.0)
def _spawn(self, fn, name):
t = threading.Thread(target=fn, name=name, daemon=True)
t.start()
self.threads.append(t)
def _broadcast(self, payload: dict):
if not self.udp_socket:
return
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
try:
self.udp_socket.sendto(data, ("255.255.255.255", UDP_DISCOVERY_PORT))
except OSError:
pass
def _broadcast_files_removed(self, file_ids):
if not file_ids:
return
payload = {
"type": "FILES_REMOVED",
"node_id": self.node_id,
"node_name": self.node_name,
"http_port": self.http_port,
"owner_id": self.node_id,
"file_ids": list(file_ids),
"ts": now_ts(),
}
# Best effort: send a few times before stopping network threads.
for _ in range(2):
self._broadcast(payload)
time.sleep(0.05)
def _cleanup_shared_dir(self):
removed = 0
try:
for p in self.shared_dir.iterdir():
if p.is_file():
try:
p.unlink()
removed += 1
except OSError:
pass
except OSError:
return
if removed > 0:
self.log(f"退出清理完成：删除共享副本 {removed} 个。")
def _hello_loop(self):
while not self.stop_event.is_set():
self._broadcast(
{
"type": "HELLO",
"node_id": self.node_id,
"node_name": self.node_name,
"http_port": self.http_port,
"ts": now_ts(),
}
)
self.stop_event.wait(HELLO_INTERVAL_SEC)
def _peer_gc_loop(self):
while not self.stop_event.is_set():
stats_changed = False
files_changed = False
cutoff = now_ts() - PEER_TIMEOUT_SEC
dead_endpoints = []
with self.lock:
dead = [pid for pid, p in self.peers.items() if p["last_seen"] < cutoff]
for pid in dead:
p = self.peers.get(pid) or {}
dead_endpoints.append((p.get("host"), int(p.get("http_port", 0))))
self.peers.pop(pid, None)
stats_changed = True
for host, port in dead_endpoints:
if not host or port <= 0:
continue
for fid, meta in list(self.files.items()):
if meta.get("is_local"):
continue
sources = meta.get("sources", [])
filtered = []
dropped = False
for s in sources:
s_host = s.get("host")
s_port = int(s.get("port", 0))
if s_host == host and s_port == port:
dropped = True
continue
filtered.append({"host": s_host, "port": s_port})
if not dropped:
continue
meta["sources"] = filtered
local_path = str(meta.get("local_path", ""))
has_local = bool(local_path and os.path.exists(local_path))
if has_local:
meta["status"] = "ready"
elif filtered:
meta["status"] = "remote"
else:
self.files.pop(fid, None)
self.local_file_paths.pop(fid, None)
files_changed = True
if stats_changed:
self.log("部分在线节点已离线。")
self._emit_stats()
if files_changed:
self.files_changed.emit()
self._emit_stats()
self.stop_event.wait(2.0)
def _udp_recv_loop(self):
while not self.stop_event.is_set():
try:
data, addr = self.udp_socket.recvfrom(UDP_BUFFER_SIZE)
except socket.timeout:
continue
except OSError:
break
try:
msg = json.loads(data.decode("utf-8", errors="ignore"))
except json.JSONDecodeError:
continue
mtype = msg.get("type")
if msg.get("node_id") == self.node_id:
continue
if mtype == "HELLO":
self._handle_hello(msg, addr[0])
elif mtype == "NEW_FILE":
self._handle_new_file(msg, addr[0])
elif mtype == "FILES_REMOVED":
self._handle_files_removed(msg, addr[0])
def _handle_hello(self, msg: dict, host: str):
node_id = msg.get("node_id")
port = int(msg.get("http_port", 0))
if not node_id or port <= 0:
return
is_new = False
with self.lock:
old = self.peers.get(node_id)
self.peers[node_id] = {
"node_name": msg.get("node_name", node_id),
"host": host,
"http_port": port,
"last_seen": now_ts(),
}
if old is None:
is_new = True
if is_new:
self.log(f"节点上线: {msg.get('node_name', node_id)} @ {host}:{port}")
self._emit_stats()
self._spawn(lambda: self._sync_from_peer(host, port), f"sync-{node_id}")
def _handle_new_file(self, msg: dict, host: str):
port = int(msg.get("http_port", 0))
if port <= 0:
return
meta = msg.get("meta", {})
if not isinstance(meta, dict):
return
self._ingest_remote(meta, host, port)
def _handle_files_removed(self, msg: dict, host: str):
file_ids = msg.get("file_ids", [])
if not isinstance(file_ids, list):
return
port = int(msg.get("http_port", 0))
changed = False
removed_count = 0
with self.lock:
for raw_fid in file_ids:
fid = str(raw_fid)
meta = self.files.get(fid)
if not meta or meta.get("is_local"):
continue
sources = meta.get("sources", [])
filtered = []
for s in sources:
s_host = s.get("host")
s_port = int(s.get("port", 0))
if s_host == host and (port <= 0 or s_port == port):
continue
filtered.append({"host": s_host, "port": s_port})
meta["sources"] = filtered
local_path = str(meta.get("local_path", ""))
has_local = bool(local_path and os.path.exists(local_path))
if has_local:
meta["status"] = "ready"
changed = True
continue
if not filtered:
self.files.pop(fid, None)
self.local_file_paths.pop(fid, None)
removed_count += 1
else:
meta["status"] = "remote"
changed = True
if changed:
if removed_count > 0:
owner_name = msg.get("node_name", msg.get("owner_id", host))
self.log(f"节点下线通知：已移除 {removed_count} 个不可用文件（{owner_name}）。")
self.files_changed.emit()
self._emit_stats()
def _add_source_locked(self, meta: dict, host: str, port: int):
sources = meta.setdefault("sources", [])
for s in sources:
if s.get("host") == host and int(s.get("port", 0)) == int(port):
return
sources.append({"host": host, "port": int(port)})
def _sync_from_peer(self, host: str, port: int):
url = f"http://{host}:{port}/index"
req = Request(url, method="GET")
try:
with urlopen(req, timeout=HTTP_TIMEOUT_SEC) as resp:
payload = resp.read().decode("utf-8", errors="replace")
entries = json.loads(payload)
if not isinstance(entries, list):
return
except Exception:
return
for entry in entries:
if isinstance(entry, dict):
self._ingest_remote(entry, host, port)
def _ingest_remote(self, remote_meta: dict, host: str, port: int):
file_id = remote_meta.get("file_id")
name = remote_meta.get("name", "未知文件")
size = int(remote_meta.get("size", 0))
sha1 = remote_meta.get("sha1", "")
owner_id = remote_meta.get("owner_id", "")
owner_name = remote_meta.get("owner_name", owner_id)
added_at = float(remote_meta.get("added_at", now_ts()))
if not file_id or owner_id == self.node_id:
return
mirror_path = self.mirror_dir / f"{file_id}_{safe_name(name)}"
auto_path = self.auto_sync_dir / f"{file_id}_{safe_name(name)}"
changed = False
should_auto_receive = False
with self.lock:
existing = self.files.get(file_id)
if existing:
self._add_source_locked(existing, host, port)
if not existing.get("owner_host"):
existing["owner_host"] = host
existing["owner_port"] = port
if (mirror_path.exists() or auto_path.exists()) and (not existing.get("local_path")):
ready_path = mirror_path if mirror_path.exists() else auto_path
existing["status"] = "ready"
existing["local_path"] = str(ready_path)
self.local_file_paths[file_id] = str(ready_path)
changed = True
if self.auto_sync_enabled and existing.get("status") in ("remote", "error"):
should_auto_receive = True
else:
ready_path = None
if mirror_path.exists():
ready_path = mirror_path
elif auto_path.exists():
ready_path = auto_path
status = "ready" if ready_path else "remote"
self.files[file_id] = {
"file_id": file_id,
"name": name,
"size": size,
"sha1": sha1,
"added_at": added_at,
"owner_id": owner_id,
"owner_name": owner_name,
"owner_host": host,
"owner_port": port,
"status": status,
"is_local": False,
"local_path": str(ready_path) if ready_path else "",
"sources": [{"host": host, "port": int(port)}],
}
if ready_path:
self.local_file_paths[file_id] = str(ready_path)
changed = True
if self.auto_sync_enabled and status == "remote":
should_auto_receive = True
if changed:
self.files_changed.emit()
self._emit_stats()
if should_auto_receive:
self._spawn(
lambda: self._download_remote(
file_id, target_dir=self.auto_sync_dir, reason="auto-receive"
),
f"auto-{file_id[:8]}",
)
def _download_remote(self, file_id: str, target_dir: Path = None, reason: str = "manual"):
emit_progress = False
with self.lock:
if file_id in self.downloading:
return
self.downloading.add(file_id)
meta = self.files.get(file_id)
if not meta:
self.downloading.discard(file_id)
return
host = meta["owner_host"]
port = meta["owner_port"]
name = meta["name"]
expected_size = int(meta.get("size", 0))
expected_sha1 = meta.get("sha1", "")
sources = list(meta.get("sources", []))
if host and port:
sources.insert(0, {"host": host, "port": port})
unique = []
seen = set()
for s in sources:
key = (s.get("host"), int(s.get("port", 0)))
if key[0] and key[1] > 0 and key not in seen:
seen.add(key)
unique.append({"host": key[0], "port": key[1]})
sources = unique
if meta.get("status") != "ready":
meta["status"] = "downloading"
emit_progress = True
if emit_progress:
self.files_changed.emit()
final_dir = Path(target_dir) if target_dir else self.mirror_dir
final_dir.mkdir(parents=True, exist_ok=True)
final_path = final_dir / f"{file_id}_{safe_name(name)}"
temp_path = final_path.with_suffix(final_path.suffix + ".part")
ok = False
err = ""
if final_path.exists():
try:
if (not expected_size) or final_path.stat().st_size == expected_size:
ok = True
except OSError:
ok = False
for src in sources:
if ok:
break
try:
if temp_path.exists():
temp_path.unlink()
url = f"http://{src['host']}:{src['port']}/file/{quote(file_id)}"
req = Request(url, method="GET")
with urlopen(req, timeout=HTTP_TIMEOUT_SEC) as resp, temp_path.open("wb") as out:
h = hashlib.sha1()
total = 0
while True:
chunk = resp.read(1024 * 1024)
if not chunk:
break
out.write(chunk)
h.update(chunk)
total += len(chunk)
if expected_size and total != expected_size:
raise RuntimeError(f"size mismatch ({total} != {expected_size})")
if expected_sha1 and h.hexdigest() != expected_sha1:
raise RuntimeError("sha1 mismatch")
temp_path.replace(final_path)
ok = True
except Exception as e:
err = str(e)
if temp_path.exists():
try:
temp_path.unlink()
except OSError:
pass
with self.lock:
self.downloading.discard(file_id)
meta = self.files.get(file_id)
if not meta:
return
if ok:
meta["status"] = "ready"
meta["local_path"] = str(final_path)
self.local_file_paths[file_id] = str(final_path)
else:
meta["status"] = "error"
reason_text = {
"manual": "手动",
"on-demand": "按需拖拽",
"auto-receive": "自动接收",
}.get(reason, reason)
if ok:
self.log(f"下载完成（{reason_text}）: {name}")
else:
self.log(f"下载失败（{reason_text}）: {name} ({err})")
self.files_changed.emit()
def ensure_local(self, file_id: str):
with self.lock:
meta = self.files.get(file_id)
if not meta:
return False, "", "文件不存在"
p = meta.get("local_path", "")
if p and os.path.exists(p):
return True, p, ""
in_progress = file_id in self.downloading
if in_progress:
deadline = time.time() + 20
while time.time() < deadline:
time.sleep(0.2)
with self.lock:
meta = self.files.get(file_id)
if not meta:
return False, "", "等待期间文件不存在"
p = meta.get("local_path", "")
if p and os.path.exists(p):
return True, p, ""
if file_id not in self.downloading:
break
self._download_remote(file_id, target_dir=self.mirror_dir, reason="on-demand")
with self.lock:
meta = self.files.get(file_id)
if not meta:
return False, "", "下载后文件不存在"
p = meta.get("local_path", "")
if p and os.path.exists(p):
return True, p, ""
return False, "", "下载失败或源节点离线"
def _emit_stats(self):
self.stats_changed.emit(self.get_stats())
class DropFrame(QFrame):
files_dropped = pyqtSignal(list)
def __init__(self):
super().__init__()
self.setObjectName("DropFrame")
self.setAcceptDrops(True)
layout = QVBoxLayout(self)
layout.setContentsMargins(22, 18, 22, 18)
self.title = QLabel("把文件拖到这里即可共享")
self.title.setAlignment(Qt.AlignCenter)
self.subtitle = QLabel(
"默认只同步文件列表；开启“自动接收”后会自动下载到你设置的目录。"
)
self.subtitle.setWordWrap(True)
self.subtitle.setAlignment(Qt.AlignCenter)
layout.addWidget(self.title)
layout.addWidget(self.subtitle)
def dragEnterEvent(self, event):
if event.mimeData().hasUrls():
event.acceptProposedAction()
else:
event.ignore()
def dragMoveEvent(self, event):
if event.mimeData().hasUrls():
event.acceptProposedAction()
else:
event.ignore()
def dropEvent(self, event):
paths = []
for url in event.mimeData().urls():
if url.isLocalFile():
p = url.toLocalFile()
if os.path.isfile(p):
paths.append(p)
if paths:
self.files_dropped.emit(paths)
event.acceptProposedAction()
else:
event.ignore()
class SharedListWidget(QListWidget):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.download_callback = None
def startDrag(self, supportedActions):
item = self.currentItem()
if not item:
return
meta = item.data(Qt.UserRole) or {}
local_path = meta.get("local_path", "")
if not local_path or not os.path.exists(local_path):
if not self.download_callback:
QMessageBox.information(self, "文件未就绪", "该文件尚未在本地。")
return
ok, local_path, err = self.download_callback(meta)
if not ok:
QMessageBox.warning(self, "下载失败", err or "当前无法下载该文件。")
return
drag = QDrag(self)
mime = self.mimeData(self.selectedItems())
mime.setUrls([QUrl.fromLocalFile(local_path)])
drag.setMimeData(mime)
drag.exec_(Qt.CopyAction)
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.core = ShareCore()
self.setWindowTitle("局域网文件共享工具")
self.resize(980, 700)
self._build_ui()
self._bind()
self.core.start()
self.refresh_stats(self.core.get_stats())
def _build_ui(self):
self.setStyleSheet(
"""
QMainWindow {
background: #0f1422;
}
QWidget {
color: #e8efff;
font-size: 13px;
}
QFrame#TopCard, QFrame#PanelCard {
background: #1a2237;
border: 1px solid #2e3858;
border-radius: 14px;
}
#DropFrame {
background: #141c2e;
border: 2px dashed #3c4e76;
border-radius: 14px;
}
#DropFrame QLabel:first-child {
font-size: 18px;
font-weight: 600;
color: #f0f4ff;
}
#DropFrame QLabel:last-child {
color: #9ab0de;
}
QLabel#MainTitle {
font-size: 24px;
font-weight: 700;
color: #f7f9ff;
}
QLabel#SubInfo {
color: #9bb1df;
}
QLabel#DirLabel {
color: #bdd1ff;
}
QLabel#SectionTitle {
font-size: 15px;
font-weight: 600;
color: #dce7ff;
}
QPushButton {
min-height: 32px;
padding: 0 14px;
border: 1px solid #4b5f93;
border-radius: 10px;
background: #2f4eb9;
color: #f4f7ff;
font-weight: 600;
}
QPushButton:hover {
background: #3b5fd8;
border-color: #6b84da;
}
QPushButton:pressed {
background: #2848a5;
}
QPushButton:disabled {
background: #263252;
border-color: #364264;
color: #7f8fb3;
}
QPushButton#GhostButton {
background: #202c4c;
border-color: #4b5d8c;
}
QPushButton#GhostButton:hover {
background: #27365d;
}
QPushButton#GhostButton:pressed {
background: #1d2a4a;
}
QCheckBox {
spacing: 8px;
color: #dce6ff;
}
QCheckBox::indicator {
width: 16px;
height: 16px;
border-radius: 4px;
border: 1px solid #4a5d90;
background: #121a2b;
}
QCheckBox::indicator:checked {
background: #3b5fd8;
border-color: #89a0e8;
}
QListWidget {
background: #121a2a;
border: 1px solid #34405f;
border-radius: 12px;
padding: 6px;
}
QListWidget::item {
background: #1a2541;
border: 1px solid #304068;
border-radius: 10px;
margin: 4px;
padding: 10px;
}
QListWidget::item:selected {
background: #263b74;
border: 1px solid #6887e0;
}
QPlainTextEdit {
background: #121a2a;
border: 1px solid #34405f;
border-radius: 12px;
padding: 8px;
color: #dce7ff;
selection-background-color: #3b5fd8;
}
QScrollBar:vertical {
background: transparent;
width: 10px;
margin: 2px;
}
QScrollBar::handle:vertical {
background: #3a4d7a;
min-height: 28px;
border-radius: 5px;
}
QScrollBar::handle:vertical:hover {
background: #4d66a3;
}
QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical {
height: 0;
}
QScrollBar:horizontal {
background: transparent;
height: 10px;
margin: 2px;
}
QScrollBar::handle:horizontal {
background: #3a4d7a;
min-width: 28px;
border-radius: 5px;
}
QScrollBar::handle:horizontal:hover {
background: #4d66a3;
}
QScrollBar::add-line:horizontal, QScrollBar::sub-line:horizontal {
width: 0;
}
"""
)
root = QWidget()
self.setCentralWidget(root)
outer = QVBoxLayout(root)
outer.setContentsMargins(18, 18, 18, 18)
outer.setSpacing(14)
top = QFrame()
top.setObjectName("TopCard")
top_layout = QVBoxLayout(top)
top_layout.setContentsMargins(16, 14, 16, 14)
top_layout.setSpacing(8)
self.title_label = QLabel("局域网文件共享")
self.title_label.setObjectName("MainTitle")
self.addr_label = QLabel("地址: -")
self.addr_label.setObjectName("SubInfo")
self.stats_label = QLabel("在线: 0 | 文件: 0")
self.stats_label.setObjectName("SubInfo")
self.auto_sync_check = QCheckBox("自动接收并保存他人上传的文件")
self.cleanup_on_exit_check = QCheckBox("退出软件时自动删除共享副本")
self.cleanup_on_exit_check.setChecked(True)
self.auto_sync_btn = QPushButton("选择目录")
self.auto_sync_btn.setObjectName("GhostButton")
self.help_btn = QPushButton("使用帮助")
self.help_btn.setObjectName("GhostButton")
self.auto_sync_dir_label = QLabel("自动接收目录: -")
self.auto_sync_dir_label.setObjectName("DirLabel")
self.auto_sync_dir_label.setTextInteractionFlags(Qt.TextSelectableByMouse)
sync_row = QHBoxLayout()
sync_row.setSpacing(8)
sync_row.addWidget(self.auto_sync_check)
sync_row.addWidget(self.cleanup_on_exit_check)
sync_row.addWidget(self.auto_sync_btn)
sync_row.addWidget(self.help_btn)
sync_row.addStretch(1)
top_layout.addWidget(self.title_label)
top_layout.addWidget(self.addr_label)
top_layout.addWidget(self.stats_label)
top_layout.addLayout(sync_row)
top_layout.addWidget(self.auto_sync_dir_label)
outer.addWidget(top)
self.drop_frame = DropFrame()
outer.addWidget(self.drop_frame)
list_card = QFrame()
list_card.setObjectName("PanelCard")
list_layout = QVBoxLayout(list_card)
list_layout.setContentsMargins(14, 12, 14, 14)
list_layout.setSpacing(10)
list_title = QLabel("共享文件列表（拖出时按需下载，可保存到任意位置）")
list_title.setObjectName("SectionTitle")
list_layout.addWidget(list_title)
self.file_list = SharedListWidget()
self.file_list.setSelectionMode(self.file_list.SingleSelection)
self.file_list.setDragEnabled(True)
self.file_list.download_callback = self._download_for_drag
list_layout.addWidget(self.file_list, 1)
outer.addWidget(list_card, 1)
log_card = QFrame()
log_card.setObjectName("PanelCard")
log_layout = QVBoxLayout(log_card)
log_layout.setContentsMargins(14, 12, 14, 14)
log_layout.setSpacing(10)
log_title = QLabel("传输日志")
log_title.setObjectName("SectionTitle")
log_layout.addWidget(log_title)
self.log_box = QPlainTextEdit()
self.log_box.setReadOnly(True)
self.log_box.setMaximumBlockCount(400)
self.log_box.setFixedHeight(130)
log_layout.addWidget(self.log_box)
outer.addWidget(log_card)
def _bind(self):
self.drop_frame.files_dropped.connect(self.core.share_paths)
self.core.files_changed.connect(self.refresh_files)
self.core.stats_changed.connect(self.refresh_stats)
self.core.log_event.connect(self.append_log)
self.auto_sync_check.toggled.connect(self._on_auto_sync_toggled)
self.cleanup_on_exit_check.toggled.connect(self._on_cleanup_on_exit_toggled)
self.auto_sync_btn.clicked.connect(self._on_choose_auto_sync_dir)
self.help_btn.clicked.connect(self._show_help_dialog)
def append_log(self, text: str):
self.log_box.appendPlainText(text)
def _on_auto_sync_toggled(self, checked: bool):
self.core.set_auto_sync_enabled(bool(checked))
def _on_cleanup_on_exit_toggled(self, checked: bool):
self.core.set_cleanup_shared_on_exit(bool(checked))
def _on_choose_auto_sync_dir(self):
current = str(self.core.auto_sync_dir)
folder = QFileDialog.getExistingDirectory(self, "选择自动接收目录", current)
if not folder:
return
if not self.core.set_auto_sync_dir(folder):
QMessageBox.warning(self, "目录错误", "无法使用所选目录。")
def _show_help_dialog(self):
help_text = (
"一、基础使用\n"
"1. 所有电脑打开同一个软件，并处于同一局域网。\n"
"2. 把文件拖到主窗口上方拖拽区域，即可共享给在线用户。\n"
"3. 其他人会先看到文件列表（不自动下载，除非开启自动接收）。\n\n"
"二、如何获取文件\n"
"1. 在文件列表中选中一个文件，直接拖到桌面或任意文件夹。\n"
"2. 如果本地还没有该文件，软件会先下载，再完成拖放保存。\n\n"
"三、自动接收模式\n"
"1. 勾选“自动接收并保存他人上传的文件”。\n"
"2. 点击“选择目录”设置保存位置。\n"
"3. 开启后，他人新上传文件会自动下载到该目录。\n\n"
"四、注意事项\n"
"1. 下载文件时，至少要有一个持有该文件的在线节点。\n"
"2. 文件过大时会占用网络带宽，请按需使用。\n"
"3. 默认开启“退出软件时自动删除共享副本”，退出会广播下线清单给在线节点。\n"
)
dialog = QDialog(self)
dialog.setWindowTitle("使用帮助")
dialog.resize(700, 560)
dialog.setModal(True)
dialog.setStyleSheet(
"""
QDialog {
background: #101729;
color: #e8efff;
}
QLabel#HelpTitle {
font-size: 22px;
font-weight: 700;
color: #f3f7ff;
}
QLabel#HelpDesc {
color: #9db2df;
font-size: 13px;
}
QPlainTextEdit#HelpBody {
background: #121a2a;
border: 1px solid #34405f;
border-radius: 12px;
color: #dce7ff;
padding: 10px;
selection-background-color: #3b5fd8;
font-size: 13px;
}
QPushButton {
min-height: 34px;
min-width: 110px;
padding: 0 14px;
border: 1px solid #4b5f93;
border-radius: 10px;
background: #2f4eb9;
color: #f4f7ff;
font-weight: 600;
}
QPushButton:hover {
background: #3b5fd8;
border-color: #6b84da;
}
QPushButton:pressed {
background: #2848a5;
}
"""
)
layout = QVBoxLayout(dialog)
layout.setContentsMargins(16, 14, 16, 14)
layout.setSpacing(10)
title = QLabel("局域网文件共享工具 使用帮助")
title.setObjectName("HelpTitle")
desc = QLabel("支持拖入共享、按需下载拖出、自动接收落地。")
desc.setObjectName("HelpDesc")
body = QPlainTextEdit()
body.setObjectName("HelpBody")
body.setReadOnly(True)
body.setPlainText(help_text)
btn_row = QHBoxLayout()
btn_row.addStretch(1)
close_btn = QPushButton("关闭")
close_btn.clicked.connect(dialog.accept)
btn_row.addWidget(close_btn)
layout.addWidget(title)
layout.addWidget(desc)
layout.addWidget(body, 1)
layout.addLayout(btn_row)
dialog.exec_()
def _download_for_drag(self, meta: dict):
file_id = meta.get("file_id")
name = meta.get("name", "未知文件")
if not file_id:
return False, "", "文件ID无效"
self.append_log(f"[{datetime.now().strftime('%H:%M:%S')}] 按需下载: {name}")
QApplication.setOverrideCursor(Qt.WaitCursor)
try:
ok, local_path, err = self.core.ensure_local(file_id)
finally:
QApplication.restoreOverrideCursor()
self.refresh_files()
if ok:
self.append_log(f"[{datetime.now().strftime('%H:%M:%S')}] 已就绪，可拖出保存: {name}")
return True, local_path, ""
self.append_log(f"[{datetime.now().strftime('%H:%M:%S')}] 下载失败: {name} ({err})")
return False, "", err
def refresh_stats(self, stats: dict):
self.addr_label.setText(
f"地址: {stats['local_ip']}:{stats['http_port']} | 节点: {stats['node_name']} ({stats['node_id']})"
)
mode = "自动接收: 开启" if stats.get("auto_sync_enabled") else "自动接收: 关闭"
self.stats_label.setText(f"在线: {stats['peer_count']} | 文件: {stats['file_count']} | {mode}")
self.auto_sync_dir_label.setText(f"自动接收目录: {stats.get('auto_sync_dir', '-')}")
checked = bool(stats.get("auto_sync_enabled"))
if self.auto_sync_check.isChecked() != checked:
self.auto_sync_check.blockSignals(True)
self.auto_sync_check.setChecked(checked)
self.auto_sync_check.blockSignals(False)
cleanup_checked = bool(stats.get("cleanup_shared_on_exit", True))
if self.cleanup_on_exit_check.isChecked() != cleanup_checked:
self.cleanup_on_exit_check.blockSignals(True)
self.cleanup_on_exit_check.setChecked(cleanup_checked)
self.cleanup_on_exit_check.blockSignals(False)
def refresh_files(self):
selected_id = None
current = self.file_list.currentItem()
if current:
m = current.data(Qt.UserRole) or {}
selected_id = m.get("file_id")
self.file_list.clear()
for f in self.core.get_files_snapshot():
status = f.get("status", "unknown")
status_text = {
"ready": "已就绪",
"downloading": "下载中",
"remote": "仅列表",
"error": "错误",
}.get(status, "未知")
is_local = bool(f.get("is_local"))
prefix = "我方" if is_local else "远端"
if is_local:
source_label = "本机"
else:
owner_name = f.get("owner_name", "-")
owner_host = str(f.get("owner_host", "") or "-")
owner_id = str(f.get("owner_id", "") or "")
owner_short = owner_id[-4:] if owner_id else "--"
source_label = f"{owner_name}({owner_short})@{owner_host}"
sources = f.get("sources", [])
source_count = len(sources) if isinstance(sources, list) else 0
text = (
f"{f.get('name', '未知文件')}\n"
f"{prefix} {source_label}"
f" | 源:{source_count}"
f" | {fmt_size(int(f.get('size', 0)))}"
f" | {status_text}"
f" | {fmt_time(float(f.get('added_at', now_ts())))}"
)
item = QListWidgetItem(text)
item.setData(Qt.UserRole, f)
item.setSizeHint(QSize(0, 56))
if status == "ready":
item.setForeground(QColor("#7ee2a6"))
elif status == "downloading":
item.setForeground(QColor("#f2cd7a"))
elif status == "remote":
item.setForeground(QColor("#89b7ff"))
elif status == "error":
item.setForeground(QColor("#ff9a9a"))
self.file_list.addItem(item)
if selected_id and f.get("file_id") == selected_id:
self.file_list.setCurrentItem(item)
def closeEvent(self, event):
self.core.stop()
super().closeEvent(event)
def main():
app = QApplication(sys.argv)
app.setApplicationName("局域网文件共享工具")
win = MainWindow()
win.show()
sys.exit(app.exec_())
if __name__ == "__main__":
main()
[复制链接]
发表于 2026-2-28 15:57
