79 lines
3.2 KiB
Python
79 lines
3.2 KiB
Python
# -----------------------------------------------------------------------------
|
|
# 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
|