nngo/WebChecker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

04/24 Fixed bulk import check for duplicate

Browse Source
This commit is contained in:
nngo
2026-04-24 14:26:07 -04:00
parent 88b5e85c5b
commit f2a8d3f15b
4 changed files with 170 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
views/admin_shifts_view.py
+39 -7
View File
@@ -450,12 +450,26 @@ class ShiftDialog(tk.Toplevel):
)
self._user_picker.grid(row=0, column=0, sticky="nsew", pady=(0, 8))
# Websites picker
# Websites picker — include check_type suffix in label and colour map
_TYPE_SUFFIX = {"daily": " [D]", "weekly": " [W]"}
_TYPE_COLOUR = {
"daily": COLOURS.get("accent", "#5B4DE8"),
"weekly": COLOURS.get("warning", "#F57C00"),
}
site_items = [
(w["id"], w["name"] + _TYPE_SUFFIX.get(w.get("check_type", "daily"), ""))
for w in self._all_websites
]
site_colours = {
w["id"]: _TYPE_COLOUR.get(w.get("check_type", "daily"), COLOURS["text"])
for w in self._all_websites
}
self._site_picker = _DualListPicker(
pane,
title="Assigned Websites",
all_items=[(w["id"], w["name"]) for w in self._all_websites],
all_items=site_items,
allow_reorder=True,
item_colours=site_colours,
)
self._site_picker.grid(row=1, column=0, sticky="nsew")
@@ -541,11 +555,13 @@ class _DualListPicker(ttk.Frame):
Supports optional drag-to-reorder on the assigned list.
"""
def __init__(self, parent, title: str, all_items: list,
allow_reorder=False):
allow_reorder=False, item_colours: dict = None):
super().__init__(parent)
self._all_items = all_items # [(id, label), ...]
self._allow_reorder = allow_reorder
self._drag_start = None
# item_colours: {id: colour_str} — applied per-item after refresh
self._item_colours = item_colours or {}
self._build(title)
def _build(self, title: str):
@@ -555,7 +571,19 @@ class _DualListPicker(ttk.Frame):
ttk.Label(self, text=title,
style="Heading.TLabel").grid(
row=0, column=0, columnspan=3, sticky="w", pady=(4, 6))
row=0, column=0, columnspan=3, sticky="w", pady=(4, 2))
# Colour legend — only shown when item_colours are provided
if self._item_colours:
legend = tk.Frame(self, bg=COLOURS["bg"])
legend.grid(row=0, column=0, columnspan=3, sticky="e", pady=(4, 2))
for lbl_text, colour in [
("● Daily", COLOURS.get("accent", "#5B4DE8")),
("● Weekly", COLOURS.get("warning", "#F57C00")),
]:
tk.Label(legend, text=lbl_text,
bg=COLOURS["bg"], fg=colour,
font=FONT_SMALL).pack(side="left", padx=(0, 10))
# ── Available list ────────────────────────────────────────────────────
avail_frame = tk.Frame(self, bg=COLOURS["surface"])
@@ -632,12 +660,16 @@ class _DualListPicker(ttk.Frame):
def _refresh_listboxes(self):
self._avail_lb.delete(0, "end")
for _, lbl in self._avail_data:
for idx, (id_, lbl) in enumerate(self._avail_data):
self._avail_lb.insert("end", lbl)
if id_ in self._item_colours:
self._avail_lb.itemconfig(idx, fg=self._item_colours[id_])
self._assign_lb.delete(0, "end")
for _, lbl in self._assign_data:
for idx, (id_, lbl) in enumerate(self._assign_data):
self._assign_lb.insert("end", lbl)
if id_ in self._item_colours:
self._assign_lb.itemconfig(idx, fg=self._item_colours[id_])
def _add(self):
sel = list(self._avail_lb.curselection())
@@ -747,4 +779,4 @@ def _valid_time(s: str) -> bool:
h, m = int(parts[0]), int(parts[1])
return 0 <= h <= 23 and 0 <= m <= 59
except ValueError:
return False
return False
views/admin_websites_view.py
+95 -30
View File
@@ -608,15 +608,17 @@ class BulkImportDialog(tk.Toplevel):
tree_frame = tk.Frame(self, bg=C["bg"])
tree_frame.pack(fill="both", expand=True, padx=24)
cols = ("name", "url", "check_type", "note", "visibility")
widths = [160, 240, 90, 180, 90]
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)
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("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)
@@ -672,8 +674,18 @@ class BulkImportDialog(tk.Toplevel):
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()
@@ -687,27 +699,47 @@ class BulkImportDialog(tk.Toplevel):
if vis not in self._ALLOWED_VIS:
vis = "all"
skip = not name or not url
tag = "skip" if skip else "valid"
# 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[:60], vis))
values=(name, url, ct, note[:50], vis, status))
self._rows.append({
"name": name, "url": url,
"check_type": ct, "note": note, "visibility": vis,
"_skip": skip,
"_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=(f"Preview: {valid_count} valid row(s), "
f"{skip_count} skipped (missing name/url).{more}"),
text="Preview: " + ", ".join(parts) + more + ". "
"(Duplicates are highlighted in orange.)",
fg=COLOURS["text_dim"],
)
if valid_count:
@@ -762,20 +794,55 @@ class BulkImportDialog(tk.Toplevel):
# ── Import ────────────────────────────────────────────────────────────────
def _do_import(self):
"""Insert all valid rows via create_website(). Skips duplicates by URL."""
from models import create_website
import re
"""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.")
show_error("No valid rows to import.\n\n"
"All rows are either missing required fields "
"or are duplicates of existing websites.")
return
inserted = 0
skipped = 0
errors = []
# 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"],
@@ -786,26 +853,24 @@ class BulkImportDialog(tk.Toplevel):
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:
err_str = str(e)
# Duplicate entry (MySQL error 1062) → count as skip, not error
if "1062" in err_str or "Duplicate" in err_str:
skipped += 1
logger.info(
f"[IMPORT] Skipped duplicate URL: {r['url']}")
else:
errors.append(f"{r['name']}: {e}")
logger.error(
f"[IMPORT] Failed to import '{r['name']}': {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 skipped:
msg += f"\n⏭ {skipped} duplicate URL(s) skipped."
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)