3654 lines
117 KiB
Python
3654 lines
117 KiB
Python
"""HTMX-driven admin dashboard for users, roles, and thread standards."""
|
||
|
||
import math
|
||
import urllib.parse
|
||
|
||
from fastapi import (
|
||
APIRouter,
|
||
Depends,
|
||
File,
|
||
Form,
|
||
HTTPException,
|
||
Request,
|
||
UploadFile,
|
||
status,
|
||
)
|
||
from fastapi.responses import HTMLResponse, RedirectResponse, Response
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.database import AsyncSessionLocal, get_db
|
||
from app.crud import audit as audit_crud
|
||
from app.crud import augmented_part as augmented_part_crud
|
||
from app.crud import augmented_part_note as augmented_part_note_crud
|
||
from app.crud import imported_part as imported_part_crud
|
||
from app.crud import imported_part_config as imported_part_config_crud
|
||
from app.crud import iso_thread as iso_thread_crud
|
||
from app.crud import manufacturer as manufacturer_crud
|
||
from app.crud import part_colour as part_colour_crud
|
||
from app.crud import part_type as part_type_crud
|
||
from app.crud import role as role_crud
|
||
from app.crud import scope_info as scope_info_crud
|
||
from app.crud import scope_model as scope_model_crud
|
||
from app.crud import attachment_point as attachment_point_crud
|
||
from app.crud import attachment_point_name as apn_crud
|
||
from app.crud import clamp_connection as clamp_connection_crud
|
||
from app.crud import fit as fit_crud
|
||
from app.crud import fit_note as fit_note_crud
|
||
from app.crud import static_connection as static_connection_crud
|
||
from app.crud import sub_part_type as sub_part_type_crud
|
||
from app.crud import unc_thread as unc_thread_crud
|
||
from app.crud import user as user_crud
|
||
from app.deps import get_current_web_user
|
||
from app.models.fit import FitLabel
|
||
from app.models.unc_thread import UncSeries
|
||
from app.models.user import User
|
||
from app.services import augmented_part_sync as augmented_part_sync_service
|
||
from app.services import danger_zone as danger_zone_service
|
||
from app.services import fit_matching as fit_matching_service
|
||
from app.services import jobs
|
||
from app.services import odata as odata_service
|
||
from app.services import scope_import as scope_import_service
|
||
from app.web.templates import templates
|
||
|
||
router = APIRouter(prefix="/admin", include_in_schema=False)
|
||
|
||
_PAGE_SIZE = 50
|
||
|
||
|
||
def _query_url(base_path: str, params: dict) -> str:
|
||
clean = {k: v for k, v in params.items() if v not in (None, "", 0)}
|
||
return f"{base_path}?{urllib.parse.urlencode(clean)}"
|
||
|
||
|
||
def _paginate(
|
||
base_path: str,
|
||
filter_params: dict,
|
||
page: int,
|
||
page_size: int,
|
||
total: int,
|
||
target_id: str,
|
||
) -> dict:
|
||
total_pages = max(1, math.ceil(total / page_size))
|
||
page = max(1, min(page, total_pages))
|
||
skip = (page - 1) * page_size
|
||
|
||
def _url(p: int) -> str:
|
||
return _query_url(
|
||
base_path, {**filter_params, "page": p, "page_size": page_size}
|
||
)
|
||
|
||
return {
|
||
"page": page,
|
||
"page_size": page_size,
|
||
"total": total,
|
||
"total_pages": total_pages,
|
||
"skip": skip,
|
||
"showing_from": skip + 1 if total > 0 else 0,
|
||
"showing_to": min(skip + page_size, total),
|
||
"prev_url": _url(page - 1) if page > 1 else None,
|
||
"next_url": _url(page + 1) if page < total_pages else None,
|
||
"target_id": target_id,
|
||
}
|
||
|
||
|
||
def _sort_headers(
|
||
base_path: str,
|
||
filter_params: dict,
|
||
columns: list[str],
|
||
sort: str,
|
||
direction: str,
|
||
target_id: str,
|
||
) -> dict:
|
||
"""Per-column header link context for click-to-sort table headers.
|
||
|
||
``filter_params`` should NOT include "sort"/"dir"/"page" — those are added here.
|
||
"""
|
||
headers = {}
|
||
for key in columns:
|
||
next_dir = "desc" if sort == key and direction == "asc" else "asc"
|
||
headers[key] = {
|
||
"url": _query_url(
|
||
base_path,
|
||
{**filter_params, "sort": key, "dir": next_dir, "page": 1},
|
||
),
|
||
"active": sort == key,
|
||
"dir": direction if sort == key else None,
|
||
"target_id": target_id,
|
||
}
|
||
return headers
|
||
|
||
|
||
def web_require(*perms: str):
|
||
"""Web dependency: allow superusers or holders of all given permissions."""
|
||
|
||
async def checker(user: User = Depends(get_current_web_user)) -> User:
|
||
if user.is_superuser or set(perms).issubset(user.permissions):
|
||
return user
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="You do not have access to this page.",
|
||
)
|
||
|
||
return checker
|
||
|
||
|
||
async def web_require_superuser(user: User = Depends(get_current_web_user)) -> User:
|
||
"""Web dependency: superusers only, no permission-string bypass."""
|
||
if not user.is_superuser:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="You do not have access to this page.",
|
||
)
|
||
return user
|
||
|
||
|
||
async def _role_perm_ctx(db: AsyncSession) -> dict:
|
||
all_perms = await role_crud.list_permissions(db)
|
||
scopes = sorted({p.name.split(":")[0] for p in all_perms})
|
||
perm_map = {p.name: p for p in all_perms}
|
||
return {
|
||
"permissions": all_perms,
|
||
"scopes": scopes,
|
||
"perm_map": perm_map,
|
||
"actions": ["read", "write", "delete"],
|
||
}
|
||
|
||
|
||
@router.get("", response_class=HTMLResponse)
|
||
async def admin_home():
|
||
return RedirectResponse(url="/admin/users", status_code=status.HTTP_302_FOUND)
|
||
|
||
|
||
# --- Users -------------------------------------------------------------------
|
||
|
||
|
||
@router.get("/users", response_class=HTMLResponse)
|
||
async def users_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("users:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/users.html",
|
||
{
|
||
"user": user,
|
||
"users": await user_crud.list_users(db),
|
||
"roles": await role_crud.list_roles(db),
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/users", response_class=HTMLResponse)
|
||
async def create_user(
|
||
request: Request,
|
||
email: str = Form(...),
|
||
password: str = Form(...),
|
||
full_name: str = Form(""),
|
||
is_superuser: bool = Form(False),
|
||
role_ids: list[int] = Form(default=[]),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("users:write")),
|
||
):
|
||
if not await user_crud.get_by_email(db, email):
|
||
created = await user_crud.create(
|
||
db,
|
||
email=email,
|
||
password=password,
|
||
full_name=full_name or None,
|
||
is_superuser=is_superuser,
|
||
role_ids=role_ids,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="User",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("User", created.id, created.email),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_users_table.html",
|
||
{
|
||
"users": await user_crud.list_users(db),
|
||
"roles": await role_crud.list_roles(db),
|
||
},
|
||
)
|
||
|
||
|
||
@router.get("/users/{user_id}/edit", response_class=HTMLResponse)
|
||
async def edit_user_row(
|
||
request: Request,
|
||
user_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("users:write")),
|
||
):
|
||
target = await user_crud.get(db, user_id)
|
||
if target is None:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_user_row_edit.html",
|
||
{"u": target, "roles": await role_crud.list_roles(db)},
|
||
)
|
||
|
||
|
||
@router.get("/users/{user_id}/row", response_class=HTMLResponse)
|
||
async def user_row(
|
||
request: Request,
|
||
user_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("users:read")),
|
||
):
|
||
target = await user_crud.get(db, user_id)
|
||
if target is None:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
return templates.TemplateResponse(request, "admin/_user_row.html", {"u": target})
|
||
|
||
|
||
@router.post("/users/{user_id}", response_class=HTMLResponse)
|
||
async def update_user(
|
||
request: Request,
|
||
user_id: int,
|
||
full_name: str = Form(""),
|
||
password: str = Form(""),
|
||
is_active: bool = Form(False),
|
||
is_superuser: bool = Form(False),
|
||
role_ids: list[int] = Form(default=[]),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("users:write")),
|
||
):
|
||
target = await user_crud.get(db, user_id)
|
||
if target is None:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
changes = {
|
||
"full_name": full_name or None,
|
||
"password": password or None,
|
||
"is_active": is_active,
|
||
"is_superuser": is_superuser,
|
||
"role_ids": role_ids,
|
||
}
|
||
await user_crud.update(db, target, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="User",
|
||
resource_id=user_id,
|
||
description=audit_crud.describe_update("User", user_id, changes),
|
||
)
|
||
return templates.TemplateResponse(request, "admin/_user_row.html", {"u": target})
|
||
|
||
|
||
@router.delete("/users/{user_id}")
|
||
async def delete_user(
|
||
user_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("users:delete")),
|
||
):
|
||
target = await user_crud.get(db, user_id)
|
||
if target is not None:
|
||
email = target.email
|
||
await user_crud.delete(db, target)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="User",
|
||
resource_id=user_id,
|
||
description=audit_crud.describe_delete("User", user_id, email),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Roles -------------------------------------------------------------------
|
||
|
||
|
||
@router.get("/roles", response_class=HTMLResponse)
|
||
async def roles_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("roles:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/roles.html",
|
||
{
|
||
"user": user,
|
||
"roles": await role_crud.list_roles(db),
|
||
**(await _role_perm_ctx(db)),
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/roles", response_class=HTMLResponse)
|
||
async def create_role(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
description: str = Form(""),
|
||
permission_ids: list[int] = Form(default=[]),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("roles:write")),
|
||
):
|
||
if not await role_crud.get_by_name(db, name):
|
||
created = await role_crud.create(
|
||
db,
|
||
name=name,
|
||
description=description or None,
|
||
permission_ids=permission_ids,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="Role",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("Role", created.id, created.name),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_roles_table.html",
|
||
{"roles": await role_crud.list_roles(db), **(await _role_perm_ctx(db))},
|
||
)
|
||
|
||
|
||
@router.get("/roles/{role_id}/edit", response_class=HTMLResponse)
|
||
async def edit_role_row(
|
||
request: Request,
|
||
role_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("roles:write")),
|
||
):
|
||
role = await role_crud.get(db, role_id)
|
||
if role is None:
|
||
raise HTTPException(status_code=404, detail="Role not found")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_role_row_edit.html",
|
||
{"r": role, **(await _role_perm_ctx(db))},
|
||
)
|
||
|
||
|
||
@router.get("/roles/{role_id}/row", response_class=HTMLResponse)
|
||
async def role_row(
|
||
request: Request,
|
||
role_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("roles:read")),
|
||
):
|
||
role = await role_crud.get(db, role_id)
|
||
if role is None:
|
||
raise HTTPException(status_code=404, detail="Role not found")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_role_row.html",
|
||
{"r": role, **(await _role_perm_ctx(db))},
|
||
)
|
||
|
||
|
||
@router.post("/roles/{role_id}", response_class=HTMLResponse)
|
||
async def update_role(
|
||
request: Request,
|
||
role_id: int,
|
||
permission_ids: list[int] = Form(default=[]),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("roles:write")),
|
||
):
|
||
role = await role_crud.get(db, role_id)
|
||
if role is None:
|
||
raise HTTPException(status_code=404, detail="Role not found")
|
||
changes = {"permission_ids": permission_ids}
|
||
await role_crud.update(db, role, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="Role",
|
||
resource_id=role_id,
|
||
description=audit_crud.describe_update("Role", role_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_role_row.html",
|
||
{"r": role, **(await _role_perm_ctx(db))},
|
||
)
|
||
|
||
|
||
@router.post("/roles/{role_id}/default", response_class=HTMLResponse)
|
||
async def make_default_role(
|
||
request: Request,
|
||
role_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("roles:write")),
|
||
):
|
||
role = await role_crud.get(db, role_id)
|
||
if role is None:
|
||
raise HTTPException(status_code=404, detail="Role not found")
|
||
await role_crud.set_default(db, role)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="Role",
|
||
resource_id=role_id,
|
||
description=f"Set role {role.name} (#{role_id}) as the default role",
|
||
)
|
||
# Re-render the whole table so every row's default badge refreshes.
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_roles_table.html",
|
||
{"roles": await role_crud.list_roles(db), **(await _role_perm_ctx(db))},
|
||
)
|
||
|
||
|
||
@router.delete("/roles/{role_id}")
|
||
async def delete_role(
|
||
role_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("roles:write")),
|
||
):
|
||
role = await role_crud.get(db, role_id)
|
||
if role is not None:
|
||
name = role.name
|
||
await role_crud.delete(db, role)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="Role",
|
||
resource_id=role_id,
|
||
description=audit_crud.describe_delete("Role", role_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- ISO Threads -------------------------------------------------------------
|
||
|
||
|
||
@router.get("/iso-threads", response_class=HTMLResponse)
|
||
async def iso_threads_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("iso_threads:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/iso_threads.html",
|
||
{"user": user, "threads": await iso_thread_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/iso-threads", response_class=HTMLResponse)
|
||
async def create_iso_thread(
|
||
request: Request,
|
||
diameter_mm: float = Form(...),
|
||
pitch_mm: float = Form(...),
|
||
depth_mm: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("iso_threads:write")),
|
||
):
|
||
thread = await iso_thread_crud.create(
|
||
db,
|
||
diameter_mm=diameter_mm,
|
||
pitch_mm=pitch_mm,
|
||
depth_mm=float(depth_mm) if depth_mm.strip() else None,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="IsoThread",
|
||
resource_id=thread.id,
|
||
description=audit_crud.describe_create(
|
||
"IsoThread", thread.id, f"M{thread.diameter_mm}×{thread.pitch_mm}"
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_iso_threads_table.html",
|
||
{"threads": await iso_thread_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/iso-threads/{thread_id}/edit", response_class=HTMLResponse)
|
||
async def edit_iso_thread_row(
|
||
request: Request,
|
||
thread_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("iso_threads:write")),
|
||
):
|
||
thread = await iso_thread_crud.get(db, thread_id)
|
||
if thread is None:
|
||
raise HTTPException(status_code=404, detail="ISO thread not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_iso_thread_row_edit.html", {"t": thread}
|
||
)
|
||
|
||
|
||
@router.get("/iso-threads/{thread_id}/row", response_class=HTMLResponse)
|
||
async def iso_thread_row(
|
||
request: Request,
|
||
thread_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("iso_threads:read")),
|
||
):
|
||
thread = await iso_thread_crud.get(db, thread_id)
|
||
if thread is None:
|
||
raise HTTPException(status_code=404, detail="ISO thread not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_iso_thread_row.html", {"t": thread}
|
||
)
|
||
|
||
|
||
@router.post("/iso-threads/{thread_id}", response_class=HTMLResponse)
|
||
async def update_iso_thread(
|
||
request: Request,
|
||
thread_id: int,
|
||
diameter_mm: float = Form(...),
|
||
pitch_mm: float = Form(...),
|
||
depth_mm: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("iso_threads:write")),
|
||
):
|
||
thread = await iso_thread_crud.get(db, thread_id)
|
||
if thread is None:
|
||
raise HTTPException(status_code=404, detail="ISO thread not found")
|
||
changes = {
|
||
"diameter_mm": diameter_mm,
|
||
"pitch_mm": pitch_mm,
|
||
"depth_mm": float(depth_mm) if depth_mm.strip() else None,
|
||
}
|
||
await iso_thread_crud.update(db, thread, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="IsoThread",
|
||
resource_id=thread_id,
|
||
description=audit_crud.describe_update("IsoThread", thread_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_iso_thread_row.html", {"t": thread}
|
||
)
|
||
|
||
|
||
@router.delete("/iso-threads/{thread_id}")
|
||
async def delete_iso_thread(
|
||
thread_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("iso_threads:delete")),
|
||
):
|
||
thread = await iso_thread_crud.get(db, thread_id)
|
||
if thread is not None:
|
||
name = f"M{thread.diameter_mm}×{thread.pitch_mm}"
|
||
await iso_thread_crud.delete(db, thread)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="IsoThread",
|
||
resource_id=thread_id,
|
||
description=audit_crud.describe_delete("IsoThread", thread_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- UNC Threads -------------------------------------------------------------
|
||
|
||
|
||
@router.get("/unc-threads", response_class=HTMLResponse)
|
||
async def unc_threads_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("unc_threads:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/unc_threads.html",
|
||
{
|
||
"user": user,
|
||
"threads": await unc_thread_crud.list_all(db),
|
||
"series_options": list(UncSeries),
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/unc-threads", response_class=HTMLResponse)
|
||
async def create_unc_thread(
|
||
request: Request,
|
||
diameter_fraction: str = Form(...),
|
||
tpi: int = Form(...),
|
||
series: UncSeries = Form(...),
|
||
depth_in: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("unc_threads:write")),
|
||
):
|
||
thread = await unc_thread_crud.create(
|
||
db,
|
||
diameter_fraction=diameter_fraction,
|
||
tpi=tpi,
|
||
series=series,
|
||
depth_in=float(depth_in) if depth_in.strip() else None,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="UncThread",
|
||
resource_id=thread.id,
|
||
description=audit_crud.describe_create(
|
||
"UncThread",
|
||
thread.id,
|
||
f"{thread.diameter_fraction}-{thread.tpi} {thread.series.value}",
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_unc_threads_table.html",
|
||
{"threads": await unc_thread_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/unc-threads/{thread_id}/edit", response_class=HTMLResponse)
|
||
async def edit_unc_thread_row(
|
||
request: Request,
|
||
thread_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("unc_threads:write")),
|
||
):
|
||
thread = await unc_thread_crud.get(db, thread_id)
|
||
if thread is None:
|
||
raise HTTPException(status_code=404, detail="UNC thread not found")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_unc_thread_row_edit.html",
|
||
{"t": thread, "series_options": list(UncSeries)},
|
||
)
|
||
|
||
|
||
@router.get("/unc-threads/{thread_id}/row", response_class=HTMLResponse)
|
||
async def unc_thread_row(
|
||
request: Request,
|
||
thread_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("unc_threads:read")),
|
||
):
|
||
thread = await unc_thread_crud.get(db, thread_id)
|
||
if thread is None:
|
||
raise HTTPException(status_code=404, detail="UNC thread not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_unc_thread_row.html", {"t": thread}
|
||
)
|
||
|
||
|
||
@router.post("/unc-threads/{thread_id}", response_class=HTMLResponse)
|
||
async def update_unc_thread(
|
||
request: Request,
|
||
thread_id: int,
|
||
diameter_fraction: str = Form(...),
|
||
tpi: int = Form(...),
|
||
series: UncSeries = Form(...),
|
||
depth_in: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("unc_threads:write")),
|
||
):
|
||
thread = await unc_thread_crud.get(db, thread_id)
|
||
if thread is None:
|
||
raise HTTPException(status_code=404, detail="UNC thread not found")
|
||
changes = {
|
||
"diameter_fraction": diameter_fraction,
|
||
"tpi": tpi,
|
||
"series": series,
|
||
"depth_in": float(depth_in) if depth_in.strip() else None,
|
||
}
|
||
await unc_thread_crud.update(db, thread, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="UncThread",
|
||
resource_id=thread_id,
|
||
description=audit_crud.describe_update("UncThread", thread_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_unc_thread_row.html", {"t": thread}
|
||
)
|
||
|
||
|
||
@router.delete("/unc-threads/{thread_id}")
|
||
async def delete_unc_thread(
|
||
thread_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("unc_threads:delete")),
|
||
):
|
||
thread = await unc_thread_crud.get(db, thread_id)
|
||
if thread is not None:
|
||
name = f"{thread.diameter_fraction}-{thread.tpi} {thread.series.value}"
|
||
await unc_thread_crud.delete(db, thread)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="UncThread",
|
||
resource_id=thread_id,
|
||
description=audit_crud.describe_delete("UncThread", thread_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Manufacturers -----------------------------------------------------------
|
||
|
||
|
||
@router.get("/manufacturers", response_class=HTMLResponse)
|
||
async def manufacturers_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("manufacturers:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/manufacturers.html",
|
||
{"user": user, "manufacturers": await manufacturer_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/manufacturers", response_class=HTMLResponse)
|
||
async def create_manufacturer(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
url: str = Form(...),
|
||
logo_url: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("manufacturers:write")),
|
||
):
|
||
created = await manufacturer_crud.create(
|
||
db,
|
||
name=name,
|
||
url=url,
|
||
logo_url=logo_url.strip() or None,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="Manufacturer",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create(
|
||
"Manufacturer", created.id, created.name
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_manufacturers_table.html",
|
||
{"manufacturers": await manufacturer_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/manufacturers/{manufacturer_id}/edit", response_class=HTMLResponse)
|
||
async def edit_manufacturer_row(
|
||
request: Request,
|
||
manufacturer_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("manufacturers:write")),
|
||
):
|
||
m = await manufacturer_crud.get(db, manufacturer_id)
|
||
if m is None:
|
||
raise HTTPException(status_code=404, detail="Manufacturer not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_manufacturer_row_edit.html", {"m": m}
|
||
)
|
||
|
||
|
||
@router.get("/manufacturers/{manufacturer_id}/row", response_class=HTMLResponse)
|
||
async def manufacturer_row(
|
||
request: Request,
|
||
manufacturer_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("manufacturers:read")),
|
||
):
|
||
m = await manufacturer_crud.get(db, manufacturer_id)
|
||
if m is None:
|
||
raise HTTPException(status_code=404, detail="Manufacturer not found")
|
||
return templates.TemplateResponse(request, "admin/_manufacturer_row.html", {"m": m})
|
||
|
||
|
||
@router.post("/manufacturers/{manufacturer_id}", response_class=HTMLResponse)
|
||
async def update_manufacturer(
|
||
request: Request,
|
||
manufacturer_id: int,
|
||
name: str = Form(...),
|
||
url: str = Form(...),
|
||
logo_url: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("manufacturers:write")),
|
||
):
|
||
m = await manufacturer_crud.get(db, manufacturer_id)
|
||
if m is None:
|
||
raise HTTPException(status_code=404, detail="Manufacturer not found")
|
||
changes = {"name": name, "url": url, "logo_url": logo_url.strip() or None}
|
||
await manufacturer_crud.update(db, m, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="Manufacturer",
|
||
resource_id=manufacturer_id,
|
||
description=audit_crud.describe_update(
|
||
"Manufacturer", manufacturer_id, changes
|
||
),
|
||
)
|
||
return templates.TemplateResponse(request, "admin/_manufacturer_row.html", {"m": m})
|
||
|
||
|
||
@router.delete("/manufacturers/{manufacturer_id}")
|
||
async def delete_manufacturer(
|
||
manufacturer_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("manufacturers:delete")),
|
||
):
|
||
m = await manufacturer_crud.get(db, manufacturer_id)
|
||
if m is not None:
|
||
name = m.name
|
||
await manufacturer_crud.delete(db, m)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="Manufacturer",
|
||
resource_id=manufacturer_id,
|
||
description=audit_crud.describe_delete(
|
||
"Manufacturer", manufacturer_id, name
|
||
),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Scope Models ------------------------------------------------------------
|
||
|
||
|
||
@router.get("/scope-models", response_class=HTMLResponse)
|
||
async def scope_models_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("scope_models:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/scope_models.html",
|
||
{
|
||
"user": user,
|
||
"scope_models": await scope_model_crud.list_all(db),
|
||
"manufacturers": await manufacturer_crud.list_all(db),
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/scope-models", response_class=HTMLResponse)
|
||
async def create_scope_model(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
manufacturer_id: int = Form(...),
|
||
url: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("scope_models:write")),
|
||
):
|
||
created = await scope_model_crud.create(
|
||
db,
|
||
name=name,
|
||
manufacturer_id=manufacturer_id,
|
||
url=url.strip() or None,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="ScopeModel",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create("ScopeModel", created.id, created.name),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_scope_models_table.html",
|
||
{"scope_models": await scope_model_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/scope-models/{scope_model_id}/edit", response_class=HTMLResponse)
|
||
async def edit_scope_model_row(
|
||
request: Request,
|
||
scope_model_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("scope_models:write")),
|
||
):
|
||
sm = await scope_model_crud.get(db, scope_model_id)
|
||
if sm is None:
|
||
raise HTTPException(status_code=404, detail="Scope model not found")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_scope_model_row_edit.html",
|
||
{"sm": sm, "manufacturers": await manufacturer_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/scope-models/{scope_model_id}/row", response_class=HTMLResponse)
|
||
async def scope_model_row(
|
||
request: Request,
|
||
scope_model_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("scope_models:read")),
|
||
):
|
||
sm = await scope_model_crud.get(db, scope_model_id)
|
||
if sm is None:
|
||
raise HTTPException(status_code=404, detail="Scope model not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_scope_model_row.html", {"sm": sm}
|
||
)
|
||
|
||
|
||
@router.post("/scope-models/{scope_model_id}", response_class=HTMLResponse)
|
||
async def update_scope_model(
|
||
request: Request,
|
||
scope_model_id: int,
|
||
name: str = Form(...),
|
||
manufacturer_id: int = Form(...),
|
||
url: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("scope_models:write")),
|
||
):
|
||
sm = await scope_model_crud.get(db, scope_model_id)
|
||
if sm is None:
|
||
raise HTTPException(status_code=404, detail="Scope model not found")
|
||
changes = {
|
||
"name": name,
|
||
"manufacturer_id": manufacturer_id,
|
||
"url": url.strip() or None,
|
||
}
|
||
await scope_model_crud.update(db, sm, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="ScopeModel",
|
||
resource_id=scope_model_id,
|
||
description=audit_crud.describe_update("ScopeModel", scope_model_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_scope_model_row.html", {"sm": sm}
|
||
)
|
||
|
||
|
||
@router.delete("/scope-models/{scope_model_id}")
|
||
async def delete_scope_model(
|
||
scope_model_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("scope_models:delete")),
|
||
):
|
||
sm = await scope_model_crud.get(db, scope_model_id)
|
||
if sm is not None:
|
||
name = sm.name
|
||
await scope_model_crud.delete(db, sm)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="ScopeModel",
|
||
resource_id=scope_model_id,
|
||
description=audit_crud.describe_delete("ScopeModel", scope_model_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Part Types --------------------------------------------------------------
|
||
|
||
|
||
@router.get("/part-types", response_class=HTMLResponse)
|
||
async def part_types_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("part_types:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/part_types.html",
|
||
{"user": user, "part_types": await part_type_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/part-types", response_class=HTMLResponse)
|
||
async def create_part_type(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_types:write")),
|
||
):
|
||
if not await part_type_crud.get_by_name(db, name):
|
||
created = await part_type_crud.create(db, name=name)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="PartType",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create(
|
||
"PartType", created.id, created.name
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_part_types_table.html",
|
||
{"part_types": await part_type_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/part-types/{part_type_id}/edit", response_class=HTMLResponse)
|
||
async def edit_part_type_row(
|
||
request: Request,
|
||
part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("part_types:write")),
|
||
):
|
||
pt = await part_type_crud.get(db, part_type_id)
|
||
if pt is None:
|
||
raise HTTPException(status_code=404, detail="Part type not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_type_row_edit.html", {"pt": pt}
|
||
)
|
||
|
||
|
||
@router.get("/part-types/{part_type_id}/row", response_class=HTMLResponse)
|
||
async def part_type_row(
|
||
request: Request,
|
||
part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("part_types:read")),
|
||
):
|
||
pt = await part_type_crud.get(db, part_type_id)
|
||
if pt is None:
|
||
raise HTTPException(status_code=404, detail="Part type not found")
|
||
return templates.TemplateResponse(request, "admin/_part_type_row.html", {"pt": pt})
|
||
|
||
|
||
@router.post("/part-types/{part_type_id}", response_class=HTMLResponse)
|
||
async def update_part_type(
|
||
request: Request,
|
||
part_type_id: int,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_types:write")),
|
||
):
|
||
pt = await part_type_crud.get(db, part_type_id)
|
||
if pt is None:
|
||
raise HTTPException(status_code=404, detail="Part type not found")
|
||
changes = {"name": name}
|
||
await part_type_crud.update(db, pt, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="PartType",
|
||
resource_id=part_type_id,
|
||
description=audit_crud.describe_update("PartType", part_type_id, changes),
|
||
)
|
||
return templates.TemplateResponse(request, "admin/_part_type_row.html", {"pt": pt})
|
||
|
||
|
||
@router.delete("/part-types/{part_type_id}")
|
||
async def delete_part_type(
|
||
part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_types:delete")),
|
||
):
|
||
pt = await part_type_crud.get(db, part_type_id)
|
||
if pt is not None:
|
||
name = pt.name
|
||
await part_type_crud.delete(db, pt)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="PartType",
|
||
resource_id=part_type_id,
|
||
description=audit_crud.describe_delete("PartType", part_type_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Sub Part Types ----------------------------------------------------------
|
||
|
||
|
||
@router.get("/sub-part-types", response_class=HTMLResponse)
|
||
async def sub_part_types_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("sub_part_types:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/sub_part_types.html",
|
||
{
|
||
"user": user,
|
||
"sub_part_types": await sub_part_type_crud.list_all(db),
|
||
"part_types": await part_type_crud.list_all(db),
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/sub-part-types", response_class=HTMLResponse)
|
||
async def create_sub_part_type(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
part_type_id: int = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("sub_part_types:write")),
|
||
):
|
||
if not await sub_part_type_crud.get_by_name_and_part_type(db, name, part_type_id):
|
||
created = await sub_part_type_crud.create(
|
||
db, name=name, part_type_id=part_type_id
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="SubPartType",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create(
|
||
"SubPartType", created.id, created.name
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_sub_part_types_table.html",
|
||
{"sub_part_types": await sub_part_type_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/sub-part-types/{sub_part_type_id}/edit", response_class=HTMLResponse)
|
||
async def edit_sub_part_type_row(
|
||
request: Request,
|
||
sub_part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("sub_part_types:write")),
|
||
):
|
||
spt = await sub_part_type_crud.get(db, sub_part_type_id)
|
||
if spt is None:
|
||
raise HTTPException(status_code=404, detail="Sub part type not found")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_sub_part_type_row_edit.html",
|
||
{"spt": spt, "part_types": await part_type_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/sub-part-types/{sub_part_type_id}/row", response_class=HTMLResponse)
|
||
async def sub_part_type_row(
|
||
request: Request,
|
||
sub_part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("sub_part_types:read")),
|
||
):
|
||
spt = await sub_part_type_crud.get(db, sub_part_type_id)
|
||
if spt is None:
|
||
raise HTTPException(status_code=404, detail="Sub part type not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_sub_part_type_row.html", {"spt": spt}
|
||
)
|
||
|
||
|
||
@router.post("/sub-part-types/{sub_part_type_id}", response_class=HTMLResponse)
|
||
async def update_sub_part_type(
|
||
request: Request,
|
||
sub_part_type_id: int,
|
||
name: str = Form(...),
|
||
part_type_id: int = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("sub_part_types:write")),
|
||
):
|
||
spt = await sub_part_type_crud.get(db, sub_part_type_id)
|
||
if spt is None:
|
||
raise HTTPException(status_code=404, detail="Sub part type not found")
|
||
changes = {"name": name, "part_type_id": part_type_id}
|
||
await sub_part_type_crud.update(db, spt, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="SubPartType",
|
||
resource_id=sub_part_type_id,
|
||
description=audit_crud.describe_update(
|
||
"SubPartType", sub_part_type_id, changes
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_sub_part_type_row.html", {"spt": spt}
|
||
)
|
||
|
||
|
||
@router.delete("/sub-part-types/{sub_part_type_id}")
|
||
async def delete_sub_part_type(
|
||
sub_part_type_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("sub_part_types:delete")),
|
||
):
|
||
spt = await sub_part_type_crud.get(db, sub_part_type_id)
|
||
if spt is not None:
|
||
name = spt.name
|
||
await sub_part_type_crud.delete(db, spt)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="SubPartType",
|
||
resource_id=sub_part_type_id,
|
||
description=audit_crud.describe_delete(
|
||
"SubPartType", sub_part_type_id, name
|
||
),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Imported Parts ----------------------------------------------------------
|
||
|
||
_IMPORTED_PARTS_SORT_COLUMNS = [
|
||
"InventoryID",
|
||
"Description",
|
||
"TBXColour",
|
||
"ProductMaxThreadDepthmm",
|
||
"ProductMaxThreadDepthin",
|
||
"ProductODmm",
|
||
"ProductODin",
|
||
"TBXThreadAttribute",
|
||
]
|
||
|
||
|
||
async def _default_imported_parts_table_ctx(db: AsyncSession) -> dict:
|
||
"""Table-card context (default sort, no filter, page 1) after a refresh."""
|
||
total = await imported_part_crud.count(db)
|
||
pagination = _paginate(
|
||
"/admin/imported-parts/filter",
|
||
{},
|
||
1,
|
||
_PAGE_SIZE,
|
||
total,
|
||
"#imported-parts-table-card",
|
||
)
|
||
return {
|
||
"parts": await imported_part_crud.list_filtered(db, limit=_PAGE_SIZE),
|
||
"pagination": pagination,
|
||
"sort_headers": _sort_headers(
|
||
"/admin/imported-parts/filter",
|
||
{},
|
||
_IMPORTED_PARTS_SORT_COLUMNS,
|
||
"InventoryID",
|
||
"asc",
|
||
"#imported-parts-table-card",
|
||
),
|
||
"sort": "InventoryID",
|
||
"dir": "asc",
|
||
}
|
||
|
||
|
||
@router.get("/imported-parts", response_class=HTMLResponse)
|
||
async def imported_parts_page(
|
||
request: Request,
|
||
sort: str = "InventoryID",
|
||
dir: str = "asc",
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("imported_parts:read")),
|
||
):
|
||
config = await imported_part_config_crud.get(db)
|
||
config_view = {
|
||
"odata_url": config.odata_url if config else None,
|
||
"username": config.username if config else None,
|
||
"has_password": bool(config and config.password),
|
||
}
|
||
total = await imported_part_crud.count(db)
|
||
parts = await imported_part_crud.list_filtered(
|
||
db, sort=sort, direction=dir, limit=_PAGE_SIZE
|
||
)
|
||
pagination = _paginate(
|
||
"/admin/imported-parts/filter",
|
||
{"sort": sort, "dir": dir},
|
||
1,
|
||
_PAGE_SIZE,
|
||
total,
|
||
"#imported-parts-table-card",
|
||
)
|
||
sort_headers = _sort_headers(
|
||
"/admin/imported-parts/filter",
|
||
{},
|
||
_IMPORTED_PARTS_SORT_COLUMNS,
|
||
sort,
|
||
dir,
|
||
"#imported-parts-table-card",
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/imported_parts.html",
|
||
{
|
||
"user": user,
|
||
"config": config_view,
|
||
"parts": parts,
|
||
"pagination": pagination,
|
||
"sort_headers": sort_headers,
|
||
"sort": sort,
|
||
"dir": dir,
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/imported-parts/config", response_class=HTMLResponse)
|
||
async def save_imported_part_config(
|
||
request: Request,
|
||
odata_url: str = Form(""),
|
||
username: str = Form(""),
|
||
password: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("imported_parts:write")),
|
||
):
|
||
current = await imported_part_config_crud.get(db)
|
||
# Keep existing password when field submitted blank
|
||
kept_password = password.strip() or (current.password if current else None)
|
||
await imported_part_config_crud.upsert(
|
||
db,
|
||
odata_url=odata_url.strip() or None,
|
||
username=username.strip() or None,
|
||
password=kept_password,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="ImportedPartConfig",
|
||
resource_id=1,
|
||
description="Updated ImportedPart OData configuration",
|
||
)
|
||
return HTMLResponse("Configuration saved.", status_code=200)
|
||
|
||
|
||
@router.post("/imported-parts/refresh", response_class=HTMLResponse)
|
||
async def refresh_imported_parts(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("imported_parts:write")),
|
||
):
|
||
config = await imported_part_config_crud.get(db)
|
||
|
||
async def _imported_table_ctx(sync_result=None, sync_error=None):
|
||
return {
|
||
**await _default_imported_parts_table_ctx(db),
|
||
"sync_result": sync_result,
|
||
"sync_error": sync_error,
|
||
}
|
||
|
||
if not config or not config.odata_url:
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_imported_parts_table.html",
|
||
await _imported_table_ctx(sync_error="OData URL is not configured."),
|
||
)
|
||
try:
|
||
records = await odata_service.fetch(
|
||
config.odata_url, config.username, config.password
|
||
)
|
||
created, updated = await imported_part_crud.upsert_bulk(db, records)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="ImportedPart",
|
||
resource_id=None,
|
||
description=f"OData sync: {created} created, {updated} updated ({len(records)} total records)",
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_imported_parts_table.html",
|
||
await _imported_table_ctx(
|
||
sync_result=f"Sync complete: {created} created, {updated} updated."
|
||
),
|
||
)
|
||
except Exception as exc:
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_imported_parts_table.html",
|
||
await _imported_table_ctx(sync_error=f"Sync failed: {exc}"),
|
||
)
|
||
|
||
|
||
@router.get("/imported-parts/filter", response_class=HTMLResponse)
|
||
async def filter_imported_parts(
|
||
request: Request,
|
||
description: str = "",
|
||
sort: str = "InventoryID",
|
||
dir: str = "asc",
|
||
page: int = 1,
|
||
page_size: int = _PAGE_SIZE,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("imported_parts:read")),
|
||
):
|
||
total = await imported_part_crud.count_filtered(db, description=description)
|
||
filter_params = {"description": description}
|
||
pagination = _paginate(
|
||
"/admin/imported-parts/filter",
|
||
{**filter_params, "sort": sort, "dir": dir},
|
||
page,
|
||
page_size,
|
||
total,
|
||
"#imported-parts-table-card",
|
||
)
|
||
parts = await imported_part_crud.list_filtered(
|
||
db,
|
||
description=description,
|
||
sort=sort,
|
||
direction=dir,
|
||
skip=pagination["skip"],
|
||
limit=page_size,
|
||
)
|
||
sort_headers = _sort_headers(
|
||
"/admin/imported-parts/filter",
|
||
filter_params,
|
||
_IMPORTED_PARTS_SORT_COLUMNS,
|
||
sort,
|
||
dir,
|
||
"#imported-parts-table-card",
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_imported_parts_table_card.html",
|
||
{
|
||
"parts": parts,
|
||
"pagination": pagination,
|
||
"sort_headers": sort_headers,
|
||
"sort": sort,
|
||
"dir": dir,
|
||
},
|
||
)
|
||
|
||
|
||
# --- Scope Import ------------------------------------------------------------
|
||
|
||
|
||
@router.get("/scope-import", response_class=HTMLResponse)
|
||
async def scope_import_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/scope_import.html",
|
||
{"user": user, "expected_columns": scope_import_service.EXPECTED_COLUMNS},
|
||
)
|
||
|
||
|
||
_SCOPE_IMPORT_STATUS_ID = "scope-import-status"
|
||
|
||
|
||
def _scope_import_progress_ctx(job: jobs.Job) -> dict:
|
||
return {
|
||
"container_id": _SCOPE_IMPORT_STATUS_ID,
|
||
"poll_url": f"/admin/scope-import/jobs/{job.id}",
|
||
"job": job,
|
||
}
|
||
|
||
|
||
@router.post("/scope-import", response_class=HTMLResponse)
|
||
async def upload_scope_import(
|
||
request: Request,
|
||
file: UploadFile = File(...),
|
||
actor: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
filename = file.filename or ""
|
||
if not filename.lower().endswith(".xlsx"):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_scope_import_result.html",
|
||
{
|
||
"error": "Expected an .xlsx spreadsheet.",
|
||
"container_id": _SCOPE_IMPORT_STATUS_ID,
|
||
},
|
||
)
|
||
data = await file.read()
|
||
actor_id = actor.id
|
||
|
||
async def _run(job: jobs.Job) -> None:
|
||
async with AsyncSessionLocal() as db:
|
||
try:
|
||
result = await scope_import_service.import_from_bytes(db, data)
|
||
except Exception as exc: # noqa: BLE001
|
||
job.status = "error"
|
||
job.message = f"Could not read spreadsheet: {exc}"
|
||
return
|
||
actor_row = await user_crud.get(db, actor_id)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor_row,
|
||
action="update",
|
||
resource_type="AugmentedPart",
|
||
resource_id=None,
|
||
description=(
|
||
f"Scope import ({filename}): {result.created} created, "
|
||
f"{result.updated} updated, {result.skipped} skipped, "
|
||
f"{result.errors} errors"
|
||
),
|
||
)
|
||
job.status = "done"
|
||
job.result = result
|
||
|
||
job = jobs.start("scope_import", _run)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_job_progress.html", _scope_import_progress_ctx(job)
|
||
)
|
||
|
||
|
||
@router.get("/scope-import/jobs/{job_id}", response_class=HTMLResponse)
|
||
async def scope_import_job_status(
|
||
request: Request,
|
||
job_id: str,
|
||
_: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
job = jobs.get(job_id)
|
||
if job is None:
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_scope_import_result.html",
|
||
{
|
||
"error": "Import job not found — it may have expired.",
|
||
"container_id": _SCOPE_IMPORT_STATUS_ID,
|
||
},
|
||
)
|
||
if job.status == "running":
|
||
return templates.TemplateResponse(
|
||
request, "admin/_job_progress.html", _scope_import_progress_ctx(job)
|
||
)
|
||
if job.status == "error":
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_scope_import_result.html",
|
||
{"error": job.message, "container_id": _SCOPE_IMPORT_STATUS_ID},
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_scope_import_result.html",
|
||
{"result": job.result, "container_id": _SCOPE_IMPORT_STATUS_ID},
|
||
)
|
||
|
||
|
||
# --- Augmented Parts ---------------------------------------------------------
|
||
|
||
|
||
async def _skipped_colours(db: AsyncSession) -> list[str]:
|
||
return await augmented_part_crud.distinct_unlinked_colours(db)
|
||
|
||
|
||
async def _augmented_part_form_ctx(db: AsyncSession) -> dict:
|
||
return {
|
||
"part_colours": await part_colour_crud.list_all(db),
|
||
"manufacturers": await manufacturer_crud.list_all(db),
|
||
"scope_models": await scope_model_crud.list_all(db),
|
||
"part_types": await part_type_crud.list_all(db),
|
||
"sub_part_types": await sub_part_type_crud.list_all(db),
|
||
}
|
||
|
||
|
||
_AUGMENTED_PARTS_SORT_COLUMNS = [
|
||
"id",
|
||
"sku",
|
||
"description",
|
||
"colour",
|
||
"manufacturer",
|
||
"scope_model",
|
||
"part_type",
|
||
"sub_part_type",
|
||
]
|
||
|
||
|
||
async def _default_augmented_parts_table_ctx(db: AsyncSession) -> dict:
|
||
"""Table-card context (default sort, no filters, page 1) after create/sync actions."""
|
||
total = await augmented_part_crud.count_filtered(db)
|
||
pagination = _paginate(
|
||
"/admin/augmented-parts/filter",
|
||
{},
|
||
1,
|
||
_PAGE_SIZE,
|
||
total,
|
||
"#augmented-parts-table-card",
|
||
)
|
||
return {
|
||
"augmented_parts": await augmented_part_crud.list_filtered(
|
||
db, limit=_PAGE_SIZE
|
||
),
|
||
"pagination": pagination,
|
||
"sort_headers": _sort_headers(
|
||
"/admin/augmented-parts/filter",
|
||
{},
|
||
_AUGMENTED_PARTS_SORT_COLUMNS,
|
||
"description",
|
||
"asc",
|
||
"#augmented-parts-table-card",
|
||
),
|
||
"sort": "description",
|
||
"dir": "asc",
|
||
}
|
||
|
||
|
||
@router.get("/augmented-parts", response_class=HTMLResponse)
|
||
async def augmented_parts_page(
|
||
request: Request,
|
||
sort: str = "description",
|
||
dir: str = "asc",
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("augmented_parts:read")),
|
||
):
|
||
ctx = await _augmented_part_form_ctx(db)
|
||
total = await augmented_part_crud.count_filtered(db)
|
||
ctx["augmented_parts"] = await augmented_part_crud.list_filtered(
|
||
db, sort=sort, direction=dir, limit=_PAGE_SIZE
|
||
)
|
||
ctx["pagination"] = _paginate(
|
||
"/admin/augmented-parts/filter",
|
||
{"sort": sort, "dir": dir},
|
||
1,
|
||
_PAGE_SIZE,
|
||
total,
|
||
"#augmented-parts-table-card",
|
||
)
|
||
ctx["sort_headers"] = _sort_headers(
|
||
"/admin/augmented-parts/filter",
|
||
{},
|
||
_AUGMENTED_PARTS_SORT_COLUMNS,
|
||
sort,
|
||
dir,
|
||
"#augmented-parts-table-card",
|
||
)
|
||
ctx["sort"] = sort
|
||
ctx["dir"] = dir
|
||
ctx["user"] = user
|
||
return templates.TemplateResponse(request, "admin/augmented_parts.html", ctx)
|
||
|
||
|
||
@router.post("/augmented-parts", response_class=HTMLResponse)
|
||
async def create_augmented_part(
|
||
request: Request,
|
||
sku: str = Form(""),
|
||
description: str = Form(...),
|
||
colour_id: int = Form(...),
|
||
manufacturer_id: int = Form(...),
|
||
scope_model_id: str = Form(""),
|
||
part_type_id: int = Form(...),
|
||
sub_part_type_id: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
created = await augmented_part_crud.create(
|
||
db,
|
||
sku=sku.strip() or None,
|
||
description=description,
|
||
colour_id=colour_id,
|
||
manufacturer_id=manufacturer_id,
|
||
scope_model_id=int(scope_model_id) if scope_model_id else None,
|
||
part_type_id=part_type_id,
|
||
sub_part_type_id=int(sub_part_type_id) if sub_part_type_id else None,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="AugmentedPart",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create(
|
||
"AugmentedPart", created.id, created.sku or created.description
|
||
),
|
||
)
|
||
form_ctx = await _augmented_part_form_ctx(db)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_augmented_parts_table.html",
|
||
{**form_ctx, **await _default_augmented_parts_table_ctx(db)},
|
||
)
|
||
|
||
|
||
_AUGMENTED_PARTS_SYNC_TARGET_ID = "augmented-part-list"
|
||
|
||
|
||
@router.post("/augmented-parts/sync", response_class=HTMLResponse)
|
||
async def sync_augmented_parts(
|
||
request: Request,
|
||
actor: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
actor_id = actor.id
|
||
|
||
async def _run(job: jobs.Job) -> None:
|
||
async with AsyncSessionLocal() as db:
|
||
try:
|
||
result = await augmented_part_sync_service.sync(db)
|
||
except Exception as exc:
|
||
job.status = "error"
|
||
job.message = f"Sync failed: {exc}"
|
||
return
|
||
actor_row = await user_crud.get(db, actor_id)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor_row,
|
||
action="update",
|
||
resource_type="AugmentedPart",
|
||
resource_id=None,
|
||
description=(
|
||
f"Sync: {result.created} created, {result.deleted} deleted, "
|
||
f"{result.skipped} skipped (no type match)"
|
||
),
|
||
)
|
||
job.status = "done"
|
||
job.message = (
|
||
f"Sync complete: {result.created} created, {result.deleted} deleted"
|
||
+ (
|
||
f", {result.skipped} skipped (type not recognised)"
|
||
if result.skipped
|
||
else ""
|
||
)
|
||
+ "."
|
||
)
|
||
|
||
job = jobs.start("augmented_part_sync", _run)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_job_progress.html",
|
||
{
|
||
"container_id": _AUGMENTED_PARTS_SYNC_TARGET_ID,
|
||
"poll_url": f"/admin/augmented-parts/sync/jobs/{job.id}",
|
||
"job": job,
|
||
},
|
||
)
|
||
|
||
|
||
@router.get("/augmented-parts/sync/jobs/{job_id}", response_class=HTMLResponse)
|
||
async def augmented_parts_sync_job_status(
|
||
request: Request,
|
||
job_id: str,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
job = jobs.get(job_id)
|
||
if job is None:
|
||
form_ctx = await _augmented_part_form_ctx(db)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_augmented_parts_table.html",
|
||
{
|
||
**form_ctx,
|
||
**await _default_augmented_parts_table_ctx(db),
|
||
"sync_result": None,
|
||
"sync_error": "Sync job not found — it may have expired.",
|
||
},
|
||
)
|
||
if job.status == "running":
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_job_progress.html",
|
||
{
|
||
"container_id": _AUGMENTED_PARTS_SYNC_TARGET_ID,
|
||
"poll_url": f"/admin/augmented-parts/sync/jobs/{job_id}",
|
||
"job": job,
|
||
},
|
||
)
|
||
form_ctx = await _augmented_part_form_ctx(db)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_augmented_parts_table.html",
|
||
{
|
||
**form_ctx,
|
||
**await _default_augmented_parts_table_ctx(db),
|
||
"sync_result": job.message if job.status == "done" else None,
|
||
"sync_error": job.message if job.status == "error" else None,
|
||
},
|
||
)
|
||
|
||
|
||
@router.get("/augmented-parts/filter", response_class=HTMLResponse)
|
||
async def filter_augmented_parts(
|
||
request: Request,
|
||
description: str = "",
|
||
colour_id: str = "",
|
||
manufacturer_id: str = "",
|
||
scope_model_id: str = "",
|
||
part_type_id: str = "",
|
||
sub_part_type_id: str = "",
|
||
sort: str = "description",
|
||
dir: str = "asc",
|
||
page: int = 1,
|
||
page_size: int = _PAGE_SIZE,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("augmented_parts:read")),
|
||
):
|
||
colour_id_int = int(colour_id) if colour_id else None
|
||
manufacturer_id_int = int(manufacturer_id) if manufacturer_id else None
|
||
scope_model_id_int = int(scope_model_id) if scope_model_id else None
|
||
part_type_id_int = int(part_type_id) if part_type_id else None
|
||
sub_part_type_id_int = int(sub_part_type_id) if sub_part_type_id else None
|
||
total = await augmented_part_crud.count_filtered(
|
||
db,
|
||
description=description,
|
||
colour_id=colour_id_int,
|
||
manufacturer_id=manufacturer_id_int,
|
||
scope_model_id=scope_model_id_int,
|
||
part_type_id=part_type_id_int,
|
||
sub_part_type_id=sub_part_type_id_int,
|
||
)
|
||
filter_params = {
|
||
"description": description,
|
||
"colour_id": colour_id,
|
||
"manufacturer_id": manufacturer_id,
|
||
"scope_model_id": scope_model_id,
|
||
"part_type_id": part_type_id,
|
||
"sub_part_type_id": sub_part_type_id,
|
||
}
|
||
pagination = _paginate(
|
||
"/admin/augmented-parts/filter",
|
||
{**filter_params, "sort": sort, "dir": dir},
|
||
page,
|
||
page_size,
|
||
total,
|
||
"#augmented-parts-table-card",
|
||
)
|
||
parts = await augmented_part_crud.list_filtered(
|
||
db,
|
||
description=description,
|
||
colour_id=colour_id_int,
|
||
manufacturer_id=manufacturer_id_int,
|
||
scope_model_id=scope_model_id_int,
|
||
part_type_id=part_type_id_int,
|
||
sub_part_type_id=sub_part_type_id_int,
|
||
sort=sort,
|
||
direction=dir,
|
||
skip=pagination["skip"],
|
||
limit=page_size,
|
||
)
|
||
sort_headers = _sort_headers(
|
||
"/admin/augmented-parts/filter",
|
||
filter_params,
|
||
_AUGMENTED_PARTS_SORT_COLUMNS,
|
||
sort,
|
||
dir,
|
||
"#augmented-parts-table-card",
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_augmented_parts_table_card.html",
|
||
{
|
||
"augmented_parts": parts,
|
||
"pagination": pagination,
|
||
"sort_headers": sort_headers,
|
||
"sort": sort,
|
||
"dir": dir,
|
||
},
|
||
)
|
||
|
||
|
||
@router.get("/augmented-parts/skipped", response_class=HTMLResponse)
|
||
async def skipped_parts_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("augmented_parts:read")),
|
||
):
|
||
form_ctx = await _augmented_part_form_ctx(db)
|
||
total_unlinked = await augmented_part_crud.count_unlinked_filtered(db)
|
||
skipped_colours = await _skipped_colours(db)
|
||
pagination = _paginate(
|
||
"/admin/augmented-parts/skipped/filter",
|
||
{},
|
||
1,
|
||
_PAGE_SIZE,
|
||
total_unlinked,
|
||
"#skipped-parts-table-card",
|
||
)
|
||
unlinked = await augmented_part_crud.get_unlinked_imported_parts_filtered(
|
||
db, limit=_PAGE_SIZE
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/augmented_parts_skipped.html",
|
||
{
|
||
"user": user,
|
||
**form_ctx,
|
||
"unlinked_parts": unlinked,
|
||
"any_unlinked": bool(total_unlinked),
|
||
"skipped_colours": skipped_colours,
|
||
"filter_description": "",
|
||
"filter_colour": "",
|
||
"pagination": pagination,
|
||
},
|
||
)
|
||
|
||
|
||
@router.post("/augmented-parts/skipped", response_class=HTMLResponse)
|
||
async def sync_skipped_parts(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
form_data = await request.form()
|
||
inventory_ids = [str(v) for v in form_data.getlist("inventory_ids")]
|
||
part_type_id_str = str(form_data.get("part_type_id", ""))
|
||
sub_part_type_id_str = str(form_data.get("sub_part_type_id", ""))
|
||
|
||
part_type_id = int(part_type_id_str) if part_type_id_str else None
|
||
sub_part_type_id = int(sub_part_type_id_str) if sub_part_type_id_str else None
|
||
|
||
sync_result = None
|
||
sync_error = None
|
||
|
||
if not inventory_ids:
|
||
sync_error = "No parts selected."
|
||
elif part_type_id is None:
|
||
sync_error = "No type selected."
|
||
else:
|
||
synced = 0
|
||
errors: list[str] = []
|
||
for inv_id in inventory_ids:
|
||
try:
|
||
ap = await augmented_part_sync_service.sync_one(
|
||
db,
|
||
inventory_id=inv_id,
|
||
part_type_id=part_type_id,
|
||
sub_part_type_id=sub_part_type_id,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="AugmentedPart",
|
||
resource_id=ap.id,
|
||
description=audit_crud.describe_create(
|
||
"AugmentedPart", ap.id, ap.description
|
||
),
|
||
)
|
||
synced += 1
|
||
except Exception as exc:
|
||
errors.append(f"{inv_id}: {exc}")
|
||
if errors:
|
||
sync_error = "Some parts failed: " + "; ".join(errors)
|
||
if synced:
|
||
sync_result = f"{synced} part(s) synced successfully."
|
||
|
||
form_ctx = await _augmented_part_form_ctx(db)
|
||
total_unlinked = await augmented_part_crud.count_unlinked_filtered(db)
|
||
colours = await _skipped_colours(db)
|
||
pagination = _paginate(
|
||
"/admin/augmented-parts/skipped/filter",
|
||
{},
|
||
1,
|
||
_PAGE_SIZE,
|
||
total_unlinked,
|
||
"#skipped-parts-table-card",
|
||
)
|
||
unlinked = await augmented_part_crud.get_unlinked_imported_parts_filtered(
|
||
db, limit=_PAGE_SIZE
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_skipped_parts_table.html",
|
||
{
|
||
**form_ctx,
|
||
"unlinked_parts": unlinked,
|
||
"any_unlinked": bool(total_unlinked),
|
||
"skipped_colours": colours,
|
||
"filter_description": "",
|
||
"filter_colour": "",
|
||
"pagination": pagination,
|
||
"sync_result": sync_result,
|
||
"sync_error": sync_error,
|
||
},
|
||
)
|
||
|
||
|
||
@router.get("/augmented-parts/skipped/filter", response_class=HTMLResponse)
|
||
async def filter_skipped_parts(
|
||
request: Request,
|
||
description: str = "",
|
||
colour: str = "",
|
||
page: int = 1,
|
||
page_size: int = _PAGE_SIZE,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("augmented_parts:read")),
|
||
):
|
||
total = await augmented_part_crud.count_unlinked_filtered(
|
||
db, description=description, colour=colour
|
||
)
|
||
filter_params = {"description": description, "colour": colour}
|
||
pagination = _paginate(
|
||
"/admin/augmented-parts/skipped/filter",
|
||
filter_params,
|
||
page,
|
||
page_size,
|
||
total,
|
||
"#skipped-parts-table-card",
|
||
)
|
||
unlinked = await augmented_part_crud.get_unlinked_imported_parts_filtered(
|
||
db,
|
||
description=description,
|
||
colour=colour,
|
||
skip=pagination["skip"],
|
||
limit=page_size,
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_skipped_parts_table_card.html",
|
||
{"unlinked_parts": unlinked, "pagination": pagination},
|
||
)
|
||
|
||
|
||
@router.get("/augmented-parts/{augmented_part_id}", response_class=HTMLResponse)
|
||
async def augmented_part_detail(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("augmented_parts:read")),
|
||
):
|
||
ap = await augmented_part_crud.get(db, augmented_part_id)
|
||
if ap is None:
|
||
raise HTTPException(status_code=404, detail="Augmented part not found")
|
||
can_read_fits = user.is_superuser or "fits:read" in user.permissions
|
||
can_read_attachment_points = (
|
||
user.is_superuser or "attachment_points:read" in user.permissions
|
||
)
|
||
ctx = {
|
||
"user": user,
|
||
"ap": ap,
|
||
"scope_info": await scope_info_crud.get_by_augmented_part(
|
||
db, augmented_part_id
|
||
),
|
||
"can_read_fits": can_read_fits,
|
||
"can_read_attachment_points": can_read_attachment_points,
|
||
"augmented_part_id": augmented_part_id,
|
||
"notes": ap.notes,
|
||
"can_write_notes": _can_write_augmented_part_notes(user),
|
||
}
|
||
if can_read_attachment_points:
|
||
ctx.update(
|
||
await _augmented_part_ap_ctx(
|
||
db,
|
||
augmented_part_id,
|
||
can_write=_can_write_attachment_points(user),
|
||
)
|
||
)
|
||
if can_read_fits:
|
||
ctx.update(
|
||
await _augmented_part_fits_ctx(
|
||
db, augmented_part_id, can_write=_can_write_fits(user)
|
||
)
|
||
)
|
||
return templates.TemplateResponse(request, "admin/augmented_part_detail.html", ctx)
|
||
|
||
|
||
@router.get("/augmented-parts/{augmented_part_id}/edit", response_class=HTMLResponse)
|
||
async def edit_augmented_part_row(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
ap = await augmented_part_crud.get(db, augmented_part_id)
|
||
if ap is None:
|
||
raise HTTPException(status_code=404, detail="Augmented part not found")
|
||
ctx = await _augmented_part_form_ctx(db)
|
||
ctx["ap"] = ap
|
||
return templates.TemplateResponse(
|
||
request, "admin/_augmented_part_row_edit.html", ctx
|
||
)
|
||
|
||
|
||
@router.get("/augmented-parts/{augmented_part_id}/row", response_class=HTMLResponse)
|
||
async def augmented_part_row(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("augmented_parts:read")),
|
||
):
|
||
ap = await augmented_part_crud.get(db, augmented_part_id)
|
||
if ap is None:
|
||
raise HTTPException(status_code=404, detail="Augmented part not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_augmented_part_row.html", {"ap": ap}
|
||
)
|
||
|
||
|
||
@router.post("/augmented-parts/{augmented_part_id}", response_class=HTMLResponse)
|
||
async def update_augmented_part(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
sku: str = Form(""),
|
||
description: str = Form(...),
|
||
colour_id: int = Form(...),
|
||
manufacturer_id: int = Form(...),
|
||
scope_model_id: str = Form(""),
|
||
part_type_id: int = Form(...),
|
||
sub_part_type_id: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
ap = await augmented_part_crud.get(db, augmented_part_id)
|
||
if ap is None:
|
||
raise HTTPException(status_code=404, detail="Augmented part not found")
|
||
changes = {
|
||
"sku": sku.strip() or None,
|
||
"description": description,
|
||
"colour_id": colour_id,
|
||
"manufacturer_id": manufacturer_id,
|
||
"scope_model_id": int(scope_model_id) if scope_model_id else None,
|
||
"part_type_id": part_type_id,
|
||
"sub_part_type_id": int(sub_part_type_id) if sub_part_type_id else None,
|
||
}
|
||
await augmented_part_crud.update(db, ap, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="AugmentedPart",
|
||
resource_id=augmented_part_id,
|
||
description=audit_crud.describe_update(
|
||
"AugmentedPart", augmented_part_id, changes
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_augmented_part_row.html", {"ap": ap}
|
||
)
|
||
|
||
|
||
@router.delete("/augmented-parts/{augmented_part_id}")
|
||
async def delete_augmented_part(
|
||
augmented_part_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("augmented_parts:delete")),
|
||
):
|
||
ap = await augmented_part_crud.get(db, augmented_part_id)
|
||
if ap is not None:
|
||
label = ap.sku or ap.description
|
||
await augmented_part_crud.delete(db, ap)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="AugmentedPart",
|
||
resource_id=augmented_part_id,
|
||
description=audit_crud.describe_delete(
|
||
"AugmentedPart", augmented_part_id, label
|
||
),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
def _can_write_attachment_points(user: User) -> bool:
|
||
return user.is_superuser or "attachment_points:write" in user.permissions
|
||
|
||
|
||
async def _augmented_part_ap_ctx(
|
||
db: AsyncSession, augmented_part_id: int, *, can_write: bool = True
|
||
) -> dict:
|
||
return {
|
||
"augmented_part_id": augmented_part_id,
|
||
"attachment_points_list": await attachment_point_crud.list_for_augmented_part(
|
||
db, augmented_part_id
|
||
),
|
||
"attachment_point_names": await apn_crud.list_all(db),
|
||
"iso_threads": await iso_thread_crud.list_all(db),
|
||
"unc_threads": await unc_thread_crud.list_all(db),
|
||
"clamp_connections": await clamp_connection_crud.list_all(db),
|
||
"static_connections": await static_connection_crud.list_all(db),
|
||
"can_write_attachment_points": can_write,
|
||
}
|
||
|
||
|
||
@router.get(
|
||
"/augmented-parts/{augmented_part_id}/attachment-points",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def augmented_part_attachment_points(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("attachment_points:read")),
|
||
):
|
||
ctx = await _augmented_part_ap_ctx(
|
||
db, augmented_part_id, can_write=_can_write_attachment_points(user)
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_augmented_part_attachment_points.html", ctx
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/augmented-parts/{augmented_part_id}/attachment-points",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def create_augmented_part_attachment_point(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
attachment_point_name_id: int = Form(...),
|
||
connection_type: str = Form(""),
|
||
iso_thread_id: str = Form(""),
|
||
unc_thread_id: str = Form(""),
|
||
clamp_connection_id: str = Form(""),
|
||
static_connection_id: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("attachment_points:write")),
|
||
):
|
||
apoint = await attachment_point_crud.create(
|
||
db,
|
||
augmented_part_id=augmented_part_id,
|
||
attachment_point_name_id=attachment_point_name_id,
|
||
iso_thread_id=int(iso_thread_id) if iso_thread_id else None,
|
||
unc_thread_id=int(unc_thread_id) if unc_thread_id else None,
|
||
clamp_connection_id=int(clamp_connection_id) if clamp_connection_id else None,
|
||
static_connection_id=(
|
||
int(static_connection_id) if static_connection_id else None
|
||
),
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="AttachmentPoint",
|
||
resource_id=apoint.id,
|
||
description=audit_crud.describe_create(
|
||
"AttachmentPoint", apoint.id, f"part {augmented_part_id}"
|
||
),
|
||
)
|
||
ctx = await _augmented_part_ap_ctx(
|
||
db, augmented_part_id, can_write=_can_write_attachment_points(actor)
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_augmented_part_attachment_points.html", ctx
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/augmented-parts/{augmented_part_id}/attachment-points/{attachment_point_id}/edit",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def edit_augmented_part_attachment_point(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
attachment_point_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("attachment_points:write")),
|
||
):
|
||
apoint = await attachment_point_crud.get(db, attachment_point_id)
|
||
if apoint is None:
|
||
raise HTTPException(status_code=404, detail="Attachment point not found")
|
||
ctx = await _augmented_part_ap_ctx(db, augmented_part_id)
|
||
ctx["apoint"] = apoint
|
||
if apoint.iso_thread_id:
|
||
ctx["connection_type"] = "iso_thread"
|
||
elif apoint.unc_thread_id:
|
||
ctx["connection_type"] = "unc_thread"
|
||
elif apoint.clamp_connection_id:
|
||
ctx["connection_type"] = "clamp_connection"
|
||
elif apoint.static_connection_id:
|
||
ctx["connection_type"] = "static_connection"
|
||
else:
|
||
ctx["connection_type"] = ""
|
||
return templates.TemplateResponse(
|
||
request, "admin/_augmented_part_ap_row_edit.html", ctx
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/augmented-parts/{augmented_part_id}/attachment-points/{attachment_point_id}/row",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def augmented_part_attachment_point_row(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
attachment_point_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("attachment_points:read")),
|
||
):
|
||
apoint = await attachment_point_crud.get(db, attachment_point_id)
|
||
if apoint is None:
|
||
raise HTTPException(status_code=404, detail="Attachment point not found")
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_augmented_part_ap_row.html",
|
||
{
|
||
"apoint": apoint,
|
||
"can_write_attachment_points": _can_write_attachment_points(user),
|
||
},
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/augmented-parts/{augmented_part_id}/attachment-points/{attachment_point_id}",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def update_augmented_part_attachment_point(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
attachment_point_id: int,
|
||
attachment_point_name_id: int = Form(...),
|
||
connection_type: str = Form(""),
|
||
iso_thread_id: str = Form(""),
|
||
unc_thread_id: str = Form(""),
|
||
clamp_connection_id: str = Form(""),
|
||
static_connection_id: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("attachment_points:write")),
|
||
):
|
||
apoint = await attachment_point_crud.get(db, attachment_point_id)
|
||
if apoint is None:
|
||
raise HTTPException(status_code=404, detail="Attachment point not found")
|
||
changes = {
|
||
"attachment_point_name_id": attachment_point_name_id,
|
||
"iso_thread_id": int(iso_thread_id) if iso_thread_id else None,
|
||
"unc_thread_id": int(unc_thread_id) if unc_thread_id else None,
|
||
"clamp_connection_id": (
|
||
int(clamp_connection_id) if clamp_connection_id else None
|
||
),
|
||
"static_connection_id": (
|
||
int(static_connection_id) if static_connection_id else None
|
||
),
|
||
}
|
||
apoint = await attachment_point_crud.update(db, apoint, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="AttachmentPoint",
|
||
resource_id=attachment_point_id,
|
||
description=audit_crud.describe_update(
|
||
"AttachmentPoint", attachment_point_id, changes
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_augmented_part_ap_row.html",
|
||
{
|
||
"apoint": apoint,
|
||
"can_write_attachment_points": _can_write_attachment_points(actor),
|
||
},
|
||
)
|
||
|
||
|
||
@router.delete(
|
||
"/augmented-parts/{augmented_part_id}/attachment-points/{attachment_point_id}",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def delete_augmented_part_attachment_point(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
attachment_point_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("attachment_points:delete")),
|
||
):
|
||
apoint = await attachment_point_crud.get(db, attachment_point_id)
|
||
if apoint is not None:
|
||
await attachment_point_crud.delete(db, apoint)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="AttachmentPoint",
|
||
resource_id=attachment_point_id,
|
||
description=audit_crud.describe_delete(
|
||
"AttachmentPoint", attachment_point_id, f"part {augmented_part_id}"
|
||
),
|
||
)
|
||
ctx = await _augmented_part_ap_ctx(
|
||
db, augmented_part_id, can_write=_can_write_attachment_points(actor)
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_augmented_part_attachment_points.html", ctx
|
||
)
|
||
|
||
|
||
# --- Fits ----------------------------------------------------------------
|
||
|
||
|
||
async def _fit_view(db: AsyncSession, fit, part_id: int) -> dict:
|
||
if fit.augmented_part_a_id == part_id:
|
||
own_ap, other_ap, other_part_id = (
|
||
fit.attachment_point_a,
|
||
fit.attachment_point_b,
|
||
fit.augmented_part_b_id,
|
||
)
|
||
else:
|
||
own_ap, other_ap, other_part_id = (
|
||
fit.attachment_point_b,
|
||
fit.attachment_point_a,
|
||
fit.augmented_part_a_id,
|
||
)
|
||
other_part = await augmented_part_crud.get(db, other_part_id)
|
||
return {
|
||
"fit": fit,
|
||
"own_ap": own_ap,
|
||
"other_ap": other_ap,
|
||
"other_part": other_part,
|
||
}
|
||
|
||
|
||
def _can_write_fits(user: User) -> bool:
|
||
return user.is_superuser or "fits:write" in user.permissions
|
||
|
||
|
||
async def _augmented_part_fits_ctx(
|
||
db: AsyncSession,
|
||
augmented_part_id: int,
|
||
*,
|
||
can_write: bool = True,
|
||
message: str | None = None,
|
||
error: str | None = None,
|
||
) -> dict:
|
||
fits = await fit_crud.list_for_part(db, augmented_part_id)
|
||
other_parts = [
|
||
p
|
||
for p in await augmented_part_crud.list_all(db, limit=500)
|
||
if p.id != augmented_part_id
|
||
]
|
||
return {
|
||
"augmented_part_id": augmented_part_id,
|
||
"fit_views": [await _fit_view(db, f, augmented_part_id) for f in fits],
|
||
"own_points": await attachment_point_crud.list_for_augmented_part(
|
||
db, augmented_part_id
|
||
),
|
||
"other_parts": other_parts,
|
||
"can_write": can_write,
|
||
"message": message,
|
||
"error": error,
|
||
}
|
||
|
||
|
||
@router.get("/augmented-parts/{augmented_part_id}/fits", response_class=HTMLResponse)
|
||
async def augmented_part_fits(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("fits:read")),
|
||
):
|
||
ctx = await _augmented_part_fits_ctx(
|
||
db, augmented_part_id, can_write=_can_write_fits(user)
|
||
)
|
||
return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
|
||
|
||
|
||
def _augmented_part_fits_target_id(augmented_part_id: int) -> str:
|
||
return f"ap-{augmented_part_id}-fits"
|
||
|
||
|
||
@router.post(
|
||
"/augmented-parts/{augmented_part_id}/fits/recompute", response_class=HTMLResponse
|
||
)
|
||
async def recompute_augmented_part_fits(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
actor: User = Depends(web_require("fits:write")),
|
||
):
|
||
actor_id = actor.id
|
||
|
||
async def _run(job: jobs.Job) -> None:
|
||
async with AsyncSessionLocal() as db:
|
||
result = await fit_matching_service.recompute_for_part(
|
||
db, augmented_part_id
|
||
)
|
||
actor_row = await user_crud.get(db, actor_id)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor_row,
|
||
action="update",
|
||
resource_type="Fit",
|
||
resource_id=augmented_part_id,
|
||
description=(
|
||
f"Recomputed fits for AugmentedPart #{augmented_part_id}: "
|
||
f"{result.created} created, {result.removed} removed"
|
||
),
|
||
)
|
||
job.status = "done"
|
||
job.message = f"{result.created} created, {result.removed} removed."
|
||
|
||
job = jobs.start("fit_recompute_part", _run)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_job_progress.html",
|
||
{
|
||
"container_id": _augmented_part_fits_target_id(augmented_part_id),
|
||
"poll_url": f"/admin/augmented-parts/{augmented_part_id}/fits/recompute/jobs/{job.id}",
|
||
"job": job,
|
||
"extra_class": "ap-inline-section",
|
||
},
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/augmented-parts/{augmented_part_id}/fits/recompute/jobs/{job_id}",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def augmented_part_fits_recompute_job_status(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
job_id: str,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("fits:write")),
|
||
):
|
||
job = jobs.get(job_id)
|
||
if job is None:
|
||
ctx = await _augmented_part_fits_ctx(
|
||
db,
|
||
augmented_part_id,
|
||
can_write=_can_write_fits(actor),
|
||
error="Recompute job not found — it may have expired.",
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_augmented_part_fits.html", ctx
|
||
)
|
||
if job.status == "running":
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_job_progress.html",
|
||
{
|
||
"container_id": _augmented_part_fits_target_id(augmented_part_id),
|
||
"poll_url": (
|
||
f"/admin/augmented-parts/{augmented_part_id}/fits/recompute/jobs/{job_id}"
|
||
),
|
||
"job": job,
|
||
"extra_class": "ap-inline-section",
|
||
},
|
||
)
|
||
ctx = await _augmented_part_fits_ctx(
|
||
db,
|
||
augmented_part_id,
|
||
can_write=_can_write_fits(actor),
|
||
message=job.message if job.status == "done" else None,
|
||
error=job.message if job.status == "error" else None,
|
||
)
|
||
return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
|
||
|
||
|
||
@router.get(
|
||
"/augmented-parts/{augmented_part_id}/fits/other-points",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def fit_other_points(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
other_augmented_part_id: str = "",
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("fits:write")),
|
||
):
|
||
points = (
|
||
await attachment_point_crud.list_for_augmented_part(
|
||
db, int(other_augmented_part_id)
|
||
)
|
||
if other_augmented_part_id
|
||
else []
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_fit_other_points_select.html", {"points": points}
|
||
)
|
||
|
||
|
||
@router.post("/augmented-parts/{augmented_part_id}/fits", response_class=HTMLResponse)
|
||
async def create_augmented_part_fit(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
own_attachment_point_id: int = Form(...),
|
||
other_augmented_part_id: str = Form(""),
|
||
other_attachment_point_id: str = Form(""),
|
||
label: FitLabel = Form(FitLabel.EXACT),
|
||
note: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("fits:write")),
|
||
):
|
||
error = None
|
||
auto_created_ap = None
|
||
ap_a = await attachment_point_crud.get(db, own_attachment_point_id)
|
||
ap_b = None
|
||
if ap_a is None:
|
||
error = "Attachment point not found."
|
||
elif other_attachment_point_id:
|
||
ap_b = await attachment_point_crud.get(db, int(other_attachment_point_id))
|
||
if ap_b is None:
|
||
error = "Attachment point not found."
|
||
elif other_augmented_part_id:
|
||
other_part = await augmented_part_crud.get(db, int(other_augmented_part_id))
|
||
if other_part is None:
|
||
error = "Part not found."
|
||
elif await attachment_point_crud.list_for_augmented_part(db, other_part.id):
|
||
error = "Select the other part's attachment point."
|
||
else:
|
||
ap_b = await attachment_point_crud.create(
|
||
db,
|
||
augmented_part_id=other_part.id,
|
||
attachment_point_name_id=ap_a.attachment_point_name_id,
|
||
iso_thread_id=ap_a.iso_thread_id,
|
||
unc_thread_id=ap_a.unc_thread_id,
|
||
clamp_connection_id=ap_a.clamp_connection_id,
|
||
static_connection_id=ap_a.static_connection_id,
|
||
)
|
||
auto_created_ap = ap_b
|
||
else:
|
||
error = "Select the other part's attachment point."
|
||
|
||
if error is None and ap_b is not None:
|
||
if ap_a.augmented_part_id == ap_b.augmented_part_id:
|
||
error = "A part cannot fit itself."
|
||
elif await fit_crud.get_by_attachment_points(db, ap_a, ap_b) is not None:
|
||
error = "That fit already exists."
|
||
else:
|
||
if auto_created_ap is not None:
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="AttachmentPoint",
|
||
resource_id=auto_created_ap.id,
|
||
description=audit_crud.describe_create(
|
||
"AttachmentPoint",
|
||
auto_created_ap.id,
|
||
f"auto-created on AugmentedPart #{auto_created_ap.augmented_part_id} "
|
||
f"to match attachment point {ap_a.id}",
|
||
),
|
||
)
|
||
fit = await fit_crud.create(db, ap_a=ap_a, ap_b=ap_b, label=label)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="Fit",
|
||
resource_id=fit.id,
|
||
description=audit_crud.describe_create(
|
||
"Fit",
|
||
fit.id,
|
||
f"attachment points {ap_a.id}↔{ap_b.id} ({label.value})",
|
||
),
|
||
)
|
||
if note.strip():
|
||
await fit_note_crud.create(db, fit=fit, author=actor, body=note.strip())
|
||
await db.refresh(fit, attribute_names=["notes"])
|
||
ctx = await _augmented_part_fits_ctx(
|
||
db, augmented_part_id, can_write=_can_write_fits(actor), error=error
|
||
)
|
||
return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
|
||
|
||
|
||
@router.post(
|
||
"/augmented-parts/{augmented_part_id}/fits/{fit_id}/label",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def update_augmented_part_fit_label(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
fit_id: int,
|
||
label: FitLabel = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("fits:write")),
|
||
):
|
||
fit = await fit_crud.get(db, fit_id)
|
||
if fit is None:
|
||
raise HTTPException(status_code=404, detail="Fit not found")
|
||
fit = await fit_crud.set_label(db, fit, label)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="Fit",
|
||
resource_id=fit.id,
|
||
description=audit_crud.describe_update("Fit", fit.id, {"label": label.value}),
|
||
)
|
||
view = await _fit_view(db, fit, augmented_part_id)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_augmented_part_fit_row.html",
|
||
{
|
||
"augmented_part_id": augmented_part_id,
|
||
"can_write": _can_write_fits(actor),
|
||
**view,
|
||
},
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/augmented-parts/{augmented_part_id}/fits/{fit_id}/notes",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def create_augmented_part_fit_note(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
fit_id: int,
|
||
body: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("fits:write")),
|
||
):
|
||
fit = await fit_crud.get(db, fit_id)
|
||
if fit is None:
|
||
raise HTTPException(status_code=404, detail="Fit not found")
|
||
if body.strip():
|
||
await fit_note_crud.create(db, fit=fit, author=actor, body=body.strip())
|
||
await db.refresh(fit, attribute_names=["notes"])
|
||
view = await _fit_view(db, fit, augmented_part_id)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_augmented_part_fit_row.html",
|
||
{
|
||
"augmented_part_id": augmented_part_id,
|
||
"can_write": _can_write_fits(actor),
|
||
**view,
|
||
},
|
||
)
|
||
|
||
|
||
@router.delete(
|
||
"/augmented-parts/{augmented_part_id}/fits/{fit_id}", response_class=HTMLResponse
|
||
)
|
||
async def delete_augmented_part_fit(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
fit_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("fits:delete")),
|
||
):
|
||
fit = await fit_crud.get(db, fit_id)
|
||
if fit is not None:
|
||
label = (
|
||
f"attachment points {fit.attachment_point_a_id}↔{fit.attachment_point_b_id}"
|
||
)
|
||
await fit_crud.delete(db, fit)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="Fit",
|
||
resource_id=fit_id,
|
||
description=audit_crud.describe_delete("Fit", fit_id, label),
|
||
)
|
||
ctx = await _augmented_part_fits_ctx(
|
||
db, augmented_part_id, can_write=_can_write_fits(actor)
|
||
)
|
||
return templates.TemplateResponse(request, "admin/_augmented_part_fits.html", ctx)
|
||
|
||
|
||
_ALL_FITS_RECOMPUTE_STATUS_ID = "fits-recompute-status"
|
||
|
||
|
||
@router.post("/augmented-parts/fits/recompute", response_class=HTMLResponse)
|
||
async def recompute_all_fits(
|
||
request: Request,
|
||
actor: User = Depends(web_require("fits:write")),
|
||
):
|
||
actor_id = actor.id
|
||
|
||
async def _run(job: jobs.Job) -> None:
|
||
async with AsyncSessionLocal() as db:
|
||
result = await fit_matching_service.recompute_all(db)
|
||
actor_row = await user_crud.get(db, actor_id)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor_row,
|
||
action="update",
|
||
resource_type="Fit",
|
||
resource_id=None,
|
||
description=(
|
||
f"Recomputed fits for all parts: {result.created} created, "
|
||
f"{result.removed} removed"
|
||
),
|
||
)
|
||
job.status = "done"
|
||
job.message = (
|
||
f"Recompute complete: {result.created} created, "
|
||
f"{result.removed} removed."
|
||
)
|
||
|
||
job = jobs.start("fit_recompute_all", _run)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_job_progress_inline.html",
|
||
{
|
||
"container_id": _ALL_FITS_RECOMPUTE_STATUS_ID,
|
||
"poll_url": f"/admin/augmented-parts/fits/recompute/jobs/{job.id}",
|
||
"job": job,
|
||
},
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/augmented-parts/fits/recompute/jobs/{job_id}", response_class=HTMLResponse
|
||
)
|
||
async def recompute_all_fits_job_status(
|
||
request: Request,
|
||
job_id: str,
|
||
_: User = Depends(web_require("fits:write")),
|
||
):
|
||
job = jobs.get(job_id)
|
||
if job is None:
|
||
return HTMLResponse("Recompute job not found — it may have expired.")
|
||
if job.status == "running":
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_job_progress_inline.html",
|
||
{
|
||
"container_id": _ALL_FITS_RECOMPUTE_STATUS_ID,
|
||
"poll_url": f"/admin/augmented-parts/fits/recompute/jobs/{job_id}",
|
||
"job": job,
|
||
},
|
||
)
|
||
return HTMLResponse(job.message)
|
||
|
||
|
||
# --- Augmented Part Notes -------------------------------------------------
|
||
|
||
|
||
def _can_write_augmented_part_notes(user: User) -> bool:
|
||
return user.is_superuser or "augmented_parts:write" in user.permissions
|
||
|
||
|
||
@router.post(
|
||
"/augmented-parts/{augmented_part_id}/notes",
|
||
response_class=HTMLResponse,
|
||
)
|
||
async def create_augmented_part_note(
|
||
request: Request,
|
||
augmented_part_id: int,
|
||
body: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("augmented_parts:write")),
|
||
):
|
||
ap = await augmented_part_crud.get(db, augmented_part_id)
|
||
if ap is None:
|
||
raise HTTPException(status_code=404, detail="Augmented part not found")
|
||
if body.strip():
|
||
await augmented_part_note_crud.create(
|
||
db, augmented_part=ap, author=actor, body=body.strip()
|
||
)
|
||
await db.refresh(ap, attribute_names=["notes"])
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_augmented_part_notes.html",
|
||
{
|
||
"augmented_part_id": augmented_part_id,
|
||
"notes": ap.notes,
|
||
"can_write_notes": _can_write_augmented_part_notes(actor),
|
||
},
|
||
)
|
||
|
||
|
||
# --- Part Colours ------------------------------------------------------------
|
||
|
||
|
||
@router.get("/part-colours", response_class=HTMLResponse)
|
||
async def part_colours_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("part_colours:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/part_colours.html",
|
||
{"user": user, "part_colours": await part_colour_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/part-colours", response_class=HTMLResponse)
|
||
async def create_part_colour(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_colours:write")),
|
||
):
|
||
if not await part_colour_crud.get_by_name(db, name):
|
||
created = await part_colour_crud.create(db, name=name)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="PartColour",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create(
|
||
"PartColour", created.id, created.name
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_part_colours_table.html",
|
||
{"part_colours": await part_colour_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/part-colours/{part_colour_id}/edit", response_class=HTMLResponse)
|
||
async def edit_part_colour_row(
|
||
request: Request,
|
||
part_colour_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("part_colours:write")),
|
||
):
|
||
pc = await part_colour_crud.get(db, part_colour_id)
|
||
if pc is None:
|
||
raise HTTPException(status_code=404, detail="Part colour not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_colour_row_edit.html", {"pc": pc}
|
||
)
|
||
|
||
|
||
@router.get("/part-colours/{part_colour_id}/row", response_class=HTMLResponse)
|
||
async def part_colour_row(
|
||
request: Request,
|
||
part_colour_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("part_colours:read")),
|
||
):
|
||
pc = await part_colour_crud.get(db, part_colour_id)
|
||
if pc is None:
|
||
raise HTTPException(status_code=404, detail="Part colour not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_colour_row.html", {"pc": pc}
|
||
)
|
||
|
||
|
||
@router.post("/part-colours/{part_colour_id}", response_class=HTMLResponse)
|
||
async def update_part_colour(
|
||
request: Request,
|
||
part_colour_id: int,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_colours:write")),
|
||
):
|
||
pc = await part_colour_crud.get(db, part_colour_id)
|
||
if pc is None:
|
||
raise HTTPException(status_code=404, detail="Part colour not found")
|
||
changes = {"name": name}
|
||
await part_colour_crud.update(db, pc, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="PartColour",
|
||
resource_id=part_colour_id,
|
||
description=audit_crud.describe_update("PartColour", part_colour_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_part_colour_row.html", {"pc": pc}
|
||
)
|
||
|
||
|
||
@router.delete("/part-colours/{part_colour_id}")
|
||
async def delete_part_colour(
|
||
part_colour_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("part_colours:delete")),
|
||
):
|
||
pc = await part_colour_crud.get(db, part_colour_id)
|
||
if pc is not None:
|
||
name = pc.name
|
||
await part_colour_crud.delete(db, pc)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="PartColour",
|
||
resource_id=part_colour_id,
|
||
description=audit_crud.describe_delete("PartColour", part_colour_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Attachment Point Names ---------------------------------------------------
|
||
|
||
|
||
@router.get("/attachment-point-names", response_class=HTMLResponse)
|
||
async def attachment_point_names_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("attachment_point_names:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/attachment_point_names.html",
|
||
{"user": user, "attachment_point_names": await apn_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/attachment-point-names", response_class=HTMLResponse)
|
||
async def create_attachment_point_name(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("attachment_point_names:write")),
|
||
):
|
||
if not await apn_crud.get_by_name(db, name):
|
||
created = await apn_crud.create(db, name=name)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="AttachmentPointName",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create(
|
||
"AttachmentPointName", created.id, name
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_attachment_point_names_table.html",
|
||
{"attachment_point_names": await apn_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get("/attachment-point-names/{apn_id}/edit", response_class=HTMLResponse)
|
||
async def edit_attachment_point_name_row(
|
||
request: Request,
|
||
apn_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("attachment_point_names:write")),
|
||
):
|
||
apn = await apn_crud.get(db, apn_id)
|
||
if apn is None:
|
||
raise HTTPException(status_code=404, detail="Attachment point name not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_attachment_point_name_row_edit.html", {"apn": apn}
|
||
)
|
||
|
||
|
||
@router.get("/attachment-point-names/{apn_id}/row", response_class=HTMLResponse)
|
||
async def attachment_point_name_row(
|
||
request: Request,
|
||
apn_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("attachment_point_names:read")),
|
||
):
|
||
apn = await apn_crud.get(db, apn_id)
|
||
if apn is None:
|
||
raise HTTPException(status_code=404, detail="Attachment point name not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_attachment_point_name_row.html", {"apn": apn}
|
||
)
|
||
|
||
|
||
@router.post("/attachment-point-names/{apn_id}", response_class=HTMLResponse)
|
||
async def update_attachment_point_name(
|
||
request: Request,
|
||
apn_id: int,
|
||
name: str = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("attachment_point_names:write")),
|
||
):
|
||
apn = await apn_crud.get(db, apn_id)
|
||
if apn is None:
|
||
raise HTTPException(status_code=404, detail="Attachment point name not found")
|
||
changes = {"name": name}
|
||
await apn_crud.update(db, apn, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="AttachmentPointName",
|
||
resource_id=apn_id,
|
||
description=audit_crud.describe_update("AttachmentPointName", apn_id, changes),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_attachment_point_name_row.html", {"apn": apn}
|
||
)
|
||
|
||
|
||
@router.delete("/attachment-point-names/{apn_id}")
|
||
async def delete_attachment_point_name(
|
||
apn_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("attachment_point_names:delete")),
|
||
):
|
||
apn = await apn_crud.get(db, apn_id)
|
||
if apn is not None:
|
||
name = apn.name
|
||
await apn_crud.delete(db, apn)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="AttachmentPointName",
|
||
resource_id=apn_id,
|
||
description=audit_crud.describe_delete("AttachmentPointName", apn_id, name),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Attachment Points --------------------------------------------------------
|
||
|
||
|
||
async def _attachment_point_form_ctx(db: AsyncSession) -> dict:
|
||
return {
|
||
"augmented_parts": await augmented_part_crud.list_all(db),
|
||
"attachment_point_names": await apn_crud.list_all(db),
|
||
"iso_threads": await iso_thread_crud.list_all(db),
|
||
"unc_threads": await unc_thread_crud.list_all(db),
|
||
"clamp_connections": await clamp_connection_crud.list_all(db),
|
||
"static_connections": await static_connection_crud.list_all(db),
|
||
}
|
||
|
||
|
||
def _ap_connection_label(ap) -> str:
|
||
if ap.iso_thread:
|
||
return f"M{ap.iso_thread.diameter_mm}×{ap.iso_thread.pitch_mm}"
|
||
if ap.unc_thread:
|
||
return f"{ap.unc_thread.diameter_fraction}-{ap.unc_thread.tpi} {ap.unc_thread.series.value}"
|
||
if ap.clamp_connection:
|
||
cc = ap.clamp_connection
|
||
if cc.min_mm is not None:
|
||
return f"{cc.min_mm}–{cc.max_mm} mm"
|
||
return f"{cc.min_in}–{cc.max_in} in"
|
||
if ap.static_connection:
|
||
return f"{ap.static_connection.od_mm} mm OD"
|
||
return "—"
|
||
|
||
|
||
@router.get("/attachment-points", response_class=HTMLResponse)
|
||
async def attachment_points_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("attachment_points:read")),
|
||
):
|
||
ctx = await _attachment_point_form_ctx(db)
|
||
ctx["user"] = user
|
||
ctx["attachment_points"] = await attachment_point_crud.list_all(db)
|
||
return templates.TemplateResponse(request, "admin/attachment_points.html", ctx)
|
||
|
||
|
||
# Must come before /{attachment_point_id}
|
||
@router.get("/attachment-points/connection-options", response_class=HTMLResponse)
|
||
async def attachment_point_connection_options(
|
||
request: Request,
|
||
connection_type: str = "",
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("attachment_points:write")),
|
||
):
|
||
ctx = await _attachment_point_form_ctx(db)
|
||
ctx["connection_type"] = connection_type
|
||
ctx["ap"] = None
|
||
return templates.TemplateResponse(
|
||
request, "admin/_attachment_point_connection_select.html", ctx
|
||
)
|
||
|
||
|
||
@router.post("/attachment-points", response_class=HTMLResponse)
|
||
async def create_attachment_point(
|
||
request: Request,
|
||
augmented_part_id: int = Form(...),
|
||
attachment_point_name_id: int = Form(...),
|
||
connection_type: str = Form(""),
|
||
iso_thread_id: str = Form(""),
|
||
unc_thread_id: str = Form(""),
|
||
clamp_connection_id: str = Form(""),
|
||
static_connection_id: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("attachment_points:write")),
|
||
):
|
||
iso_id = int(iso_thread_id) if iso_thread_id else None
|
||
unc_id = int(unc_thread_id) if unc_thread_id else None
|
||
clamp_id = int(clamp_connection_id) if clamp_connection_id else None
|
||
static_id = int(static_connection_id) if static_connection_id else None
|
||
ap = await attachment_point_crud.create(
|
||
db,
|
||
augmented_part_id=augmented_part_id,
|
||
attachment_point_name_id=attachment_point_name_id,
|
||
iso_thread_id=iso_id,
|
||
unc_thread_id=unc_id,
|
||
clamp_connection_id=clamp_id,
|
||
static_connection_id=static_id,
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="AttachmentPoint",
|
||
resource_id=ap.id,
|
||
description=audit_crud.describe_create(
|
||
"AttachmentPoint", ap.id, _ap_connection_label(ap)
|
||
),
|
||
)
|
||
ctx = await _attachment_point_form_ctx(db)
|
||
ctx["attachment_points"] = await attachment_point_crud.list_all(db)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_attachment_points_table.html", ctx
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/attachment-points/{attachment_point_id}/edit", response_class=HTMLResponse
|
||
)
|
||
async def edit_attachment_point_row(
|
||
request: Request,
|
||
attachment_point_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("attachment_points:write")),
|
||
):
|
||
ap = await attachment_point_crud.get(db, attachment_point_id)
|
||
if ap is None:
|
||
raise HTTPException(status_code=404, detail="Attachment point not found")
|
||
ctx = await _attachment_point_form_ctx(db)
|
||
ctx["ap"] = ap
|
||
if ap.iso_thread_id:
|
||
ctx["connection_type"] = "iso_thread"
|
||
elif ap.unc_thread_id:
|
||
ctx["connection_type"] = "unc_thread"
|
||
elif ap.clamp_connection_id:
|
||
ctx["connection_type"] = "clamp_connection"
|
||
elif ap.static_connection_id:
|
||
ctx["connection_type"] = "static_connection"
|
||
else:
|
||
ctx["connection_type"] = ""
|
||
return templates.TemplateResponse(
|
||
request, "admin/_attachment_point_row_edit.html", ctx
|
||
)
|
||
|
||
|
||
@router.get("/attachment-points/{attachment_point_id}/row", response_class=HTMLResponse)
|
||
async def attachment_point_row(
|
||
request: Request,
|
||
attachment_point_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("attachment_points:read")),
|
||
):
|
||
ap = await attachment_point_crud.get(db, attachment_point_id)
|
||
if ap is None:
|
||
raise HTTPException(status_code=404, detail="Attachment point not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_attachment_point_row.html", {"ap": ap}
|
||
)
|
||
|
||
|
||
@router.post("/attachment-points/{attachment_point_id}", response_class=HTMLResponse)
|
||
async def update_attachment_point(
|
||
request: Request,
|
||
attachment_point_id: int,
|
||
augmented_part_id: int = Form(...),
|
||
attachment_point_name_id: int = Form(...),
|
||
connection_type: str = Form(""),
|
||
iso_thread_id: str = Form(""),
|
||
unc_thread_id: str = Form(""),
|
||
clamp_connection_id: str = Form(""),
|
||
static_connection_id: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("attachment_points:write")),
|
||
):
|
||
ap = await attachment_point_crud.get(db, attachment_point_id)
|
||
if ap is None:
|
||
raise HTTPException(status_code=404, detail="Attachment point not found")
|
||
changes = {
|
||
"augmented_part_id": augmented_part_id,
|
||
"attachment_point_name_id": attachment_point_name_id,
|
||
"iso_thread_id": int(iso_thread_id) if iso_thread_id else None,
|
||
"unc_thread_id": int(unc_thread_id) if unc_thread_id else None,
|
||
"clamp_connection_id": (
|
||
int(clamp_connection_id) if clamp_connection_id else None
|
||
),
|
||
"static_connection_id": (
|
||
int(static_connection_id) if static_connection_id else None
|
||
),
|
||
}
|
||
ap = await attachment_point_crud.update(db, ap, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="AttachmentPoint",
|
||
resource_id=attachment_point_id,
|
||
description=audit_crud.describe_update(
|
||
"AttachmentPoint", attachment_point_id, changes
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_attachment_point_row.html", {"ap": ap}
|
||
)
|
||
|
||
|
||
@router.delete("/attachment-points/{attachment_point_id}")
|
||
async def delete_attachment_point(
|
||
attachment_point_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("attachment_points:delete")),
|
||
):
|
||
ap = await attachment_point_crud.get(db, attachment_point_id)
|
||
if ap is not None:
|
||
label = _ap_connection_label(ap)
|
||
await attachment_point_crud.delete(db, ap)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="AttachmentPoint",
|
||
resource_id=attachment_point_id,
|
||
description=audit_crud.describe_delete(
|
||
"AttachmentPoint", attachment_point_id, label
|
||
),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Static Connections -------------------------------------------------------
|
||
|
||
|
||
@router.get("/static-connections", response_class=HTMLResponse)
|
||
async def static_connections_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("static_connections:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/static_connections.html",
|
||
{"user": user, "static_connections": await static_connection_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/static-connections", response_class=HTMLResponse)
|
||
async def create_static_connection(
|
||
request: Request,
|
||
od_mm: float = Form(...),
|
||
od_in: float = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("static_connections:write")),
|
||
):
|
||
if not await static_connection_crud.get_by_od_mm(db, od_mm):
|
||
created = await static_connection_crud.create(db, od_mm=od_mm, od_in=od_in)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="StaticConnection",
|
||
resource_id=created.id,
|
||
description=audit_crud.describe_create(
|
||
"StaticConnection", created.id, f"{created.od_mm} mm"
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_static_connections_table.html",
|
||
{"static_connections": await static_connection_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/static-connections/{static_connection_id}/edit", response_class=HTMLResponse
|
||
)
|
||
async def edit_static_connection_row(
|
||
request: Request,
|
||
static_connection_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("static_connections:write")),
|
||
):
|
||
sc = await static_connection_crud.get(db, static_connection_id)
|
||
if sc is None:
|
||
raise HTTPException(status_code=404, detail="Static connection not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_static_connection_row_edit.html", {"sc": sc}
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/static-connections/{static_connection_id}/row", response_class=HTMLResponse
|
||
)
|
||
async def static_connection_row(
|
||
request: Request,
|
||
static_connection_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("static_connections:read")),
|
||
):
|
||
sc = await static_connection_crud.get(db, static_connection_id)
|
||
if sc is None:
|
||
raise HTTPException(status_code=404, detail="Static connection not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_static_connection_row.html", {"sc": sc}
|
||
)
|
||
|
||
|
||
@router.post("/static-connections/{static_connection_id}", response_class=HTMLResponse)
|
||
async def update_static_connection(
|
||
request: Request,
|
||
static_connection_id: int,
|
||
od_mm: float = Form(...),
|
||
od_in: float = Form(...),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("static_connections:write")),
|
||
):
|
||
sc = await static_connection_crud.get(db, static_connection_id)
|
||
if sc is None:
|
||
raise HTTPException(status_code=404, detail="Static connection not found")
|
||
changes = {"od_mm": od_mm, "od_in": od_in}
|
||
await static_connection_crud.update(db, sc, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="StaticConnection",
|
||
resource_id=static_connection_id,
|
||
description=audit_crud.describe_update(
|
||
"StaticConnection", static_connection_id, changes
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_static_connection_row.html", {"sc": sc}
|
||
)
|
||
|
||
|
||
@router.delete("/static-connections/{static_connection_id}")
|
||
async def delete_static_connection(
|
||
static_connection_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("static_connections:delete")),
|
||
):
|
||
sc = await static_connection_crud.get(db, static_connection_id)
|
||
if sc is not None:
|
||
od_mm = sc.od_mm
|
||
await static_connection_crud.delete(db, sc)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="StaticConnection",
|
||
resource_id=static_connection_id,
|
||
description=audit_crud.describe_delete(
|
||
"StaticConnection", static_connection_id, f"{od_mm} mm"
|
||
),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Clamp Connections -------------------------------------------------------
|
||
|
||
|
||
@router.get("/clamp-connections", response_class=HTMLResponse)
|
||
async def clamp_connections_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("clamp_connections:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/clamp_connections.html",
|
||
{"user": user, "clamp_connections": await clamp_connection_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.post("/clamp-connections", response_class=HTMLResponse)
|
||
async def create_clamp_connection(
|
||
request: Request,
|
||
min_mm: str = Form(""),
|
||
max_mm: str = Form(""),
|
||
min_in: str = Form(""),
|
||
max_in: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("clamp_connections:write")),
|
||
):
|
||
cc = await clamp_connection_crud.create(
|
||
db,
|
||
min_mm=float(min_mm) if min_mm else None,
|
||
max_mm=float(max_mm) if max_mm else None,
|
||
min_in=float(min_in) if min_in else None,
|
||
max_in=float(max_in) if max_in else None,
|
||
)
|
||
label = (
|
||
f"{cc.min_mm}–{cc.max_mm} mm"
|
||
if cc.min_mm is not None
|
||
else f"{cc.min_in}–{cc.max_in} in"
|
||
)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="create",
|
||
resource_type="ClampConnection",
|
||
resource_id=cc.id,
|
||
description=audit_crud.describe_create("ClampConnection", cc.id, label),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_clamp_connections_table.html",
|
||
{"clamp_connections": await clamp_connection_crud.list_all(db)},
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/clamp-connections/{clamp_connection_id}/edit", response_class=HTMLResponse
|
||
)
|
||
async def edit_clamp_connection_row(
|
||
request: Request,
|
||
clamp_connection_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("clamp_connections:write")),
|
||
):
|
||
cc = await clamp_connection_crud.get(db, clamp_connection_id)
|
||
if cc is None:
|
||
raise HTTPException(status_code=404, detail="Clamp connection not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_clamp_connection_row_edit.html", {"cc": cc}
|
||
)
|
||
|
||
|
||
@router.get("/clamp-connections/{clamp_connection_id}/row", response_class=HTMLResponse)
|
||
async def clamp_connection_row(
|
||
request: Request,
|
||
clamp_connection_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
_: User = Depends(web_require("clamp_connections:read")),
|
||
):
|
||
cc = await clamp_connection_crud.get(db, clamp_connection_id)
|
||
if cc is None:
|
||
raise HTTPException(status_code=404, detail="Clamp connection not found")
|
||
return templates.TemplateResponse(
|
||
request, "admin/_clamp_connection_row.html", {"cc": cc}
|
||
)
|
||
|
||
|
||
@router.post("/clamp-connections/{clamp_connection_id}", response_class=HTMLResponse)
|
||
async def update_clamp_connection(
|
||
request: Request,
|
||
clamp_connection_id: int,
|
||
min_mm: str = Form(""),
|
||
max_mm: str = Form(""),
|
||
min_in: str = Form(""),
|
||
max_in: str = Form(""),
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("clamp_connections:write")),
|
||
):
|
||
cc = await clamp_connection_crud.get(db, clamp_connection_id)
|
||
if cc is None:
|
||
raise HTTPException(status_code=404, detail="Clamp connection not found")
|
||
changes = {
|
||
"min_mm": float(min_mm) if min_mm else None,
|
||
"max_mm": float(max_mm) if max_mm else None,
|
||
"min_in": float(min_in) if min_in else None,
|
||
"max_in": float(max_in) if max_in else None,
|
||
}
|
||
await clamp_connection_crud.update(db, cc, changes)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="update",
|
||
resource_type="ClampConnection",
|
||
resource_id=clamp_connection_id,
|
||
description=audit_crud.describe_update(
|
||
"ClampConnection", clamp_connection_id, changes
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request, "admin/_clamp_connection_row.html", {"cc": cc}
|
||
)
|
||
|
||
|
||
@router.delete("/clamp-connections/{clamp_connection_id}")
|
||
async def delete_clamp_connection(
|
||
clamp_connection_id: int,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require("clamp_connections:delete")),
|
||
):
|
||
cc = await clamp_connection_crud.get(db, clamp_connection_id)
|
||
if cc is not None:
|
||
label = (
|
||
f"{cc.min_mm}–{cc.max_mm} mm"
|
||
if cc.min_mm is not None
|
||
else f"{cc.min_in}–{cc.max_in} in"
|
||
)
|
||
await clamp_connection_crud.delete(db, cc)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="ClampConnection",
|
||
resource_id=clamp_connection_id,
|
||
description=audit_crud.describe_delete(
|
||
"ClampConnection", clamp_connection_id, label
|
||
),
|
||
)
|
||
return Response(status_code=status.HTTP_200_OK)
|
||
|
||
|
||
# --- Audit Log ---------------------------------------------------------------
|
||
|
||
|
||
@router.get("/audit-log", response_class=HTMLResponse)
|
||
async def audit_log_page(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
user: User = Depends(web_require("audit_log:read")),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/audit_log.html",
|
||
{"user": user, "entries": await audit_crud.list_recent(db)},
|
||
)
|
||
|
||
|
||
# --- Danger Zone ---------------------------------------------------------------
|
||
|
||
|
||
@router.get("/danger-zone", response_class=HTMLResponse)
|
||
async def danger_zone_page(
|
||
request: Request,
|
||
user: User = Depends(web_require_superuser),
|
||
):
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/danger_zone.html",
|
||
{"user": user},
|
||
)
|
||
|
||
|
||
@router.post("/danger-zone/purge", response_class=HTMLResponse)
|
||
async def purge_database(
|
||
request: Request,
|
||
db: AsyncSession = Depends(get_db),
|
||
actor: User = Depends(web_require_superuser),
|
||
):
|
||
counts = await danger_zone_service.purge_domain_data(db)
|
||
total = sum(counts.values())
|
||
breakdown = ", ".join(f"{label} {n}" for label, n in counts.items() if n)
|
||
await audit_crud.log(
|
||
db,
|
||
actor=actor,
|
||
action="delete",
|
||
resource_type="Database",
|
||
resource_id=None,
|
||
description=(
|
||
f"Purged all domain data via Danger Zone ({total} row(s) deleted"
|
||
+ (f": {breakdown})" if breakdown else ")")
|
||
),
|
||
)
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"admin/_danger_zone_result.html",
|
||
{"total": total, "counts": counts},
|
||
)
|