From df6f5cce34300c02854f6a0e2f68256d07a987e8 Mon Sep 17 00:00:00 2001 From: ymq Date: Mon, 17 Aug 2026 07:10:28 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20feature=5Fcapability.py=20=E5=8A=9F?= =?UTF-8?q?=E8=83=BD/=E9=9C=80=E6=B1=82=E7=8A=B6=E6=80=81=E6=9C=BA?= =?UTF-8?q?=E8=83=BD=E5=8A=9B(CAS=E8=BF=81=E7=A7=BB+=E5=AE=A1=E8=AE=A1,sco?= =?UTF-8?q?pe=3Dproject=5Fid)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_service/feature_capability.py | 197 +++++++++++++++++++++++++ 1 file changed, 197 insertions(+) create mode 100644 pipeline_service/feature_capability.py diff --git a/pipeline_service/feature_capability.py b/pipeline_service/feature_capability.py new file mode 100644 index 0000000..2a51648 --- /dev/null +++ b/pipeline_service/feature_capability.py @@ -0,0 +1,197 @@ +"""功能/需求能力 — sd_features 的状态机语义化迁移（通用、产线无关）。 + +定位：本模块是功能（feature/需求）的状态机语义化操作层。CRUD（新增/编辑字段）走 +xls2ui 生成的 CRUD 端点；本模块只做「状态流转」——每个迁移 CAS 原子（防并发/越权）+ 租户隔离 + 审计。 + +功能的状态机/流转规则在 feature skill 里（LLM 读 skill 判断合法性）；本模块只固化操作原语。 + +角色规范：role 参数用 agent.{role}（无前缀自动补 agent.）；人角色用 {orgtype}.{role}。 + +scope 约定：sd_features 用 project_id 做范围校验（CAS 的 WHERE 里带 project_id）， +与 pipeline_tasks/pipeline_agent_questions 用 tenant_id 不同——sd_* 业务表用自己的归属列。 +""" + +import logging +from sqlor.dbpools import DBPools +from appPublic.uniqueID import getID +from .audit import record_audit + +DBNAME = "pipeline" +logger = logging.getLogger("pipeline.feature_capability") + +TABLE = "sd_features" + +# 功能状态（SDLC 默认，状态机语义见 feature skill） +S_PROPOSED = "proposed" # 已提出（待审批） +S_APPROVED = "approved" # 已批准（待开发） +S_IN_PROGRESS = "in_progress" # 开发中 +S_DELIVERED = "delivered" # 已交付（待验证） +S_VERIFIED = "verified" # 已验证 +S_REJECTED = "rejected" # 已驳回 + + +def _get_db(): + db = DBPools() + if not db.databases: + from appPublic.jsonConfig import getConfig + config = getConfig() + if config.databases: + db.databases = config.databases + return db, DBNAME + + +def _normalize_role(role): + """角色规范：agent 角色补 agent. 前缀；人角色 {orgtype}.{role} 保留原样。""" + role = (role or "").strip() + if not role: + return "" + if "." in role: + return role + return f"agent.{role}" + + +async def _transition(feature_id, project_id, from_state, to_state, action, + who=None, agent_id=None, detail=None): + """CAS 状态迁移 + 审计。返回 (ok, message)。""" + if not feature_id or not project_id: + return False, "缺少 feature_id 或 project_id" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + await sor.sqlExe( + f"UPDATE {TABLE} SET status=${{to}}$, updated_at=NOW() " + "WHERE id=${fid}$ AND project_id=${pid}$ AND status=${from}$", + {"to": to_state, "fid": feature_id, "pid": project_id, "from": from_state}) + recs = await sor.R(TABLE, {'id': feature_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return False, "功能不存在" + cur = getattr(recs[0], 'status', '') + if cur != to_state: + return False, f"状态迁移失败(CAS): 期望 from={from_state} 实际 status={cur}" + await record_audit(project_id, TABLE, feature_id, action, + from_state=from_state, to_state=to_state, + who=who, agent_id=agent_id, detail=detail, sor=sor) + return True, to_state + + +async def propose_feature(project_id, feature_name, description="", + iteration_id="", priority="P2", feature_type="new_feature", + created_by="", who=None, agent_id=None): + """提出功能：新建记录，status=proposed。返回 (ok, feature_id_or_message)。""" + if not project_id: + return False, "缺少 project_id" + if not feature_name or not feature_name.strip(): + return False, "缺少 feature_name" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + fid = getID() + await sor.C(TABLE, { + 'id': fid, + 'project_id': project_id, + 'iteration_id': iteration_id or '', + 'feature_name': feature_name.strip(), + 'description': description or '', + 'feature_type': feature_type or 'new_feature', + 'priority': priority or 'P2', + 'status': S_PROPOSED, + 'created_by': created_by or '', + }) + await record_audit(project_id, TABLE, fid, 'propose', + to_state=S_PROPOSED, who=_normalize_role(who), + agent_id=agent_id, sor=sor) + logger.info("propose_feature: %s project=%s", fid, project_id) + return True, fid + + +async def approve_feature(feature_id, project_id, who=None, agent_id=None, comment=None): + """审批通过：proposed → approved。""" + return await _transition(feature_id, project_id, S_PROPOSED, S_APPROVED, 'approve', + who=who, agent_id=agent_id, detail=comment) + + +async def reject_feature(feature_id, project_id, who=None, agent_id=None, comment=None): + """驳回：proposed → rejected（也兼容 approved → rejected）。""" + ok, msg = await _transition(feature_id, project_id, S_PROPOSED, S_REJECTED, 'reject', + who=who, agent_id=agent_id, detail=comment) + if ok: + return ok, msg + return await _transition(feature_id, project_id, S_APPROVED, S_REJECTED, 'reject', + who=who, agent_id=agent_id, detail=comment) + + +async def start_feature(feature_id, project_id, who=None, agent_id=None): + """开始开发：approved → in_progress。""" + return await _transition(feature_id, project_id, S_APPROVED, S_IN_PROGRESS, 'start', + who=who, agent_id=agent_id) + + +async def deliver_feature(feature_id, project_id, who=None, agent_id=None): + """交付：in_progress → delivered。""" + return await _transition(feature_id, project_id, S_IN_PROGRESS, S_DELIVERED, 'deliver', + who=who, agent_id=agent_id) + + +async def verify_feature(feature_id, project_id, who=None, agent_id=None): + """验证通过：delivered → verified。""" + return await _transition(feature_id, project_id, S_DELIVERED, S_VERIFIED, 'verify', + who=who, agent_id=agent_id) + + +async def reopen_feature(feature_id, project_id, who=None, agent_id=None, comment=None): + """重新打开：rejected → proposed（也兼容 verified → proposed 回炉）。""" + ok, msg = await _transition(feature_id, project_id, S_REJECTED, S_PROPOSED, 'reopen', + who=who, agent_id=agent_id, detail=comment) + if ok: + return ok, msg + return await _transition(feature_id, project_id, S_VERIFIED, S_PROPOSED, 'reopen', + who=who, agent_id=agent_id, detail=comment) + + +async def set_feature_state(feature_id, project_id, from_state, to_state, + who=None, agent_id=None, detail=None): + """通用 CAS 状态迁移兜底（跨产线自定义状态机用）。""" + return await _transition(feature_id, project_id, from_state, to_state, 'set_state', + who=who, agent_id=agent_id, detail=detail) + + +async def list_features(project_id, status=None, iteration_id=None, limit=50) -> list: + """列出功能（可按状态/迭代过滤）。""" + db, dbname = _get_db() + async with db.sqlorContext(dbname) as sor: + conditions = ["project_id=${pid}$"] + params = {"pid": project_id} + if status: + conditions.append("status=${status}$") + params["status"] = status + if iteration_id: + conditions.append("iteration_id=${iid}$") + params["iid"] = iteration_id + where = " AND ".join(conditions) + try: + limit = int(limit) + except (TypeError, ValueError): + limit = 50 + sql = (f"SELECT * FROM {TABLE} WHERE {where} " + f"ORDER BY created_at DESC LIMIT {limit}") + recs = await sor.sqlExe(sql, params) + # 释放 SELECT 元数据锁 + await sor.sqlExe("COMMIT", {}) + result = [] + for rec in (recs or []): + result.append(_rec_to_dict(rec)) + return result + + +def _rec_to_dict(rec): + """把 sqlor 记录对象转成 dict（sqlor 行是 DictObject，必须 dict(rec) 取列）。""" + if isinstance(rec, dict): + return dict(rec) + try: + return dict(rec) + except (TypeError, ValueError): + if hasattr(rec, 'to_dict'): + try: + return rec.to_dict() + except Exception: + pass + return {}