"""按「服务名 + 上线时间」定位交付单并保存配置（纯 HTTP，禁止经 MCP）。 改顶部 SERVER_NAME / ONLINE_DATE / OVERRIDES 即可；账号密码走环境变量 DEVOPS_USER/DEVOPS_PWD 或运行时交互输入（密码不回显、不落盘）。启动后会回读 4 项输入让你确认，确认后才登录。 定位到交付单先调 checkEditDeliver 校验录制权限，非 M0200 直接终止。默认只 save，不 commit。 每个页签的 OVERRIDES 结构见 ../tabs/ 下对应文件。 前置：交付单必须已存在（本脚本只编辑、不创建）。找不到会提示去平台建单。 合并语义：OVERRIDES 里的页签默认与交付单**现值合并**（按自然键去重：nacos 按 dataId+key、 mq 按 topic、cronJobs 按 jobHandler、网关按 type 一类一份、sqlFile 追加去重）。 要整批替换某页签，把它的字段名加进 REPLACE。 ⛔ 本脚本只做 save，绝不 commit：commitDeliver 严禁调用，提交只能由用户本人在平台页面上手动点击。""" import hashlib, datetime, getpass, os, requests BASE = "http://devops.test.yljr.com" # 换成你司持续交付平台地址 # 账号密码：环境变量优先，取不到就交互输入。⛔ 不要把密码写进本文件（会被提交/落盘）。 USERNAME = os.environ.get("DEVOPS_USER") or input("持续交付平台账号: ").strip() PASSWORD = os.environ.get("DEVOPS_PWD") or getpass.getpass("密码(不回显): ") # ==== 只改这里 ==== SERVER_NAME = "<服务名称>" # 必须是用户确认过的平台服务名 ONLINE_DATE = "<上线时间 YYYY-MM-DD>" # 必须是用户确认过的上线日期 CREATE_NAME = "" # 同名多条时填创建人唯一定位 OVERRIDES = {} # 本次要配的页签（结构见 ../tabs/）；默认与现值合并 REPLACE = set() # 需整批替换的页签字段名，如 {"nacos"} # ================== def calc_pwd(pwd): salt = "CHIANABEIJINGXINSANYOURUANJIANZHONGLIANXINXINGFILSOFT20081127XSY" p = pwd.translate(str.maketrans("0123456789", "~$!@:][{}`")) return hashlib.md5((salt[:len(salt) - len(p)] + p).encode()).hexdigest().upper() def date_to_ts(d): # YYYY-MM-DD 当天 0 点(东八区) -> 毫秒 dt = datetime.datetime.strptime(d, "%Y-%m-%d").replace( tzinfo=datetime.timezone(datetime.timedelta(hours=8))) return str(int(dt.timestamp() * 1000)) def upload(sess, path, biz_type): # 网关/附件文件上传 with open(path, "rb") as f: return sess.post(f"{BASE}/zqyl-devops-file-api/file/uploadFile", files={"file": f}, data={"bizType": biz_type}).json()["data"] # 网关行示例（一个网关类型一份文件；type: 路由2/参数3/限流4/熔断5）： # u = upload(s, "路由.xls", "cicd-outGateway") # OVERRIDES["outGateway"] = [{"type": 2, "fileName": u["showName"], "fileNo": u["fileNo"], # "gatewayContent": "", "downloadUrl": u["downLoadUrl"], "viewUrl": u["downLoadUrl"]}] # ---- 合并 ---- KEY = { # 数组页签的自然键（None = 追加去重） "nacos": lambda x: (x.get("groups"), x.get("dataId")), "sqlFile": None, "file": None, "resourceImport": None, "message": None, "firewall": lambda x: (x.get("srcServerId"), x.get("serverId")), # 一对「源→目标」一条 "mq": lambda x: x.get("mqType"), "cronJobs": lambda x: (x.get("appEnName"), x.get("serverId")), "innerGateway": lambda x: x.get("type"), "outGateway": lambda x: x.get("type"), "midGateway": lambda x: x.get("type"), "outFileGateway": lambda x: x.get("type"), } def _merge_sub(cur, new, kname, sub): # 深合并对象里的一个子数组（按 kname 去重） out = list(cur.get(sub) or []) idx = {r.get(kname): i for i, r in enumerate(out)} for r in (new.get(sub) or []): k = r.get(kname) if k in idx: out[idx[k]] = r else: idx[k] = len(out); out.append(r) m = dict(cur); m.update({kk: vv for kk, vv in new.items() if kk != sub}); m[sub] = out return m def merge_list(field, cur, new): cur, new = list(cur or []), list(new or []) kf = KEY.get(field) if kf is None: # 无自然键：追加、去掉完全相同项 out = list(cur) for it in new: if it not in out: out.append(it) return out out = list(cur); idx = {kf(x): i for i, x in enumerate(out)} for it in new: k = kf(it) if k not in idx: idx[k] = len(out); out.append(it) elif field == "nacos": out[idx[k]] = _merge_sub(out[idx[k]], it, "key", "nacosKeys") elif field == "mq": out[idx[k]] = _merge_sub(out[idx[k]], it, "topic", "mqsubInfo") elif field == "cronJobs": out[idx[k]] = _merge_sub(out[idx[k]], it, "jobHandler", "jobs") else: out[idx[k]] = it # 网关：同 type 换成新文件 return out EMPTY = {"nacos": [], "sqlFile": [], "mq": [], "cronJobs": [], "file": [], "innerGateway": [], "outGateway": [], "midGateway": [], "outFileGateway": [], "firewall": [], "errorCenter": [], "resourceImport": [], "message": [], "dataware": {"content": None, "files": []}, "nginx": {"protocolGroup": "", "serverId": None, "confId": None, "nginxServerId": None, "upstream": [], "location": []}, "webpermission": {}, "midpermission": {}, "managerpermission": {}, "contentConfig": {}, "business": {}, "others": {}, "onlinevalid": {}, "outserver": {}} def build_payload(d): p = {"deliverId": d["deliverId"], "projectId": d["projectId"], "jiraCode": d["jiraCode"], "projectName": d["projectName"], "projectCode": d["projectCode"], "onlineTime": d["onlineTime"], "serverInfo": {"serverId": d["serverId"], "serverName": d["serverName"]}, "serverId": d["serverId"], "serverName": d["serverName"], "optimistic": d["optimistic"], "version": d["version"], "configtypes": d["configtypes"], "configtypeNames": d["configtypeNames"]} for k, empty in EMPTY.items(): # 先回填现值，缺失补空 p[k] = d.get(k) if d.get(k) is not None else empty if d.get("upgrade"): # 数据升级：有现值才带，空则不带（否则会丢） p["upgrade"] = d["upgrade"] for field, newval in OVERRIDES.items(): # 覆盖：默认合并，REPLACE 里的整批替换 if field in REPLACE or not isinstance(newval, list): p[field] = newval else: p[field] = merge_list(field, p.get(field), newval) return p # 注：configtypes 由后端按报文内容重算，无需手动维护。 ts = date_to_ts(ONLINE_DATE) # 跑之前把 4 项输入回读一遍，让人眼确认（服务名/上线时间填错会写到别人的交付单上） print(f"服务名 : {SERVER_NAME}\n上线时间: {ONLINE_DATE} (ts={ts})\n平台账号: {USERNAME}") assert input("以上确认无误？(y/N) ").strip().lower() == "y", "已取消：请修正 SERVER_NAME / ONLINE_DATE 后重跑。" s = requests.Session() lg = s.post(f"{BASE}/zqyl-devops-base-api/user/login", json={"username": USERNAME, "password": calc_pwd(PASSWORD)}) lg.raise_for_status() assert lg.json().get("status") == "M0200", f"登录失败：{lg.json().get('msg')}（请核对账号密码后重跑）" rows = s.get(f"{BASE}/zqyl-devops-deliver-api/deliver/queryDeliverList", params={"page": 1, "pageRow": 50, "serverName": SERVER_NAME, "onlineTime": ts, "createName": CREATE_NAME}).json().get("data") or [] assert rows, (f"未找到交付单：{SERVER_NAME} @ {ONLINE_DATE}。" f"本脚本只编辑不创建——请先到平台「交付管理」为该服务/上线窗口建单后再运行。") assert len(rows) == 1, f"匹配到 {len(rows)} 条，请补 CREATE_NAME 唯一定位" deliver_id = rows[0]["id"] # 录制权限校验：当前账号有没有权限录制这张单（同时兼作编辑锁校验），非 M0200 一律终止 chk = s.get(f"{BASE}/zqyl-devops-deliver-api/deliver/checkEditDeliver", params={"deliverId": deliver_id}).json() assert chk.get("status") == "M0200", ( f"账号 {USERNAME} 无法录制交付单 {deliver_id}：{chk.get('msg')}。" f"可能无权限、别人正在编辑或单据已锁定——请到平台确认权限或联系单据创建人，不要绕过本校验。") # 解析测试环境（供 Nacos/XXL-Job/网关平台读现值） wins = s.get(f"{BASE}/zqyl-devops-base-api/onlinetime/queryOnlineTimelist", params={"page": 1, "pageRow": 20, "time": ts, "environment": "", "type": "", "onlineCommander": ""}).json().get("data") or [] print("测试环境:", wins[0]["environmentType"] if wins else None) # 保存：每次都先取最新详情再合并，乐观锁冲突自动重取重试（避免覆盖并发改动） r = None for _ in range(5): d = s.get(f"{BASE}/zqyl-devops-deliver-api/deliver/getDeliverDetail", params={"deliverId": deliver_id}).json()["data"] r = s.post(f"{BASE}/zqyl-devops-deliver-api/deliver/saveDeliver", json=build_payload(d)).json() if r.get("status") != "M0600": break print(deliver_id, r) # 期望 {'status':'M0200', ...} # commit（提交）是对外不可逆动作：请人工确认后自行到页面点击，严禁在此调用 commitDeliver。