67e35c298a
- NC User and NC Pass are now required fields when adding a site - Auto-provision Nextcloud user on site creation - Client-side validation enforces all fields in both Add and Edit forms - Updated placeholder text to reflect required status - Sites without NC creds now show Not set badge instead of Global Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1142 lines
45 KiB
Python
1142 lines
45 KiB
Python
##############################################################################
|
||
# APtool - AP Installation Data Collection Web App
|
||
#
|
||
# A mobile-friendly Flask web application for collecting access point (AP)
|
||
# installation data at job sites. Technicians log in with a PIN and 4-digit
|
||
# site number, then submit AP details (location, serial number, cable length)
|
||
# along with 6 required photos per AP.
|
||
#
|
||
# Data is saved locally and synced to Nextcloud via WebDAV:
|
||
# Local: uploads/{site}/ — photos + data.xlsx
|
||
# Nextcloud: APtool/{site}/uploads/ — photos
|
||
# APtool/{site}/data.xlsx — spreadsheet
|
||
#
|
||
# Configuration is via environment variables (with defaults):
|
||
# NC_URL, NC_USER, NC_PASS — Nextcloud connection
|
||
# NC_FOLDER — Nextcloud root folder (default: "APtool")
|
||
# sites.conf — per-site PIN + optional NC creds (site:pin[:nc_user:nc_pass])
|
||
# SECRET_KEY — Flask session secret (auto-generated if unset)
|
||
##############################################################################
|
||
|
||
import os
|
||
import re
|
||
import glob
|
||
import logging
|
||
import secrets
|
||
from pathlib import Path
|
||
|
||
import requests
|
||
from flask import (
|
||
Flask, render_template, request, jsonify, session, redirect, url_for,
|
||
send_file,
|
||
)
|
||
from openpyxl import Workbook, load_workbook
|
||
# Font – controls text style (bold, italic, color, size)
|
||
# PatternFill – controls cell background color
|
||
# Alignment – controls text alignment (left, center, right)
|
||
# These are used in get_or_create_workbook() to style the header row,
|
||
# and in submit() to apply alternating row colors.
|
||
from openpyxl.styles import Font, PatternFill, Alignment
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Flask app initialization
|
||
# ---------------------------------------------------------------------------
|
||
# secret_key is required for session cookies. If SECRET_KEY env var is not
|
||
# set, a random key is generated at startup (sessions won't survive restarts).
|
||
app = Flask(__name__)
|
||
app.secret_key = os.environ.get("SECRET_KEY", secrets.token_hex(32))
|
||
log = logging.getLogger(__name__)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Filesystem paths
|
||
# ---------------------------------------------------------------------------
|
||
# BASE_DIR — the directory containing this script (/home/kamaji/aptool/)
|
||
# UPLOAD_DIR — root directory for all site uploads; each site gets a
|
||
# subdirectory named by its 4-digit site number, e.g.
|
||
# uploads/5001/AP010.jpg, uploads/5001/data.xlsx
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
UPLOAD_DIR = BASE_DIR / "uploads"
|
||
|
||
# Create the uploads directory if it doesn't already exist
|
||
UPLOAD_DIR.mkdir(exist_ok=True)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Nextcloud configuration
|
||
# ---------------------------------------------------------------------------
|
||
# NC_URL — base URL of the Nextcloud instance (e.g. https://cloud.example.com)
|
||
# NC_USER — Nextcloud username for WebDAV authentication
|
||
# NC_PASS — Nextcloud app password (generate at Settings > Security > App passwords)
|
||
# NC_FOLDER — top-level folder in Nextcloud where site data is stored;
|
||
# each site gets a subfolder: APtool/5001/uploads/, APtool/5001/data.xlsx
|
||
NC_URL = os.environ.get("NC_URL", "")
|
||
NC_USER = os.environ.get("NC_USER", "")
|
||
NC_PASS = os.environ.get("NC_PASS", "")
|
||
NC_FOLDER = os.environ.get("NC_FOLDER", "APtool")
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Site authentication — loaded from sites.conf
|
||
# ---------------------------------------------------------------------------
|
||
# Each site has its own PIN, defined in sites.conf (one per line, format:
|
||
# site_number:pin[:nc_user:nc_pass]).
|
||
# The optional nc_user:nc_pass fields allow per-site Nextcloud credentials.
|
||
# If omitted, the global NC_USER / NC_PASS defaults are used.
|
||
# The file is re-read on every login attempt so that an admin can add/remove
|
||
# sites without restarting the app.
|
||
#
|
||
# Example sites.conf:
|
||
# 5001:1234 (uses global NC credentials)
|
||
# 5002:5678:alice:AppPass-12345 (uses per-site NC user "alice")
|
||
SITES_FILE = BASE_DIR / "sites.conf"
|
||
|
||
|
||
def load_sites() -> dict[str, dict]:
|
||
"""Read sites.conf and return a dict mapping site number → site config.
|
||
|
||
Each value is a dict with keys:
|
||
pin — the site's login PIN (always present)
|
||
nc_user — per-site Nextcloud username (None → use global NC_USER)
|
||
nc_pass — per-site Nextcloud password (None → use global NC_PASS)
|
||
|
||
Skips blank lines and lines starting with #.
|
||
If the file doesn't exist, returns an empty dict (no sites can log in).
|
||
Re-read on every call so edits take effect immediately.
|
||
"""
|
||
sites = {}
|
||
if not SITES_FILE.exists():
|
||
log.warning("sites.conf not found — no sites can log in")
|
||
return sites
|
||
for line in SITES_FILE.read_text().splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
if ":" not in line:
|
||
continue
|
||
parts = line.split(":", maxsplit=3)
|
||
site_num = parts[0].strip()
|
||
pin = parts[1].strip() if len(parts) > 1 else ""
|
||
nc_user = parts[2].strip() if len(parts) > 2 else None
|
||
nc_pass = parts[3].strip() if len(parts) > 3 else None
|
||
if site_num and pin:
|
||
sites[site_num] = {
|
||
"pin": pin,
|
||
"nc_user": nc_user or None,
|
||
"nc_pass": nc_pass or None,
|
||
}
|
||
return sites
|
||
|
||
|
||
def save_sites(sites: dict):
|
||
"""Write the sites dict back to sites.conf with the standard header.
|
||
|
||
Sorts sites numerically by site number. Only appends :nc_user:nc_pass
|
||
when both are present.
|
||
|
||
Args:
|
||
sites: dict mapping site number → {pin, nc_user, nc_pass}
|
||
"""
|
||
header = (
|
||
"# APtool Site Configuration\n"
|
||
"#\n"
|
||
"# Each line defines a site that technicians can log into.\n"
|
||
"# Format: site_number:pin[:nc_user:nc_pass]\n"
|
||
"#\n"
|
||
"# - site_number must be exactly 4 digits\n"
|
||
"# - pin can be any string (digits recommended for mobile entry)\n"
|
||
"# - nc_user and nc_pass are optional Nextcloud credentials for this site\n"
|
||
"# If omitted, the global NC_USER / NC_PASS defaults are used.\n"
|
||
"# - blank lines and lines starting with # are ignored\n"
|
||
"#\n"
|
||
"# Examples:\n"
|
||
"# 5001:1234 (uses global Nextcloud credentials)\n"
|
||
"# 5002:5678:alice:AppPass-12345 (uses per-site Nextcloud user \"alice\")\n"
|
||
"# 9999:0000:bob:AppPass-67890 (uses per-site Nextcloud user \"bob\")\n"
|
||
"#\n"
|
||
"# To add a site: add a new line with site_number:pin[:nc_user:nc_pass]\n"
|
||
"# To remove a site: delete or comment out the line\n"
|
||
"# To change a PIN: edit the pin after the colon\n"
|
||
"#\n"
|
||
"# The app reloads this file on every login attempt, so changes\n"
|
||
"# take effect immediately — no restart needed.\n"
|
||
"\n"
|
||
)
|
||
lines = []
|
||
for site_num in sorted(sites.keys(), key=lambda s: int(s) if s.isdigit() else 0):
|
||
cfg = sites[site_num]
|
||
line = f"{site_num}:{cfg['pin']}"
|
||
if cfg.get("nc_user") and cfg.get("nc_pass"):
|
||
line += f":{cfg['nc_user']}:{cfg['nc_pass']}"
|
||
lines.append(line)
|
||
SITES_FILE.write_text(header + "\n".join(lines) + "\n")
|
||
|
||
|
||
def get_nc_creds(site: str) -> tuple[str, str]:
|
||
"""Return the (nc_user, nc_pass) to use for a given site.
|
||
|
||
Checks sites.conf for per-site Nextcloud credentials. If the site has
|
||
its own nc_user and nc_pass, those are returned. Otherwise, the global
|
||
NC_USER / NC_PASS defaults are used.
|
||
|
||
Args:
|
||
site: 4-digit site number (e.g. "5001")
|
||
|
||
Returns:
|
||
(username, password) tuple for Nextcloud authentication
|
||
"""
|
||
sites = load_sites()
|
||
site_cfg = sites.get(site, {})
|
||
user = site_cfg.get("nc_user") or NC_USER
|
||
passwd = site_cfg.get("nc_pass") or NC_PASS
|
||
return user, passwd
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Photo field definitions
|
||
# ---------------------------------------------------------------------------
|
||
# Each tuple is (form_field_name, filename_suffix).
|
||
# For AP number "010", the resulting filenames would be:
|
||
# photo_ap → AP010.jpg (close-up of the AP)
|
||
# photo_far → AP010F.jpg (distant/wide shot of the AP)
|
||
# photo_length → AP010_length.jpg (cable length measurement)
|
||
# photo_cont → AP010_cont.jpg (continuity test result)
|
||
# photo_rate → AP010_rate.jpg (speed test result)
|
||
# photo_box → AP010_box.jpg (box label / packaging)
|
||
PHOTO_FIELDS = [
|
||
("photo_ap", ""), # APxxx
|
||
("photo_far", "F"), # APxxxF
|
||
("photo_length", "_length"),# APxxx_length
|
||
("photo_cont", "_cont"), # APxxx_cont
|
||
("photo_rate", "_rate"), # APxxx_rate
|
||
("photo_box", "_box"), # APxxx_box
|
||
]
|
||
|
||
# Maps URL-friendly suffix names → file suffixes used in PHOTO_FIELDS.
|
||
# Used by the /photo routes to translate e.g. "close" → "" (AP027.jpg).
|
||
SUFFIX_MAP = {
|
||
"close": "",
|
||
"far": "F",
|
||
"length": "_length",
|
||
"cont": "_cont",
|
||
"rate": "_rate",
|
||
"box": "_box",
|
||
}
|
||
|
||
# Human-readable labels for each photo suffix (used in the entries template)
|
||
SUFFIX_LABELS = {
|
||
"close": "Close-up",
|
||
"far": "Far shot",
|
||
"length": "Cable length",
|
||
"cont": "Continuity",
|
||
"rate": "Speed test",
|
||
"box": "Box label",
|
||
}
|
||
|
||
|
||
def find_photo(site: str, ap_number: str, file_suffix: str) -> Path | None:
|
||
"""Find a photo file regardless of extension.
|
||
|
||
Globs for AP{ap_number}{file_suffix}.* in the site's upload directory.
|
||
Returns the first match as a Path, or None if no file is found.
|
||
|
||
Args:
|
||
site: 4-digit site number (e.g. "1102")
|
||
ap_number: 3-digit AP number (e.g. "027")
|
||
file_suffix: filename suffix (e.g. "" for close-up, "F" for far)
|
||
"""
|
||
site_dir = UPLOAD_DIR / site
|
||
pattern = str(site_dir / f"AP{ap_number}{file_suffix}.*")
|
||
matches = glob.glob(pattern)
|
||
if matches:
|
||
return Path(matches[0])
|
||
return None
|
||
|
||
|
||
##############################################################################
|
||
# Nextcloud OCS Provisioning API helpers
|
||
##############################################################################
|
||
# The OCS Provisioning API allows admins to create and manage users:
|
||
# https://<host>/ocs/v1.php/cloud/users
|
||
# All OCS calls require the header OCS-APIRequest: true and are
|
||
# authenticated with the global admin credentials (NC_USER / NC_PASS).
|
||
# A successful response has ocs.meta.statuscode == 100.
|
||
##############################################################################
|
||
|
||
def nc_user_exists(username: str) -> bool:
|
||
"""Check whether a user account exists on Nextcloud.
|
||
|
||
Uses the OCS Provisioning API to query user details.
|
||
Authenticated with the global admin credentials.
|
||
|
||
Args:
|
||
username: The Nextcloud user ID to check (e.g. "alice")
|
||
|
||
Returns:
|
||
True if the user exists, False otherwise (including on errors).
|
||
"""
|
||
url = f"{NC_URL.rstrip('/')}/ocs/v1.php/cloud/users/{username}?format=json"
|
||
try:
|
||
resp = requests.get(
|
||
url,
|
||
auth=(NC_USER, NC_PASS),
|
||
headers={"OCS-APIRequest": "true"},
|
||
timeout=30,
|
||
)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
return data.get("ocs", {}).get("meta", {}).get("statuscode") == 100
|
||
except Exception:
|
||
log.exception("Failed to check if NC user %s exists", username)
|
||
return False
|
||
|
||
|
||
def nc_create_user(username: str, password: str) -> bool:
|
||
"""Create a new user account on Nextcloud.
|
||
|
||
Uses the OCS Provisioning API. Authenticated with the global admin
|
||
credentials (NC_USER / NC_PASS must be a Nextcloud admin account).
|
||
|
||
Args:
|
||
username: The user ID to create (e.g. "alice")
|
||
password: The password for the new account
|
||
|
||
Returns:
|
||
True on success (statuscode 100), False on failure.
|
||
"""
|
||
url = f"{NC_URL.rstrip('/')}/ocs/v1.php/cloud/users?format=json"
|
||
try:
|
||
resp = requests.post(
|
||
url,
|
||
auth=(NC_USER, NC_PASS),
|
||
headers={"OCS-APIRequest": "true"},
|
||
data={"userid": username, "password": password},
|
||
timeout=30,
|
||
)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
status = data.get("ocs", {}).get("meta", {}).get("statuscode")
|
||
if status == 100:
|
||
return True
|
||
msg = data.get("ocs", {}).get("meta", {}).get("message", "unknown error")
|
||
log.error("Failed to create NC user %s: %s (status %s)", username, msg, status)
|
||
else:
|
||
log.error("Failed to create NC user %s: HTTP %s", username, resp.status_code)
|
||
except Exception:
|
||
log.exception("Failed to create NC user %s", username)
|
||
return False
|
||
|
||
|
||
def nc_change_password(username: str, new_password: str) -> bool:
|
||
"""Change the password for an existing Nextcloud user.
|
||
|
||
Uses the OCS Provisioning API. Authenticated with the global admin
|
||
credentials (NC_USER / NC_PASS must be a Nextcloud admin account).
|
||
|
||
Args:
|
||
username: The Nextcloud user ID whose password to change
|
||
new_password: The new password to set
|
||
|
||
Returns:
|
||
True on success (statuscode 100), False on failure.
|
||
"""
|
||
url = f"{NC_URL.rstrip('/')}/ocs/v1.php/cloud/users/{username}?format=json"
|
||
try:
|
||
resp = requests.put(
|
||
url,
|
||
auth=(NC_USER, NC_PASS),
|
||
headers={"OCS-APIRequest": "true"},
|
||
data={"key": "password", "value": new_password},
|
||
timeout=30,
|
||
)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
status = data.get("ocs", {}).get("meta", {}).get("statuscode")
|
||
if status == 100:
|
||
return True
|
||
msg = data.get("ocs", {}).get("meta", {}).get("message", "unknown error")
|
||
log.error("Failed to change NC password for %s: %s (status %s)", username, msg, status)
|
||
else:
|
||
log.error("Failed to change NC password for %s: HTTP %s", username, resp.status_code)
|
||
except Exception:
|
||
log.exception("Failed to change NC password for %s", username)
|
||
return False
|
||
|
||
|
||
def nc_provision_site_users():
|
||
"""Ensure all per-site Nextcloud users from sites.conf exist.
|
||
|
||
Iterates over every site in sites.conf that has per-site nc_user/nc_pass.
|
||
For each one, checks whether the user already exists on Nextcloud;
|
||
if not, creates the account using the OCS Provisioning API.
|
||
|
||
Uses the global admin credentials (NC_USER / NC_PASS) for the API calls.
|
||
The per-site nc_pass becomes the new user's password.
|
||
|
||
Called at app startup. To pick up newly added sites, restart the service.
|
||
"""
|
||
if not all([NC_URL, NC_USER, NC_PASS]):
|
||
log.info("Nextcloud not configured — skipping user provisioning")
|
||
return
|
||
|
||
sites = load_sites()
|
||
for site_num, cfg in sites.items():
|
||
nc_user = cfg.get("nc_user")
|
||
nc_pass = cfg.get("nc_pass")
|
||
if not nc_user or not nc_pass:
|
||
continue # no per-site credentials — uses global, skip
|
||
|
||
if nc_user_exists(nc_user):
|
||
print(f" Site {site_num}: NC user '{nc_user}' exists ✓")
|
||
else:
|
||
print(f" Site {site_num}: NC user '{nc_user}' not found — creating...")
|
||
if nc_create_user(nc_user, nc_pass):
|
||
print(f" Site {site_num}: NC user '{nc_user}' created ✓")
|
||
else:
|
||
print(f" Site {site_num}: FAILED to create NC user '{nc_user}'")
|
||
|
||
|
||
##############################################################################
|
||
# Nextcloud WebDAV helpers
|
||
##############################################################################
|
||
# Nextcloud exposes a WebDAV endpoint at:
|
||
# https://<host>/remote.php/dav/files/<username>/
|
||
# Files are uploaded with PUT, directories created with MKCOL.
|
||
# All Nextcloud errors are logged but treated as non-fatal — local saves
|
||
# always succeed even if Nextcloud is unreachable.
|
||
##############################################################################
|
||
|
||
def nc_enabled(nc_user: str, nc_pass: str) -> bool:
|
||
"""Check whether Nextcloud credentials are configured.
|
||
|
||
Returns True only if NC_URL and the given user/pass are all non-empty.
|
||
When False, nc_sync() is a no-op.
|
||
|
||
Args:
|
||
nc_user: Nextcloud username (per-site or global)
|
||
nc_pass: Nextcloud password (per-site or global)
|
||
"""
|
||
return all([NC_URL, nc_user, nc_pass])
|
||
|
||
|
||
def nc_webdav_url(path: str, nc_user: str) -> str:
|
||
"""Build the full WebDAV URL for a path relative to the user's files root.
|
||
|
||
The WebDAV path includes the username, so per-site users get their own
|
||
file namespace on the same Nextcloud instance.
|
||
|
||
Args:
|
||
path: Remote path (e.g. "APtool/5001/uploads/AP010.jpg")
|
||
nc_user: Nextcloud username whose files root to use
|
||
|
||
Example:
|
||
nc_webdav_url("APtool/5001/uploads/AP010.jpg", "alice")
|
||
→ "https://nextcloud.example.com/remote.php/dav/files/alice/APtool/5001/uploads/AP010.jpg"
|
||
"""
|
||
base = NC_URL.rstrip("/")
|
||
return f"{base}/remote.php/dav/files/{nc_user}/{path.lstrip('/')}"
|
||
|
||
|
||
def nc_ensure_folder(folder_path: str, nc_user: str, nc_pass: str):
|
||
"""Create a folder (and all parent folders) on Nextcloud via MKCOL.
|
||
|
||
Walks the path segments one by one, issuing MKCOL for each.
|
||
HTTP 201 = created, 405 = already exists — both are acceptable.
|
||
|
||
Args:
|
||
folder_path: Remote folder path (e.g. "APtool/5001/uploads")
|
||
nc_user: Nextcloud username for authentication
|
||
nc_pass: Nextcloud password for authentication
|
||
|
||
Example:
|
||
nc_ensure_folder("APtool/5001/uploads", "alice", "pass")
|
||
→ MKCOL APtool, then MKCOL APtool/5001, then MKCOL APtool/5001/uploads
|
||
"""
|
||
parts = folder_path.strip("/").split("/")
|
||
current = ""
|
||
for part in parts:
|
||
current = f"{current}/{part}" if current else part
|
||
url = nc_webdav_url(current, nc_user)
|
||
resp = requests.request("MKCOL", url, auth=(nc_user, nc_pass), timeout=30)
|
||
# 201 = created, 405 = already exists — both are fine
|
||
if resp.status_code not in (201, 405):
|
||
log.warning("MKCOL %s returned %s", url, resp.status_code)
|
||
|
||
|
||
def nc_upload_file(local_path: Path, remote_path: str, nc_user: str, nc_pass: str):
|
||
"""Upload a single local file to Nextcloud via WebDAV PUT.
|
||
|
||
Args:
|
||
local_path: Path to the file on disk (e.g. uploads/5001/AP010.jpg)
|
||
remote_path: Destination path in Nextcloud (e.g. APtool/5001/uploads/AP010.jpg)
|
||
nc_user: Nextcloud username for authentication
|
||
nc_pass: Nextcloud password for authentication
|
||
|
||
Returns:
|
||
True on success (HTTP 200/201/204), False on failure.
|
||
"""
|
||
url = nc_webdav_url(remote_path, nc_user)
|
||
with open(local_path, "rb") as f:
|
||
resp = requests.put(url, data=f, auth=(nc_user, nc_pass), timeout=120)
|
||
if resp.status_code not in (200, 201, 204):
|
||
log.error("Nextcloud upload failed: PUT %s → %s", url, resp.status_code)
|
||
return False
|
||
return True
|
||
|
||
|
||
def nc_sync(site: str, photo_paths: list[Path], excel_path: Path):
|
||
"""Sync a site's photos and Excel file to Nextcloud.
|
||
|
||
Called after every successful form submission. Looks up per-site Nextcloud
|
||
credentials from sites.conf (falling back to global NC_USER/NC_PASS),
|
||
creates the remote folder structure if needed, uploads all photos into
|
||
the site's uploads/ subfolder, and uploads the updated data.xlsx.
|
||
|
||
All errors are caught and logged — a Nextcloud failure will not cause
|
||
the submission to fail for the technician.
|
||
|
||
Args:
|
||
site: 4-digit site number (e.g. "5001")
|
||
photo_paths: list of local photo file Paths just saved
|
||
excel_path: Path to the site's data.xlsx
|
||
"""
|
||
# Resolve per-site or global Nextcloud credentials
|
||
nc_user, nc_pass = get_nc_creds(site)
|
||
|
||
if not nc_enabled(nc_user, nc_pass):
|
||
return
|
||
|
||
try:
|
||
# Ensure APtool/{site}/uploads/ exists on Nextcloud
|
||
nc_ensure_folder(f"{NC_FOLDER}/{site}/uploads", nc_user, nc_pass)
|
||
|
||
# Upload each photo
|
||
for local_path in photo_paths:
|
||
remote = f"{NC_FOLDER}/{site}/uploads/{local_path.name}"
|
||
nc_upload_file(local_path, remote, nc_user, nc_pass)
|
||
|
||
# Upload the updated Excel spreadsheet
|
||
nc_upload_file(excel_path, f"{NC_FOLDER}/{site}/data.xlsx", nc_user, nc_pass)
|
||
except Exception:
|
||
log.exception("Nextcloud sync failed for site %s", site)
|
||
|
||
|
||
##############################################################################
|
||
# Local file helpers
|
||
##############################################################################
|
||
|
||
def get_site_paths(site: str):
|
||
"""Return the local (upload_dir, excel_file) paths for a given site.
|
||
|
||
Creates the site directory if it doesn't exist.
|
||
|
||
Example:
|
||
get_site_paths("5001")
|
||
→ (Path("uploads/5001"), Path("uploads/5001/data.xlsx"))
|
||
"""
|
||
site_dir = UPLOAD_DIR / site
|
||
site_dir.mkdir(parents=True, exist_ok=True)
|
||
return site_dir, site_dir / "data.xlsx"
|
||
|
||
|
||
def get_or_create_workbook(excel_path: Path):
|
||
"""Open an existing Excel workbook or create a new one with headers.
|
||
|
||
If the file exists, it is opened and the active sheet is returned.
|
||
If not, a new workbook is created with a header row:
|
||
AP Number | AP Location | Serial Number | Cable Length
|
||
|
||
Returns:
|
||
(workbook, worksheet) tuple
|
||
"""
|
||
if excel_path.exists():
|
||
wb = load_workbook(excel_path)
|
||
ws = wb.active
|
||
else:
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "AP Data"
|
||
ws.append(["AP Number", "AP Location", "Serial Number", "MAC Address", "Cable Length"])
|
||
|
||
# ── Header row styling ──────────────────────────────────────
|
||
# Make the header row visually distinct so it's easy to spot
|
||
# in the spreadsheet. The blue color (#2563EB) matches the
|
||
# app's web UI theme for consistency.
|
||
#
|
||
# Font: bold + white text so it's readable on the blue bg
|
||
# Fill: solid blue background (start_color and end_color
|
||
# must both be set for a solid fill in openpyxl)
|
||
# Alignment: center-align header text for a clean look
|
||
header_font = Font(bold=True, color="FFFFFF")
|
||
header_fill = PatternFill(start_color="2563EB", end_color="2563EB", fill_type="solid")
|
||
header_alignment = Alignment(horizontal="center")
|
||
|
||
# ws[1] gives us all cells in row 1 (the header row).
|
||
# We loop through each cell and apply all three styles.
|
||
for cell in ws[1]:
|
||
cell.font = header_font
|
||
cell.fill = header_fill
|
||
cell.alignment = header_alignment
|
||
|
||
# ── Column widths ──────────────────────────────────────────
|
||
# By default, Excel columns are ~8 chars wide, which truncates
|
||
# most of our data. These widths are tuned to fit typical values
|
||
# without needing manual resizing. The unit is roughly "number
|
||
# of characters" (Excel's column width unit).
|
||
ws.column_dimensions["A"].width = 12 # AP Number (e.g. "001")
|
||
ws.column_dimensions["B"].width = 30 # AP Location (e.g. "Conference Room A")
|
||
ws.column_dimensions["C"].width = 20 # Serial Number(e.g. "ABCD-1234-EF56")
|
||
ws.column_dimensions["D"].width = 22 # MAC Address (e.g. "AA:BB:CC:DD:EE:FF")
|
||
ws.column_dimensions["E"].width = 14 # Cable Length (e.g. "25ft")
|
||
|
||
# ── Freeze panes ──────────────────────────────────────────
|
||
# Freezing at "A2" means everything above row 2 (i.e. the
|
||
# header in row 1) stays pinned at the top of the screen
|
||
# when you scroll down through many rows of data.
|
||
ws.freeze_panes = "A2"
|
||
|
||
wb.save(excel_path)
|
||
return wb, ws
|
||
|
||
|
||
##############################################################################
|
||
# Authentication
|
||
##############################################################################
|
||
# Each site has its own PIN defined in sites.conf. On login, the technician
|
||
# enters a site number and its corresponding PIN. The app reads sites.conf
|
||
# on every login attempt (no restart needed to add/remove sites).
|
||
#
|
||
# On successful login, two values are stored in the Flask session cookie:
|
||
# session["authenticated"] = True
|
||
# session["site"] = "5001" (the 4-digit site number)
|
||
#
|
||
# The login_required decorator checks both values are present; if not, the
|
||
# user is redirected to /login. The session persists until the browser is
|
||
# closed or the user clicks "Change site" (which hits /logout).
|
||
##############################################################################
|
||
|
||
def login_required(f):
|
||
"""Decorator that redirects to /login if the user is not authenticated
|
||
or has not selected a site."""
|
||
from functools import wraps
|
||
@wraps(f)
|
||
def decorated(*args, **kwargs):
|
||
if not session.get("authenticated") or not session.get("site"):
|
||
return redirect(url_for("login"))
|
||
return f(*args, **kwargs)
|
||
return decorated
|
||
|
||
|
||
def admin_required(f):
|
||
"""Decorator that redirects to /admin/login if admin is not authenticated.
|
||
|
||
Independent from the technician session — both can coexist.
|
||
Checks session["admin_authenticated"].
|
||
"""
|
||
from functools import wraps
|
||
@wraps(f)
|
||
def decorated(*args, **kwargs):
|
||
if not session.get("admin_authenticated"):
|
||
return redirect(url_for("admin_login"))
|
||
return f(*args, **kwargs)
|
||
return decorated
|
||
|
||
|
||
@app.route("/login", methods=["GET", "POST"])
|
||
def login():
|
||
"""Login page — accepts a site number and its PIN.
|
||
|
||
GET: renders the login form.
|
||
POST: validates the site number and PIN against sites.conf.
|
||
- Site must exist in sites.conf.
|
||
- PIN must match the site's configured PIN.
|
||
On success, sets session values and redirects to the main form.
|
||
On failure, re-renders the login page with an error message.
|
||
|
||
sites.conf is re-read on every POST so admins can add/remove sites
|
||
without restarting the app.
|
||
"""
|
||
error = None
|
||
if request.method == "POST":
|
||
pin = request.form.get("pin", "")
|
||
site = request.form.get("site", "").strip()
|
||
|
||
# Load the current site→pin mappings from sites.conf
|
||
sites = load_sites()
|
||
|
||
if site not in sites:
|
||
error = "Unknown site number"
|
||
elif pin != sites[site]["pin"]:
|
||
error = "Incorrect PIN"
|
||
else:
|
||
session["authenticated"] = True
|
||
session["site"] = site
|
||
return redirect(url_for("index"))
|
||
return render_template("login.html", error=error)
|
||
|
||
|
||
@app.route("/logout")
|
||
def logout():
|
||
"""Clear the session (PIN + site) and redirect to login.
|
||
|
||
Used by the "Change site" link on the main form page.
|
||
"""
|
||
session.clear()
|
||
return redirect(url_for("login"))
|
||
|
||
|
||
##############################################################################
|
||
# Main routes
|
||
##############################################################################
|
||
|
||
@app.route("/")
|
||
@login_required
|
||
def index():
|
||
"""Serve the main AP data collection form.
|
||
|
||
Passes the current site number to the template so it can be displayed
|
||
as a badge at the top of the page.
|
||
"""
|
||
return render_template("index.html", site=session["site"])
|
||
|
||
|
||
@app.route("/entries")
|
||
@login_required
|
||
def entries():
|
||
"""Show all submitted AP entries for the current site.
|
||
|
||
Reads uploads/{site}/data.xlsx and extracts all data rows (skipping the
|
||
header) into a list of dicts for display in cards with photo info.
|
||
"""
|
||
site = session["site"]
|
||
_, excel_path = get_site_paths(site)
|
||
|
||
rows = []
|
||
if excel_path.exists():
|
||
wb = load_workbook(excel_path, read_only=True)
|
||
ws = wb.active
|
||
for i, row in enumerate(ws.iter_rows(values_only=True)):
|
||
if i == 0:
|
||
continue # skip header
|
||
ap_num = row[0] or ""
|
||
# Check which photos exist for this AP
|
||
photos = []
|
||
if re.fullmatch(r"\d{3}", str(ap_num)):
|
||
for suffix_name, file_suffix in SUFFIX_MAP.items():
|
||
photos.append({
|
||
"suffix": suffix_name,
|
||
"label": SUFFIX_LABELS[suffix_name],
|
||
"exists": find_photo(site, ap_num, file_suffix) is not None,
|
||
})
|
||
rows.append({
|
||
"ap_number": ap_num,
|
||
"ap_location": row[1] or "",
|
||
"serial_number": row[2] or "",
|
||
"mac_address": row[3] or "",
|
||
"cable_length": row[4] or "",
|
||
"photos": photos,
|
||
})
|
||
wb.close()
|
||
|
||
return render_template("entries.html", site=site, rows=rows)
|
||
|
||
|
||
@app.route("/submit", methods=["POST"])
|
||
@login_required
|
||
def submit():
|
||
"""Handle form submission — save AP data and photos.
|
||
|
||
Expects multipart/form-data with:
|
||
- ap_number, ap_location, serial_number, cable_length (text fields)
|
||
- photo_ap, photo_far, photo_length, photo_cont, photo_rate, photo_box (files)
|
||
|
||
Processing steps:
|
||
1. Validate that all text fields are non-empty
|
||
2. Validate that all 6 photos are present
|
||
3. Save each photo to uploads/{site}/ with the structured filename
|
||
(e.g. AP010.jpg, AP010F.jpg, AP010_length.jpg, etc.)
|
||
4. Append a row to uploads/{site}/data.xlsx
|
||
5. Sync photos + Excel to Nextcloud (non-blocking on failure)
|
||
6. Return a JSON success/error response to the frontend
|
||
"""
|
||
# Get the site number from the session
|
||
site = session["site"]
|
||
site_upload_dir, site_excel = get_site_paths(site)
|
||
|
||
# Collect text fields
|
||
ap_number = request.form.get("ap_number", "").strip()
|
||
ap_location = request.form.get("ap_location", "").strip()
|
||
serial_number = request.form.get("serial_number", "").strip()
|
||
mac_address = request.form.get("mac_address", "").strip()
|
||
cable_length = request.form.get("cable_length", "").strip()
|
||
|
||
# Validate text fields — all five are required
|
||
if not all([ap_number, ap_location, serial_number, mac_address, cable_length]):
|
||
return jsonify({"success": False, "error": "All text fields are required."}), 400
|
||
|
||
# AP Number: exactly 3 digits (e.g. 001, 042, 999)
|
||
if not re.fullmatch(r"\d{3}", ap_number):
|
||
return jsonify({"success": False, "error": "AP Number must be exactly 3 digits."}), 400
|
||
|
||
# Serial Number: format xxxx-xxxx-xxxx (alphanumeric groups separated by dashes)
|
||
if not re.fullmatch(r"[A-Za-z0-9]{4}-[A-Za-z0-9]{4}-[A-Za-z0-9]{4}", serial_number):
|
||
return jsonify({"success": False, "error": "Serial Number must be in format xxxx-xxxx-xxxx."}), 400
|
||
|
||
# MAC Address: 6 hex pairs separated by colons or dashes (e.g. AA:BB:CC:DD:EE:FF)
|
||
if not re.fullmatch(r"([0-9A-Fa-f]{2}[:\-]){5}[0-9A-Fa-f]{2}", mac_address):
|
||
return jsonify({"success": False, "error": "MAC Address must be valid (e.g. AA:BB:CC:DD:EE:FF)."}), 400
|
||
|
||
# Reject duplicate AP numbers within the same site
|
||
if site_excel.exists():
|
||
wb_check = load_workbook(site_excel, read_only=True)
|
||
ws_check = wb_check.active
|
||
for i, row in enumerate(ws_check.iter_rows(values_only=True)):
|
||
if i == 0:
|
||
continue
|
||
if str(row[0]) == ap_number:
|
||
wb_check.close()
|
||
return jsonify({"success": False, "error": f"AP {ap_number} already exists for this site."}), 409
|
||
wb_check.close()
|
||
|
||
# Validate and save each of the 6 required photos
|
||
saved_files = [] # filenames for the JSON response
|
||
saved_paths = [] # full Paths for Nextcloud sync
|
||
for field_name, suffix in PHOTO_FIELDS:
|
||
photo = request.files.get(field_name)
|
||
if not photo or photo.filename == "":
|
||
return jsonify({
|
||
"success": False,
|
||
"error": f"Missing photo: {field_name}",
|
||
}), 400
|
||
|
||
# Preserve the original file extension (e.g. .jpg, .png, .heic)
|
||
ext = os.path.splitext(photo.filename)[1].lower() or ".jpg"
|
||
# Build the structured filename: AP{number}{suffix}.{ext}
|
||
filename = f"AP{ap_number}{suffix}{ext}"
|
||
save_path = site_upload_dir / filename
|
||
photo.save(str(save_path))
|
||
saved_files.append(filename)
|
||
saved_paths.append(save_path)
|
||
|
||
# Append a row to the site's Excel spreadsheet
|
||
wb, ws = get_or_create_workbook(site_excel)
|
||
ws.append([ap_number, ap_location, serial_number, mac_address, cable_length])
|
||
|
||
# ── Alternating row colors ("zebra striping") ────────────────
|
||
# Makes the spreadsheet easier to read by giving every other
|
||
# row a light blue background (#E8F0FE). The pattern is:
|
||
# Row 1 = header (blue, styled above in get_or_create_workbook)
|
||
# Row 2 = first data row → even → light blue
|
||
# Row 3 = second data row → odd → white (no fill needed)
|
||
# Row 4 = third data row → even → light blue
|
||
# ...and so on.
|
||
#
|
||
# ws.max_row gives the row number of the row we just appended.
|
||
# We only need to paint even rows; odd rows stay white by default.
|
||
row_num = ws.max_row
|
||
if row_num % 2 == 0:
|
||
fill = PatternFill(start_color="E8F0FE", end_color="E8F0FE", fill_type="solid")
|
||
for cell in ws[row_num]:
|
||
cell.fill = fill
|
||
|
||
wb.save(site_excel)
|
||
|
||
# Sync all files to Nextcloud (errors are logged but won't fail the request)
|
||
nc_sync(site, saved_paths, site_excel)
|
||
|
||
return jsonify({
|
||
"success": True,
|
||
"message": f"Site {site} — AP {ap_number} saved successfully.",
|
||
"files": saved_files,
|
||
})
|
||
|
||
|
||
##############################################################################
|
||
# Photo review & replacement routes
|
||
##############################################################################
|
||
|
||
@app.route("/photo/<ap_number>/<suffix>")
|
||
@login_required
|
||
def serve_photo(ap_number, suffix):
|
||
"""Serve a stored photo for the current site.
|
||
|
||
URL-friendly suffix names are mapped to file suffixes via SUFFIX_MAP.
|
||
The AP number is validated to prevent path traversal.
|
||
"""
|
||
if suffix not in SUFFIX_MAP:
|
||
return jsonify({"error": "Invalid photo type"}), 400
|
||
if not re.fullmatch(r"\d{3}", ap_number):
|
||
return jsonify({"error": "Invalid AP number"}), 400
|
||
|
||
site = session["site"]
|
||
file_suffix = SUFFIX_MAP[suffix]
|
||
photo_path = find_photo(site, ap_number, file_suffix)
|
||
|
||
if not photo_path or not photo_path.exists():
|
||
return jsonify({"error": "Photo not found"}), 404
|
||
|
||
return send_file(photo_path)
|
||
|
||
|
||
@app.route("/photo/<ap_number>/<suffix>/replace", methods=["POST"])
|
||
@login_required
|
||
def replace_photo(ap_number, suffix):
|
||
"""Replace (or upload) a photo for the current site.
|
||
|
||
Accepts a multipart 'photo' file field. Deletes the old file first
|
||
(handles extension changes), saves the new file, and syncs to Nextcloud.
|
||
"""
|
||
if suffix not in SUFFIX_MAP:
|
||
return jsonify({"success": False, "error": "Invalid photo type"}), 400
|
||
if not re.fullmatch(r"\d{3}", ap_number):
|
||
return jsonify({"success": False, "error": "Invalid AP number"}), 400
|
||
|
||
photo = request.files.get("photo")
|
||
if not photo or photo.filename == "":
|
||
return jsonify({"success": False, "error": "No file provided"}), 400
|
||
|
||
site = session["site"]
|
||
file_suffix = SUFFIX_MAP[suffix]
|
||
site_dir, site_excel = get_site_paths(site)
|
||
|
||
# Delete old file (may have different extension)
|
||
old_path = find_photo(site, ap_number, file_suffix)
|
||
if old_path and old_path.exists():
|
||
old_path.unlink()
|
||
|
||
# Save new file
|
||
ext = os.path.splitext(photo.filename)[1].lower() or ".jpg"
|
||
filename = f"AP{ap_number}{file_suffix}{ext}"
|
||
save_path = site_dir / filename
|
||
photo.save(str(save_path))
|
||
|
||
# Sync to Nextcloud
|
||
nc_sync(site, [save_path], site_excel)
|
||
|
||
return jsonify({"success": True, "message": f"Photo replaced: {filename}"})
|
||
|
||
|
||
##############################################################################
|
||
# Admin routes
|
||
##############################################################################
|
||
# The admin interface allows managing sites.conf from the browser.
|
||
# Admin authenticates with the global Nextcloud admin credentials
|
||
# (NC_USER / NC_PASS). The session flag "admin_authenticated" is independent
|
||
# from the technician "authenticated" flag — both can coexist.
|
||
##############################################################################
|
||
|
||
@app.route("/admin/login", methods=["GET", "POST"])
|
||
def admin_login():
|
||
"""Admin login page — validates against NC_USER / NC_PASS.
|
||
|
||
GET: renders the admin login form.
|
||
POST: checks username/password against the global NC_USER and NC_PASS.
|
||
On success, sets session["admin_authenticated"] and redirects to /admin.
|
||
On failure, re-renders with an error message.
|
||
"""
|
||
error = None
|
||
if request.method == "POST":
|
||
username = request.form.get("username", "").strip()
|
||
password = request.form.get("password", "")
|
||
if username == NC_USER and password == NC_PASS:
|
||
session["admin_authenticated"] = True
|
||
return redirect(url_for("admin_dashboard"))
|
||
else:
|
||
error = "Invalid admin credentials"
|
||
return render_template("admin_login.html", error=error)
|
||
|
||
|
||
@app.route("/admin/logout")
|
||
def admin_logout():
|
||
"""Clear admin session flag (preserves technician session)."""
|
||
session.pop("admin_authenticated", None)
|
||
return redirect(url_for("admin_login"))
|
||
|
||
|
||
@app.route("/admin")
|
||
@admin_required
|
||
def admin_dashboard():
|
||
"""Admin dashboard — lists all sites with NC user status.
|
||
|
||
Loads sites from sites.conf, checks Nextcloud user existence for
|
||
each site that has per-site credentials, and renders admin.html.
|
||
"""
|
||
sites = load_sites()
|
||
site_list = []
|
||
for site_num in sorted(sites.keys(), key=lambda s: int(s) if s.isdigit() else 0):
|
||
cfg = sites[site_num]
|
||
entry = {
|
||
"id": site_num,
|
||
"pin": cfg["pin"],
|
||
"nc_user": cfg.get("nc_user"),
|
||
"nc_pass": cfg.get("nc_pass"),
|
||
}
|
||
# Check NC user status if per-site credentials are set
|
||
if cfg.get("nc_user"):
|
||
entry["nc_exists"] = nc_user_exists(cfg["nc_user"])
|
||
else:
|
||
entry["nc_exists"] = None # uses global — no status to check
|
||
site_list.append(entry)
|
||
return render_template("admin.html", sites=site_list)
|
||
|
||
|
||
@app.route("/admin/api/sites", methods=["POST"])
|
||
@admin_required
|
||
def admin_add_site():
|
||
"""Add a new site to sites.conf.
|
||
|
||
JSON body: {site, pin, nc_user, nc_pass}
|
||
Site must be exactly 4 digits and must not already exist.
|
||
NC User and NC Pass are required.
|
||
"""
|
||
data = request.get_json(force=True)
|
||
site = data.get("site", "").strip()
|
||
pin = data.get("pin", "").strip()
|
||
nc_user = data.get("nc_user", "").strip()
|
||
nc_pass = data.get("nc_pass", "").strip()
|
||
|
||
if not re.fullmatch(r"\d{4}", site):
|
||
return jsonify({"ok": False, "error": "Site must be exactly 4 digits"}), 400
|
||
|
||
if not pin:
|
||
return jsonify({"ok": False, "error": "PIN is required"}), 400
|
||
|
||
if not nc_user:
|
||
return jsonify({"ok": False, "error": "NC User is required"}), 400
|
||
|
||
if not nc_pass:
|
||
return jsonify({"ok": False, "error": "NC Pass is required"}), 400
|
||
|
||
# Reject characters that would corrupt sites.conf (colon is the
|
||
# field delimiter, newlines would inject extra lines)
|
||
for field_name, value in [("PIN", pin), ("NC User", nc_user), ("NC Pass", nc_pass)]:
|
||
if value and ("\n" in value or "\r" in value):
|
||
return jsonify({"ok": False, "error": f"{field_name} must not contain newlines"}), 400
|
||
for field_name, value in [("PIN", pin), ("NC User", nc_user)]:
|
||
if value and ":" in value:
|
||
return jsonify({"ok": False, "error": f"{field_name} must not contain colons"}), 400
|
||
|
||
sites = load_sites()
|
||
if site in sites:
|
||
return jsonify({"ok": False, "error": f"Site {site} already exists"}), 409
|
||
|
||
sites[site] = {"pin": pin, "nc_user": nc_user, "nc_pass": nc_pass}
|
||
save_sites(sites)
|
||
|
||
# Auto-provision the Nextcloud user
|
||
nc_msg = None
|
||
if all([NC_URL, NC_USER, NC_PASS]):
|
||
if nc_user_exists(nc_user):
|
||
nc_msg = f"NC user '{nc_user}' already exists"
|
||
elif nc_create_user(nc_user, nc_pass):
|
||
nc_msg = f"NC user '{nc_user}' created"
|
||
else:
|
||
nc_msg = f"Site saved, but failed to create NC user '{nc_user}'"
|
||
|
||
return jsonify({"ok": True, "nc_msg": nc_msg})
|
||
|
||
|
||
@app.route("/admin/api/sites/<site_id>", methods=["PUT"])
|
||
@admin_required
|
||
def admin_edit_site(site_id):
|
||
"""Edit an existing site in sites.conf.
|
||
|
||
JSON body: {pin?, nc_user?, nc_pass?}
|
||
Only provided fields are updated.
|
||
"""
|
||
sites = load_sites()
|
||
if site_id not in sites:
|
||
return jsonify({"ok": False, "error": f"Site {site_id} not found"}), 404
|
||
|
||
data = request.get_json(force=True)
|
||
pin = data.get("pin", "").strip() if "pin" in data else None
|
||
nc_user = data.get("nc_user", "").strip() if "nc_user" in data else None
|
||
nc_pass = data.get("nc_pass", "").strip() if "nc_pass" in data else None
|
||
|
||
# Reject characters that would corrupt sites.conf
|
||
for field_name, value in [("PIN", pin), ("NC User", nc_user), ("NC Pass", nc_pass)]:
|
||
if value and ("\n" in value or "\r" in value):
|
||
return jsonify({"ok": False, "error": f"{field_name} must not contain newlines"}), 400
|
||
for field_name, value in [("PIN", pin), ("NC User", nc_user)]:
|
||
if value and ":" in value:
|
||
return jsonify({"ok": False, "error": f"{field_name} must not contain colons"}), 400
|
||
|
||
if pin:
|
||
sites[site_id]["pin"] = pin
|
||
if nc_user is not None:
|
||
sites[site_id]["nc_user"] = nc_user or None
|
||
if nc_pass is not None:
|
||
sites[site_id]["nc_pass"] = nc_pass or None
|
||
|
||
save_sites(sites)
|
||
|
||
# Sync password change to Nextcloud if applicable
|
||
warning = None
|
||
new_nc_pass = data.get("nc_pass", "").strip() if "nc_pass" in data else None
|
||
nc_user = sites[site_id].get("nc_user")
|
||
if new_nc_pass and nc_user and all([NC_URL, NC_USER, NC_PASS]):
|
||
if nc_user_exists(nc_user):
|
||
if not nc_change_password(nc_user, new_nc_pass):
|
||
warning = f"Site saved, but failed to update password on Nextcloud for '{nc_user}'"
|
||
|
||
if warning:
|
||
return jsonify({"ok": True, "warning": warning})
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
@app.route("/admin/api/sites/<site_id>", methods=["DELETE"])
|
||
@admin_required
|
||
def admin_delete_site(site_id):
|
||
"""Delete a site from sites.conf (preserves uploaded files)."""
|
||
sites = load_sites()
|
||
if site_id not in sites:
|
||
return jsonify({"ok": False, "error": f"Site {site_id} not found"}), 404
|
||
|
||
del sites[site_id]
|
||
save_sites(sites)
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
@app.route("/admin/api/sites/<site_id>/provision", methods=["POST"])
|
||
@admin_required
|
||
def admin_provision_site(site_id):
|
||
"""Create the per-site Nextcloud user via OCS Provisioning API.
|
||
|
||
Only works if the site has nc_user and nc_pass set.
|
||
"""
|
||
sites = load_sites()
|
||
if site_id not in sites:
|
||
return jsonify({"ok": False, "error": f"Site {site_id} not found"}), 404
|
||
|
||
cfg = sites[site_id]
|
||
nc_user = cfg.get("nc_user")
|
||
nc_pass_val = cfg.get("nc_pass")
|
||
|
||
if not nc_user or not nc_pass_val:
|
||
return jsonify({"ok": False, "error": "Site has no per-site NC credentials"}), 400
|
||
|
||
if nc_user_exists(nc_user):
|
||
return jsonify({"ok": True, "message": f"User '{nc_user}' already exists"})
|
||
|
||
if nc_create_user(nc_user, nc_pass_val):
|
||
return jsonify({"ok": True, "message": f"User '{nc_user}' created successfully"})
|
||
else:
|
||
return jsonify({"ok": False, "error": f"Failed to create user '{nc_user}'"}), 500
|
||
|
||
|
||
##############################################################################
|
||
# Entry point
|
||
##############################################################################
|
||
# When run directly (python3 app.py), starts the Flask development server
|
||
# on all interfaces (0.0.0.0) port 5000 with debug mode enabled.
|
||
# In production, this is managed by the aptool.service systemd unit.
|
||
##############################################################################
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Startup provisioning — runs on import (works with both gunicorn and dev server)
|
||
# ---------------------------------------------------------------------------
|
||
if nc_enabled(NC_USER, NC_PASS):
|
||
print(f"Nextcloud sync enabled → {NC_URL} (default user: {NC_USER}, folder: {NC_FOLDER})")
|
||
nc_provision_site_users()
|
||
else:
|
||
print("Nextcloud sync disabled (set NC_URL, NC_USER, NC_PASS to enable)")
|
||
|
||
if __name__ == "__main__":
|
||
app.run(host="0.0.0.0", port=5000, debug=True)
|