JonoRens/gr_test_app
0
1# app.py2 3 4# ========== IMPORTS ==========5import os6import pickle7from pathlib import Path8import gradio as gr9import psycopg210import psycopg2.extras # for RealDictCursor (named dict rows)11import json12import ast13 14# Encryption for DB config + active-user pickle15try:16 from cryptography.fernet import Fernet, InvalidToken17except ImportError as e:18 raise ImportError(19 "cryptography is required. Install with: pip install cryptography"20 ) from e21 22# ======== MODULE IMPORTS ========23from property_module_hf import launch_property_module24from parkingbays_module_hf import launch_parkingbays_module25from permissions_module_hf import launch_permissions_module26from accesspoints_module_hf import launch_accesspoints_module27from intercom_module_hf import launch_intercom_module28from installers_module_hf import launch_installers_module29 30# ========== DB CONNECTION ==========31 32APP_DIR = Path(__file__).parent33 34def decrypt_nhost() -> dict:35 key = os.getenv("SECRET_KEY", "").strip()36 if not key:37 raise RuntimeError("SECRET_KEY not set in this Space (Settings → Secrets).")38 39 enc_path = APP_DIR / "nhost_params.enc"40 if not enc_path.exists():41 raise RuntimeError("Encrypted DB config (nhost_params.enc) not found in repo.")42 43 token = enc_path.read_bytes()44 decrypted = Fernet(key.encode("utf-8")).decrypt(token).decode("utf-8")45 46 # Prefer JSON; allow Python-literal fallback47 try:48 params = json.loads(decrypted)49 except json.JSONDecodeError:50 params = ast.literal_eval(decrypted) # handles "{'host': ...}" format51 52 # Optional: normalize port to int53 if "port" in params:54 try:55 params["port"] = int(params["port"])56 except Exception:57 pass58 59 if not isinstance(params, dict):60 raise TypeError("Decrypted nhost config is not a dict.")61 62 return params63 64 65def get_connection():66 params = decrypt_nhost()67 if isinstance(params, str):68 raise TypeError("nhost params is a string; expected dict. Do not json.dumps() the config.")69 return psycopg2.connect(**params) # ← use params, not another decrypt_nhost()70 71 72#===========================EVERYTHING BELOW GETS COPIED TO APP.PY73 74# ========== SETTINGS ==========75PropUser1 = None # start empty; user must enter a valid ID76HERE = Path(__file__).resolve().parent77KEY_FILE = HERE / "openqr_secret.key"78DATA_FILE = HERE / "openqr_active_user.pkl.enc"79 80# ========== STYLES ==========81TAB_CSS = """82body,83.gradio-container,84.gradio-container .tabs,85.gradio-container .tab-nav,86.gradio-container .tabitem {87 background: #ffffff !important;88 box-shadow: none !important;89}90 91/* Load user button color */92#load-user-btn {93 background: #374628 !important;94 border-color: #374628 !important;95 color: #ffffff !important;96}97#load-user-btn:hover {98 filter: brightness(0.95);99}100"""101 102#=========== GET USER ID103def _ctx_get_user_id():104 """105 Returns the active Manager User ID set by the Load User flow.106 We treat '0' and empty as 'unset' so modules don't prematurely try to load data.107 """108 v = os.getenv("OPENQR_ACTIVE_USER_ID", "").strip()109 try:110 n = int(v) if v else None111 return None if (n in (None, 0)) else n112 except Exception:113 return None114 115# ========== ENCRYPTED FILE HELPERS ==========116def _ensure_key() -> bytes:117 if KEY_FILE.exists():118 return KEY_FILE.read_bytes()119 key = Fernet.generate_key()120 KEY_FILE.write_bytes(key)121 return key122 123def _write_encrypted_pickle(data: dict):124 key = _ensure_key()125 f = Fernet(key)126 blob = pickle.dumps(data, protocol=pickle.HIGHEST_PROTOCOL)127 DATA_FILE.write_bytes(f.encrypt(blob))128 129def _read_encrypted_pickle() -> dict | None:130 """131 Returns dict with (possibly stale) ActiveUserID/Name/CoName, or None on failure.132 We only use the ID; name/company are refreshed from DB.133 """134 try:135 key = KEY_FILE.read_bytes()136 f = Fernet(key)137 blob = DATA_FILE.read_bytes()138 data = pickle.loads(f.decrypt(blob))139 return data if isinstance(data, dict) else None140 except (FileNotFoundError, InvalidToken, OSError, ValueError) as e:141 print(f"_read_encrypted_pickle() warning: {e}")142 return None143 144# ========== FIELD EXTRACTION HELPERS ==========145_COMPANY_KEYS = (146 "UserCompanyName", # ← add this (likely your actual column)147 "UserCompany", # ← optional common variant148 "MUserCompanyName",149 "MUserCoName",150 "MUserCompany",151 "CompanyName",152 "Company",153 "MUserCo",154)155 156 157def _extract_user_fields(row: dict):158 """159 Accepts a dict row from managerusers (RealDictCursor).160 Returns: ActiveUserID, ActiveUserName, ActiveUserCoName, IsActive, IsInstaller161 """162 uid = row.get("MUserID")163 name = row.get("MUserName") or ""164 company = ""165 for k in _COMPANY_KEYS:166 if k in row and row[k] is not None:167 v = str(row[k]).strip()168 if v != "":169 company = v170 break171 172 is_active = bool(row.get("MUserActive") or row.get("Active") or row.get("IsActive") or False)173 installers = bool(row.get("Installers", False)) # default False if col missing174 uid = int(uid) if uid is not None else None175 return uid, name, company, is_active, installers176 177# ========== HELP MODULES ==========178def launch_help_module():179 with gr.Blocks() as help_app:180 with gr.Tabs():181 with gr.TabItem("Overview"):182 gr.Markdown("◉ **RyGo Desktop** is a desktop app for property managers")183 gr.Markdown("◉ The app has 5 modules: **Property**, **Parking Bays**, **Permissions**, **Intercoms** and **Access Points**")184 gr.Markdown("**To be developed:**")185 gr.Markdown(" ▷ manager sign-up")186 gr.Markdown(" ▷ when change / delete a LeaseID in parking bays, ensure all tenant users are marked as inactive")187 gr.Markdown(" ▷ when change / delete a BayNo in parking bays, ensure allocated bays are cleared from tenant users")188 gr.Markdown(" ▷ add intercom report to property")189 gr.Markdown(" ▷ add report identifying tenant users who have access to a particular access point")190 gr.Markdown(" ▷ add a PDF column in property table for bay drawings; provide upload/download")191 gr.Markdown(" ▷ add licence plate to tenant user & include in parking bays report")192 gr.Markdown(" ▷ export reports to Excel")193 194 with gr.TabItem("Properties"):195 gr.Markdown("◉ Manage properties (add, edit, delete) and property-related reports.")196 gr.Markdown("◉ Reports include:")197 gr.Markdown(" ▷ Bay summary (loaded / let / allocated)")198 gr.Markdown(" ▷ All parking bays at a property with details of lets & allocations")199 gr.Markdown(" ▷ List of intercoms")200 gr.Markdown(" ▷ List of access points (AP)")201 gr.Markdown(" ▷ Individual activity report")202 203 204 with gr.TabItem("Permissions"):205 gr.Markdown("◉ Manage tenant users connected to a property (add, edit, delete).")206 gr.Markdown("◉ Enter tenant's name & mobile, connect to lease, bay, and other access points.")207 208 with gr.TabItem("Access Points"):209 gr.Markdown("◉ Manage access points at a property, set Unrestricted/Restricted (permissions required if Restricted).")210 211# with gr.TabItem("Intercoms"):212# gr.Markdown("◉ Intercoms module overview (coming features).")213 214 215 with gr.TabItem("ays & Leases"):216 gr.Markdown("◉ Manage parking bays at a property (add, edit, delete).")217 gr.Markdown("◉ Assign bays to tenants via Lease ID and connect restricted access points.")218 219 220 return help_app221 222def launch_installer_help_module():223 with gr.Blocks() as help_app:224 with gr.Tabs():225 with gr.TabItem("Installer overview"):226 gr.Markdown("◉ The Installer app gives the installer access to all Access Points (AP) allocated to the installer across properties")227 gr.Markdown("◉ Basic steps:")228 gr.Markdown("1) Choose Property tab.\n"229 "2) Choose row to edit.\n"230 "3) Insert / edit Name of device at AP, the location of access point and the API for the AP.\n"231 "4) Return to main menu when done.")232 return help_app233 234# ========== MAIN APP ==========235def main_app():236 237 # Ensure no stale value at process start; user must enter a UserID each run.238 os.environ.pop("OPENQR_ACTIVE_USER_ID", None)239 240 with gr.Blocks() as app:241 # Global CSS242 gr.HTML(f"<style>{TAB_CSS}</style>")243 244 # Header + status245 header_md = gr.Markdown('<div style="color:#374628;"><h2 style="margin:0;">RyGo Admin Dashboard</h2></div>')246 status_md = gr.Markdown("", visible=False)247 248 # User input row249 with gr.Row():250 muser_in = gr.Number(label="Manager User ID", precision=0, value=PropUser1)251 with gr.Row():252 load_btn = gr.Button("Load user", variant="primary", elem_id="load-user-btn")253 254 # --- Active User ID (edit step, shown only when an ID already exists) ---255 with gr.Row():256 curr_uid_md = gr.Markdown("") # shows "Current Active User ID: `...`" when present257 with gr.Row():258 edit_uid_tb = gr.Textbox(label="Active User ID (edit if needed)", value="", visible=False)259 with gr.Row():260 confirm_uid_btn = gr.Button("Enter / Edit User ID", variant="secondary", visible=False)261 with gr.Row():262 uid_note_md = gr.Markdown("")263 264 # App-wide state (kept for compatibility)265 s_user_id = gr.State(None)266 s_user_name = gr.State("")267 s_user_coname = gr.State("")268 269 # --- init current-id display & prefill edit box ---270 def _init_uid_display(uid):271 has_uid = uid not in (None, "")272 md_text = f"**Current Active User ID:** `{uid}`" if has_uid else ""273 return (274 md_text,275 gr.update(value=(uid if has_uid else ""), visible=has_uid), # edit_uid_tb276 gr.update(visible=has_uid), # confirm_uid_btn277 gr.update(value="") # uid_note_md cleared278 )279 280 # ===== Standard Manager Tabs (hidden until user is validated) =====281 tabs_group = gr.Group(visible=False)282 with tabs_group:283 with gr.Tabs():284 285 286 with gr.TabItem("🏢 Property"):287 prop_ui = launch_property_module(get_user_id=_ctx_get_user_id)288 289 with gr.TabItem("👷 Permissions"):290 tenant_ui = launch_permissions_module(get_user_id=_ctx_get_user_id)291 292 with gr.TabItem("🚥 Access Points"):293 access_ui = launch_accesspoints_module(get_user_id=_ctx_get_user_id)294 295 with gr.TabItem("📱 Intercoms"):296 intercom_ui = launch_intercom_module(get_user_id=_ctx_get_user_id)297 298 with gr.TabItem("🅿️ Bays & Leases"):299 bays_ui = launch_parkingbays_module(get_user_id=_ctx_get_user_id)300 301 302 with gr.TabItem("Help"):303 _ = launch_help_module()304 305 # ===== Installer Tabs (hidden until user is validated and flagged as installer) =====306 installer_tabs_group = gr.Group(visible=False)307 with installer_tabs_group:308 with gr.Tabs():309 with gr.TabItem("Installer's APs"):310 # IMPORTANT: pass the get_user_id callback so the module can resolve after load311 installers_ui = launch_installers_module(get_user_id=_ctx_get_user_id)312 with gr.TabItem("Installer help"):313 _ = launch_installer_help_module()314 315 # ===== Load user handler (manual flow) =====316 def load_user(muser_id):317 # 1) Validate numeric input318 try:319 muser_id = int(muser_id)320 except (TypeError, ValueError):321 return (322 gr.update(value='<div style="color:#374628;"><h2 style="margin:0;">RyGo Admin Dashboard</h2></div>\n⚠️ Enter a valid numeric Manager User ID.'),323 gr.update(value="", visible=True),324 gr.update(visible=False), # tabs_group325 gr.update(visible=False), # installer_tabs_group326 None, "", "",327 gr.update(), # keep input328 gr.update() # keep button329 )330 331 # 2) Check user existence & active status using named dict rows332 with get_connection() as conn:333 with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:334 cur.execute(335 'SELECT * FROM managerusers WHERE "MUserID" = %s',336 (muser_id,),337 )338 row = cur.fetchone()339 340 if not row:341 return (342 gr.update(value='<div style="color:#374628;"><h2 style="margin:0;">RyGo Admin Dashboard</h2></div>\nManager user does not exist. Please enter a correct User ID.'),343 gr.update(value="", visible=True),344 gr.update(visible=False),345 gr.update(visible=False),346 None, "", "",347 gr.update(), gr.update()348 )349 350 ActiveUserID, ActiveUserName, ActiveUserCoName, is_active, is_installer = _extract_user_fields(row)351 352 if not is_active:353 return (354 gr.update(value='<div style="color:#374628;"><h2 style="margin:0;">RyGo Admin Dashboard</h2></div>\nManager user exists but is not active.'),355 gr.update(value="", visible=True),356 gr.update(visible=False),357 gr.update(visible=False),358 None, "", "",359 gr.update(), gr.update()360 )361 362 # 3) Success → write encrypted pickle, set env var, show appropriate tabs, clear+hide inputs363 try:364 _write_encrypted_pickle(365 {366 "ActiveUserID": ActiveUserID,367 "ActiveUserName": ActiveUserName,368 "ActiveUserCoName": ActiveUserCoName,369 "Installers": is_installer,370 }371 )372 except Exception as e:373 print(f"WARNING: failed to write encrypted active user file: {e}")374 375 if ActiveUserID is not None:376 os.environ["OPENQR_ACTIVE_USER_ID"] = str(ActiveUserID) # cached for modules377 378 header = (379 '<div style="color:#374628;"><h2 style="margin:0;">RyGo Admin Dashboard</h2></div>\n'380 f"**User:** {ActiveUserName} (ID: {ActiveUserID} ) **Company:** {ActiveUserCoName}"381 )382 383 # Toggle groups per Installers flag384 show_manager = not is_installer385 show_installer = is_installer386 387 return (388 gr.update(value=header), # header_md389 gr.update(value="", visible=False), # status_md hidden to remove gap390 gr.update(visible=show_manager), # tabs_group391 gr.update(visible=show_installer), # installer_tabs_group392 ActiveUserID, ActiveUserName, ActiveUserCoName,393 gr.update(value=None, visible=False), # muser_in → clear+hide394 gr.update(visible=False), # load_btn → hide395 )396 397 # Wire the button (NOTE: +1 output for installer_tabs_group)398 load_btn.click(399 load_user,400 inputs=[muser_in],401 outputs=[402 header_md, status_md,403 tabs_group, installer_tabs_group, # both groups404 s_user_id, s_user_name, s_user_coname,405 muser_in, load_btn406 ]407 )408 409 # --- confirm & check flow (reuse the existing load_user pipeline) ---410 def _set_muser_in_from_text(txt_value):411 try:412 return int(str(txt_value).strip())413 except Exception:414 return None415 416 def _post_check_ui(s_uid):417 """418 After calling load_user(), decide whether to show an error note or hide the edit UI.419 - If s_user_id is None/empty → user doesn't exist: show note and keep input/button visible.420 - If s_user_id is valid → user exists: clear note, hide input + button.421 """422 if s_uid in (None, ""):423 return (424 "User ID does not exist - try again",425 gr.update(visible=True), # edit_uid_tb stays visible for retry426 gr.update(visible=True) # confirm button stays visible427 )428 else:429 return (430 "",431 gr.update(value="", visible=False), # hide & clear input432 gr.update(visible=False) # hide button433 )434 435 confirm_uid_btn.click(436 _set_muser_in_from_text,437 inputs=[edit_uid_tb],438 outputs=[muser_in],439 ).then(440 load_user,441 inputs=[muser_in],442 outputs=[443 header_md, status_md,444 tabs_group, installer_tabs_group,445 s_user_id, s_user_name, s_user_coname,446 muser_in, load_btn447 ]448 ).then(449 _post_check_ui,450 inputs=[s_user_id],451 outputs=[uid_note_md, edit_uid_tb, confirm_uid_btn],452 )453 454 # We intentionally do not auto-load any saved user on startup:455 # user must enter a UserID each run per your (a)-(e) flow.456 app.load(457 _init_uid_display,458 inputs=[s_user_id],459 outputs=[curr_uid_md, edit_uid_tb, confirm_uid_btn, uid_note_md],460 )461 462 # IMPORTANT: return the Blocks object so app is not None463 return app464 465# ========== ENTRY POINT ==========466if __name__ == "__main__":467 app = main_app()468 app.launch()469 # app.launch(server_name="127.0.0.1", server_port=7860, inbrowser=True, show_error=True)470 471 472 473