Ross
1096 lines
38 KiB
Python
1096 lines
38 KiB
Python
#!/usr/bin/env python3
|
|
import asyncio
|
|
import pynng
|
|
|
|
# asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
from bs4 import BeautifulSoup
|
|
|
|
from nicegui import ui, app, html, run, Client, native
|
|
|
|
from datetime import datetime
|
|
|
|
from loguru import logger
|
|
import sys
|
|
import os
|
|
|
|
from anonymiser import Anonymizer
|
|
import pydicom
|
|
|
|
import requests
|
|
import time
|
|
|
|
from multiprocessing import Manager, Queue, freeze_support
|
|
|
|
from functools import partial
|
|
|
|
from local_file_picker import local_file_picker
|
|
from local_folder_picker import local_folder_picker
|
|
|
|
from typing import Optional
|
|
import tempfile
|
|
import glob
|
|
|
|
from fastapi import Request
|
|
from fastapi.responses import RedirectResponse
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
import typer
|
|
import logging
|
|
import subprocess
|
|
import shutil
|
|
from urllib.parse import urlparse
|
|
import socket
|
|
import platform
|
|
import importlib.util
|
|
|
|
# traceback was previously imported for debugging but is unused
|
|
|
|
# SSL verification setting. Can be a boolean (True/False) or a path to
|
|
# a CA bundle (PEM file) used by requests' 'verify' parameter.
|
|
# Default is True. It can be overridden via CLI options (--ca-bundle,
|
|
# --insecure) or environment variables: NICE_UPLOADER_CA_BUNDLE,
|
|
# REQUESTS_CA_BUNDLE or SSL_CERT_FILE.
|
|
VERIFY_SSL = True
|
|
|
|
SOCKET_PATH = "tcp://localhost:9976"
|
|
|
|
WORK_DIR = Path("C:/uploader")
|
|
|
|
EXPORT_PATH = WORK_DIR / Path("export/")
|
|
PROCESS_PATH = WORK_DIR / Path("to_process/")
|
|
ANON_PATH = WORK_DIR / Path("anon/")
|
|
|
|
EXPORT_PATH.mkdir(exist_ok=True, parents=True)
|
|
PROCESS_PATH.mkdir(exist_ok=True, parents=True)
|
|
ANON_PATH.mkdir(exist_ok=True, parents=True)
|
|
|
|
BASE_SITE_URL = "https://www.penracourses.org.uk"
|
|
|
|
LOGIN_URL = f"{BASE_SITE_URL}/accounts/login/"
|
|
|
|
LOGIN_SUCCESS = False
|
|
|
|
LOGGED_IN_USER = "None"
|
|
|
|
|
|
REFRESH_TRIGGERED = True
|
|
RELOAD_ANON_FILES = False
|
|
SHUTDOWN = False
|
|
|
|
DELETE_FILES_ON_UPLOAD = True
|
|
|
|
DEBUG = False
|
|
|
|
logging.getLogger('niceGUI').setLevel(logging.INFO)
|
|
|
|
# Choose a log file path that works both during development and when the
|
|
# application is frozen into a single executable (PyInstaller --onefile).
|
|
try:
|
|
if getattr(sys, "frozen", False):
|
|
# When frozen by PyInstaller, data files are extracted to a temporary
|
|
# folder available as sys._MEIPASS. Fall back to the executable
|
|
# directory when _MEIPASS is not present.
|
|
meipass = getattr(sys, "_MEIPASS", None)
|
|
if meipass:
|
|
base_path = Path(meipass)
|
|
else:
|
|
base_path = Path(sys.executable).resolve().parent
|
|
else:
|
|
base_path = Path(__file__).resolve().parent
|
|
except Exception:
|
|
base_path = Path.cwd()
|
|
|
|
log_path = base_path / "uploader.log"
|
|
logger.add(str(log_path), rotation="10 MB")
|
|
|
|
if DEBUG:
|
|
BASE_SITE_URL = "http://localhost:8000"
|
|
EXPORT_PATH = Path("./test/export")
|
|
PROCESS_PATH = Path("./test/to_process")
|
|
ANON_PATH = Path("./test/anon")
|
|
|
|
rqst = None
|
|
|
|
|
|
loaded_files: dict = {}
|
|
loaded_series = defaultdict(list)
|
|
loaded_series_data = {}
|
|
uploaded_files: dict = {}
|
|
duplicate_series = set()
|
|
|
|
|
|
async def read_nng_messages():
|
|
print("start nng")
|
|
try:
|
|
with pynng.Pair0(listen=SOCKET_PATH) as sub:
|
|
# sub.subscribe('')
|
|
while not app.is_stopped:
|
|
msg = await sub.arecv_msg()
|
|
data = msg.bytes.decode()
|
|
logger.debug(f"Received msg {data}")
|
|
match data:
|
|
case "loaded":
|
|
for client in Client.instances.values():
|
|
if not client.has_socket_connection:
|
|
continue
|
|
with client:
|
|
ui.notify("Export done", color="positive")
|
|
global REFRESH_TRIGGERED
|
|
REFRESH_TRIGGERED = True
|
|
await sub.asend(b"done")
|
|
logger.debug("loaded")
|
|
case _:
|
|
logger.error("invalid message")
|
|
except pynng.exceptions.AddressInUse:
|
|
logger.debug("Address in use")
|
|
with pynng.Pair0(dial=SOCKET_PATH) as sub:
|
|
await sub.asend(b"loaded")
|
|
|
|
# Trigger shutdown from in the async loop
|
|
# This shouldn't be necessary
|
|
global SHUTDOWN
|
|
SHUTDOWN = True
|
|
logger.debug("Triggering shutdown due to socket in use")
|
|
return
|
|
|
|
# asyncio.create_task(read_nng_messages())
|
|
|
|
|
|
anonymizer = Anonymizer(ANON_PATH)
|
|
|
|
|
|
async def upload_files_start(progress, upload_queue):
|
|
global uploaded_files, duplicate_series
|
|
progress.visible = True
|
|
|
|
results = await run.cpu_bound(upload_files, upload_queue, rqst)
|
|
progress.visible = False
|
|
|
|
if results:
|
|
(
|
|
new_uploaded_files,
|
|
upload_file_list,
|
|
duplicate_file_list,
|
|
failed,
|
|
duplicate_series,
|
|
) = results
|
|
else:
|
|
ui.notify("No files to upload", color="negative")
|
|
return
|
|
|
|
uploaded_files.update(new_uploaded_files)
|
|
|
|
ui.notify(f"Uploaded {len(upload_file_list)} files", color="positive")
|
|
|
|
if duplicate_file_list:
|
|
ui.notify(f"Duplicate {len(duplicate_file_list)} files", color="warning")
|
|
|
|
if failed:
|
|
ui.notify(f"Failed {len(failed)} files", color="negative")
|
|
|
|
if DELETE_FILES_ON_UPLOAD:
|
|
clear_anonymized_files()
|
|
|
|
loaded_series_ui.refresh()
|
|
loaded_files_ui.refresh()
|
|
uploaded_files_ui.refresh()
|
|
|
|
|
|
async def load_files_start(progress_bar, queue, custom_path=None):
|
|
logger.debug("load files start")
|
|
global loaded_files, loaded_series, loaded_series_data
|
|
progress_bar.visible = True
|
|
|
|
new_loaded_files, new_loaded_series, new_loaded_series_data = await run.cpu_bound(
|
|
load_files, queue, custom_path
|
|
)
|
|
|
|
for client in Client.instances.values():
|
|
if not client.has_socket_connection:
|
|
continue
|
|
with client:
|
|
ui.notify(f"Files loaded: {len(new_loaded_files)}, [Series: {len(new_loaded_series)}] ", color="positive")
|
|
|
|
loaded_files.update(new_loaded_files)
|
|
loaded_series.update(new_loaded_series)
|
|
loaded_series_data.update(new_loaded_series_data)
|
|
|
|
progress_bar.visible = False
|
|
# load_series_view(loaded_series, loaded_series_data)
|
|
loaded_series_ui.refresh()
|
|
loaded_files_ui.refresh()
|
|
|
|
|
|
async def reload_anonymized_start(progress_bar, queue):
|
|
global loaded_files, loaded_series, loaded_series_data
|
|
progress_bar.visible = True
|
|
loaded_files, loaded_series, loaded_series_data = await run.cpu_bound(
|
|
reload_anonymized, queue
|
|
)
|
|
progress_bar.visible = False
|
|
|
|
for client in Client.instances.values():
|
|
if not client.has_socket_connection:
|
|
continue
|
|
with client:
|
|
ui.notify(f"Files reloaded: {len(loaded_files)}, [Series: {len(loaded_series)}] ", color="positive")
|
|
# load_series_view(loaded_series, loaded_series_data)
|
|
loaded_series_ui.refresh()
|
|
loaded_files_ui.refresh()
|
|
|
|
|
|
@ui.refreshable
|
|
def user_info_ui():
|
|
global LOGGED_IN_USER
|
|
user_label = ui.label(f"User: {LOGGED_IN_USER}")
|
|
# user_label.bind_visibility_from(globals(), "LOGIN_SUCCESS")
|
|
|
|
|
|
@ui.refreshable
|
|
def loaded_series_ui() -> None:
|
|
global loaded_series, loaded_series_data
|
|
|
|
series_title = ui.label("Series to upload").classes("text-h3")
|
|
series_title.visible = False
|
|
|
|
series = ui.row()
|
|
|
|
# def load_series_view(loaded_series, loaded_series_data):
|
|
logger.debug("load series view")
|
|
logger.debug(loaded_series)
|
|
for key in loaded_series:
|
|
logger.debug(f"load series: {key}")
|
|
with series:
|
|
with ui.card():
|
|
ui.label(loaded_series_data[key][0])
|
|
ui.label(loaded_series_data[key][1])
|
|
ui.label(f"Images: {len(loaded_series[key])}")
|
|
|
|
if loaded_series:
|
|
series_title.visible = True
|
|
|
|
|
|
@ui.refreshable
|
|
def loaded_files_ui() -> None:
|
|
with ui.expansion("Loaded files", icon="drive_folder_upload").classes("w-full"):
|
|
ui.label(f"Items: {len(loaded_files)}")
|
|
with ui.list():
|
|
for file, data in loaded_files.items():
|
|
ui.item(f"{file} - {data[0]} - {data[1]}")
|
|
|
|
|
|
@ui.refreshable
|
|
def uploaded_files_ui() -> None:
|
|
with ui.expansion("Uploaded files", icon="file_upload").classes("w-full"):
|
|
ui.label(f"Items: {len(uploaded_files)}")
|
|
with ui.list():
|
|
for series in duplicate_series:
|
|
ui.item(
|
|
f"Duplicate series: {series}",
|
|
on_click=lambda: ui.navigate.to(
|
|
f"{BASE_SITE_URL}{series}", new_tab=True
|
|
),
|
|
).classes("text-red-500")
|
|
with ui.list():
|
|
for file, data in uploaded_files.items():
|
|
ui.item(f"{file} - {data}")
|
|
|
|
|
|
def load_files(q: Queue, src_path: Path | None = None):
|
|
loaded_files: dict = {}
|
|
loaded_series = defaultdict(list)
|
|
loaded_series_data = {}
|
|
# Move files
|
|
if src_path is None:
|
|
src_path = EXPORT_PATH
|
|
|
|
to_process = []
|
|
for file in src_path.glob("**/*.dcm"):
|
|
logger.debug(f"Moving {file} to {PROCESS_PATH}")
|
|
|
|
file_to_process = PROCESS_PATH / file.name
|
|
file.rename(file_to_process)
|
|
|
|
to_process.append(file_to_process)
|
|
|
|
to_process_len = len(to_process)
|
|
for n, file in enumerate(to_process):
|
|
dataset, output_file = anonymizer.anonymize_file(file, remove_original=True)
|
|
# with processed_files:
|
|
# ui.item(f"{datetime.now().strftime('%H:%M:%S')} - {file.name} -> {output_file.name}")
|
|
try:
|
|
series_description = dataset.SeriesDescription
|
|
except AttributeError:
|
|
series_description = ""
|
|
|
|
loaded_files[output_file] = (
|
|
dataset.StudyDescription,
|
|
series_description,
|
|
str(dataset.SeriesInstanceUID),
|
|
)
|
|
|
|
loaded_series[dataset.SeriesInstanceUID].append(output_file)
|
|
loaded_series_data[dataset.SeriesInstanceUID] = (
|
|
dataset.StudyDescription,
|
|
series_description
|
|
)
|
|
|
|
q.put_nowait((n + 1) / to_process_len)
|
|
|
|
# time.sleep(0.01)
|
|
|
|
return loaded_files, loaded_series, loaded_series_data
|
|
|
|
|
|
def reload_anonymized(q: Queue):
|
|
loaded_files = {}
|
|
loaded_series = defaultdict(list)
|
|
loaded_series_data = {}
|
|
|
|
# processed_files.clear()
|
|
|
|
to_process = list(ANON_PATH.glob("**/*.dcm"))
|
|
to_process_len = len(to_process)
|
|
for n, file in enumerate(to_process):
|
|
with pydicom.dcmread(file) as dataset:
|
|
loaded_files[file] = (
|
|
dataset.StudyDescription,
|
|
dataset.SeriesDescription,
|
|
str(dataset.SeriesInstanceUID),
|
|
)
|
|
|
|
loaded_series[dataset.SeriesInstanceUID].append(file)
|
|
loaded_series_data[dataset.SeriesInstanceUID] = (
|
|
dataset.StudyDescription,
|
|
dataset.SeriesDescription,
|
|
)
|
|
|
|
# with processed_files:
|
|
# ui.item(f"{datetime.now().strftime('%H:%M:%S')} - {file.name} -> Reloaded")
|
|
|
|
q.put_nowait(f"{(n + 1) / to_process_len * 100:.0f}%")
|
|
logger.debug(file)
|
|
|
|
# time.sleep(0.01)
|
|
|
|
logger.debug("end reload")
|
|
logger.debug(loaded_files)
|
|
logger.debug(loaded_series)
|
|
logger.debug("-----")
|
|
|
|
return loaded_files, loaded_series, loaded_series_data
|
|
|
|
|
|
def clear_anonymized_files():
|
|
global RELOAD_ANON_FILES
|
|
for file in ANON_PATH.iterdir():
|
|
if file.is_file():
|
|
file.unlink()
|
|
|
|
logger.debug("Deleted all files in ANON_PATH")
|
|
|
|
RELOAD_ANON_FILES = True
|
|
|
|
return
|
|
|
|
|
|
def ssl_check_cmd(url: str) -> str:
|
|
"""Run openssl s_client -showcerts against the given URL host and return output.
|
|
|
|
Falls back to a minimal Python-based peer certificate fetch if openssl is not available.
|
|
"""
|
|
try:
|
|
parsed = urlparse(url)
|
|
host = parsed.hostname or url
|
|
port = parsed.port or (443 if parsed.scheme in ("https", "") else 80)
|
|
except Exception:
|
|
host = url
|
|
port = 443
|
|
|
|
logger.debug(f"ssl_check_cmd: host={host} port={port}")
|
|
openssl_path = shutil.which("openssl")
|
|
if openssl_path:
|
|
cmd = [openssl_path, "s_client", "-showcerts", "-connect", f"{host}:{port}", "-servername", host]
|
|
logger.debug(f"Running openssl: {' '.join(cmd)}")
|
|
try:
|
|
# Provide a small stdin so s_client can complete instead of
|
|
# waiting for interactive input. Capture both stdout and stderr.
|
|
proc = subprocess.run(cmd, input='\n', capture_output=True, text=True, timeout=60)
|
|
out = proc.stdout + "\n" + proc.stderr
|
|
logger.debug(f"OpenSSL finished, {len(out)} bytes returned, rc={proc.returncode}")
|
|
return out
|
|
except subprocess.TimeoutExpired:
|
|
logger.debug("OpenSSL timed out")
|
|
return (
|
|
"OpenSSL check timed out after 60s\n"
|
|
"This can happen if the connection is blocked by a proxy or firewall. "
|
|
"Try running the openssl command manually or check your network/proxy settings.\n"
|
|
)
|
|
except Exception as e:
|
|
logger.exception("OpenSSL check failed")
|
|
return f"OpenSSL check failed: {e}\n"
|
|
else:
|
|
# Fallback: use Python ssl to fetch the peer certificate (leaf only)
|
|
import ssl
|
|
|
|
logger.debug("OpenSSL not found; using Python ssl fallback")
|
|
try:
|
|
ctx = ssl.create_default_context()
|
|
s = socket.socket(socket.AF_INET)
|
|
with ctx.wrap_socket(s, server_hostname=host) as ss:
|
|
ss.settimeout(15)
|
|
ss.connect((host, port))
|
|
der = ss.getpeercert(True)
|
|
pem = ssl.DER_cert_to_PEM_cert(der)
|
|
logger.debug("Python ssl fallback succeeded")
|
|
return pem
|
|
except Exception as e:
|
|
logger.exception("Python SSL fallback failed")
|
|
return f"Python SSL fallback failed: {e}\n"
|
|
|
|
|
|
def upload_files(q, rqst):
|
|
# global rqst#, uploaded_files
|
|
|
|
uploaded_files = {}
|
|
|
|
files_to_upload = []
|
|
for file in ANON_PATH.iterdir():
|
|
files_to_upload.append(("files", open(str(file), "rb")))
|
|
|
|
if not files_to_upload:
|
|
return None
|
|
|
|
# chunck files
|
|
n = 10
|
|
chunked_files = [
|
|
files_to_upload[i : i + n] for i in range(0, len(files_to_upload), n)
|
|
]
|
|
|
|
upload_file_list = []
|
|
duplicate_file_list = []
|
|
failed = []
|
|
duplicate_series = set()
|
|
|
|
logger.debug("START UPLOAD")
|
|
logger.debug(files_to_upload)
|
|
|
|
to_process_len = len(chunked_files)
|
|
|
|
for n, files in enumerate(chunked_files):
|
|
|
|
def upload_files_(files):
|
|
rqst.headers["X-CSRFToken"] = rqst.cookies["csrftoken"]
|
|
# print(self.rqst.headers)
|
|
# print(self.rqst.cookies)
|
|
resp = rqst.post(
|
|
f"{BASE_SITE_URL}/api/atlas/upload_dicom",
|
|
# data=data,
|
|
files=files,
|
|
verify=VERIFY_SSL
|
|
)
|
|
|
|
logger.debug(f"Resp: {resp}")
|
|
logger.debug(f"{resp.content}")
|
|
return resp
|
|
|
|
# progress_dialog.Update(n, f"Uploading batch {n}/{len(chunked_files)}")
|
|
|
|
logger.debug(f"n: {n}")
|
|
# try to upload the files
|
|
|
|
for i in range(3):
|
|
resp = upload_files_(files)
|
|
if resp.status_code == 200:
|
|
upload_file_list.extend(resp.json()["uploaded"])
|
|
duplicate_file_list.extend(resp.json()["duplicates"])
|
|
failed.extend(resp.json()["failed"])
|
|
duplicate_series.update(resp.json()["duplicate_series"])
|
|
|
|
break
|
|
|
|
logger.error(f"i: {i}")
|
|
logger.debug(f"n: {n} fail (attempt {i})")
|
|
|
|
q.put_nowait((n + 1) / to_process_len)
|
|
# progress_dialog.Destroy()
|
|
print(upload_file_list)
|
|
print("dup", duplicate_file_list)
|
|
print("failed", failed)
|
|
|
|
for f, hash in upload_file_list:
|
|
uploaded_files[f] = "success"
|
|
|
|
for f, hash in duplicate_file_list:
|
|
uploaded_files[f] = "duplicate"
|
|
|
|
for f in failed:
|
|
uploaded_files[f] = "failed"
|
|
|
|
return (
|
|
uploaded_files,
|
|
upload_file_list,
|
|
duplicate_file_list,
|
|
failed,
|
|
duplicate_series,
|
|
)
|
|
|
|
|
|
def perform_login(user: str, pw: str) -> dict:
|
|
"""Module-level blocking login function suitable for run.cpu_bound.
|
|
|
|
Returns a dict with keys:
|
|
- error: str on error
|
|
- session: requests.Session on success
|
|
- token: csrf token if found
|
|
- success: bool
|
|
"""
|
|
s = requests.session()
|
|
try:
|
|
r = s.get(LOGIN_URL, verify=VERIFY_SSL, timeout=10)
|
|
except Exception as e:
|
|
logger.debug(e)
|
|
return {"error": str(e)}
|
|
|
|
try:
|
|
token = BeautifulSoup(r.content, "html.parser").find("input").attrs.get("value")
|
|
except Exception:
|
|
token = None
|
|
|
|
data = {
|
|
"username": user,
|
|
"password": pw,
|
|
"csrfmiddlewaretoken": token,
|
|
"next": "/",
|
|
}
|
|
|
|
s.headers["Referer"] = LOGIN_URL
|
|
|
|
try:
|
|
r2 = s.post(LOGIN_URL, data=data, verify=VERIFY_SSL, timeout=10)
|
|
except Exception as e:
|
|
logger.debug(e)
|
|
return {"error": str(e)}
|
|
|
|
soup = BeautifulSoup(r2.content, "html.parser")
|
|
if soup.find("button") and soup.find("button").find(string="Login"):
|
|
return {"session": s, "success": False}
|
|
else:
|
|
return {"session": s, "success": True, "token": token}
|
|
|
|
|
|
|
|
|
|
|
|
@ui.page("/login")
|
|
def login() -> Optional[RedirectResponse]:
|
|
dark = ui.dark_mode()
|
|
dark.enable()
|
|
|
|
async def try_login() -> None:
|
|
# Run blocking network calls off the UI thread to keep the page responsive.
|
|
user = username.value
|
|
pw = password.value
|
|
|
|
global rqst, LOGIN_SUCCESS, LOGGED_IN_USER
|
|
|
|
# Use the module-level perform_login function (picklable) to avoid
|
|
# multiprocessing pickling errors when using run.cpu_bound.
|
|
result = await run.cpu_bound(perform_login, user, pw)
|
|
|
|
if result.get("error"):
|
|
ui.notify("Connection error!", color="negative")
|
|
logger.debug(result.get("error"))
|
|
return
|
|
|
|
if result.get("success"):
|
|
rqst = result.get("session")
|
|
token = result.get("token")
|
|
if token:
|
|
rqst.headers["X-CSRFToken"] = token
|
|
LOGIN_SUCCESS = True
|
|
LOGGED_IN_USER = user
|
|
user_info_ui.refresh()
|
|
ui.notify("Logged in!", color="positive")
|
|
ui.navigate.to("/")
|
|
else:
|
|
ui.notify("Wrong username or password", color="negative")
|
|
|
|
# if app.storage.user.get('authenticated', False):
|
|
# return RedirectResponse('/')
|
|
with ui.card().classes("absolute-center"):
|
|
username = ui.input("Username").on("keydown.enter", try_login)
|
|
password = ui.input("Password", password=True, password_toggle_button=True).on(
|
|
"keydown.enter", try_login
|
|
)
|
|
with ui.row():
|
|
ui.button("Log in", on_click=try_login)
|
|
ui.button("Cancel", on_click=lambda: ui.navigate.to("/"), color="red")
|
|
return None
|
|
|
|
|
|
@logger.catch
|
|
@ui.page("/")
|
|
def main_page():
|
|
async def watch_for_shutdown():
|
|
global SHUTDOWN
|
|
while not app.is_stopped:
|
|
if SHUTDOWN:
|
|
logger.debug("Trigger app shutdown")
|
|
app.shutdown()
|
|
await asyncio.sleep(1)
|
|
|
|
asyncio.create_task(watch_for_shutdown())
|
|
|
|
dark = ui.dark_mode()
|
|
dark.enable()
|
|
ui.page_title("Uploader tool")
|
|
ui.label("Uploader tool").classes("text-h2")
|
|
|
|
user_info_ui()
|
|
|
|
login_button = ui.button("LOGIN", on_click=lambda: ui.navigate.to("/login"))
|
|
login_button.bind_visibility_from(
|
|
globals(), "LOGIN_SUCCESS", backward=lambda v: not v
|
|
)
|
|
|
|
# placeholder for update list (previously assigned but unused)
|
|
|
|
loaded_series_ui()
|
|
|
|
# typer_app()
|
|
# Create a queue to communicate with the heavy computation process
|
|
logger.debug("PRE QUEUE")
|
|
queue = Manager().Queue()
|
|
logger.debug("POST QUEUE")
|
|
# Update the progress bar on the main process
|
|
ui.timer(
|
|
0.1,
|
|
callback=lambda: progressbar.set_value(
|
|
queue.get() if not queue.empty() else progressbar.value
|
|
),
|
|
)
|
|
logger.debug("POST TImER")
|
|
|
|
# Create a queue to communicate with the heavy computation process
|
|
upload_queue = Manager().Queue()
|
|
# Update the progress bar on the main process
|
|
ui.timer(
|
|
0.1,
|
|
callback=lambda: upload_progressbar.set_value(
|
|
upload_queue.get() if not upload_queue.empty() else upload_progressbar.value
|
|
),
|
|
)
|
|
logger.debug("POST TImER")
|
|
|
|
upload_progress = ui.row().classes("w-full place-content-center bg-blue-900")
|
|
|
|
with upload_progress:
|
|
ui.label("Uploading files")
|
|
upload_progressbar = ui.linear_progress(value=0).props("instant-feedback")
|
|
upload_progress.visible = False
|
|
|
|
upload_button = ui.button(
|
|
"Upload", on_click=partial(upload_files_start, upload_progress, upload_queue)
|
|
)
|
|
upload_button.bind_visibility_from(globals(), "LOGIN_SUCCESS")
|
|
|
|
anon_progress = ui.row().classes("w-full place-content-center bg-blue-900")
|
|
|
|
with anon_progress:
|
|
ui.label("Anonymizing files")
|
|
progressbar = ui.linear_progress(value=0).props("instant-feedback")
|
|
anon_progress.visible = False
|
|
|
|
async def load_files_from_folder() -> None:
|
|
folder = await local_folder_picker("~")
|
|
logger.error(folder)
|
|
if folder is not None:
|
|
ui.notify(f"Selected folder: {folder}")
|
|
await load_files_start(anon_progress, queue, folder)
|
|
# return dir
|
|
else:
|
|
ui.notify("No folder selected", color="negative")
|
|
|
|
with ui.expansion("Extra", icon="build").classes("w-full"):
|
|
ui.button(
|
|
"Load files", on_click=partial(load_files_start, anon_progress, queue)
|
|
)
|
|
ui.button("Load files from folder", on_click=load_files_from_folder)
|
|
ui.button(
|
|
"Reload anon",
|
|
on_click=partial(reload_anonymized_start, anon_progress, queue),
|
|
)
|
|
|
|
with ui.expansion("File status", icon="view_list").classes("w-full"):
|
|
loaded_files_ui()
|
|
uploaded_files_ui()
|
|
|
|
# with ui.expansion('Processed files!', icon='work').classes('w-full'):
|
|
# processed_files = ui.list()
|
|
|
|
with ui.expansion("Settings", icon="settings").classes("w-full"):
|
|
ui.label("Settings").classes("text-h4")
|
|
ui.label(f"Export path: {EXPORT_PATH}")
|
|
ui.label(f"Process path: {PROCESS_PATH}")
|
|
ui.label(f"Anon path: {ANON_PATH}")
|
|
ui.button("Clear anon path", on_click=clear_anonymized_files).tooltip(
|
|
"Delete all files in ANON_PATH"
|
|
)
|
|
async def ssl_check_ui() -> None:
|
|
# Show a running dialog immediately so user sees progress
|
|
with ui.dialog() as running_dlg:
|
|
with ui.card().classes("p-4"):
|
|
ui.label("SSL check running...").classes("text-h6")
|
|
running_dlg.open()
|
|
|
|
try:
|
|
logger.debug("Starting SSL check via UI")
|
|
output = await run.cpu_bound(ssl_check_cmd, BASE_SITE_URL)
|
|
except Exception as e:
|
|
logger.exception("SSL check failed")
|
|
running_dlg.close()
|
|
ui.notify(f"SSL check failed: {e}", color="negative")
|
|
return
|
|
|
|
# close the running dialog and show the results in a new dialog
|
|
running_dlg.close()
|
|
with ui.dialog() as d:
|
|
with ui.card().classes("p-4"):
|
|
ui.label("SSL Check output").classes("text-h6")
|
|
ui.code(output).props("white-space: pre-wrap")
|
|
ui.button("Close", on_click=d.close)
|
|
d.open()
|
|
|
|
ui.button("SSL check", on_click=ssl_check_ui).tooltip("Run openssl s_client -showcerts against the configured site and show output")
|
|
|
|
def shutdown_app():
|
|
app.shutdown()
|
|
ui.run_javascript("window.close()")
|
|
|
|
async def watch_for_refresh():
|
|
global REFRESH_TRIGGERED, RELOAD_ANON_FILES
|
|
while not app.is_stopped:
|
|
if REFRESH_TRIGGERED:
|
|
REFRESH_TRIGGERED = False
|
|
await load_files_start(anon_progress, queue)
|
|
if RELOAD_ANON_FILES:
|
|
RELOAD_ANON_FILES = False
|
|
await reload_anonymized_start(anon_progress, queue)
|
|
await asyncio.sleep(1)
|
|
|
|
asyncio.create_task(watch_for_refresh())
|
|
|
|
with ui.dialog() as about_dialog:
|
|
with ui.card().classes("absolute-center"):
|
|
ui.label("Uploader tool").classes("text-h2")
|
|
ui.label("A tool to help upload files to penracourses.org.uk")
|
|
ui.image("icon/icon1.png")
|
|
ui.label("Version: 0.1")
|
|
ui.label("Author: Ross Kruger")
|
|
|
|
ui.button("Close", on_click=about_dialog.close, icon="close", color="secondary")
|
|
|
|
with ui.footer():
|
|
with ui.row():
|
|
ui.button("Shutdown", on_click=shutdown_app).props("outline color=white")
|
|
ui.button("About", on_click=about_dialog.open).props("outline color=white")
|
|
|
|
|
|
@logger.catch
|
|
def launch_app(
|
|
work_dir: Path = typer.Option(WORK_DIR, help="Working directory"),
|
|
nng: bool = typer.Option(True, help="Use nng"),
|
|
native_mode: bool = typer.Option(False, help="Use native mode"),
|
|
debug: bool = typer.Option(False, help="Debug mode"),
|
|
ca_bundle: Optional[Path] = typer.Option(
|
|
None, help="Path to a CA bundle (PEM file) to verify HTTPS connections"
|
|
),
|
|
insecure: bool = typer.Option(False, help="Disable SSL verification (insecure)"),
|
|
):
|
|
global WORK_DIR, EXPORT_PATH, PROCESS_PATH, ANON_PATH, DEBUG
|
|
|
|
WORK_DIR = work_dir
|
|
EXPORT_PATH = WORK_DIR / Path("export/")
|
|
PROCESS_PATH = WORK_DIR / Path("to_process/")
|
|
ANON_PATH = WORK_DIR / Path("anon/")
|
|
DEBUG = debug
|
|
|
|
logger.debug(f"Work dir: {WORK_DIR}")
|
|
|
|
# Configure SSL verification behaviour. VERIFY_SSL can be:
|
|
# - True (default): use system certs
|
|
# - False: disable verification (insecure)
|
|
# - path (str): path to CA bundle PEM file used by requests
|
|
global VERIFY_SSL
|
|
|
|
env_ca = (
|
|
os.environ.get("NICE_UPLOADER_CA_BUNDLE")
|
|
or os.environ.get("REQUESTS_CA_BUNDLE")
|
|
or os.environ.get("SSL_CERT_FILE")
|
|
)
|
|
|
|
if insecure:
|
|
VERIFY_SSL = False
|
|
logger.warning(
|
|
"SSL verification disabled (insecure). Only use for troubleshooting in trusted networks."
|
|
)
|
|
elif ca_bundle is not None:
|
|
VERIFY_SSL = str(ca_bundle)
|
|
logger.info(f"Using CA bundle from CLI: {VERIFY_SSL}")
|
|
elif env_ca:
|
|
VERIFY_SSL = env_ca
|
|
logger.info(f"Using CA bundle from environment: {VERIFY_SSL}")
|
|
else:
|
|
VERIFY_SSL = True
|
|
|
|
# If verification is disabled, suppress urllib3 insecure warnings to keep logs clean
|
|
if VERIFY_SSL is False:
|
|
try:
|
|
import urllib3
|
|
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
except Exception:
|
|
pass
|
|
|
|
# If a bundled cert.pem was included in the frozen app, prefer that
|
|
# when the user hasn't explicitly provided a CA bundle or disabled
|
|
# verification. This lets builders ship a corporate CA with the app.
|
|
try:
|
|
# Prefer an entire bundled cert/ folder (so users can include many
|
|
# certificate files). If present, concatenate .pem/.cer/.crt files
|
|
# into a temporary combined bundle and use that. Otherwise fall back
|
|
# to a single cert.pem file if present.
|
|
bundled_dir = base_path / "cert"
|
|
if bundled_dir.exists() and bundled_dir.is_dir() and VERIFY_SSL is True:
|
|
# Collect candidate cert files
|
|
patterns = ["*.pem", "*.cer", "*.crt"]
|
|
files = []
|
|
for p in patterns:
|
|
files.extend(sorted(bundled_dir.glob(p)))
|
|
|
|
if files:
|
|
# Create a temporary combined CA bundle file (persisted)
|
|
tmp = tempfile.NamedTemporaryFile(delete=False, prefix="nice_uploader_ca_", suffix=".pem")
|
|
included = []
|
|
try:
|
|
import base64
|
|
|
|
for f in files:
|
|
try:
|
|
with open(f, "rb") as fh:
|
|
data = fh.read()
|
|
# If the file already contains a PEM block, write as-is
|
|
if b"-----BEGIN CERTIFICATE-----" in data:
|
|
tmp.write(data)
|
|
tmp.write(b"\n")
|
|
included.append(str(f))
|
|
else:
|
|
# Assume DER/binary and convert to PEM by base64-encoding
|
|
b64 = base64.encodebytes(data)
|
|
tmp.write(b"-----BEGIN CERTIFICATE-----\n")
|
|
tmp.write(b64)
|
|
tmp.write(b"-----END CERTIFICATE-----\n")
|
|
included.append(str(f) + " (converted from DER)")
|
|
except Exception:
|
|
# skip unreadable files
|
|
logger.debug(f"Skipping unreadable cert file: {f}")
|
|
tmp.flush()
|
|
tmp.close()
|
|
VERIFY_SSL = str(Path(tmp.name))
|
|
logger.info(f"Using bundled CA bundle dir combined to {VERIFY_SSL}")
|
|
logger.debug(f"Bundled cert files: {included}")
|
|
except Exception as e:
|
|
try:
|
|
tmp.close()
|
|
except Exception:
|
|
pass
|
|
logger.debug(f"Failed to build combined CA bundle: {e}")
|
|
else:
|
|
bundled = base_path / "cert.pem"
|
|
if bundled.exists() and VERIFY_SSL is True:
|
|
VERIFY_SSL = str(bundled)
|
|
logger.info(f"Using bundled CA bundle at {VERIFY_SSL}")
|
|
except Exception:
|
|
pass
|
|
|
|
# When running as a frozen executable, the native/webview backend
|
|
# (which depends on pythonnet) can fail to load inside the bundled
|
|
# runtime. Disable native mode automatically for frozen builds so the
|
|
# app falls back to launching in the browser and remains usable.
|
|
if getattr(sys, "frozen", False):
|
|
if native_mode:
|
|
logger.debug("Running frozen executable: disabling native mode to avoid pythonnet loader issues")
|
|
native_mode = False
|
|
|
|
if nng:
|
|
socket_error = False
|
|
try:
|
|
with pynng.Pair0(listen=SOCKET_PATH) as sub:
|
|
pass
|
|
except pynng.exceptions.AddressInUse:
|
|
logger.debug("Address in use")
|
|
# Try to detect which process is holding the TCP port (helpful on Windows)
|
|
try:
|
|
port_str = SOCKET_PATH.split(':')[-1]
|
|
port = int(port_str)
|
|
except Exception:
|
|
port = None
|
|
|
|
if port:
|
|
def find_process_using_port(port: int):
|
|
"""Return a list of (pid, name) tuples for processes listening on TCP port.
|
|
|
|
Uses psutil if available; otherwise falls back to parsing `netstat -ano` and
|
|
resolving PIDs with `tasklist` on Windows. Returns empty list if none found.
|
|
"""
|
|
results = []
|
|
# Prefer psutil (fast, cross-platform) when available
|
|
try:
|
|
import psutil
|
|
|
|
for conn in psutil.net_connections(kind='tcp'):
|
|
laddr = conn.laddr
|
|
if not laddr:
|
|
continue
|
|
# laddr can be an address tuple (ip, port)
|
|
try:
|
|
conn_port = laddr.port
|
|
except Exception:
|
|
# on some platforms laddr may be a string
|
|
conn_port = int(str(laddr).split(':')[-1])
|
|
if conn_port == port and conn.status in ('LISTEN', 'LISTENING'):
|
|
pid = conn.pid
|
|
name = None
|
|
try:
|
|
if pid:
|
|
p = psutil.Process(pid)
|
|
name = p.name()
|
|
except Exception:
|
|
name = None
|
|
results.append((pid, name))
|
|
return results
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback: use netstat (Windows) and tasklist to resolve PIDs
|
|
try:
|
|
if platform.system().lower().startswith('win'):
|
|
cmd = ['netstat', '-ano', '-p', 'tcp']
|
|
proc = subprocess.run(cmd, capture_output=True, text=True)
|
|
out = proc.stdout.splitlines()
|
|
pids = set()
|
|
for line in out:
|
|
parts = line.split()
|
|
# Expected format: Proto Local Address Foreign Address State PID
|
|
if len(parts) >= 5:
|
|
local = parts[1]
|
|
state = parts[3]
|
|
pid = parts[4]
|
|
# local can be like '0.0.0.0:9976' or '[::]:9976'
|
|
if ':' in local and state.upper() in ('LISTEN', 'LISTENING'):
|
|
try:
|
|
local_port = int(local.rsplit(':', 1)[1])
|
|
except Exception:
|
|
continue
|
|
if local_port == port:
|
|
try:
|
|
pid_i = int(pid)
|
|
pids.add(pid_i)
|
|
except Exception:
|
|
pass
|
|
|
|
for pid in pids:
|
|
# Resolve process name via tasklist
|
|
try:
|
|
tl = subprocess.run(['tasklist', '/FI', f'PID eq {pid}', '/FO', 'CSV', '/NH'], capture_output=True, text=True)
|
|
line = tl.stdout.strip()
|
|
if line and '"' in line:
|
|
# CSV like "process.exe","1234","..."
|
|
cols = [c.strip('"') for c in line.split(',')]
|
|
name = cols[0] if cols else None
|
|
else:
|
|
name = None
|
|
except Exception:
|
|
name = None
|
|
results.append((pid, name))
|
|
return results
|
|
except Exception:
|
|
pass
|
|
|
|
return results
|
|
|
|
holders = find_process_using_port(port)
|
|
if holders:
|
|
for pid, name in holders:
|
|
logger.error(f"Port {port} appears in use by PID={pid} name={name}")
|
|
else:
|
|
logger.debug(f"No process detected listening on port {port} via psutil/netstat")
|
|
try:
|
|
with pynng.Pair0(dial=SOCKET_PATH, send_timeout=4000, recv_timeout=4000) as sub:
|
|
sub.send(b"loaded")
|
|
msg = sub.recv()
|
|
print(msg)
|
|
sys.exit(0)
|
|
except pynng.exceptions.Timeout:
|
|
logger.error(f"Socket timeout: {SOCKET_PATH}")
|
|
# Happens when the socket is locked
|
|
socket_error = True
|
|
pass
|
|
|
|
if not socket_error:
|
|
app.on_startup(read_nng_messages)
|
|
|
|
app.on_exception(logger.debug)
|
|
|
|
try:
|
|
port = native.find_open_port()
|
|
#port = 8001
|
|
logger.debug(f"Run on port {port}")
|
|
|
|
# If native mode was requested, ensure pywebview has a usable GUI
|
|
# backend (Qt or GTK). Use importlib.util.find_spec to check for
|
|
# available backends without importing them (avoids side-effects).
|
|
if native_mode:
|
|
try:
|
|
if importlib.util.find_spec('webview') is None:
|
|
logger.debug('pywebview package not found; disabling native mode')
|
|
native_mode = False
|
|
else:
|
|
backend_available = False
|
|
if importlib.util.find_spec('PyQt5') or importlib.util.find_spec('PySide6'):
|
|
backend_available = True
|
|
elif importlib.util.find_spec('gi'):
|
|
backend_available = True
|
|
|
|
if not backend_available:
|
|
logger.debug('pywebview backend (Qt/GTK) not available; disabling native mode')
|
|
native_mode = False
|
|
except Exception as e:
|
|
logger.debug(f'pywebview import/check failed: {e}; disabling native mode')
|
|
native_mode = False
|
|
|
|
# Try to run with the requested mode. If native startup fails at
|
|
# runtime, catch and fall back to browser mode so the app stays usable.
|
|
try:
|
|
ui.run(reload=False, show=True, port=port, native=native_mode)
|
|
except Exception as e:
|
|
logger.error(f"Failed to start UI in native mode: {e}")
|
|
if native_mode:
|
|
logger.debug("Retrying without native mode (browser fallback)")
|
|
try:
|
|
ui.run(reload=False, show=True, port=port, native=False)
|
|
except Exception as e2:
|
|
logger.error(f"Fallback to browser mode failed: {e2}")
|
|
except Exception as e:
|
|
logger.error(e)
|
|
pass
|
|
|
|
|
|
if __name__ in {"__main__", "__mp_main__"}:
|
|
freeze_support()
|
|
|
|
typer.run(launch_app)
|