573 lines
37 KiB
Python
573 lines
37 KiB
Python
"""Peripheral persistence and transactional work-order change signals."""
|
|
from contextlib import contextmanager
|
|
from copy import deepcopy
|
|
from datetime import datetime, timedelta, timezone
|
|
from math import isfinite
|
|
from typing import Any, Iterator
|
|
from uuid import UUID, uuid4
|
|
|
|
import psycopg
|
|
from psycopg.rows import dict_row
|
|
from psycopg.conninfo import conninfo_to_dict
|
|
from psycopg.types.json import Jsonb
|
|
|
|
from app.domain.schemas.shared import TimeRange
|
|
from app.services.content_digest import digest
|
|
from app.infra.db.postgresql.workbench_events import lock_event_transaction_sync
|
|
from app.infra.db.project_routing import ActiveProjectRouting
|
|
|
|
from .errors import SampleFailure, ensure
|
|
from .auth import Identity
|
|
from .config import SampleSettings
|
|
from .schemas import (
|
|
AddFeedback, CreateTask, CreateTestDataRecord, CreateWorkOrder, DeleteResource,
|
|
DemoContent, ScenarioCommand, Task, TestDataRecord, Transition,
|
|
UpdateTestDataRecord, UpdateWorkOrder, WorkOrder,
|
|
)
|
|
|
|
TRANSITIONS = {"pending": ["processing"], "processing": ["pending_review"],
|
|
"pending_review": ["processing", "completed"], "completed": []}
|
|
WORK_ORDER_FIELD_LABELS = {
|
|
"title": "标题", "summary": "说明", "status": "状态", "priority": "优先级",
|
|
"assignee": "负责人", "scheduled_at": "预约时间", "location": "位置",
|
|
"scenario_code": "场景编号", "feedback_requirements": "回执要求",
|
|
}
|
|
WORK_ORDER_CREATED_DETAILS = {
|
|
"general": "预约工单已创建。",
|
|
"water_quality_complaint": "投诉工单已创建。",
|
|
"burst_report": "爆管上报工单已创建。",
|
|
}
|
|
|
|
|
|
class PostgresStore:
|
|
def __init__(self, settings: SampleSettings, *, routing: ActiveProjectRouting | None = None) -> None:
|
|
self.routing = routing
|
|
# Explicit seed/export scripts retain their configured database. HTTP
|
|
# dependencies always supply authenticated metadata routing.
|
|
self.settings = settings if routing is None else settings.model_copy(update={
|
|
"db_name": routing.business_database_name, "project_code": routing.project_code,
|
|
})
|
|
|
|
@contextmanager
|
|
def connection(self, *, readonly: bool = False, timeseries: bool = False) -> Iterator[psycopg.Connection]:
|
|
from app.core.config import settings as product
|
|
ensure(not timeseries or readonly, "READ_ONLY_SOURCE", "历史监测源仅允许只读连接", 403)
|
|
prefix = "TIMESCALEDB_DB_" if timeseries else "DB_"
|
|
if self.routing is not None:
|
|
dsn = self.routing.timescale_dsn if timeseries else self.routing.business_dsn
|
|
ensure(bool(dsn), "PROJECT_STORAGE_NOT_CONFIGURED", "项目数据库路由尚未配置", 503)
|
|
expected_database = conninfo_to_dict(dsn).get("dbname")
|
|
ensure(bool(expected_database), "PROJECT_STORAGE_NOT_CONFIGURED", "项目数据库名称尚未配置", 503)
|
|
connect_args = {"conninfo": dsn}
|
|
else:
|
|
expected_database = self.settings.db_name
|
|
connect_args = {"host": getattr(product, prefix + "HOST"), "port": getattr(product, prefix + "PORT"),
|
|
"user": getattr(product, prefix + "USER"), "password": getattr(product, prefix + "PASSWORD"),
|
|
"dbname": expected_database}
|
|
try:
|
|
with psycopg.connect(**connect_args, connect_timeout=5,
|
|
options="-c statement_timeout=5000", row_factory=dict_row) as conn:
|
|
if readonly:
|
|
conn.execute("SET TRANSACTION READ ONLY")
|
|
actual = conn.execute("SELECT current_database() AS name").fetchone()["name"]
|
|
ensure(actual == expected_database, "DATABASE_MISMATCH", "样板数据库不匹配", 503)
|
|
yield conn
|
|
except (psycopg.errors.UndefinedTable, psycopg.errors.InvalidSchemaName) as exc:
|
|
raise SampleFailure("STORAGE_NOT_CONFIGURED", "样板存储表尚未配置", 503) from exc
|
|
except psycopg.Error as exc:
|
|
# Do not leak connection strings, SQL, passwords or business values.
|
|
raise SampleFailure("STORAGE_UNAVAILABLE", "样板数据库不可用，请查询请求回执再重试", 503, effect="unknown" if not readonly else "none") from exc
|
|
|
|
def _scenario(self, conn, who):
|
|
row = conn.execute("SELECT revision,scenario FROM peripheral_sample.scenarios WHERE project_id=%s",
|
|
(who.project_id,)).fetchone()
|
|
state = dict(row) if row else {"revision": 1, "scenario": "normal"}
|
|
if not self.settings.enable_faults:
|
|
state["scenario"] = "normal"
|
|
return state
|
|
|
|
def managed_resource_capabilities(self) -> dict[str, str]:
|
|
"""Report optional test-resource storage without touching product data."""
|
|
with self.connection(readonly=True) as conn:
|
|
row = conn.execute("""SELECT
|
|
to_regclass('peripheral_sample.requests') IS NOT NULL AS requests,
|
|
to_regclass('peripheral_sample.scenarios') IS NOT NULL AS scenarios,
|
|
to_regclass('peripheral_sample.work_orders') IS NOT NULL AS work_orders,
|
|
to_regclass('peripheral_sample.work_order_revisions') IS NOT NULL AS work_order_revisions,
|
|
to_regclass('peripheral_sample.test_data_records') IS NOT NULL AS test_data_records,
|
|
to_regclass('peripheral_sample.test_data_record_revisions') IS NOT NULL AS test_data_record_revisions
|
|
""").fetchone()
|
|
common = row["requests"] and row["scenarios"]
|
|
return {
|
|
"work_orders": "available" if common and row["work_orders"] and row["work_order_revisions"]
|
|
else "not_configured",
|
|
"test_data": "available" if common and row["test_data_records"] and row["test_data_record_revisions"]
|
|
else "not_configured",
|
|
}
|
|
|
|
def scenario(self, who: Identity) -> dict[str, Any]:
|
|
with self.connection(readonly=True) as conn:
|
|
return self._scenario(conn, who)
|
|
|
|
def _available(self, conn, who):
|
|
scenario = self._scenario(conn, who)["scenario"]
|
|
ensure(scenario != "unavailable", "SCENARIO_UNAVAILABLE", "故障演练：外围服务不可用", 503)
|
|
return scenario
|
|
|
|
def _cached(self, conn, who, key, command):
|
|
payload = command.model_dump(mode="json", exclude={"request_id"})
|
|
request_digest = digest({"command": key, "payload": payload})
|
|
row = conn.execute("""SELECT command_key,request_digest,result FROM peripheral_sample.requests
|
|
WHERE project_id=%s AND actor_id=%s AND request_id=%s""",
|
|
(who.project_id, who.user_id, command.request_id)).fetchone()
|
|
if row:
|
|
ensure(row["command_key"] == key and row["request_digest"] == request_digest,
|
|
"IDEMPOTENCY_CONFLICT", "同一 request_id 不能用于不同命令或载荷", 409)
|
|
return row["result"], request_digest
|
|
return None, request_digest
|
|
|
|
def replay(self, who: Identity, key: str, command: CreateTask) -> dict[str, Any] | None:
|
|
with self.connection(readonly=True) as conn:
|
|
return self._cached(conn, who, key, command)[0]
|
|
|
|
def _mutate(self, who, key, command, operation, *, control=False):
|
|
ensure(who.can_write, "PERMISSION_DENIED", "没有样板修改权限", 403)
|
|
lost = False
|
|
try:
|
|
with self.connection() as conn:
|
|
if "work_order" in key:
|
|
lock_event_transaction_sync(conn)
|
|
conn.execute("SELECT pg_advisory_xact_lock(hashtextextended(%s,0))",
|
|
("peripheral-sample:" + who.project_id,))
|
|
cached, request_digest = self._cached(conn, who, key, command)
|
|
if cached is not None:
|
|
return cached
|
|
scenario = "normal" if control else self._available(conn, who)
|
|
result = operation(conn, scenario)
|
|
task_id = result.get("task_id")
|
|
conn.execute("""INSERT INTO peripheral_sample.requests
|
|
(project_id,actor_id,request_id,command_key,request_digest,task_id,result)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s)""",
|
|
(who.project_id, who.user_id, command.request_id, key, request_digest, task_id, Jsonb(result)))
|
|
lost = scenario == "response_lost" and key == "create_task"
|
|
except SampleFailure as exc:
|
|
exc.request_id = str(command.request_id)
|
|
raise
|
|
if lost:
|
|
raise SampleFailure("RESPONSE_LOST", "故障演练：写入已提交但响应未知，请按 request_id 查询回执",
|
|
503, effect="unknown", request_id=command.request_id)
|
|
return result
|
|
|
|
def set_scenario(self, who: Identity, command: ScenarioCommand) -> dict[str, Any]:
|
|
ensure(command.scenario == "normal" or self.settings.enable_faults,
|
|
"FAULTS_DISABLED", "故障注入未由服务端配置启用", 403)
|
|
def write(conn, _):
|
|
state = {"revision": self._scenario(conn, who)["revision"] + 1, "scenario": command.scenario}
|
|
conn.execute("""INSERT INTO peripheral_sample.scenarios(project_id,revision,scenario) VALUES (%s,%s,%s)
|
|
ON CONFLICT(project_id) DO UPDATE SET revision=excluded.revision,scenario=excluded.scenario""",
|
|
(who.project_id, state["revision"], state["scenario"]))
|
|
return state
|
|
return self._mutate(who, "scenario", command, write, control=True)
|
|
|
|
def _task(self, conn, who, task_id):
|
|
row = conn.execute("SELECT document FROM peripheral_sample.tasks WHERE project_id=%s AND task_id=%s",
|
|
(who.project_id, task_id)).fetchone()
|
|
ensure(row is not None, "TASK_NOT_FOUND", "样板核查任务不存在", 404)
|
|
return row["document"]
|
|
|
|
def _save(self, conn, who, document):
|
|
document = Task.model_validate(document).model_dump(mode="json")
|
|
conn.execute("""INSERT INTO peripheral_sample.tasks(project_id,task_id,revision,source_task_id,document)
|
|
VALUES (%s,%s,%s,%s,%s) ON CONFLICT(project_id,task_id)
|
|
DO UPDATE SET revision=excluded.revision,document=excluded.document""",
|
|
(who.project_id, document["task_id"], document["revision"], document["source"]["task_id"], Jsonb(document)))
|
|
conn.execute("""INSERT INTO peripheral_sample.revisions(project_id,task_id,revision,document)
|
|
VALUES (%s,%s,%s,%s)""", (who.project_id, document["task_id"], document["revision"], Jsonb(document)))
|
|
return document
|
|
|
|
def create_task(self, who: Identity, command: CreateTask) -> dict[str, Any]:
|
|
def write(conn, _):
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
document = command.model_dump(mode="json", exclude={"request_id"}) | {
|
|
"task_id": str(uuid4()), "project_id": who.project_id, "revision": 1,
|
|
"status": "pending", "origin": "synthetic", "created_at": now, "updated_at": now,
|
|
"created_by": who.user_id, "feedback": [], "history": [{"sequence": 1, "from_status": None,
|
|
"to_status": "pending", "actor": who.user_id, "at": now, "reason": "提交外围样板核查，不是正式工单"}]}
|
|
return self._save(conn, who, document)
|
|
return self._mutate(who, "create_task", command, write)
|
|
|
|
def _project(self, conn, who, doc, scenario):
|
|
if scenario == "late_state" and doc["revision"] > 1:
|
|
row = conn.execute("""SELECT document FROM peripheral_sample.revisions
|
|
WHERE project_id=%s AND task_id=%s AND revision=%s""",
|
|
(who.project_id, doc["task_id"], doc["revision"] - 1)).fetchone()
|
|
if row:
|
|
return row["document"]
|
|
if scenario == "duplicate_feedback" and doc["feedback"]:
|
|
return {**doc, "feedback": doc["feedback"] + [deepcopy(doc["feedback"][-1])]}
|
|
return doc
|
|
|
|
def get_task(self, who: Identity, task_id: UUID | str) -> dict[str, Any]:
|
|
with self.connection(readonly=True) as conn:
|
|
scenario = self._available(conn, who)
|
|
return self._project(conn, who, self._task(conn, who, task_id), scenario)
|
|
|
|
def list_tasks(self, who: Identity, limit: int, cursor: UUID | None = None, source_task_id: UUID | None = None) -> dict[str, Any]:
|
|
with self.connection(readonly=True) as conn:
|
|
scenario = self._available(conn, who)
|
|
params = [who.project_id]
|
|
sql = "SELECT document FROM peripheral_sample.tasks WHERE project_id=%s"
|
|
if cursor is not None:
|
|
sql += " AND task_id>%s"
|
|
params.append(cursor)
|
|
if source_task_id is not None:
|
|
sql += " AND source_task_id=%s"
|
|
params.append(source_task_id)
|
|
rows = conn.execute(sql + " ORDER BY task_id LIMIT %s", params + [limit + 1]).fetchall()
|
|
docs = [self._project(conn, who, row["document"], scenario) for row in rows[:limit]]
|
|
return {"items": docs, "next_cursor": docs[-1]["task_id"] if len(rows) > limit else None}
|
|
|
|
def receipt(self, who: Identity, request_id: UUID | str) -> dict[str, Any]:
|
|
# Remains available during faults; no implicit re-submission.
|
|
with self.connection(readonly=True) as conn:
|
|
row = conn.execute("""SELECT task_id,result FROM peripheral_sample.requests
|
|
WHERE project_id=%s AND actor_id=%s AND request_id=%s""",
|
|
(who.project_id, who.user_id, request_id)).fetchone()
|
|
ensure(row is not None, "REQUEST_NOT_FOUND", "未找到该主体的请求回执", 404)
|
|
return {"request_id": str(request_id), "task_id": str(row["task_id"]) if row["task_id"] else None,
|
|
"result": row["result"]}
|
|
|
|
def transition(self, who: Identity, task_id: UUID | str, command: Transition) -> dict[str, Any]:
|
|
def write(conn, _):
|
|
doc = self._task(conn, who, task_id)
|
|
ensure(doc["revision"] == command.expected_revision, "REVISION_CONFLICT", "任务版本已变化，请重新读取", 409)
|
|
ensure(command.target_status in TRANSITIONS[doc["status"]], "INVALID_TRANSITION", "不允许跳过处置或审核步骤", 409)
|
|
ensure(command.target_status != "completed" or bool(doc["feedback"]),
|
|
"FEEDBACK_REQUIRED", "没有现场反馈，不能审核完成", 409)
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
doc["history"].append({"sequence": len(doc["history"]) + 1, "from_status": doc["status"],
|
|
"to_status": command.target_status, "actor": who.user_id, "at": now, "reason": command.reason})
|
|
doc.update(status=command.target_status, revision=doc["revision"] + 1, updated_at=now)
|
|
return self._save(conn, who, doc)
|
|
return self._mutate(who, f"transition:{task_id}", command, write)
|
|
|
|
def add_feedback(self, who: Identity, task_id: UUID | str, command: AddFeedback) -> dict[str, Any]:
|
|
def write(conn, scenario):
|
|
doc = self._task(conn, who, task_id)
|
|
content = command.model_dump(mode="json", exclude={"request_id", "expected_revision"})
|
|
content_digest = digest(content)
|
|
previous = conn.execute("SELECT task_id,digest FROM peripheral_sample.feedback WHERE project_id=%s AND feedback_id=%s",
|
|
(who.project_id, command.feedback_id)).fetchone()
|
|
if previous:
|
|
ensure(str(previous["task_id"]) == str(task_id) and previous["digest"] == content_digest,
|
|
"FEEDBACK_CONFLICT", "同一反馈编号不能覆盖不同内容", 409)
|
|
return doc
|
|
ensure(doc["revision"] == command.expected_revision, "REVISION_CONFLICT", "任务版本已变化，请重新读取", 409)
|
|
ensure(doc["status"] in {"processing", "pending_review"}, "INVALID_TRANSITION", "仅处理中或待审核任务可记录反馈", 409)
|
|
ensure(set(command.asset_ids) <= set(doc["asset_ids"]), "FEEDBACK_SCOPE_MISMATCH", "反馈对象不属于本核查任务")
|
|
window = TimeRange.model_validate(doc["analysis_window"])
|
|
ensure(command.applicable_time.start >= window.start and command.applicable_time.end <= window.end,
|
|
"FEEDBACK_SCOPE_MISMATCH", "反馈适用时段超出原分析范围")
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
item = {**content, "task_id": str(task_id), "recorded_by": who.user_id, "recorded_at": now, "origin": "synthetic"}
|
|
items = [(item, content_digest)]
|
|
if scenario == "conflicting_feedback" and command.finding != "uncertain":
|
|
opposite = {**item, "feedback_id": str(uuid4()), "recorded_by": "scenario:conflicting_feedback",
|
|
"finding": "no_instrument_fault" if command.finding == "instrument_fault" else "instrument_fault",
|
|
"notes": "故障演练自动注入的相反反馈，不是现场事实"}
|
|
items.append((opposite, digest(opposite)))
|
|
for item, item_digest in items:
|
|
conn.execute("INSERT INTO peripheral_sample.feedback(project_id,feedback_id,task_id,digest,document) VALUES (%s,%s,%s,%s,%s)",
|
|
(who.project_id, item["feedback_id"], task_id, item_digest, Jsonb(item)))
|
|
doc["feedback"].append(item)
|
|
doc.update(revision=doc["revision"] + 1, updated_at=now)
|
|
return self._save(conn, who, doc)
|
|
return self._mutate(who, f"feedback:{task_id}", command, write)
|
|
|
|
def _managed_document(self, conn, who: Identity, table: str, id_column: str,
|
|
resource_id: UUID | str, missing_code: str, missing_message: str) -> dict[str, Any]:
|
|
ensure(table in {"work_orders", "test_data_records"} and
|
|
id_column in {"work_order_id", "record_id"}, "SCHEMA_INVALID", "无效的样板资源")
|
|
row = conn.execute(
|
|
f"SELECT document FROM peripheral_sample.{table} WHERE project_id=%s AND {id_column}=%s AND deleted_at IS NULL",
|
|
(who.project_id, resource_id)).fetchone()
|
|
ensure(row is not None, missing_code, missing_message, 404)
|
|
return self._active_managed_document(row["document"])
|
|
|
|
@staticmethod
|
|
def _active_managed_document(document: dict[str, Any]) -> dict[str, Any]:
|
|
"""Project a current active row without legacy soft-delete metadata."""
|
|
active = {key: value for key, value in document.items()
|
|
if key not in {"deleted_at", "deleted_by"}}
|
|
if "work_order_id" in active:
|
|
active.pop("subtasks", None)
|
|
active.setdefault("feedback_requirements", [])
|
|
active.setdefault("activity", [])
|
|
return active
|
|
|
|
def _list_managed(self, who: Identity, table: str, id_column: str,
|
|
limit: int, cursor: UUID | None) -> dict[str, Any]:
|
|
ensure(table in {"work_orders", "test_data_records"} and
|
|
id_column in {"work_order_id", "record_id"}, "SCHEMA_INVALID", "无效的样板资源")
|
|
with self.connection(readonly=True) as conn:
|
|
self._available(conn, who)
|
|
params: list[Any] = [who.project_id]
|
|
sql = f"SELECT document FROM peripheral_sample.{table} WHERE project_id=%s AND deleted_at IS NULL"
|
|
if cursor is not None:
|
|
sql += f" AND {id_column}>%s"
|
|
params.append(cursor)
|
|
rows = conn.execute(sql + f" ORDER BY {id_column} LIMIT %s", params + [limit + 1]).fetchall()
|
|
docs = [self._active_managed_document(row["document"]) for row in rows[:limit]]
|
|
return {"items": docs, "next_cursor": docs[-1][id_column] if len(rows) > limit else None}
|
|
|
|
def _save_work_order(self, conn, who: Identity, document: dict[str, Any]) -> dict[str, Any]:
|
|
document = WorkOrder.model_validate(document).model_dump(mode="json")
|
|
conn.execute("""INSERT INTO peripheral_sample.work_orders
|
|
(project_id,work_order_id,revision,document,deleted_at) VALUES (%s,%s,%s,%s,NULL)
|
|
ON CONFLICT(project_id,work_order_id) DO UPDATE SET
|
|
revision=excluded.revision,document=excluded.document,deleted_at=NULL""",
|
|
(who.project_id, document["work_order_id"], document["revision"], Jsonb(document)))
|
|
conn.execute("""INSERT INTO peripheral_sample.work_order_revisions
|
|
(project_id,work_order_id,revision,document) VALUES (%s,%s,%s,%s)""",
|
|
(who.project_id, document["work_order_id"], document["revision"], Jsonb(document)))
|
|
self._work_order_event(conn, who, document)
|
|
return document
|
|
|
|
def _work_order_event(self, conn, who: Identity, document: dict[str, Any], *, deleted: bool = False) -> None:
|
|
# Publish only into this project's provisioned event log. Never create
|
|
# storage implicitly or write an event into another project's database.
|
|
if not conn.execute("SELECT to_regclass('workbench.domain_event_log') AS event_log").fetchone()["event_log"]:
|
|
return
|
|
metadata = {
|
|
"source": "peripheral_sample",
|
|
"work_order_id": document["work_order_id"],
|
|
"change": "deleted" if deleted else "created" if document["revision"] == 1 else "updated",
|
|
}
|
|
row = conn.execute(
|
|
"""INSERT INTO workbench.domain_event_log
|
|
(project_id,event_id,topic,aggregate_type,aggregate_id,
|
|
aggregate_revision,occurred_at,metadata)
|
|
VALUES (%s,%s,'workbench.work_order.changed','peripheral_work_order',%s,%s,%s,%s)
|
|
RETURNING event_seq""",
|
|
(who.project_id, uuid4(), document["work_order_id"], document["revision"],
|
|
document["updated_at"], Jsonb(metadata)),
|
|
).fetchone()
|
|
conn.execute("SELECT pg_notify('workbench_events', %s)",
|
|
(f"{who.project_id}:{row['event_seq']}",))
|
|
|
|
def list_work_orders(self, who: Identity, limit: int, cursor: UUID | None = None) -> dict[str, Any]:
|
|
return self._list_managed(who, "work_orders", "work_order_id", limit, cursor)
|
|
|
|
def get_work_order(self, who: Identity, work_order_id: UUID | str) -> dict[str, Any]:
|
|
with self.connection(readonly=True) as conn:
|
|
self._available(conn, who)
|
|
return self._managed_document(conn, who, "work_orders", "work_order_id", work_order_id,
|
|
"WORK_ORDER_NOT_FOUND", "测试工单不存在")
|
|
|
|
def create_work_order(self, who: Identity, command: CreateWorkOrder) -> dict[str, Any]:
|
|
def write(conn, _):
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
document = command.model_dump(mode="json", exclude={"request_id"}) | {
|
|
"work_order_id": str(uuid4()), "project_id": who.project_id, "revision": 1,
|
|
"origin": "synthetic", "created_at": now, "updated_at": now,
|
|
"created_by": who.user_id, "updated_by": who.user_id,
|
|
"activity": [{"sequence": 1, "occurred_at": now, "actor": who.user_id,
|
|
"kind": "created", "title": "工单创建",
|
|
"detail": WORK_ORDER_CREATED_DETAILS[command.work_order_type]}],
|
|
}
|
|
return self._save_work_order(conn, who, document)
|
|
return self._mutate(who, "create_work_order", command, write)
|
|
|
|
def update_work_order(self, who: Identity, work_order_id: UUID | str,
|
|
command: UpdateWorkOrder) -> dict[str, Any]:
|
|
def write(conn, _):
|
|
document = self._managed_document(conn, who, "work_orders", "work_order_id", work_order_id,
|
|
"WORK_ORDER_NOT_FOUND", "测试工单不存在")
|
|
ensure(document["revision"] == command.expected_revision, "REVISION_CONFLICT", "工单版本已变化，请重新读取", 409)
|
|
changes = command.model_dump(mode="json", exclude_unset=True,
|
|
exclude={"request_id", "expected_revision", "clear_assignee", "clear_scheduled_at",
|
|
"clear_location", "clear_scenario_code", "progress_note"})
|
|
if command.clear_assignee:
|
|
changes["assignee"] = None
|
|
if command.clear_scheduled_at:
|
|
changes["scheduled_at"] = None
|
|
if command.clear_location:
|
|
changes["location"] = None
|
|
if command.clear_scenario_code:
|
|
changes["scenario_code"] = None
|
|
changed_fields = [key for key, value in changes.items() if document.get(key) != value]
|
|
document.update(changes)
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
revision = document["revision"] + 1
|
|
if command.progress_note:
|
|
title, detail, kind = "现场反馈", command.progress_note, "feedback"
|
|
elif changed_fields:
|
|
field_labels = WORK_ORDER_FIELD_LABELS | ({"scheduled_at": "发生时间"}
|
|
if document["work_order_type"] != "general" else {})
|
|
title, detail, kind = "工单信息更新", "更新了" + "、".join(
|
|
field_labels.get(field, field) for field in changed_fields
|
|
) + "。", "updated"
|
|
else:
|
|
title, detail, kind = "工单信息确认", "工单信息已确认，未记录新的现场进展。", "updated"
|
|
document["activity"].append({"sequence": revision, "occurred_at": now,
|
|
"actor": who.user_id, "kind": kind,
|
|
"title": title, "detail": detail})
|
|
document.update(revision=revision, updated_at=now,
|
|
updated_by=who.user_id)
|
|
return self._save_work_order(conn, who, document)
|
|
return self._mutate(who, f"update_work_order:{work_order_id}", command, write)
|
|
|
|
def delete_work_order(self, who: Identity, work_order_id: UUID | str,
|
|
command: DeleteResource) -> dict[str, Any]:
|
|
def write(conn, _):
|
|
document = self._managed_document(conn, who, "work_orders", "work_order_id", work_order_id,
|
|
"WORK_ORDER_NOT_FOUND", "测试工单不存在")
|
|
ensure(document["revision"] == command.expected_revision, "REVISION_CONFLICT", "工单版本已变化，请重新读取", 409)
|
|
revision = document["revision"] + 1
|
|
deleted_at = datetime.now(timezone.utc).isoformat()
|
|
result = {"resource_type": "work_order", "resource_id": str(work_order_id),
|
|
"revision": revision, "deleted": True}
|
|
tombstone = {**document, "revision": revision, "updated_at": deleted_at,
|
|
"updated_by": who.user_id, "deleted_at": deleted_at, "deleted_by": who.user_id}
|
|
conn.execute("""UPDATE peripheral_sample.work_orders SET revision=%s,document=%s,deleted_at=%s,deleted_by=%s
|
|
WHERE project_id=%s AND work_order_id=%s AND deleted_at IS NULL""",
|
|
(revision, Jsonb(tombstone), deleted_at, who.user_id, who.project_id, work_order_id))
|
|
conn.execute("""INSERT INTO peripheral_sample.work_order_revisions
|
|
(project_id,work_order_id,revision,document) VALUES (%s,%s,%s,%s)""",
|
|
(who.project_id, work_order_id, revision, Jsonb(tombstone)))
|
|
self._work_order_event(conn, who, tombstone, deleted=True)
|
|
return result
|
|
return self._mutate(who, f"delete_work_order:{work_order_id}", command, write)
|
|
|
|
def _save_test_data(self, conn, who: Identity, document: dict[str, Any]) -> dict[str, Any]:
|
|
document = TestDataRecord.model_validate(document).model_dump(mode="json")
|
|
conn.execute("""INSERT INTO peripheral_sample.test_data_records
|
|
(project_id,record_id,revision,document,deleted_at) VALUES (%s,%s,%s,%s,NULL)
|
|
ON CONFLICT(project_id,record_id) DO UPDATE SET
|
|
revision=excluded.revision,document=excluded.document,deleted_at=NULL""",
|
|
(who.project_id, document["record_id"], document["revision"], Jsonb(document)))
|
|
conn.execute("""INSERT INTO peripheral_sample.test_data_record_revisions
|
|
(project_id,record_id,revision,document) VALUES (%s,%s,%s,%s)""",
|
|
(who.project_id, document["record_id"], document["revision"], Jsonb(document)))
|
|
return document
|
|
|
|
def list_test_data(self, who: Identity, limit: int, cursor: UUID | None = None) -> dict[str, Any]:
|
|
return self._list_managed(who, "test_data_records", "record_id", limit, cursor)
|
|
|
|
def get_test_data(self, who: Identity, record_id: UUID | str) -> dict[str, Any]:
|
|
with self.connection(readonly=True) as conn:
|
|
self._available(conn, who)
|
|
return self._managed_document(conn, who, "test_data_records", "record_id", record_id,
|
|
"TEST_DATA_NOT_FOUND", "测试数据不存在")
|
|
|
|
def create_test_data(self, who: Identity, command: CreateTestDataRecord) -> dict[str, Any]:
|
|
def write(conn, _):
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
document = command.model_dump(mode="json", exclude={"request_id"}) | {
|
|
"record_id": str(uuid4()), "project_id": who.project_id, "revision": 1,
|
|
"origin": "synthetic", "created_at": now, "updated_at": now,
|
|
"created_by": who.user_id, "updated_by": who.user_id,
|
|
}
|
|
return self._save_test_data(conn, who, document)
|
|
return self._mutate(who, "create_test_data", command, write)
|
|
|
|
def update_test_data(self, who: Identity, record_id: UUID | str,
|
|
command: UpdateTestDataRecord) -> dict[str, Any]:
|
|
def write(conn, _):
|
|
document = self._managed_document(conn, who, "test_data_records", "record_id", record_id,
|
|
"TEST_DATA_NOT_FOUND", "测试数据不存在")
|
|
ensure(document["revision"] == command.expected_revision, "REVISION_CONFLICT", "数据版本已变化，请重新读取", 409)
|
|
changes = command.model_dump(mode="json", exclude_unset=True,
|
|
exclude={"request_id", "expected_revision", "clear_value"})
|
|
if command.clear_value:
|
|
changes["value"] = None
|
|
document.update(changes)
|
|
document.update(revision=document["revision"] + 1, updated_at=datetime.now(timezone.utc).isoformat(),
|
|
updated_by=who.user_id)
|
|
return self._save_test_data(conn, who, document)
|
|
return self._mutate(who, f"update_test_data:{record_id}", command, write)
|
|
|
|
def delete_test_data(self, who: Identity, record_id: UUID | str,
|
|
command: DeleteResource) -> dict[str, Any]:
|
|
def write(conn, _):
|
|
document = self._managed_document(conn, who, "test_data_records", "record_id", record_id,
|
|
"TEST_DATA_NOT_FOUND", "测试数据不存在")
|
|
ensure(document["revision"] == command.expected_revision, "REVISION_CONFLICT", "数据版本已变化，请重新读取", 409)
|
|
revision = document["revision"] + 1
|
|
deleted_at = datetime.now(timezone.utc).isoformat()
|
|
result = {"resource_type": "test_data_record", "resource_id": str(record_id),
|
|
"revision": revision, "deleted": True}
|
|
tombstone = {**document, "revision": revision, "updated_at": deleted_at,
|
|
"updated_by": who.user_id, "deleted_at": deleted_at, "deleted_by": who.user_id}
|
|
conn.execute("""UPDATE peripheral_sample.test_data_records SET revision=%s,document=%s,deleted_at=%s,deleted_by=%s
|
|
WHERE project_id=%s AND record_id=%s AND deleted_at IS NULL""",
|
|
(revision, Jsonb(tombstone), deleted_at, who.user_id, who.project_id, record_id))
|
|
conn.execute("""INSERT INTO peripheral_sample.test_data_record_revisions
|
|
(project_id,record_id,revision,document) VALUES (%s,%s,%s,%s)""",
|
|
(who.project_id, record_id, revision, Jsonb(tombstone)))
|
|
return result
|
|
return self._mutate(who, f"delete_test_data:{record_id}", command, write)
|
|
|
|
def points(self, who: Identity, limit: int = 100, cursor: str | None = None) -> dict[str, Any]:
|
|
with self.connection(readonly=True) as conn:
|
|
self._available(conn, who)
|
|
# Existing project-owned table; absolutely no DDL/DML on it.
|
|
exists = conn.execute("SELECT to_regclass('asset.scada_devices') AS table_name").fetchone()["table_name"]
|
|
ensure(exists is not None, "SOURCE_NOT_CONFIGURED", "项目测点目录尚未映射", 503)
|
|
rows = conn.execute("""SELECT device_id,device_type,measurement_unit FROM asset.scada_devices
|
|
WHERE (%s::text IS NULL OR device_id::text>%s) ORDER BY device_id::text LIMIT %s""",
|
|
(cursor, cursor, limit + 1)).fetchall()
|
|
return {"source_id": self.settings.db_name + ":asset.scada_devices", "origin": "historical_replay",
|
|
"items": [{"point_id": str(r["device_id"]).strip(), "label": str(r["device_id"]).strip() + " · " + str(r["device_type"]),
|
|
"unit": str(r["measurement_unit"]), "status": "unknown", "origin": "historical_replay"} for r in rows[:limit]],
|
|
"next_cursor": str(rows[limit-1]["device_id"]).strip() if len(rows) > limit else None}
|
|
|
|
def observations(self, who: Identity, point_id: str, start: str, end: str) -> dict[str, Any]:
|
|
try:
|
|
window = TimeRange.model_validate({"start": start, "end": end})
|
|
except ValueError as exc:
|
|
raise SampleFailure("SCHEMA_INVALID", "需要带时区的有效起止时间") from exc
|
|
ensure(window.end - window.start <= timedelta(days=7), "WINDOW_TOO_LARGE", "样板查询窗口最多 7 天")
|
|
with self.connection(readonly=True) as conn:
|
|
scenario = self._available(conn, who)
|
|
point = conn.execute("SELECT measurement_unit,api_query_id FROM asset.scada_devices WHERE device_id=%s", (point_id,)).fetchone()
|
|
ensure(point is not None, "POINT_NOT_FOUND", "测点不存在", 404)
|
|
ensure(point["api_query_id"], "SOURCE_NOT_CONFIGURED", "测点历史查询标识未配置", 503)
|
|
with self.connection(readonly=True, timeseries=True) as conn:
|
|
rows = conn.execute("""SELECT time,monitored_value FROM scada.measurements
|
|
WHERE device_id=%s AND time>=%s AND time<%s ORDER BY time LIMIT 1001""",
|
|
(point["api_query_id"], window.start, window.end)).fetchall()
|
|
items = [{"record_id": digest({"point": point_id, "time": r["time"].isoformat()}),
|
|
"point_id": point_id, "observed_at": r["time"].isoformat(),
|
|
"value": r["monitored_value"] if r["monitored_value"] is not None and isfinite(r["monitored_value"]) else None,
|
|
"quality": "unknown" if r["monitored_value"] is not None and isfinite(r["monitored_value"]) else "missing",
|
|
"unit": point["measurement_unit"]} for r in rows[:1000]]
|
|
if scenario == "missing_data":
|
|
for item in items[::2]:
|
|
item.update(value=None, quality="missing")
|
|
source_database = conninfo_to_dict(self.routing.timescale_dsn)["dbname"] if self.routing else self.settings.db_name
|
|
return {"origin": "synthetic" if scenario == "missing_data" else "historical_replay",
|
|
"source_id": source_database + ":scada.measurements", "source_revision": digest(items),
|
|
"items": items, "truncated": len(rows) > 1000}
|
|
|
|
def alerts(self, who: Identity) -> dict[str, Any]:
|
|
with self.connection(readonly=True) as conn:
|
|
self._available(conn, who)
|
|
return {"items": [], "source_id": self.settings.db_name, "origin": "historical_replay",
|
|
"availability": "not_configured", "reason": "告警来源未映射；不从压力读数猜测告警或阈值"}
|
|
|
|
def demo_data(self, who: Identity) -> dict[str, Any]:
|
|
with self.connection(readonly=True) as conn:
|
|
self._available(conn, who)
|
|
row = conn.execute("""SELECT document,digest FROM peripheral_sample.demo_datasets
|
|
WHERE project_id=%s AND dataset_id='water-demo-v1'""", (who.project_id,)).fetchone()
|
|
ensure(row is not None, "SOURCE_NOT_CONFIGURED", "当前项目尚未登记合成样本", 503)
|
|
ensure(digest(row["document"]) == row["digest"], "INTEGRITY_MISMATCH", "合成样本摘要不匹配", 503)
|
|
try:
|
|
content = DemoContent.model_validate(row["document"]).model_dump(mode="json")
|
|
except ValueError as exc:
|
|
raise SampleFailure("INTEGRITY_MISMATCH", "合成样本格式不匹配", 503) from exc
|
|
return {**content, "project_id": who.project_id,
|
|
"source_id": self.settings.db_name + ":peripheral_sample.demo_datasets:water-demo-v1",
|
|
"source_revision": row["digest"]}
|