902 lines
37 KiB
Python
902 lines
37 KiB
Python
"""
|
||
views/admin_websites_view.py — Admin panel: Website Link Management tab.
|
||
Changes: added Check Type (Daily / Weekly) field and treeview column.
|
||
"""
|
||
|
||
import tkinter as tk
|
||
from tkinter import ttk
|
||
import logging
|
||
from utils.ui_helpers import (
|
||
COLOURS, FONT, FONT_BOLD, FONT_HEADING, FONT_SMALL,
|
||
scrolled_text, show_error, show_info, confirm_delete
|
||
)
|
||
|
||
logger = logging.getLogger("admin_websites_view")
|
||
|
||
CHECK_TYPE_OPTIONS = ["daily", "weekly"]
|
||
CHECK_TYPE_LABELS = {"daily": "📅 Daily", "weekly": "🗓 Weekly"}
|
||
|
||
|
||
class AdminWebsitesView(ttk.Frame):
|
||
def __init__(self, parent, current_user: dict):
|
||
super().__init__(parent)
|
||
self.current_user = current_user
|
||
self._build_ui()
|
||
self._load_websites()
|
||
|
||
# ─── Layout ───────────────────────────────────────────────────────────────
|
||
|
||
def _build_ui(self):
|
||
toolbar = ttk.Frame(self)
|
||
toolbar.pack(fill="x", pady=(0, 10))
|
||
ttk.Label(toolbar, text="Website Link Management",
|
||
style="Heading.TLabel").pack(side="left")
|
||
|
||
ttk.Button(toolbar, text="＋ Add Website",
|
||
command=self._open_add).pack(side="right", padx=(4, 0))
|
||
ttk.Button(toolbar, text="📥 Import",
|
||
style="Ghost.TButton",
|
||
command=self._open_import).pack(side="right", padx=(4, 0))
|
||
ttk.Button(toolbar, text="✎ Edit",
|
||
style="Ghost.TButton",
|
||
command=self._open_edit).pack(side="right", padx=(4, 0))
|
||
ttk.Button(toolbar, text="✕ Delete",
|
||
style="Danger.TButton",
|
||
command=self._delete_selected).pack(side="right")
|
||
|
||
cols = ("ID", "Name", "Type", "URL", "Note", "Created By")
|
||
widths = [40, 170, 80, 230, 150, 100]
|
||
self.tree = ttk.Treeview(self, columns=cols, show="headings",
|
||
selectmode="browse")
|
||
for col, w in zip(cols, widths):
|
||
self.tree.heading(col, text=col)
|
||
self.tree.column(col, width=w,
|
||
anchor="center" if w <= 80 else "w")
|
||
self.tree.pack(fill="both", expand=True)
|
||
self.tree.bind("<Double-1>", lambda _: self._open_edit())
|
||
|
||
# Tag weekly rows with a subtle colour
|
||
self.tree.tag_configure("weekly", foreground=COLOURS["warning"])
|
||
|
||
vsb = ttk.Scrollbar(self, orient="vertical", command=self.tree.yview)
|
||
self.tree.configure(yscrollcommand=vsb.set)
|
||
vsb.place(relx=1, rely=0, relheight=1, anchor="ne")
|
||
|
||
# ─── Data ─────────────────────────────────────────────────────────────────
|
||
|
||
def _load_websites(self):
|
||
from models import get_all_websites
|
||
self.tree.delete(*self.tree.get_children())
|
||
try:
|
||
for w in get_all_websites():
|
||
ct = w.get("check_type") or "daily"
|
||
tag = "weekly" if ct == "weekly" else ""
|
||
self.tree.insert(
|
||
"", "end", iid=str(w["id"]), tags=(tag,),
|
||
values=(
|
||
w["id"],
|
||
w["name"],
|
||
CHECK_TYPE_LABELS.get(ct, ct),
|
||
w["url"],
|
||
(w["note"] or "")[:60],
|
||
w["creator"] or "",
|
||
)
|
||
)
|
||
except Exception as e:
|
||
show_error(f"Failed to load websites:\n{e}")
|
||
|
||
def _get_selected_id(self):
|
||
sel = self.tree.selection()
|
||
return int(sel[0]) if sel else None
|
||
|
||
# ─── Actions ──────────────────────────────────────────────────────────────
|
||
|
||
def _open_import(self):
|
||
"""Open the bulk import dialog."""
|
||
BulkImportDialog(self, self.current_user, on_complete=self._load_websites)
|
||
|
||
def _open_add(self):
|
||
WebsiteDialog(self, self.current_user, website_data=None,
|
||
on_save=self._load_websites)
|
||
|
||
def _open_edit(self):
|
||
wid = self._get_selected_id()
|
||
if not wid:
|
||
show_error("Please select a website to edit.")
|
||
return
|
||
from models import get_website_by_id, get_website_credentials, get_website_assigned_users
|
||
data = get_website_by_id(wid)
|
||
creds = get_website_credentials(wid)
|
||
data["credentials"] = creds
|
||
data["assigned_users"] = get_website_assigned_users(wid)
|
||
WebsiteDialog(self, self.current_user, website_data=data,
|
||
on_save=self._load_websites)
|
||
|
||
def _delete_selected(self):
|
||
wid = self._get_selected_id()
|
||
if not wid:
|
||
show_error("Please select a website to delete.")
|
||
return
|
||
vals = self.tree.item(wid, "values")
|
||
name = vals[1] if vals else str(wid)
|
||
if confirm_delete(name):
|
||
try:
|
||
from models import delete_website
|
||
delete_website(self.current_user["id"], wid)
|
||
logger.info(f"Website id={wid} deleted by admin.")
|
||
show_info(f"Website '{name}' deleted.")
|
||
self._load_websites()
|
||
except Exception as e:
|
||
show_error(f"Delete failed:\n{e}")
|
||
|
||
|
||
# ─── Website Dialog (Add / Edit) ──────────────────────────────────────────────
|
||
|
||
class WebsiteDialog(tk.Toplevel):
|
||
def __init__(self, parent, current_user, website_data, on_save):
|
||
super().__init__(parent)
|
||
self.current_user = current_user
|
||
self.website_data = website_data
|
||
self.on_save = on_save
|
||
self.is_edit = website_data is not None
|
||
self.cred_rows = []
|
||
|
||
self.title("Edit Website" if self.is_edit else "Add Website")
|
||
self.configure(bg=COLOURS["bg"])
|
||
self.grab_set()
|
||
self._build_ui()
|
||
self._centre()
|
||
|
||
def _centre(self):
|
||
self.update_idletasks()
|
||
w, h = 600, 680
|
||
x = (self.winfo_screenwidth() - w) // 2
|
||
y = (self.winfo_screenheight() - h) // 2
|
||
self.geometry(f"{w}x{h}+{x}+{y}")
|
||
|
||
def _build_ui(self):
|
||
# Scrollable container
|
||
canvas = tk.Canvas(self, bg=COLOURS["bg"], highlightthickness=0)
|
||
vsb = ttk.Scrollbar(self, orient="vertical", command=canvas.yview)
|
||
canvas.configure(yscrollcommand=vsb.set)
|
||
vsb.pack(side="right", fill="y")
|
||
canvas.pack(fill="both", expand=True)
|
||
|
||
self.inner = ttk.Frame(canvas)
|
||
self.inner.bind("<Configure>",
|
||
lambda e: canvas.configure(scrollregion=canvas.bbox("all")))
|
||
canvas.create_window((0, 0), window=self.inner, anchor="nw")
|
||
|
||
# Scope mousewheel to canvas hover — avoids stale-widget errors on close
|
||
def _enter(e):
|
||
canvas.bind_all("<MouseWheel>",
|
||
lambda ev: _safe_scroll(ev))
|
||
def _leave(e):
|
||
try:
|
||
canvas.unbind_all("<MouseWheel>")
|
||
except Exception:
|
||
pass
|
||
def _safe_scroll(ev):
|
||
try:
|
||
canvas.yview_scroll(int(-1 * (ev.delta / 120)), "units")
|
||
except Exception:
|
||
pass
|
||
|
||
canvas.bind("<Enter>", _enter)
|
||
canvas.bind("<Leave>", _leave)
|
||
self.protocol("WM_DELETE_WINDOW", lambda: (_leave(None), self.destroy()))
|
||
|
||
ttk.Label(self.inner,
|
||
text="Edit Website" if self.is_edit else "New Website",
|
||
style="Heading.TLabel").pack(anchor="w", padx=24, pady=(20, 4))
|
||
|
||
form = ttk.Frame(self.inner)
|
||
form.pack(fill="x", padx=24, pady=4)
|
||
form.columnconfigure(1, weight=1)
|
||
|
||
def text_row(label, r):
|
||
ttk.Label(form, text=label).grid(row=r, column=0,
|
||
sticky="nw", padx=(0, 10), pady=6)
|
||
var = tk.StringVar()
|
||
ent = ttk.Entry(form, textvariable=var)
|
||
ent.grid(row=r, column=1, sticky="ew", pady=6)
|
||
return var
|
||
|
||
self.name_var = text_row("Name *", 0)
|
||
self.url_var = text_row("URL *", 1)
|
||
|
||
# ── Check Type ────────────────────────────────────────────────────────
|
||
ttk.Label(form, text="Check Type").grid(
|
||
row=2, column=0, sticky="w", padx=(0, 10), pady=6)
|
||
|
||
type_frame = tk.Frame(form, bg=COLOURS["bg"])
|
||
type_frame.grid(row=2, column=1, sticky="w", pady=6)
|
||
|
||
self.check_type_var = tk.StringVar(value="daily")
|
||
for val, lbl in [("daily", "📅 Daily"), ("weekly", "🗓 Weekly")]:
|
||
rb = tk.Radiobutton(
|
||
type_frame,
|
||
text=lbl,
|
||
variable=self.check_type_var,
|
||
value=val,
|
||
bg=COLOURS["bg"],
|
||
fg=COLOURS["text"],
|
||
activebackground=COLOURS["bg"],
|
||
activeforeground=COLOURS["accent"],
|
||
selectcolor=COLOURS["surface2"],
|
||
font=FONT,
|
||
cursor="hand2",
|
||
)
|
||
rb.pack(side="left", padx=(0, 16))
|
||
|
||
# ── Note ──────────────────────────────────────────────────────────────
|
||
ttk.Label(form, text="Note").grid(row=3, column=0, sticky="nw",
|
||
padx=(0, 10), pady=6)
|
||
note_frame, self.note_txt = scrolled_text(form, height=4)
|
||
note_frame.grid(row=3, column=1, sticky="ew", pady=6)
|
||
|
||
# ── Visibility ────────────────────────────────────────────────────────
|
||
ttk.Label(form, text="Visibility").grid(
|
||
row=4, column=0, sticky="w", padx=(0, 10), pady=6)
|
||
|
||
vis_frame = tk.Frame(form, bg=COLOURS["bg"])
|
||
vis_frame.grid(row=4, column=1, sticky="w", pady=6)
|
||
|
||
self.visibility_var = tk.StringVar(value="all")
|
||
for val, lbl in [("all", "👥 All Users"), ("assigned", "🔒 Assigned Only")]:
|
||
rb = tk.Radiobutton(
|
||
vis_frame, text=lbl,
|
||
variable=self.visibility_var, value=val,
|
||
bg=COLOURS["bg"], fg=COLOURS["text"],
|
||
activebackground=COLOURS["bg"],
|
||
activeforeground=COLOURS["accent"],
|
||
selectcolor=COLOURS["surface2"],
|
||
font=FONT, cursor="hand2",
|
||
command=self._on_visibility_change,
|
||
)
|
||
rb.pack(side="left", padx=(0, 16))
|
||
|
||
# ── Assigned users (shown only when visibility='assigned') ─────────────
|
||
self._user_assign_frame = ttk.Frame(self.inner)
|
||
self._user_assign_frame.pack(fill="x", padx=24, pady=(0, 4))
|
||
|
||
ttk.Label(self._user_assign_frame, text="Assigned Users",
|
||
style="Heading.TLabel").pack(anchor="w", pady=(4, 6))
|
||
|
||
# Load all active regular users for the picker
|
||
from models import get_all_users
|
||
try:
|
||
all_users = [u for u in get_all_users()
|
||
if u["is_active"] and u["role"] == "user"]
|
||
except Exception:
|
||
all_users = []
|
||
|
||
self._user_vars = {} # user_id -> BooleanVar
|
||
user_grid = tk.Frame(self._user_assign_frame, bg=COLOURS["surface"],
|
||
padx=12, pady=8)
|
||
user_grid.pack(fill="x")
|
||
|
||
for i, u in enumerate(all_users):
|
||
var = tk.BooleanVar(value=False)
|
||
self._user_vars[u["id"]] = var
|
||
col, row = i % 3, i // 3
|
||
tk.Checkbutton(
|
||
user_grid,
|
||
text=f"{u['username']} ({u['full_name'] or ''})",
|
||
variable=var,
|
||
bg=COLOURS["surface"], fg=COLOURS["text"],
|
||
activebackground=COLOURS["surface"],
|
||
activeforeground=COLOURS["accent"],
|
||
selectcolor=COLOURS["surface2"],
|
||
font=FONT_SMALL, cursor="hand2",
|
||
anchor="w",
|
||
).grid(row=row, column=col, sticky="w", padx=8, pady=2)
|
||
|
||
# Hide initially; shown when visibility='assigned'
|
||
self._user_assign_frame.pack_forget()
|
||
|
||
# ── Credentials section (collapsed by default) ───────────────────────
|
||
ttk.Separator(self.inner, orient="horizontal").pack(
|
||
fill="x", padx=24, pady=12)
|
||
|
||
# Toggle header row
|
||
cred_toggle = ttk.Frame(self.inner)
|
||
cred_toggle.pack(fill="x", padx=24)
|
||
|
||
ttk.Label(cred_toggle, text="Login Credentials",
|
||
style="Heading.TLabel").pack(side="left")
|
||
|
||
self._cred_expanded = False
|
||
self._cred_toggle_btn = ttk.Button(
|
||
cred_toggle,
|
||
text="🔑 Show Credentials",
|
||
style="Ghost.TButton",
|
||
command=self._toggle_credentials,
|
||
)
|
||
self._cred_toggle_btn.pack(side="right")
|
||
|
||
ttk.Button(cred_toggle, text="＋ Add Credential",
|
||
command=self._add_cred_row_visible).pack(side="right", padx=(0, 6))
|
||
|
||
# Collapsible body
|
||
self._cred_body = ttk.Frame(self.inner)
|
||
self.creds_container = ttk.Frame(self._cred_body)
|
||
self.creds_container.pack(fill="x")
|
||
# _cred_body is NOT packed initially — hidden by default
|
||
|
||
# ── Buttons ───────────────────────────────────────────────────────────
|
||
ttk.Separator(self.inner, orient="horizontal").pack(
|
||
fill="x", padx=24, pady=12)
|
||
btn_frame = ttk.Frame(self.inner)
|
||
btn_frame.pack(fill="x", padx=24, pady=(0, 20))
|
||
ttk.Button(btn_frame, text="Save",
|
||
command=self._save).pack(side="right", padx=(6, 0))
|
||
ttk.Button(btn_frame, text="Cancel", style="Ghost.TButton",
|
||
command=self.destroy).pack(side="right")
|
||
|
||
# Pre-populate if editing
|
||
if self.is_edit:
|
||
d = self.website_data
|
||
self.name_var.set(d.get("name") or "")
|
||
self.url_var.set(d.get("url") or "")
|
||
self.check_type_var.set(d.get("check_type") or "daily")
|
||
self.note_txt.insert("1.0", d.get("note") or "")
|
||
vis = d.get("visibility") or "all"
|
||
self.visibility_var.set(vis)
|
||
# Pre-tick assigned users
|
||
assigned_ids = {u["id"] for u in d.get("assigned_users", [])}
|
||
for uid, var in self._user_vars.items():
|
||
var.set(uid in assigned_ids)
|
||
self._on_visibility_change()
|
||
existing_creds = d.get("credentials", [])
|
||
for cred in existing_creds:
|
||
self._add_cred_row(cred)
|
||
# Auto-expand credentials section if creds already exist
|
||
if existing_creds:
|
||
self._expand_credentials()
|
||
# For new websites: credentials section stays collapsed
|
||
|
||
def _on_visibility_change(self):
|
||
"""Show or hide the user assignment panel based on visibility selection."""
|
||
if self.visibility_var.get() == "assigned":
|
||
# Use before=self._cred_body (not self.creds_container) because
|
||
# pack's `before` argument requires the reference widget to share
|
||
# the same parent. _user_assign_frame and _cred_body are both
|
||
# children of self.inner; creds_container is a child of _cred_body,
|
||
# so referencing it across that boundary raises a TclError.
|
||
self._user_assign_frame.pack(fill="x", padx=24, pady=(0, 4),
|
||
before=self._cred_body)
|
||
else:
|
||
self._user_assign_frame.pack_forget()
|
||
|
||
def _expand_credentials(self):
|
||
"""Show the credentials body and update the toggle button label."""
|
||
self._cred_expanded = True
|
||
self._cred_body.pack(fill="x", padx=24, pady=(4, 0))
|
||
self._cred_toggle_btn.config(text="🔒 Hide Credentials")
|
||
|
||
def _toggle_credentials(self):
|
||
"""Show or hide the collapsible credentials section."""
|
||
if self._cred_expanded:
|
||
self._cred_expanded = False
|
||
self._cred_body.pack_forget()
|
||
self._cred_toggle_btn.config(text="🔑 Show Credentials")
|
||
else:
|
||
self._expand_credentials()
|
||
|
||
def _add_cred_row_visible(self):
|
||
"""Expand the section (if collapsed) then add an empty credential row."""
|
||
if not self._cred_expanded:
|
||
self._expand_credentials()
|
||
self._add_cred_row()
|
||
|
||
def _add_cred_row(self, cred=None):
|
||
frame = ttk.Frame(self.creds_container, style="Surface.TFrame")
|
||
frame.pack(fill="x", pady=4, ipady=4)
|
||
|
||
label_var = tk.StringVar(value=cred.get("label", "") if cred else "")
|
||
user_var = tk.StringVar(value=cred.get("username", "") if cred else "")
|
||
pass_var = tk.StringVar(value=cred.get("password", "") if cred else "")
|
||
|
||
for col_n, (lbl, var, show) in enumerate([
|
||
("Label", label_var, None),
|
||
("Username", user_var, None),
|
||
("Password", pass_var, "•"),
|
||
]):
|
||
tk.Label(frame, text=lbl, bg=COLOURS["surface"],
|
||
fg=COLOURS["text_dim"], font=FONT_SMALL).grid(
|
||
row=0, column=col_n*2, sticky="w", padx=(8, 2))
|
||
ent = ttk.Entry(frame, textvariable=var, show=show or "", width=14)
|
||
ent.grid(row=0, column=col_n*2+1, sticky="ew", padx=(0, 8), pady=4)
|
||
|
||
ttk.Button(frame, text="✕", style="Danger.TButton", width=3,
|
||
command=lambda f=frame: self._remove_cred_row(f)).grid(
|
||
row=0, column=6, padx=(0, 8))
|
||
|
||
frame.columnconfigure(1, weight=1)
|
||
frame.columnconfigure(3, weight=1)
|
||
frame.columnconfigure(5, weight=1)
|
||
self.cred_rows.append((label_var, user_var, pass_var, frame))
|
||
|
||
def _remove_cred_row(self, frame):
|
||
self.cred_rows = [(l, u, p, f) for l, u, p, f in self.cred_rows
|
||
if f is not frame]
|
||
frame.destroy()
|
||
|
||
def _save(self):
|
||
name = self.name_var.get().strip()
|
||
url = self.url_var.get().strip()
|
||
check_type = self.check_type_var.get()
|
||
note = self.note_txt.get("1.0", "end-1c").strip()
|
||
visibility = self.visibility_var.get()
|
||
assigned_user_ids = [uid for uid, var in self._user_vars.items() if var.get()]
|
||
|
||
if not name or not url:
|
||
show_error("Name and URL are required.")
|
||
return
|
||
|
||
# Normalise and validate URL before writing to the database.
|
||
# Reject schemes that could be executed in-browser (javascript:, data:, etc.)
|
||
# and enforce that a host is present so the health-check thread and the
|
||
# "open in browser" action both receive a usable address.
|
||
url = _validate_and_normalise_url(url)
|
||
if url is None:
|
||
show_error(
|
||
"The URL entered is not valid.\n\n"
|
||
"Please enter a full web address, e.g.:\n"
|
||
" https://www.example.com\n"
|
||
" http://intranet.local/app"
|
||
)
|
||
return
|
||
|
||
if visibility == "assigned" and not assigned_user_ids:
|
||
show_error("Please assign at least one user, or set visibility to All Users.")
|
||
return
|
||
|
||
credentials = []
|
||
for label_var, user_var, pass_var, _ in self.cred_rows:
|
||
u = user_var.get().strip()
|
||
if u:
|
||
credentials.append({
|
||
"label": label_var.get().strip(),
|
||
"username": u,
|
||
"password": pass_var.get().strip(),
|
||
})
|
||
|
||
try:
|
||
if self.is_edit:
|
||
from models import update_website
|
||
update_website(self.current_user["id"],
|
||
self.website_data["id"],
|
||
name, url, check_type, note, credentials,
|
||
visibility, assigned_user_ids)
|
||
logger.info(f"Website id={self.website_data['id']} updated "
|
||
f"check_type='{check_type}' visibility='{visibility}'.")
|
||
show_info("Website updated successfully.")
|
||
else:
|
||
from models import create_website
|
||
create_website(self.current_user["id"],
|
||
name, url, check_type, note, credentials,
|
||
visibility, assigned_user_ids)
|
||
logger.info(f"New website '{name}' created "
|
||
f"check_type='{check_type}' visibility='{visibility}'.")
|
||
show_info("Website created successfully.")
|
||
self.on_save()
|
||
self.destroy()
|
||
except Exception as e:
|
||
show_error(f"Save failed:\n{e}")
|
||
|
||
|
||
# ─── URL validation helper ────────────────────────────────────────────────────
|
||
|
||
def _validate_and_normalise_url(raw: str) -> "str | None":
|
||
"""
|
||
Validate and normalise a user-supplied URL string.
|
||
|
||
Rules:
|
||
- If no scheme is present, prepend 'https://'.
|
||
- Only 'http' and 'https' schemes are accepted.
|
||
- A non-empty netloc (host) must be present.
|
||
- Returns the normalised URL string on success, None on failure.
|
||
|
||
This prevents dangerous schemes (javascript:, data:, file:, etc.) from
|
||
reaching the database, the health-check thread, or webbrowser.open().
|
||
"""
|
||
import urllib.parse
|
||
|
||
if not raw:
|
||
return None
|
||
|
||
# Prepend https:// if no scheme is given so urlparse can parse the host
|
||
if "://" not in raw:
|
||
raw = "https://" + raw
|
||
|
||
try:
|
||
parts = urllib.parse.urlparse(raw)
|
||
except Exception:
|
||
return None
|
||
|
||
if parts.scheme.lower() not in ("http", "https"):
|
||
return None
|
||
|
||
if not parts.netloc:
|
||
return None
|
||
|
||
# Reconstruct a clean URL (strips any leading/trailing whitespace artefacts)
|
||
return urllib.parse.urlunparse(parts)
|
||
|
||
|
||
# ─── Bulk Import Dialog ────────────────────────────────────────────────────────
|
||
|
||
class BulkImportDialog(tk.Toplevel):
|
||
"""
|
||
Modal dialog for bulk-importing websites from a CSV or Excel file.
|
||
|
||
Workflow:
|
||
1. User picks a .csv or .xlsx/.xls file (or downloads the template).
|
||
2. File is parsed and previewed in a treeview (up to 200 rows shown).
|
||
3. User confirms → rows are inserted via create_website(); duplicates skipped.
|
||
|
||
Expected columns (case-insensitive, order-independent):
|
||
name * — website display name (required)
|
||
url * — full URL (required)
|
||
check_type — 'daily' or 'weekly' (default: daily)
|
||
note — optional description
|
||
visibility — 'all' or 'assigned' (default: all)
|
||
"""
|
||
|
||
_TEMPLATE_PATH = "website_import_template.xlsx"
|
||
_REQUIRED_COLS = {"name", "url"}
|
||
_ALLOWED_TYPES = {"daily", "weekly"}
|
||
_ALLOWED_VIS = {"all", "assigned"}
|
||
|
||
def __init__(self, parent, current_user: dict, on_complete):
|
||
super().__init__(parent)
|
||
self.current_user = current_user
|
||
self.on_complete = on_complete
|
||
self._rows: list = [] # parsed preview rows
|
||
|
||
self.title("Import Websites")
|
||
self.configure(bg=COLOURS["bg"])
|
||
self.resizable(True, True)
|
||
self.grab_set()
|
||
self._build_ui()
|
||
self._centre()
|
||
|
||
def _centre(self):
|
||
self.update_idletasks()
|
||
w, h = 780, 540
|
||
x = (self.winfo_screenwidth() - w) // 2
|
||
y = (self.winfo_screenheight() - h) // 2
|
||
self.geometry(f"{w}x{h}+{x}+{y}")
|
||
|
||
# ── UI ────────────────────────────────────────────────────────────────────
|
||
|
||
def _build_ui(self):
|
||
C = COLOURS
|
||
|
||
ttk.Label(self, text="Import Websites",
|
||
style="Heading.TLabel").pack(anchor="w", padx=24, pady=(20, 4))
|
||
|
||
# ── File picker row ───────────────────────────────────────────────────
|
||
picker = tk.Frame(self, bg=C["bg"])
|
||
picker.pack(fill="x", padx=24, pady=(0, 8))
|
||
|
||
tk.Label(picker, text="File:", bg=C["bg"], fg=C["text"],
|
||
font=FONT_SMALL).pack(side="left")
|
||
|
||
self._file_var = tk.StringVar()
|
||
tk.Entry(picker, textvariable=self._file_var, state="readonly",
|
||
readonlybackground=C["surface2"], fg=C["text"],
|
||
relief="flat", width=50, font=FONT_SMALL).pack(
|
||
side="left", padx=(6, 6), fill="x", expand=True)
|
||
|
||
ttk.Button(picker, text="Browse…",
|
||
command=self._browse).pack(side="left", padx=(0, 6))
|
||
|
||
ttk.Button(picker, text="⬇ Download Template",
|
||
style="Ghost.TButton",
|
||
command=self._download_template).pack(side="left")
|
||
|
||
# ── Status label ──────────────────────────────────────────────────────
|
||
self._status_lbl = tk.Label(
|
||
self, text="Select a .xlsx or .csv file to preview.",
|
||
bg=C["bg"], fg=C["text_dim"], font=FONT_SMALL, anchor="w")
|
||
self._status_lbl.pack(fill="x", padx=24, pady=(0, 6))
|
||
|
||
# ── Preview treeview ──────────────────────────────────────────────────
|
||
tree_frame = tk.Frame(self, bg=C["bg"])
|
||
tree_frame.pack(fill="both", expand=True, padx=24)
|
||
|
||
cols = ("name", "url", "check_type", "note", "visibility", "status")
|
||
widths = [150, 210, 80, 150, 80, 110]
|
||
self._tree = ttk.Treeview(tree_frame, columns=cols,
|
||
show="headings", selectmode="none", height=12)
|
||
for col, w in zip(cols, widths):
|
||
self._tree.heading(col, text=col.capitalize())
|
||
self._tree.column(col, width=w, anchor="w")
|
||
self._tree.tag_configure("skip", foreground=C["text_dim"])
|
||
self._tree.tag_configure("valid", foreground=C["text"])
|
||
self._tree.tag_configure("dup_url", foreground=C["warning"])
|
||
self._tree.tag_configure("dup_name", foreground=C["warning"])
|
||
|
||
vsb = ttk.Scrollbar(tree_frame, orient="vertical",
|
||
command=self._tree.yview)
|
||
self._tree.configure(yscrollcommand=vsb.set)
|
||
vsb.pack(side="right", fill="y")
|
||
self._tree.pack(side="left", fill="both", expand=True)
|
||
|
||
# ── Footer buttons ────────────────────────────────────────────────────
|
||
ttk.Separator(self, orient="horizontal").pack(fill="x", padx=0, pady=8)
|
||
btn_row = ttk.Frame(self)
|
||
btn_row.pack(fill="x", padx=24, pady=(0, 16))
|
||
|
||
self._import_btn = ttk.Button(
|
||
btn_row, text="Import",
|
||
command=self._do_import, state="disabled")
|
||
self._import_btn.pack(side="right", padx=(6, 0))
|
||
ttk.Button(btn_row, text="Cancel", style="Ghost.TButton",
|
||
command=self.destroy).pack(side="right")
|
||
|
||
# ── File handling ─────────────────────────────────────────────────────────
|
||
|
||
def _browse(self):
|
||
from tkinter import filedialog
|
||
path = filedialog.askopenfilename(
|
||
title="Select import file",
|
||
filetypes=[
|
||
("Spreadsheets", "*.xlsx *.xls *.csv"),
|
||
("Excel", "*.xlsx *.xls"),
|
||
("CSV", "*.csv"),
|
||
("All files", "*.*"),
|
||
],
|
||
)
|
||
if not path:
|
||
return
|
||
self._file_var.set(path)
|
||
self._parse_file(path)
|
||
|
||
def _parse_file(self, path: str):
|
||
"""Parse the selected file and populate the preview treeview."""
|
||
self._tree.delete(*self._tree.get_children())
|
||
self._rows.clear()
|
||
self._import_btn.config(state="disabled")
|
||
|
||
try:
|
||
rows = self._read_file(path)
|
||
except Exception as e:
|
||
self._status_lbl.config(
|
||
text=f"Error reading file: {e}", fg=COLOURS["danger"])
|
||
return
|
||
|
||
if not rows:
|
||
self._status_lbl.config(
|
||
text="No data rows found.", fg=COLOURS["warning"])
|
||
return
|
||
|
||
# Fetch existing URLs and names once for the entire preview pass
|
||
try:
|
||
from models import get_existing_website_urls, get_existing_website_names
|
||
existing_urls = get_existing_website_urls()
|
||
existing_names = get_existing_website_names()
|
||
except Exception:
|
||
existing_urls = set()
|
||
existing_names = set()
|
||
|
||
valid_count = 0
|
||
skip_count = 0
|
||
dup_count = 0
|
||
for row in rows[:200]: # preview cap
|
||
name = (row.get("name") or "").strip()
|
||
url = (row.get("url") or "").strip()
|
||
ct = (row.get("check_type") or "daily").strip().lower()
|
||
note = (row.get("note") or "").strip()
|
||
vis = (row.get("visibility") or "all").strip().lower()
|
||
|
||
# Normalise / default
|
||
if ct not in self._ALLOWED_TYPES:
|
||
ct = "daily"
|
||
if vis not in self._ALLOWED_VIS:
|
||
vis = "all"
|
||
|
||
# Classify the row
|
||
skip = not name or not url
|
||
dup_url = not skip and url.lower() in existing_urls
|
||
dup_name = not skip and not dup_url and name.lower() in existing_names
|
||
|
||
if skip:
|
||
tag = "skip"
|
||
status = "⚠ Missing field"
|
||
skip_count += 1
|
||
elif dup_url:
|
||
tag = "dup_url"
|
||
status = "⚠ Duplicate URL"
|
||
dup_count += 1
|
||
elif dup_name:
|
||
tag = "dup_name"
|
||
status = "⚠ Duplicate name"
|
||
dup_count += 1
|
||
else:
|
||
tag = "valid"
|
||
status = "✔ Will import"
|
||
valid_count += 1
|
||
|
||
self._tree.insert("", "end", tags=(tag,),
|
||
values=(name, url, ct, note[:50], vis, status))
|
||
self._rows.append({
|
||
"name": name, "url": url,
|
||
"check_type": ct, "note": note, "visibility": vis,
|
||
"_skip": skip or dup_url or dup_name,
|
||
})
|
||
|
||
total = len(rows)
|
||
shown = min(total, 200)
|
||
more = f" ({total - shown} more not shown)" if total > 200 else ""
|
||
parts = [f"✔ {valid_count} will import"]
|
||
if dup_count:
|
||
parts.append(f"⚠ {dup_count} duplicate(s) skipped")
|
||
if skip_count:
|
||
parts.append(f"✕ {skip_count} missing required field(s)")
|
||
self._status_lbl.config(
|
||
text="Preview: " + ", ".join(parts) + more + ". "
|
||
"(Duplicates are highlighted in orange.)",
|
||
fg=COLOURS["text_dim"],
|
||
)
|
||
if valid_count:
|
||
self._import_btn.config(state="normal")
|
||
|
||
def _read_file(self, path: str) -> list:
|
||
"""Return list of dicts from CSV or Excel. Header row normalised to lowercase."""
|
||
import os
|
||
ext = os.path.splitext(path)[1].lower()
|
||
if ext == ".csv":
|
||
return self._read_csv(path)
|
||
elif ext in (".xlsx", ".xls"):
|
||
return self._read_excel(path)
|
||
else:
|
||
raise ValueError(f"Unsupported file type: {ext}")
|
||
|
||
def _read_csv(self, path: str) -> list:
|
||
import csv
|
||
rows = []
|
||
with open(path, newline="", encoding="utf-8-sig") as fh:
|
||
reader = csv.DictReader(fh)
|
||
for row in reader:
|
||
rows.append({k.strip().lower(): v for k, v in row.items()})
|
||
return rows
|
||
|
||
def _read_excel(self, path: str) -> list:
|
||
try:
|
||
import openpyxl
|
||
except ImportError:
|
||
raise ImportError(
|
||
"openpyxl is required for Excel import.\n"
|
||
"Install it with: pip install openpyxl"
|
||
)
|
||
wb = openpyxl.load_workbook(path, read_only=True, data_only=True)
|
||
ws = wb.active
|
||
rows_iter = ws.iter_rows(values_only=True)
|
||
headers = None
|
||
result = []
|
||
for row in rows_iter:
|
||
# Skip completely empty rows
|
||
if all(v is None for v in row):
|
||
continue
|
||
if headers is None:
|
||
headers = [str(c).strip().lower() if c is not None else "" for c in row]
|
||
continue
|
||
row_dict = {headers[i]: (str(v).strip() if v is not None else "")
|
||
for i, v in enumerate(row) if i < len(headers)}
|
||
result.append(row_dict)
|
||
wb.close()
|
||
return result
|
||
|
||
# ── Import ────────────────────────────────────────────────────────────────
|
||
|
||
def _do_import(self):
|
||
"""Insert all valid, non-duplicate rows via create_website().
|
||
|
||
Duplicate detection is done in the application layer by fetching
|
||
existing URLs and names from the DB immediately before inserting.
|
||
This is necessary because there is no UNIQUE constraint on the
|
||
websites table, so relying on a DB error (1062) would not work.
|
||
"""
|
||
from models import (
|
||
create_website,
|
||
get_existing_website_urls,
|
||
get_existing_website_names,
|
||
)
|
||
|
||
to_import = [r for r in self._rows if not r["_skip"]]
|
||
if not to_import:
|
||
show_error("No valid rows to import.\n\n"
|
||
"All rows are either missing required fields "
|
||
"or are duplicates of existing websites.")
|
||
return
|
||
|
||
# Re-fetch existing state at import time — user may have been on
|
||
# the preview screen for a while; another admin could have added
|
||
# a site in the interim.
|
||
try:
|
||
existing_urls = get_existing_website_urls()
|
||
existing_names = get_existing_website_names()
|
||
except Exception as e:
|
||
show_error(f"Could not verify existing websites:\n{e}")
|
||
return
|
||
|
||
inserted = 0
|
||
dup_url = 0
|
||
dup_name = 0
|
||
errors = []
|
||
|
||
for r in to_import:
|
||
url_key = r["url"].strip().lower()
|
||
name_key = r["name"].strip().lower()
|
||
|
||
if url_key in existing_urls:
|
||
dup_url += 1
|
||
logger.info(f"[IMPORT] Skipped — duplicate URL: {r['url']}")
|
||
continue
|
||
|
||
if name_key in existing_names:
|
||
dup_name += 1
|
||
logger.info(f"[IMPORT] Skipped — duplicate name: {r['name']}")
|
||
continue
|
||
|
||
try:
|
||
create_website(
|
||
admin_id=self.current_user["id"],
|
||
name=r["name"],
|
||
url=r["url"],
|
||
check_type=r["check_type"],
|
||
note=r["note"],
|
||
credentials=[],
|
||
visibility=r["visibility"],
|
||
)
|
||
# Track inserted URL/name so intra-file duplicates are
|
||
# also caught (e.g. the same URL appears twice in the CSV).
|
||
existing_urls.add(url_key)
|
||
existing_names.add(name_key)
|
||
inserted += 1
|
||
logger.info(
|
||
f"[IMPORT] Website '{r['name']}' ({r['url']}) imported "
|
||
f"by admin_id={self.current_user['id']}."
|
||
)
|
||
except Exception as e:
|
||
errors.append(f"{r['name']}: {e}")
|
||
logger.error(f"[IMPORT] Failed to import '{r['name']}': {e}")
|
||
|
||
msg = f"Import complete.\n\n✔ {inserted} website(s) imported."
|
||
if dup_url:
|
||
msg += f"\n⚠ {dup_url} row(s) skipped — URL already exists."
|
||
if dup_name:
|
||
msg += f"\n⚠ {dup_name} row(s) skipped — name already exists."
|
||
if errors:
|
||
msg += f"\n✕ {len(errors)} error(s):\n" + "\n".join(errors[:5])
|
||
show_info(msg)
|
||
self.on_complete()
|
||
self.destroy()
|
||
|
||
# ── Template download ─────────────────────────────────────────────────────
|
||
|
||
def _download_template(self):
|
||
import os, shutil
|
||
from tkinter import filedialog
|
||
src = self._TEMPLATE_PATH
|
||
if not os.path.exists(src):
|
||
show_error(
|
||
"Template file not found.\n"
|
||
f"Expected at: {os.path.abspath(src)}"
|
||
)
|
||
return
|
||
dest = filedialog.asksaveasfilename(
|
||
title="Save import template as…",
|
||
initialfile="website_import_template.xlsx",
|
||
defaultextension=".xlsx",
|
||
filetypes=[("Excel Workbook", "*.xlsx")],
|
||
)
|
||
if not dest:
|
||
return
|
||
shutil.copy2(src, dest)
|
||
logger.info(f"Import template downloaded to: {dest}")
|
||
show_info(f"Template saved to:\n{dest}") |