# ----------------------------------------------------------------------------- # Description: Role-aware PDK access control utilities for resolving which PDK roots a user or project may use. # Inside functions: normalize_user_group, pdk_root_for_group, pdk_root_for_session, prefer_full_gds_for_session, create_export_path, cleanup_expired_exports # Developer : Qin Yue @ 2026 # Organization : OptiHK Limited # ----------------------------------------------------------------------------- import os import time import shutil import uuid # Role names control which PDK tree the backend exposes for browsing, preview, # and GDS export requests. MANAGER_GROUP = "manager" DEVELOPER_GROUP = "developers" USER_GROUP = "user" ALLOWED_GROUPS = {MANAGER_GROUP, DEVELOPER_GROUP, USER_GROUP} def normalize_user_group(user_group: str) -> str: """Normalize a session user group into the supported PDK access scope.""" group = (user_group or USER_GROUP).strip().lower() return group if group in ALLOWED_GROUPS else USER_GROUP def pdk_root_for_group(user_group: str, repo_root: str) -> str: """Resolve the PDK library root that belongs to a user group.""" group = normalize_user_group(user_group) # Managers may access the private atlas PDK tree; all other roles stay on # the public PDK tree used for normal project editing. if group == MANAGER_GROUP: return os.path.abspath(os.environ.get( "MXPIC_PDK_ATLAS_ROOT", os.path.join(repo_root, "opt_pdk_atlas", "foundries"), )) return os.path.abspath(os.environ.get( "MXPIC_PDK_PUBLIC_ROOT", os.path.join(repo_root, "opt_pdk_public", "foundries"), )) def pdk_root_for_session(session_obj, repo_root: str) -> str: """Resolve the active PDK root from the current Flask session.""" return pdk_root_for_group(session_obj.get("user_group"), repo_root) def prefer_full_gds_for_session(session_obj) -> bool: """Decide whether resolved PDK assets should prefer full GDS files.""" return normalize_user_group(session_obj.get("user_group")) == MANAGER_GROUP def create_export_path(export_root: str, project_name: str) -> tuple[str, str, str]: """Create a unique export directory and filename for generated downloads.""" export_id = uuid.uuid4().hex filename = f"{project_name}.gds" export_dir = os.path.abspath(os.path.join(export_root, export_id)) os.makedirs(export_dir, exist_ok=True) return export_id, filename, os.path.join(export_dir, filename) def cleanup_expired_exports(export_root: str, max_age_seconds: int = 86400) -> None: """Remove old export folders so temporary download storage stays bounded.""" if not os.path.isdir(export_root): return now = time.time() # Each export is stored in its own UUID directory, so old folders can be # removed independently without touching active project data. for name in os.listdir(export_root): path = os.path.join(export_root, name) if not os.path.isdir(path): continue try: age = now - os.path.getmtime(path) if age > max_age_seconds: shutil.rmtree(path, ignore_errors=True) except OSError: continue