Armor/KSP_project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: improve mission asset operations

Browse Source
This commit is contained in:
Armor
2026-05-28 11:32:11 +08:00
parent 1a8af63383
commit fee9b13ada
12 changed files with 562 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes/web.py
+315 -79
View File
@@ -161,8 +161,8 @@ MISSION_STATUS_GROUPS = [
"accent": "neutral",
"items": [
{
"name": "XH-03 Zhuque",
"asset_type": "Shuttle",
"name": "XH-03 Changxi",
"asset_type": "Exploration Mothership",
"location": "Star Port",
"mission": "Refit for Jupiter relay deployment",
"state_window": "2060-01-08 to 2060-04-12",
@@ -198,8 +198,8 @@ MISSION_LOCATION_GROUPS = [
"next_event": "Crew Arrival · 2060-03-30",
},
{
"name": "XH-03 Zhuque",
"asset_type": "Shuttle",
"name": "XH-03 Changxi",
"asset_type": "Exploration Mothership",
"state": "At Port",
"mission": "Refit for Jupiter relay deployment",
"location": "Star Port",
@@ -762,12 +762,83 @@ def _parse_datetime_form_value(
return parsed_value
class FormValidationError(ValueError):
def __init__(
self,
message: str,
*,
field_errors: dict[str, str] | None = None,
messages: list[str] | None = None,
) -> None:
super().__init__(message)
self.field_errors = field_errors or {}
self.messages = messages or [message]
def _build_validation_modal(
error: Exception | str,
*,
title: str = "本次提交未保存",
field_errors: dict[str, str] | None = None,
) -> dict[str, object]:
if isinstance(error, FormValidationError):
merged_field_errors = {**error.field_errors, **(field_errors or {})}
messages = list(error.messages)
else:
merged_field_errors = field_errors or {}
messages = [str(error)]
for message in merged_field_errors.values():
if message not in messages:
messages.append(message)
return {
"title": title,
"messages": messages,
"field_errors": merged_field_errors,
}
def _collect_submitted_form_values(
form: object,
field_specs: list[dict[str, object]],
prefix: str = "",
) -> dict[str, object]:
values: dict[str, object] = {}
for field_spec in field_specs:
field_name = field_spec["name"]
full_name = f"{prefix}{field_name}"
field_kind = field_spec["kind"]
if field_kind == "checkbox":
values[field_name] = full_name in form
continue
if field_kind == "combo":
selected_value = form.get(full_name)
custom_value = form.get(f"{full_name}__custom")
if _normalized_text(custom_value) is not None:
values[field_name] = custom_value or ""
elif selected_value == CUSTOM_OPTION_VALUE:
values[field_name] = ""
else:
values[field_name] = selected_value or ""
continue
values[field_name] = form.get(full_name) or ""
return values
def _apply_fields(
target: object,
form: object,
field_specs: list[dict[str, object]],
prefix: str = "",
) -> None:
parsed_values: dict[str, object] = {}
for field_spec in field_specs:
field_name = field_spec["name"]
full_name = f"{prefix}{field_name}"
@@ -775,27 +846,34 @@ def _apply_fields(
label = field_spec["label"]
required = bool(field_spec.get("required"))
if field_kind == "checkbox":
value = full_name in form
elif field_kind == "combo":
selected_value = _normalized_text(form.get(full_name))
custom_value = _normalized_text(form.get(f"{full_name}__custom"))
if custom_value is not None:
value = custom_value
elif selected_value == CUSTOM_OPTION_VALUE:
value = None
try:
if field_kind == "checkbox":
value = full_name in form
elif field_kind == "combo":
selected_value = _normalized_text(form.get(full_name))
custom_value = _normalized_text(form.get(f"{full_name}__custom"))
if custom_value is not None:
value = custom_value
elif selected_value == CUSTOM_OPTION_VALUE:
value = None
else:
value = selected_value
elif field_kind == "decimal":
value = _parse_decimal(form.get(full_name), label)
elif field_kind == "integer":
value = _parse_integer(form.get(full_name), label)
else:
value = selected_value
elif field_kind == "decimal":
value = _parse_decimal(form.get(full_name), label)
elif field_kind == "integer":
value = _parse_integer(form.get(full_name), label)
else:
value = _normalized_text(form.get(full_name))
value = _normalized_text(form.get(full_name))
except ValueError as exc:
raise FormValidationError(str(exc), field_errors={field_name: str(exc)}) from exc
if required and value in (None, ""):
raise ValueError(f"{label} 不能为空")
message = f"{label} 不能为空"
raise FormValidationError(message, field_errors={field_name: message})
parsed_values[field_name] = value
for field_name, value in parsed_values.items():
setattr(target, field_name, value)
@@ -822,21 +900,31 @@ def _apply_asset_entry_fields(target: AssetLogEntry, form: object) -> None:
entry_kind = _normalized_text(form.get("entry_kind")) or "event"
allowed_kinds = {option["value"] for option in ASSET_ENTRY_KIND_OPTIONS}
if entry_kind not in allowed_kinds:
raise ValueError("Entry Type 无效")
message = "Entry Type 无效"
raise FormValidationError(message, field_errors={"entry_kind": message})
title = _normalized_text(form.get("title"))
if title is None:
raise ValueError("Title 不能为空")
message = "Title 不能为空"
raise FormValidationError(message, field_errors={"title": message})
start_at = _parse_datetime_form_value(form, "start_at", "Start At", required=True)
try:
start_at = _parse_datetime_form_value(form, "start_at", "Start At", required=True)
except ValueError as exc:
raise FormValidationError(str(exc), field_errors={"start_at": str(exc)}) from exc
if start_at is None:
raise ValueError("Start At 不能为空")
message = "Start At 不能为空"
raise FormValidationError(message, field_errors={"start_at": message})
end_at = _parse_datetime_form_value(form, "end_at", "End At")
try:
end_at = _parse_datetime_form_value(form, "end_at", "End At")
except ValueError as exc:
raise FormValidationError(str(exc), field_errors={"end_at": str(exc)}) from exc
if entry_kind == "event":
end_at = None
elif end_at is not None and end_at <= start_at:
raise ValueError("End At 必须晚于 Start At")
message = "End At 必须晚于 Start At"
raise FormValidationError(message, field_errors={"start_at": message, "end_at": message})
state_node_rows = []
if entry_kind == "state":
@@ -883,6 +971,40 @@ def _asset_entry_form_values(entry: AssetLogEntry) -> dict[str, object]:
}
def _asset_entry_form_values_from_form(form: object) -> dict[str, object]:
node_ids = form.getlist("state_node_id")
node_titles = form.getlist("state_node_title")
node_details = form.getlist("state_node_detail")
node_times = form.getlist("state_node_at")
row_count = max(len(node_ids), len(node_titles), len(node_details), len(node_times))
state_nodes = []
for index in range(row_count):
state_nodes.append(
{
"id": node_ids[index] if index < len(node_ids) else "",
"title": node_titles[index] if index < len(node_titles) else "",
"detail": node_details[index] if index < len(node_details) else "",
"at": node_times[index] if index < len(node_times) else "",
}
)
return {
"entry_kind": form.get("entry_kind") or "event",
"title": form.get("title") or "",
"state_label": form.get("state_label") or "",
"mission_label": form.get("mission_label") or "",
"location": form.get("location") or "",
"start_at": form.get("start_at") or "",
"start_at_parts": {"year": "", "month": "", "day": "", "hour": "", "minute": ""},
"end_at": form.get("end_at") or "",
"end_at_parts": {"year": "", "month": "", "day": "", "hour": "", "minute": ""},
"summary": form.get("summary") or "",
"note": form.get("note") or "",
"state_nodes": state_nodes,
}
def _parse_state_node_rows(
form: object,
start_at: datetime,
@@ -901,7 +1023,8 @@ def _parse_state_node_rows(
or len(node_titles) != len(node_details)
or len(node_details) != len(node_times)
):
raise ValueError("State Node 数据不完整")
message = "State Node 数据不完整"
raise FormValidationError(message, field_errors={"state_nodes": message})
state_nodes = []
for index, (raw_id, raw_title, raw_detail, raw_at_time) in enumerate(
@@ -911,16 +1034,38 @@ def _parse_state_node_rows(
node_id = _normalized_text(raw_id)
title = _normalized_text(raw_title)
detail = _normalized_text(raw_detail)
at_time = _parse_datetime(raw_at_time, f"State Node {index} At")
row_index = index - 1
try:
at_time = _parse_datetime(raw_at_time, f"State Node {index} At")
except ValueError as exc:
raise FormValidationError(
str(exc),
field_errors={f"state_node_at:{row_index}": str(exc)},
) from exc
if title is None and detail is None and at_time is None:
continue
if title is None or at_time is None:
raise ValueError(f"State Node {index} 需要同时填写 Title 和 At")
message = f"State Node {index} 需要同时填写 Title 和 At"
field_errors: dict[str, str] = {}
if title is None:
field_errors[f"state_node_title:{row_index}"] = message
if at_time is None:
field_errors[f"state_node_at:{row_index}"] = message
raise FormValidationError(message, field_errors=field_errors)
if at_time < start_at:
raise ValueError(f"State Node {index} 时间不能早于 Start At")
message = f"State Node {index} 时间不能早于 Start At"
raise FormValidationError(
message,
field_errors={"start_at": message, f"state_node_at:{row_index}": message},
)
if end_at is not None and at_time > end_at:
raise ValueError(f"State Node {index} 时间不能晚于 End At")
message = f"State Node {index} 时间不能晚于 End At"
raise FormValidationError(
message,
field_errors={"end_at": message, f"state_node_at:{row_index}": message},
)
state_nodes.append(
{
@@ -943,7 +1088,8 @@ def _sync_state_nodes(target: AssetLogEntry, state_node_rows: list[dict[str, obj
if isinstance(raw_node_id, str) and raw_node_id:
node = existing_nodes.pop(raw_node_id, None)
if node is None:
raise ValueError("State Node 无效")
message = "State Node 无效"
raise FormValidationError(message, field_errors={"state_nodes": message})
else:
node = AssetStateNode()
@@ -952,6 +1098,7 @@ def _sync_state_nodes(target: AssetLogEntry, state_node_rows: list[dict[str, obj
node.at_time = row["at_time"]
next_nodes.append(node)
next_nodes.sort(key=lambda item: _as_utc(item.at_time) or _default_simulation_time())
target.state_nodes = next_nodes
@@ -1335,6 +1482,68 @@ def _entry_matches_date_range(
return start_at <= window_end and effective_end >= window_start
def _build_asset_future_events(
asset: Asset,
simulation_time: datetime,
active_state: AssetLogEntry | None,
) -> list[dict[str, object]]:
future_events: list[dict[str, object]] = []
if active_state is not None:
for state_node in sorted(
active_state.state_nodes,
key=lambda item: _as_utc(item.at_time) or simulation_time,
):
node_time = _as_utc(state_node.at_time)
if node_time is None or node_time <= simulation_time:
continue
future_events.append(
{
"sort_at": node_time,
"title": state_node.title,
"when": _format_datetime_display(state_node.at_time),
"scope": active_state.title or asset.name,
"summary": state_node.detail or "在所属 State Interval 中编辑该节点。",
"kind_label": "State Node",
"status_label": "Upcoming Node",
"state_labels": _split_multi_value_text(active_state.state_label),
"mission_labels": _split_multi_value_text(active_state.mission_label),
"source_entry_id": active_state.id,
"action_label": "编辑所属区间",
"edit_hint": "State Node 需要在所属 State Interval 中修改。",
}
)
for entry in sorted(
asset.log_entries,
key=lambda item: _as_utc(item.start_at) or _default_simulation_time(),
):
start_at = _as_utc(entry.start_at)
if start_at is None or start_at <= simulation_time:
continue
status = _classify_asset_entry(entry, simulation_time)
future_events.append(
{
"sort_at": start_at,
"title": entry.title,
"when": _format_datetime_display(entry.start_at),
"scope": entry.location or _display_multi_value_text(entry.mission_label) or asset.name,
"summary": entry.summary or "未填写摘要。",
"kind_label": "Event Point" if entry.entry_kind == "event" else "State Interval",
"status_label": status["label"],
"state_labels": _split_multi_value_text(entry.state_label),
"mission_labels": _split_multi_value_text(entry.mission_label),
"source_entry_id": entry.id,
"action_label": "编辑条目",
"edit_hint": None,
}
)
future_events.sort(key=lambda item: (item["sort_at"], str(item["title"])))
return future_events
def _build_asset_snapshot(asset: Asset, simulation_time: datetime) -> dict[str, object]:
entries = sorted(asset.log_entries, key=lambda item: _as_utc(item.start_at) or _default_simulation_time())
active_state = None
@@ -1350,14 +1559,7 @@ def _build_asset_snapshot(asset: Asset, simulation_time: datetime) -> dict[str,
last_entry = past_entries[-1] if past_entries else None
future_entries = [entry for entry in entries if (_as_utc(entry.start_at) or simulation_time) > simulation_time]
next_entry = future_entries[0] if future_entries else None
pending_state_nodes = [
node
for node in sorted(
active_state.state_nodes if active_state is not None else [],
key=lambda item: _as_utc(item.at_time) or simulation_time,
)
if (_as_utc(node.at_time) or simulation_time) > simulation_time
]
future_events = _build_asset_future_events(asset, simulation_time, active_state)
if active_state is not None:
current_state = _display_multi_value_text(active_state.state_label) or active_state.title
@@ -1389,27 +1591,8 @@ def _build_asset_snapshot(asset: Asset, simulation_time: datetime) -> dict[str,
current_event = active_state.title if active_state is not None else (last_entry.title if last_entry is not None else "No Logged Timeline")
next_event_label = "No future event recorded"
upcoming_events: list[dict[str, str]] = []
for state_node in pending_state_nodes:
upcoming_events.append(
{
"when": _format_datetime_display(state_node.at_time),
"title": state_node.title,
"scope": active_state.title if active_state is not None else asset.name,
}
)
for entry in future_entries:
upcoming_events.append(
{
"when": _format_datetime_display(entry.start_at),
"title": entry.title,
"scope": entry.location or _display_multi_value_text(entry.mission_label) or asset.name,
}
)
if upcoming_events:
next_event_label = f"{upcoming_events[0]['title']} · {upcoming_events[0]['when']}"
if future_events:
next_event_label = f"{future_events[0]['title']} · {future_events[0]['when']}"
return {
"current_state": current_state,
@@ -1421,7 +1604,8 @@ def _build_asset_snapshot(asset: Asset, simulation_time: datetime) -> dict[str,
"next_entry": next_entry,
"active_state": active_state,
"last_entry": last_entry,
"upcoming_events": upcoming_events[:4],
"future_events": future_events,
"upcoming_events": future_events[:4],
}
@@ -3257,18 +3441,28 @@ def asset_new() -> str | object:
form_options = _asset_form_options()
simulation_time = _resolve_simulation_time(request.form.get("sim_time") if request.method == "POST" else request.args.get("sim_time"))
simulation_time_input = _format_datetime_input(simulation_time)
values = _collect_form_values(asset, ASSET_FIELDS)
validation_modal: dict[str, object] | None = None
field_errors: dict[str, str] = {}
if request.method == "POST":
try:
_apply_fields(asset, request.form, ASSET_FIELDS)
db.session.add(asset)
db.session.commit()
except ValueError as exc:
except FormValidationError as exc:
db.session.rollback()
flash(str(exc), "error")
validation_modal = _build_validation_modal(exc)
field_errors = dict(validation_modal["field_errors"])
values = _collect_submitted_form_values(request.form, ASSET_FIELDS)
except IntegrityError:
db.session.rollback()
flash("创建失败：Asset 名称必须唯一。", "error")
validation_modal = _build_validation_modal(
"创建失败：Asset 名称必须唯一。",
field_errors={"name": "Asset 名称必须唯一。"},
)
field_errors = dict(validation_modal["field_errors"])
values = _collect_submitted_form_values(request.form, ASSET_FIELDS)
else:
flash("资产已创建。", "success")
return redirect(url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input))
@@ -3280,9 +3474,11 @@ def asset_new() -> str | object:
heading="新增任务资产",
description=None,
fields=ASSET_FIELDS,
values=_collect_form_values(asset, ASSET_FIELDS),
values=values,
form_options=form_options,
custom_option_value=CUSTOM_OPTION_VALUE,
validation_modal=validation_modal,
field_errors=field_errors,
cancel_url=url_for("web.asset_list", sim_time=simulation_time_input),
submit_label="创建",
delete_url=None,
@@ -3299,17 +3495,27 @@ def asset_edit(asset_id: object) -> str | object:
form_options = _asset_form_options()
simulation_time = _resolve_simulation_time(request.form.get("sim_time") if request.method == "POST" else request.args.get("sim_time"))
simulation_time_input = _format_datetime_input(simulation_time)
values = _collect_form_values(asset, ASSET_FIELDS)
validation_modal: dict[str, object] | None = None
field_errors: dict[str, str] = {}
if request.method == "POST":
try:
_apply_fields(asset, request.form, ASSET_FIELDS)
db.session.commit()
except ValueError as exc:
except FormValidationError as exc:
db.session.rollback()
flash(str(exc), "error")
validation_modal = _build_validation_modal(exc)
field_errors = dict(validation_modal["field_errors"])
values = _collect_submitted_form_values(request.form, ASSET_FIELDS)
except IntegrityError:
db.session.rollback()
flash("保存失败：Asset 名称必须唯一。", "error")
validation_modal = _build_validation_modal(
"保存失败：Asset 名称必须唯一。",
field_errors={"name": "Asset 名称必须唯一。"},
)
field_errors = dict(validation_modal["field_errors"])
values = _collect_submitted_form_values(request.form, ASSET_FIELDS)
else:
flash("资产信息已保存。", "success")
return redirect(url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input))
@@ -3321,9 +3527,11 @@ def asset_edit(asset_id: object) -> str | object:
heading=f"编辑 {asset.name}",
description=None,
fields=ASSET_FIELDS,
values=_collect_form_values(asset, ASSET_FIELDS),
values=values,
form_options=form_options,
custom_option_value=CUSTOM_OPTION_VALUE,
validation_modal=validation_modal,
field_errors=field_errors,
cancel_url=url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input),
submit_label="保存",
delete_url=url_for("web.asset_delete", asset_id=asset.id),
@@ -3381,6 +3589,19 @@ def asset_detail(asset_id: object) -> str:
simulation_time = _resolve_simulation_time(request.args.get("sim_time"))
requested_page = request.args.get("page", default=1, type=int) or 1
snapshot = _build_asset_snapshot(asset, simulation_time)
simulation_time_input = _format_datetime_input(simulation_time)
future_events = [
{
**item,
"edit_url": url_for(
"web.asset_entry_edit",
asset_id=asset.id,
entry_id=item["source_entry_id"],
sim_time=simulation_time_input,
),
}
for item in snapshot["future_events"]
]
try:
entry_range_start = _parse_datetime(request.args.get("start"), "日志开始时间")
@@ -3447,8 +3668,9 @@ def asset_detail(asset_id: object) -> str:
preview_eyebrow="Mission Asset",
asset=asset,
asset_summary=snapshot,
future_events=future_events,
entries=paged_entries,
upcoming_events=snapshot["upcoming_events"],
upcoming_events=future_events[:4],
total_entry_count=len(asset.log_entries),
filtered_entry_count=len(entry_rows),
metrics=[
@@ -3468,7 +3690,7 @@ def asset_detail(asset_id: object) -> str:
preview_tab="asset",
board_nav_urls=_mission_ops_nav_urls(simulation_time),
simulation_time_display=_format_datetime_display(simulation_time),
simulation_time_input=_format_datetime_input(simulation_time),
simulation_time_input=simulation_time_input,
header_time_action=url_for("web.asset_detail", asset_id=asset.id),
header_time_field_id="asset-detail-header-sim-time",
header_time_submit_label="Apply Time",
@@ -3492,6 +3714,9 @@ def asset_entry_new(asset_id: object) -> str | object:
requested_kind = "event"
entry = AssetLogEntry(entry_kind=requested_kind)
form_values = _asset_entry_form_values(entry)
validation_modal: dict[str, object] | None = None
field_errors: dict[str, str] = {}
if request.method == "POST":
try:
@@ -3499,9 +3724,11 @@ def asset_entry_new(asset_id: object) -> str | object:
entry.asset = asset
db.session.add(entry)
db.session.commit()
except ValueError as exc:
except FormValidationError as exc:
db.session.rollback()
flash(str(exc), "error")
validation_modal = _build_validation_modal(exc)
field_errors = dict(validation_modal["field_errors"])
form_values = _asset_entry_form_values_from_form(request.form)
else:
flash("日志条目已新增。", "success")
return redirect(url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input))
@@ -3513,8 +3740,10 @@ def asset_entry_new(asset_id: object) -> str | object:
heading=f"为 {asset.name} 新增日志条目",
description=None,
submit_label="创建条目",
form_values=_asset_entry_form_values(entry),
form_values=form_values,
kind_options=ASSET_ENTRY_KIND_OPTIONS,
validation_modal=validation_modal,
field_errors=field_errors,
cancel_url=url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input),
delete_url=None,
simulation_time_input=simulation_time_input,
@@ -3529,14 +3758,19 @@ def asset_entry_edit(asset_id: object, entry_id: object) -> str | object:
entry = _load_asset_entry_or_404(asset.id, entry_id)
simulation_time = _resolve_simulation_time(request.form.get("sim_time") if request.method == "POST" else request.args.get("sim_time"))
simulation_time_input = _format_datetime_input(simulation_time)
form_values = _asset_entry_form_values(entry)
validation_modal: dict[str, object] | None = None
field_errors: dict[str, str] = {}
if request.method == "POST":
try:
_apply_asset_entry_fields(entry, request.form)
db.session.commit()
except ValueError as exc:
except FormValidationError as exc:
db.session.rollback()
flash(str(exc), "error")
validation_modal = _build_validation_modal(exc)
field_errors = dict(validation_modal["field_errors"])
form_values = _asset_entry_form_values_from_form(request.form)
else:
flash("日志条目已保存。", "success")
return redirect(url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input))
@@ -3548,8 +3782,10 @@ def asset_entry_edit(asset_id: object, entry_id: object) -> str | object:
heading=f"编辑 {asset.name} 的日志条目",
description=None,
submit_label="保存条目",
form_values=_asset_entry_form_values(entry),
form_values=form_values,
kind_options=ASSET_ENTRY_KIND_OPTIONS,
validation_modal=validation_modal,
field_errors=field_errors,
cancel_url=url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input),
delete_url=url_for("web.asset_entry_delete", asset_id=asset.id, entry_id=entry.id),
simulation_time_input=simulation_time_input,