--- /home/ray/ARMOR/apps/shire-mobile/limbforge-workspace/server.py 2026-08-06 16:33:50.696347212 +0800 +++ /tmp/shire-limbforge-edit-api.FUZG8C/server.py 2026-08-06 17:11:50.865731647 +0800 @@ -409,22 +409,27 @@ return load_project(project_id) -def save_measurement( - project_id: str, - measurement_key: str, - payload: dict, -) -> dict: - project = load_project(project_id) +# SHIRE-LIMBFORGE-EDIT-ANY-API-0001 + +EDIT_AUDIT_ACTOR = "ray" - if project is None: - raise LookupError("Project not found.") - definition = MEASUREMENT_BY_KEY.get(measurement_key) +def validate_measurement_payload( + project: dict, + measurement_key: str, + payload: dict, +) -> tuple[float, str, str]: + definition = MEASUREMENT_BY_KEY.get( + measurement_key + ) if definition is None: raise ValueError("Unknown measurement.") - if definition["oi_only"] and not project["osseointegration"]: + if ( + definition["oi_only"] + and not project["osseointegration"] + ): raise ValueError( "This measurement is available only for " "osseointegration projects." @@ -433,23 +438,467 @@ try: value = float(payload.get("value_mm")) except (TypeError, ValueError): - raise ValueError("Enter a valid measurement in millimetres.") + raise ValueError( + "Enter a valid measurement in millimetres." + ) if value <= 0 or value > 2000: raise ValueError( - "Measurement must be greater than zero and below 2000 mm." + "Measurement must be greater than zero " + "and below 2000 mm." + ) + + status = str( + payload.get("status", "completed") + ).strip() + + if status not in { + "completed", + "specialist_review", + }: + raise ValueError( + "Invalid measurement status." + ) + + notes = str( + payload.get("notes", "") + ).strip() + + return value, notes, status + + +def require_change_reason(payload: dict) -> str: + reason = str( + payload.get("change_reason", "") + ).strip() + + if not reason: + raise ValueError( + "Enter a reason for changing this measurement." + ) + + if len(reason) > 1000: + raise ValueError( + "The change reason must not exceed " + "1000 characters." + ) + + return reason + + +def insert_snapshot_supersessions( + database: sqlite3.Connection, + project_id: str, + revision_id: str, + reason: str, + timestamp: str, + session_id: str | None = None, +) -> int: + if session_id is None: + snapshots = database.execute( + """ + SELECT id + FROM design_forge_measurement_snapshots + WHERE project_id = ? + """, + (project_id,), + ).fetchall() + else: + snapshots = database.execute( + """ + SELECT id + FROM design_forge_measurement_snapshots + WHERE project_id = ? + AND ( + session_id = ? + OR session_id IS NULL + ) + """, + ( + project_id, + session_id, + ), + ).fetchall() + + inserted = 0 + + for snapshot in snapshots: + database.execute( + """ + INSERT INTO + design_forge_snapshot_supersessions ( + id, + snapshot_id, + project_id, + revision_id, + reason, + created_at + ) + VALUES (?, ?, ?, ?, ?, ?) + """, + ( + uuid.uuid4().hex, + snapshot["id"], + project_id, + revision_id, + reason, + timestamp, + ), ) - status = str(payload.get("status", "completed")) + inserted += 1 + + return inserted + + +def revision_to_dict(row: sqlite3.Row) -> dict: + revision = dict(row) + + definition = MEASUREMENT_BY_KEY.get( + revision["measurement_key"] + ) + + revision["measurement_name"] = ( + definition["name"] + if definition + else revision["measurement_key"] + ) + + revision["unit"] = "mm" + + return revision + + +def list_project_revisions( + project_id: str, +) -> list[dict]: + if load_project(project_id) is None: + raise LookupError("Project not found.") + + with DATABASE_LOCK, connect() as database: + rows = database.execute( + """ + SELECT * + FROM measurement_revisions + WHERE project_id = ? + ORDER BY changed_at DESC, id DESC + """, + (project_id,), + ).fetchall() + + return [ + revision_to_dict(row) + for row in rows + ] - if status not in {"completed", "specialist_review"}: - raise ValueError("Invalid measurement status.") +def list_session_revisions( + project_id: str, + session_id: str, +) -> list[dict]: + session = load_measurement_session( + project_id, + session_id, + ) + + if session is None: + raise LookupError( + "Measurement session not found." + ) + + with DATABASE_LOCK, connect() as database: + rows = database.execute( + """ + SELECT * + FROM measurement_revisions + WHERE project_id = ? + AND session_id = ? + ORDER BY changed_at DESC, id DESC + """, + ( + project_id, + session_id, + ), + ).fetchall() + + return [ + revision_to_dict(row) + for row in rows + ] + + +def session_to_dict( + row: sqlite3.Row, + measurements: list[sqlite3.Row] | None = None, +) -> dict: + session = dict(row) + + if measurements is not None: + session["measurements"] = { + item["measurement_key"]: { + "value_mm": item["value_mm"], + "notes": item["notes"], + "status": item["status"], + "updated_at": item["updated_at"], + } + for item in measurements + } + + return session + + +def list_measurement_sessions( + project_id: str, +) -> list[dict]: + if load_project(project_id) is None: + raise LookupError("Project not found.") + + with DATABASE_LOCK, connect() as database: + rows = database.execute( + """ + SELECT + measurement_sessions.*, + ( + SELECT COUNT(*) + FROM measurement_session_records + WHERE + measurement_session_records.session_id + = measurement_sessions.id + ) AS measurement_count + FROM measurement_sessions + WHERE project_id = ? + ORDER BY created_at DESC, id DESC + """, + (project_id,), + ).fetchall() + + return [ + dict(row) + for row in rows + ] + + +def load_measurement_session( + project_id: str, + session_id: str, +) -> dict | None: + with DATABASE_LOCK, connect() as database: + session = database.execute( + """ + SELECT * + FROM measurement_sessions + WHERE id = ? + AND project_id = ? + """, + ( + session_id, + project_id, + ), + ).fetchone() + + if session is None: + return None + + measurements = database.execute( + """ + SELECT * + FROM measurement_session_records + WHERE session_id = ? + AND project_id = ? + ORDER BY updated_at, measurement_key + """, + ( + session_id, + project_id, + ), + ).fetchall() + + return session_to_dict( + session, + measurements, + ) + + +def create_measurement_session( + project_id: str, + payload: dict, +) -> dict: + if load_project(project_id) is None: + raise LookupError("Project not found.") + + status = str( + payload.get("status", "active") + ).strip() + + if status not in { + "active", + "completed", + "cancelled", + }: + raise ValueError( + "Invalid measurement-session status." + ) + + guide_version = str( + payload.get( + "guide_version", + "limbforge-anatomy-guides-v1", + ) + ).strip() + + if not guide_version: + raise ValueError( + "Measurement guide version is required." + ) + + if len(guide_version) > 200: + raise ValueError( + "Measurement guide version is too long." + ) + + session_id = uuid.uuid4().hex timestamp = utc_now() + completed_at = ( + timestamp + if status == "completed" + else None + ) + with DATABASE_LOCK, connect() as database: database.execute( """ + INSERT INTO measurement_sessions ( + id, + project_id, + status, + guide_version, + started_at, + completed_at, + created_at, + updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + session_id, + project_id, + status, + guide_version, + timestamp, + completed_at, + timestamp, + timestamp, + ), + ) + + session = load_measurement_session( + project_id, + session_id, + ) + + if session is None: + raise RuntimeError( + "Measurement session could not be loaded." + ) + + return session + + +def save_measurement( + project_id: str, + measurement_key: str, + payload: dict, +) -> dict: + project = load_project(project_id) + + if project is None: + raise LookupError("Project not found.") + + value, notes, status = ( + validate_measurement_payload( + project, + measurement_key, + payload, + ) + ) + + timestamp = utc_now() + + with DATABASE_LOCK, connect() as database: + existing = database.execute( + """ + SELECT * + FROM measurements + WHERE project_id = ? + AND measurement_key = ? + """, + ( + project_id, + measurement_key, + ), + ).fetchone() + + changed = ( + existing is not None + and ( + float(existing["value_mm"]) != value + or existing["notes"] != notes + or existing["status"] != status + ) + ) + + if changed: + reason = require_change_reason(payload) + revision_id = uuid.uuid4().hex + + database.execute( + """ + INSERT INTO measurement_revisions ( + id, + project_id, + session_id, + measurement_key, + previous_value_mm, + new_value_mm, + previous_notes, + new_notes, + previous_status, + new_status, + change_reason, + changed_by, + changed_at + ) + VALUES ( + ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ) + """, + ( + revision_id, + project_id, + measurement_key, + existing["value_mm"], + value, + existing["notes"], + notes, + existing["status"], + status, + reason, + EDIT_AUDIT_ACTOR, + timestamp, + ), + ) + + insert_snapshot_supersessions( + database, + project_id, + revision_id, + reason, + timestamp, + ) + + database.execute( + """ INSERT INTO measurements ( project_id, measurement_key, @@ -470,7 +919,7 @@ project_id, measurement_key, value, - str(payload.get("notes", "")).strip(), + notes, status, timestamp, ), @@ -482,12 +931,197 @@ SET updated_at = ? WHERE id = ? """, - (timestamp, project_id), + ( + timestamp, + project_id, + ), ) return load_project(project_id) +def save_session_measurement( + project_id: str, + session_id: str, + measurement_key: str, + payload: dict, +) -> dict: + project = load_project(project_id) + + if project is None: + raise LookupError("Project not found.") + + session = load_measurement_session( + project_id, + session_id, + ) + + if session is None: + raise LookupError( + "Measurement session not found." + ) + + if session["status"] == "cancelled": + raise ValueError( + "Cancelled measurement sessions " + "cannot be edited." + ) + + value, notes, status = ( + validate_measurement_payload( + project, + measurement_key, + payload, + ) + ) + + timestamp = utc_now() + + with DATABASE_LOCK, connect() as database: + existing = database.execute( + """ + SELECT * + FROM measurement_session_records + WHERE session_id = ? + AND project_id = ? + AND measurement_key = ? + """, + ( + session_id, + project_id, + measurement_key, + ), + ).fetchone() + + changed = ( + existing is not None + and ( + float(existing["value_mm"]) != value + or existing["notes"] != notes + or existing["status"] != status + ) + ) + + if changed: + reason = require_change_reason(payload) + revision_id = uuid.uuid4().hex + + database.execute( + """ + INSERT INTO measurement_revisions ( + id, + project_id, + session_id, + measurement_key, + previous_value_mm, + new_value_mm, + previous_notes, + new_notes, + previous_status, + new_status, + change_reason, + changed_by, + changed_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + revision_id, + project_id, + session_id, + measurement_key, + existing["value_mm"], + value, + existing["notes"], + notes, + existing["status"], + status, + reason, + EDIT_AUDIT_ACTOR, + timestamp, + ), + ) + + insert_snapshot_supersessions( + database, + project_id, + revision_id, + reason, + timestamp, + session_id, + ) + + database.execute( + """ + INSERT INTO measurement_session_records ( + session_id, + project_id, + measurement_key, + value_mm, + notes, + status, + updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(session_id, measurement_key) + DO UPDATE SET + value_mm = excluded.value_mm, + notes = excluded.notes, + status = excluded.status, + updated_at = excluded.updated_at + """, + ( + session_id, + project_id, + measurement_key, + value, + notes, + status, + timestamp, + ), + ) + + database.execute( + """ + UPDATE measurement_sessions + SET updated_at = ? + WHERE id = ? + AND project_id = ? + """, + ( + timestamp, + session_id, + project_id, + ), + ) + + database.execute( + """ + UPDATE projects + SET updated_at = ? + WHERE id = ? + """, + ( + timestamp, + project_id, + ), + ) + + updated = load_measurement_session( + project_id, + session_id, + ) + + if updated is None: + raise RuntimeError( + "Updated measurement session " + "could not be loaded." + ) + + return updated + + + def delete_measurement( project_id: str, measurement_key: str, @@ -595,6 +1229,7 @@ "status": "ok", "service": "limbforge-shared-workspace", "database": str(DATABASE), + "edit_any_measurement_api": True, } ) return @@ -614,6 +1249,66 @@ ] if ( + len(parts) == 6 + and parts[0] == "api" + and parts[1] == "projects" + and parts[3] == "sessions" + and parts[5] == "revisions" + ): + self.send_json( + list_session_revisions( + parts[2], + parts[4], + ) + ) + return + + if ( + len(parts) == 5 + and parts[0] == "api" + and parts[1] == "projects" + and parts[3] == "sessions" + ): + session = load_measurement_session( + parts[2], + parts[4], + ) + + if session is None: + raise LookupError( + "Measurement session not found." + ) + + self.send_json(session) + return + + if ( + len(parts) == 4 + and parts[0] == "api" + and parts[1] == "projects" + and parts[3] == "sessions" + ): + self.send_json( + list_measurement_sessions( + parts[2] + ) + ) + return + + if ( + len(parts) == 4 + and parts[0] == "api" + and parts[1] == "projects" + and parts[3] == "revisions" + ): + self.send_json( + list_project_revisions( + parts[2] + ) + ) + return + + if ( len(parts) == 3 and parts[0] == "api" and parts[1] == "projects" @@ -621,7 +1316,9 @@ project = load_project(parts[2]) if project is None: - raise LookupError("Project not found.") + raise LookupError( + "Project not found." + ) self.send_json(project) return @@ -631,9 +1328,16 @@ except Exception as error: self.handle_api_error(error) + def do_POST(self) -> None: path = urlparse(self.path).path + parts = [ + unquote(part) + for part in path.strip("/").split("/") + if part + ] + try: if path == "/api/projects": self.send_json( @@ -642,6 +1346,21 @@ ) return + if ( + len(parts) == 4 + and parts[0] == "api" + and parts[1] == "projects" + and parts[3] == "sessions" + ): + self.send_json( + create_measurement_session( + parts[2], + self.read_json(), + ), + HTTPStatus.CREATED, + ) + return + self.send_json( {"error": "Route not found."}, HTTPStatus.NOT_FOUND, @@ -650,8 +1369,10 @@ except Exception as error: self.handle_api_error(error) + def do_PUT(self) -> None: path = urlparse(self.path).path + parts = [ unquote(part) for part in path.strip("/").split("/") @@ -687,6 +1408,23 @@ ) return + if ( + len(parts) == 7 + and parts[0] == "api" + and parts[1] == "projects" + and parts[3] == "sessions" + and parts[5] == "measurements" + ): + self.send_json( + save_session_measurement( + parts[2], + parts[4], + parts[6], + self.read_json(), + ) + ) + return + self.send_json( {"error": "Route not found."}, HTTPStatus.NOT_FOUND, @@ -695,6 +1433,7 @@ except Exception as error: self.handle_api_error(error) + def do_DELETE(self) -> None: path = urlparse(self.path).path parts = [