# new_secret_popup.dspy - 新增凭据弹窗(值只经表单 POST,绝不进列表/编辑对话框) import json uid = await get_user() if (params_kw or {}).get('op') != 'form': # POST:保存 if not uid: return json.dumps({"success": False, "error": "未登录"}, ensure_ascii=False) name = str((params_kw or {}).get('name', '') or '').strip() value = str((params_kw or {}).get('value', '') or '') label = str((params_kw or {}).get('label', '') or '').strip() remark = str((params_kw or {}).get('remark', '') or '').strip() stype = str((params_kw or {}).get('secret_type', '') or '').strip() if not name or not value: return json.dumps({"success": False, "error": "名称与值必填"}, ensure_ascii=False) org = (await get_userorgid()) or '' from pipeline_service import secret_vault as sv async with DBPools().sqlorContext('pipeline') as sor: r = await sv.save_secret(sor, name=name, value=value, org_id=org, user_id=uid, label=label, secret_type=stype, source='manual', remark=remark, who=uid) await sor.sqlExe("COMMIT", {}) if r.get("ok"): return json.dumps({"success": True, "message": r.get("message", "已保存")}, ensure_ascii=False) return json.dumps({"success": False, "error": r.get("message", "保存失败")}, ensure_ascii=False) # GET op=form:返回弹窗 widget if not uid: return json.dumps({"widgettype": "PopupWindow", "id": "new_secret_pw", "options": {"title": "新增凭据", "cwidth": 26, "cheight": 14, "auto_open": True}, "subwidgets": [{"widgettype": "Text", "options": {"text": "请先登录", "cfontsize": 1}}]}, ensure_ascii=False) _self_url = entire_url("/pipeline_core/secrets/new_secret_popup.dspy") def _field(fid, label, ftype, placeholder=""): # bricks 控件名:UiStr=单行 / UiPassword=口令掩码 / UiText=多行 textarea wtype = {"text": "UiStr", "pass": "UiPassword", "area": "UiText"}[ftype] return {"widgettype": "VBox", "options": {"css": "field-row"}, "subwidgets": [ {"widgettype": "Text", "options": {"text": label, "cfontsize": 0.9}}, {"widgettype": wtype, "id": fid, "options": {"placeholder": placeholder, "cheight": 3 if ftype != "area" else 5}}]} _save_js = ( "var body=new URLSearchParams();" "body.append('name',bricks.getWidgetById('ns_name',bricks.app).get_value());" "body.append('value',bricks.getWidgetById('ns_value',bricks.app).get_value());" "body.append('label',bricks.getWidgetById('ns_label',bricks.app).get_value());" "body.append('remark',bricks.getWidgetById('ns_remark',bricks.app).get_value());" "body.append('secret_type',bricks.getWidgetById('ns_type',bricks.app).get_value());" "var rp=await fetch(" + json.dumps(_self_url) + ",{method:'POST'," "headers:{'Content-Type':'application/x-www-form-urlencoded'},body:body});" "var d=await rp.json();" "if(d&&d.success){bricks.show_message({title:'凭据',message:d.message||'已保存'});" "var pw=bricks.getWidgetById('new_secret_pw',bricks.app);if(pw){pw.destroy();}" "location.reload();}" "else{bricks.show_message({title:'凭据',message:(d&&d.error)||'保存失败'});}" ) return json.dumps({ "widgettype": "PopupWindow", "id": "new_secret_pw", "options": {"title": "🔑 新增凭据", "cwidth": 26, "cheight": 20, "auto_open": True}, "subwidgets": [ {"widgettype": "Text", "options": { "text": "值加密存储、永不回显;会话用 @@sec:名称@@ 引用,命令用 $PIPELINE_SEC_名称。", "cfontsize": 0.85}}, _field("ns_name", "名称(大写字母/数字/下划线)", "text", "如 GITHUB_TOKEN"), _field("ns_value", "值(明文,仅本次传输)", "pass", "粘贴 apikey/token/口令"), _field("ns_label", "标签(可选)", "text", "如 我的GitHub"), _field("ns_type", "类型(可选)", "text", "如 api_key/password/token"), _field("ns_remark", "备注(可选)", "area", "用途说明"), {"widgettype": "Button", "options": {"label": "保存", "css": "primary"}, "script": _save_js}, ]}, ensure_ascii=False)