1073 lines
25 KiB
Python
1073 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
from fastapi import (
|
|
FastAPI,
|
|
Form,
|
|
HTTPException,
|
|
Request,
|
|
status,
|
|
)
|
|
from fastapi.responses import (
|
|
HTMLResponse,
|
|
RedirectResponse,
|
|
)
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from psycopg2.extras import RealDictCursor
|
|
|
|
from .auth import (
|
|
SESSION_COOKIE,
|
|
authenticate,
|
|
build_session_token,
|
|
can_approve_context,
|
|
can_manage_clarification,
|
|
can_review_owner,
|
|
can_view_owner,
|
|
read_current_user,
|
|
require_user,
|
|
verify_csrf,
|
|
)
|
|
from .db import get_connection
|
|
from .labels import label
|
|
from .proposal_service import (
|
|
ProposalActionInput,
|
|
ProposalError,
|
|
approve_proposal,
|
|
change_proposal_status,
|
|
)
|
|
from .review_service import (
|
|
ConsultationInput,
|
|
ReviewError,
|
|
ReviewInput,
|
|
apply_production_review,
|
|
mark_in_consultation,
|
|
resume_from_consultation,
|
|
)
|
|
from .settings import get_settings
|
|
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
|
|
app = FastAPI(
|
|
title="MESAVAULT Governance Review UI",
|
|
version="0.3.15",
|
|
)
|
|
|
|
app.mount(
|
|
"/static",
|
|
StaticFiles(
|
|
directory=BASE_DIR / "static"
|
|
),
|
|
name="static",
|
|
)
|
|
|
|
templates = Jinja2Templates(
|
|
directory=BASE_DIR / "templates"
|
|
)
|
|
|
|
|
|
def format_datetime(value: Any) -> str:
|
|
if value is None:
|
|
return "—"
|
|
try:
|
|
return value.strftime(
|
|
"%d/%m/%Y %H:%M:%S"
|
|
)
|
|
except AttributeError:
|
|
return str(value)
|
|
|
|
|
|
templates.env.filters[
|
|
"datetime_es"
|
|
] = format_datetime
|
|
templates.env.filters[
|
|
"label"
|
|
] = label
|
|
|
|
|
|
def redirect_login() -> RedirectResponse:
|
|
return RedirectResponse(
|
|
"/login",
|
|
status_code=status.HTTP_303_SEE_OTHER,
|
|
)
|
|
|
|
|
|
def visible_queue(user) -> list[dict[str, Any]]:
|
|
with get_connection() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_ui_queue_v2
|
|
ORDER BY agenda_rank
|
|
"""
|
|
)
|
|
rows = [
|
|
dict(row)
|
|
for row in cursor.fetchall()
|
|
]
|
|
|
|
return [
|
|
row
|
|
for row in rows
|
|
if can_view_owner(
|
|
user,
|
|
str(row["review_owner"]),
|
|
)
|
|
]
|
|
|
|
|
|
def recent_activity(
|
|
limit: int = 100,
|
|
) -> list[dict[str, Any]]:
|
|
with get_connection() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_ui_activity_v1
|
|
ORDER BY
|
|
created_at DESC,
|
|
activity_type,
|
|
activity_id DESC
|
|
LIMIT %s
|
|
""",
|
|
(limit,),
|
|
)
|
|
return [
|
|
dict(row)
|
|
for row in cursor.fetchall()
|
|
]
|
|
|
|
|
|
def load_decision(
|
|
decision_group_key: str,
|
|
) -> tuple[
|
|
dict[str, Any],
|
|
list[dict[str, Any]],
|
|
dict[str, Any] | None,
|
|
]:
|
|
with get_connection() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_decision_evidence_pack_v1
|
|
WHERE decision_group_key = %s
|
|
""",
|
|
(decision_group_key,),
|
|
)
|
|
decision = cursor.fetchone()
|
|
|
|
if not decision:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Decisión no encontrada.",
|
|
)
|
|
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_ui_case_timeline_v1
|
|
WHERE decision_group_key = %s
|
|
ORDER BY
|
|
first_seen_at,
|
|
case_id
|
|
""",
|
|
(decision_group_key,),
|
|
)
|
|
cases = [
|
|
dict(row)
|
|
for row in cursor.fetchall()
|
|
]
|
|
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_ui_consultation_current_v1
|
|
WHERE decision_group_key = %s
|
|
""",
|
|
(decision_group_key,),
|
|
)
|
|
consultation = cursor.fetchone()
|
|
|
|
return (
|
|
dict(decision),
|
|
cases,
|
|
dict(consultation)
|
|
if consultation
|
|
else None,
|
|
)
|
|
|
|
|
|
def answer_options(
|
|
decision_family: str,
|
|
) -> list[tuple[str, str]]:
|
|
if decision_family == "UNCONFIRMED_ACTIVITY":
|
|
codes = [
|
|
"AUTHORIZED_LEGACY_PRODUCTION",
|
|
"ODOO_START_OMITTED",
|
|
"TEST_OR_SETUP",
|
|
"RESIDUAL_MATERIAL",
|
|
"MAINTENANCE",
|
|
"NOT_PRODUCTION",
|
|
"OTHER",
|
|
]
|
|
elif decision_family == "ODOO_ORDER_START_CONFLICT":
|
|
codes = [
|
|
"ODOO_DOUBLE_START_ONE_VALID",
|
|
"ODOO_DOUBLE_START_BOTH_INVALID",
|
|
"ODOO_CONSECUTIVE_ORDERS_NOT_CLOSED",
|
|
"DATA_ISSUE",
|
|
"OTHER",
|
|
]
|
|
else:
|
|
codes = [
|
|
"DATA_ISSUE",
|
|
"OTHER",
|
|
]
|
|
|
|
return [
|
|
(code, label(code))
|
|
for code in codes
|
|
]
|
|
|
|
|
|
def load_approval_queue() -> list[dict[str, Any]]:
|
|
with get_connection() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_context_approval_queue_v1
|
|
ORDER BY approval_rank
|
|
"""
|
|
)
|
|
return [
|
|
dict(row)
|
|
for row in cursor.fetchall()
|
|
]
|
|
|
|
|
|
def load_proposal(
|
|
proposal_id: int,
|
|
) -> dict[str, Any]:
|
|
with get_connection() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_context_proposals_v1
|
|
WHERE proposal_id = %s
|
|
""",
|
|
(proposal_id,),
|
|
)
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Propuesta no encontrada.",
|
|
)
|
|
return dict(row)
|
|
|
|
|
|
def load_clarifications() -> list[dict[str, Any]]:
|
|
with get_connection() as connection:
|
|
with connection.cursor(
|
|
cursor_factory=RealDictCursor,
|
|
) as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT *
|
|
FROM
|
|
mv_reports_ucepsa_prod
|
|
.li_context_governance_context_clarification_queue_v1
|
|
ORDER BY
|
|
last_action_at,
|
|
proposal_id
|
|
"""
|
|
)
|
|
return [
|
|
dict(row)
|
|
for row in cursor.fetchall()
|
|
]
|
|
|
|
|
|
@app.get(
|
|
"/health",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def health() -> str:
|
|
with get_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SELECT 1")
|
|
cursor.fetchone()
|
|
return "ok"
|
|
|
|
|
|
@app.get(
|
|
"/login",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def login_page(request: Request):
|
|
if read_current_user(request):
|
|
return RedirectResponse(
|
|
"/",
|
|
status_code=303,
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="login.html",
|
|
context={
|
|
"settings": get_settings(),
|
|
"error": None,
|
|
},
|
|
)
|
|
|
|
|
|
@app.post(
|
|
"/login",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def login_submit(
|
|
request: Request,
|
|
username: str = Form(...),
|
|
password: str = Form(...),
|
|
):
|
|
user = authenticate(
|
|
username.strip(),
|
|
password,
|
|
)
|
|
|
|
if user is None:
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="login.html",
|
|
status_code=401,
|
|
context={
|
|
"settings": get_settings(),
|
|
"error":
|
|
"Usuario o contraseña incorrectos.",
|
|
},
|
|
)
|
|
|
|
response = RedirectResponse(
|
|
"/",
|
|
status_code=303,
|
|
)
|
|
response.set_cookie(
|
|
SESSION_COOKIE,
|
|
build_session_token(user),
|
|
max_age=get_settings().session_max_age_s,
|
|
httponly=True,
|
|
secure=get_settings().cookie_secure,
|
|
samesite="lax",
|
|
path="/",
|
|
)
|
|
return response
|
|
|
|
|
|
@app.post("/logout")
|
|
def logout(
|
|
request: Request,
|
|
csrf_token: str = Form(...),
|
|
):
|
|
user = require_user(request)
|
|
verify_csrf(
|
|
user,
|
|
csrf_token,
|
|
)
|
|
|
|
response = RedirectResponse(
|
|
"/login",
|
|
status_code=303,
|
|
)
|
|
response.delete_cookie(
|
|
SESSION_COOKIE,
|
|
path="/",
|
|
)
|
|
return response
|
|
|
|
|
|
@app.get(
|
|
"/",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def queue_page(
|
|
request: Request,
|
|
message: str | None = None,
|
|
):
|
|
user = read_current_user(request)
|
|
if user is None:
|
|
return redirect_login()
|
|
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="queue.html",
|
|
context={
|
|
"settings": get_settings(),
|
|
"user": user,
|
|
"rows": visible_queue(user),
|
|
"message": message,
|
|
"can_approve":
|
|
can_approve_context(user),
|
|
"clarification_count":
|
|
len(load_clarifications())
|
|
if can_manage_clarification(user)
|
|
else 0,
|
|
},
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/decisions/{decision_group_key:path}",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def decision_page(
|
|
request: Request,
|
|
decision_group_key: str,
|
|
error: str | None = None,
|
|
):
|
|
user = read_current_user(request)
|
|
if user is None:
|
|
return redirect_login()
|
|
|
|
decision, cases, consultation = load_decision(
|
|
decision_group_key
|
|
)
|
|
|
|
if not can_view_owner(
|
|
user,
|
|
str(decision["review_owner"]),
|
|
):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="No tiene acceso a esta decisión.",
|
|
)
|
|
|
|
pending_cases = [
|
|
case
|
|
for case in cases
|
|
if case["case_review_status"] ==
|
|
"PENDING"
|
|
]
|
|
|
|
order_pairs = sorted({
|
|
(
|
|
str(item["production_order"]),
|
|
int(item["odoo_workorder_id"]),
|
|
)
|
|
for item in (
|
|
decision.get(
|
|
"episode_evidence"
|
|
)
|
|
or []
|
|
)
|
|
if item.get(
|
|
"production_order"
|
|
)
|
|
and item.get(
|
|
"odoo_workorder_id"
|
|
) is not None
|
|
})
|
|
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="decision.html",
|
|
context={
|
|
"settings": get_settings(),
|
|
"user": user,
|
|
"decision": decision,
|
|
"cases": cases,
|
|
"pending_cases": pending_cases,
|
|
"consultation": consultation,
|
|
"answer_options": answer_options(
|
|
str(
|
|
decision[
|
|
"decision_family"
|
|
]
|
|
)
|
|
),
|
|
"order_pairs": order_pairs,
|
|
"can_review": (
|
|
get_settings().write_enabled
|
|
and can_review_owner(
|
|
user,
|
|
str(decision["review_owner"]),
|
|
)
|
|
),
|
|
"write_enabled":
|
|
get_settings().write_enabled,
|
|
"review_submission_key":
|
|
str(uuid.uuid4()),
|
|
"consultation_submission_key":
|
|
str(uuid.uuid4()),
|
|
"resume_submission_key":
|
|
str(uuid.uuid4()),
|
|
"error": error,
|
|
},
|
|
)
|
|
|
|
|
|
@app.post(
|
|
"/decisions/{decision_group_key:path}/review"
|
|
)
|
|
def review_decision(
|
|
request: Request,
|
|
decision_group_key: str,
|
|
csrf_token: str = Form(...),
|
|
submission_key: uuid.UUID = Form(...),
|
|
case_ids: list[int] = Form(default=[]),
|
|
answer_code: str = Form(...),
|
|
notes: str = Form(""),
|
|
selected_order: str = Form(""),
|
|
corrected_in_odoo: bool = Form(False),
|
|
confirmation: bool = Form(False),
|
|
):
|
|
user = require_user(request)
|
|
verify_csrf(user, csrf_token)
|
|
|
|
if not get_settings().write_enabled:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="La interfaz está en modo solo lectura.",
|
|
)
|
|
|
|
decision, _, _ = load_decision(
|
|
decision_group_key
|
|
)
|
|
|
|
if not can_review_owner(
|
|
user,
|
|
str(decision["review_owner"]),
|
|
):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="No puede resolver esta decisión.",
|
|
)
|
|
|
|
selected_order_ref = None
|
|
selected_workorder_id = None
|
|
|
|
if selected_order:
|
|
try:
|
|
selected_order_ref, raw_id = (
|
|
selected_order.split(
|
|
"|",
|
|
1,
|
|
)
|
|
)
|
|
selected_workorder_id = int(
|
|
raw_id
|
|
)
|
|
except (
|
|
ValueError,
|
|
TypeError,
|
|
):
|
|
target = (
|
|
"/decisions/"
|
|
+ quote(
|
|
decision_group_key,
|
|
safe="",
|
|
)
|
|
+ "?error="
|
|
+ quote(
|
|
"Orden seleccionada no válida."
|
|
)
|
|
)
|
|
return RedirectResponse(
|
|
target,
|
|
status_code=303,
|
|
)
|
|
|
|
review = ReviewInput(
|
|
submission_key=submission_key,
|
|
decision_group_key=
|
|
decision_group_key,
|
|
selected_case_ids=tuple(
|
|
sorted(
|
|
set(
|
|
int(case_id)
|
|
for case_id in case_ids
|
|
)
|
|
)
|
|
),
|
|
answer_code=answer_code,
|
|
notes=notes.strip(),
|
|
selected_order_ref=
|
|
selected_order_ref,
|
|
selected_workorder_id=
|
|
selected_workorder_id,
|
|
corrected_in_odoo=
|
|
corrected_in_odoo,
|
|
confirmation=confirmation,
|
|
)
|
|
|
|
try:
|
|
result = apply_production_review(
|
|
review=review,
|
|
user=user,
|
|
source_ip=(
|
|
request.client.host
|
|
if request.client
|
|
else None
|
|
),
|
|
user_agent=request.headers.get(
|
|
"user-agent"
|
|
),
|
|
)
|
|
except ReviewError as exc:
|
|
target = (
|
|
"/decisions/"
|
|
+ quote(
|
|
decision_group_key,
|
|
safe="",
|
|
)
|
|
+ "?error="
|
|
+ quote(str(exc))
|
|
)
|
|
return RedirectResponse(
|
|
target,
|
|
status_code=303,
|
|
)
|
|
|
|
message = (
|
|
"Respuesta registrada. "
|
|
f"Acción UI {result['ui_action_id']}."
|
|
)
|
|
|
|
proposal_ids = result.get(
|
|
"context_proposal_ids"
|
|
) or []
|
|
if proposal_ids:
|
|
message += (
|
|
" Se han enviado "
|
|
f"{len(proposal_ids)} propuestas "
|
|
"a validación técnica."
|
|
)
|
|
|
|
return RedirectResponse(
|
|
"/?message="
|
|
+ quote(message),
|
|
status_code=303,
|
|
)
|
|
|
|
|
|
@app.post(
|
|
"/decisions/{decision_group_key:path}/consultation"
|
|
)
|
|
def consultation_submit(
|
|
request: Request,
|
|
decision_group_key: str,
|
|
csrf_token: str = Form(...),
|
|
submission_key: uuid.UUID = Form(...),
|
|
reason: str = Form(...),
|
|
follow_up_with: str = Form(""),
|
|
):
|
|
user = require_user(request)
|
|
verify_csrf(user, csrf_token)
|
|
|
|
if not get_settings().write_enabled:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="La interfaz está en modo solo lectura.",
|
|
)
|
|
|
|
decision, _, _ = load_decision(
|
|
decision_group_key
|
|
)
|
|
if not can_review_owner(
|
|
user,
|
|
str(decision["review_owner"]),
|
|
):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="No puede gestionar esta decisión.",
|
|
)
|
|
|
|
try:
|
|
mark_in_consultation(
|
|
ConsultationInput(
|
|
submission_key=
|
|
submission_key,
|
|
decision_group_key=
|
|
decision_group_key,
|
|
reason=reason,
|
|
follow_up_with=
|
|
follow_up_with,
|
|
),
|
|
user,
|
|
)
|
|
except ReviewError as exc:
|
|
return RedirectResponse(
|
|
"/decisions/"
|
|
+ quote(
|
|
decision_group_key,
|
|
safe="",
|
|
)
|
|
+ "?error="
|
|
+ quote(str(exc)),
|
|
status_code=303,
|
|
)
|
|
|
|
return RedirectResponse(
|
|
"/?message="
|
|
+ quote(
|
|
"La decisión queda marcada En consulta."
|
|
),
|
|
status_code=303,
|
|
)
|
|
|
|
|
|
@app.post(
|
|
"/decisions/{decision_group_key:path}/resume"
|
|
)
|
|
def consultation_resume(
|
|
request: Request,
|
|
decision_group_key: str,
|
|
csrf_token: str = Form(...),
|
|
submission_key: uuid.UUID = Form(...),
|
|
):
|
|
user = require_user(request)
|
|
verify_csrf(user, csrf_token)
|
|
|
|
if not get_settings().write_enabled:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="La interfaz está en modo solo lectura.",
|
|
)
|
|
|
|
try:
|
|
resume_from_consultation(
|
|
submission_key,
|
|
decision_group_key,
|
|
user,
|
|
)
|
|
except ReviewError as exc:
|
|
return RedirectResponse(
|
|
"/decisions/"
|
|
+ quote(
|
|
decision_group_key,
|
|
safe="",
|
|
)
|
|
+ "?error="
|
|
+ quote(str(exc)),
|
|
status_code=303,
|
|
)
|
|
|
|
return RedirectResponse(
|
|
"/decisions/"
|
|
+ quote(
|
|
decision_group_key,
|
|
safe="",
|
|
),
|
|
status_code=303,
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/approvals",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def approvals_page(
|
|
request: Request,
|
|
message: str | None = None,
|
|
):
|
|
user = read_current_user(request)
|
|
if user is None:
|
|
return redirect_login()
|
|
if not can_approve_context(user):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="No tiene permisos de aprobación técnica.",
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="approvals.html",
|
|
context={
|
|
"settings": get_settings(),
|
|
"user": user,
|
|
"proposals": load_approval_queue(),
|
|
"message": message,
|
|
},
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/approvals/{proposal_id}",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def approval_detail(
|
|
request: Request,
|
|
proposal_id: int,
|
|
error: str | None = None,
|
|
):
|
|
user = read_current_user(request)
|
|
if user is None:
|
|
return redirect_login()
|
|
if not can_approve_context(user):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="No tiene permisos de aprobación técnica.",
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="approval_detail.html",
|
|
context={
|
|
"settings": get_settings(),
|
|
"user": user,
|
|
"proposal": load_proposal(
|
|
proposal_id
|
|
),
|
|
"write_enabled":
|
|
get_settings().write_enabled,
|
|
"submission_key":
|
|
str(uuid.uuid4()),
|
|
"error": error,
|
|
},
|
|
)
|
|
|
|
|
|
@app.post(
|
|
"/approvals/{proposal_id}/{action_code}"
|
|
)
|
|
def approval_action(
|
|
request: Request,
|
|
proposal_id: int,
|
|
action_code: str,
|
|
csrf_token: str = Form(...),
|
|
submission_key: uuid.UUID = Form(...),
|
|
notes: str = Form(""),
|
|
):
|
|
user = require_user(request)
|
|
verify_csrf(user, csrf_token)
|
|
|
|
if not get_settings().write_enabled:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="La interfaz está en modo solo lectura.",
|
|
)
|
|
if not can_approve_context(user):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="No tiene permisos de aprobación técnica.",
|
|
)
|
|
|
|
normalized = action_code.upper()
|
|
if normalized not in {
|
|
"APPROVE",
|
|
"REJECT",
|
|
"REQUEST_CLARIFICATION",
|
|
}:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Acción no permitida.",
|
|
)
|
|
|
|
action = ProposalActionInput(
|
|
submission_key=submission_key,
|
|
proposal_id=proposal_id,
|
|
action_code=normalized,
|
|
notes=notes,
|
|
)
|
|
|
|
try:
|
|
if normalized == "APPROVE":
|
|
result = approve_proposal(
|
|
action,
|
|
user,
|
|
(
|
|
request.client.host
|
|
if request.client
|
|
else None
|
|
),
|
|
request.headers.get(
|
|
"user-agent"
|
|
),
|
|
)
|
|
message = (
|
|
"Contexto aprobado y creado: "
|
|
f"{result['context_session_id']}."
|
|
)
|
|
else:
|
|
result = change_proposal_status(
|
|
action,
|
|
user,
|
|
(
|
|
request.client.host
|
|
if request.client
|
|
else None
|
|
),
|
|
request.headers.get(
|
|
"user-agent"
|
|
),
|
|
)
|
|
message = (
|
|
"Propuesta actualizada: "
|
|
+ label(
|
|
result[
|
|
"resulting_status"
|
|
]
|
|
)
|
|
+ "."
|
|
)
|
|
except ProposalError as exc:
|
|
return RedirectResponse(
|
|
f"/approvals/{proposal_id}?error="
|
|
+ quote(str(exc)),
|
|
status_code=303,
|
|
)
|
|
|
|
return RedirectResponse(
|
|
"/approvals?message="
|
|
+ quote(message),
|
|
status_code=303,
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/clarifications",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def clarifications_page(
|
|
request: Request,
|
|
message: str | None = None,
|
|
):
|
|
user = read_current_user(request)
|
|
if user is None:
|
|
return redirect_login()
|
|
if not can_manage_clarification(user):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="No tiene acceso a aclaraciones de Producción.",
|
|
)
|
|
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="clarifications.html",
|
|
context={
|
|
"settings": get_settings(),
|
|
"user": user,
|
|
"proposals": load_clarifications(),
|
|
"submission_keys": {
|
|
int(row["proposal_id"]):
|
|
str(uuid.uuid4())
|
|
for row in load_clarifications()
|
|
},
|
|
"message": message,
|
|
"write_enabled":
|
|
get_settings().write_enabled,
|
|
},
|
|
)
|
|
|
|
|
|
@app.post(
|
|
"/clarifications/{proposal_id}/respond"
|
|
)
|
|
def clarification_respond(
|
|
request: Request,
|
|
proposal_id: int,
|
|
csrf_token: str = Form(...),
|
|
submission_key: uuid.UUID = Form(...),
|
|
notes: str = Form(...),
|
|
):
|
|
user = require_user(request)
|
|
verify_csrf(user, csrf_token)
|
|
|
|
if not get_settings().write_enabled:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="La interfaz está en modo solo lectura.",
|
|
)
|
|
if not can_manage_clarification(user):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="No puede responder esta aclaración.",
|
|
)
|
|
|
|
try:
|
|
change_proposal_status(
|
|
ProposalActionInput(
|
|
submission_key=submission_key,
|
|
proposal_id=proposal_id,
|
|
action_code=
|
|
"CLARIFICATION_RESPONSE",
|
|
notes=notes,
|
|
),
|
|
user,
|
|
(
|
|
request.client.host
|
|
if request.client
|
|
else None
|
|
),
|
|
request.headers.get(
|
|
"user-agent"
|
|
),
|
|
)
|
|
except ProposalError as exc:
|
|
return RedirectResponse(
|
|
"/clarifications?message="
|
|
+ quote(str(exc)),
|
|
status_code=303,
|
|
)
|
|
|
|
return RedirectResponse(
|
|
"/clarifications?message="
|
|
+ quote(
|
|
"Aclaración enviada a revisión técnica."
|
|
),
|
|
status_code=303,
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/history",
|
|
response_class=HTMLResponse,
|
|
)
|
|
def history_page(request: Request):
|
|
user = read_current_user(request)
|
|
if user is None:
|
|
return redirect_login()
|
|
|
|
return templates.TemplateResponse(
|
|
request=request,
|
|
name="history.html",
|
|
context={
|
|
"settings": get_settings(),
|
|
"user": user,
|
|
"activities": recent_activity(),
|
|
},
|
|
)
|