"""只读验证渠道画像路由及现有数据的缺表处理,不修改业务记录。""" import argparse import json import urllib.parse import urllib.request from pathlib import Path def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base-url", default="http://127.0.0.1:18000") parser.add_argument("--token-file", required=True, type=Path) parser.add_argument("--output", required=True, type=Path) args = parser.parse_args() token = args.token_file.read_text(encoding="utf-8").strip() checks = [] def get(path, expected=0): req = urllib.request.Request(args.base_url + path, headers={"x-token": token}) with urllib.request.urlopen(req, timeout=30) as response: result = json.load(response) item = {"path": path, "http": response.status, "code": result.get("code"), "msg": result.get("msg"), "passed": response.status == 200 and result.get("code") == expected} data = result.get("data", {}) if isinstance(data, dict) and "available" in data: item["available"] = data["available"] checks.append(item) return data channels = get("/sw/promote/list?pageNum=1&pageSize=1") if channels.get("available") is False: get("/sw/promote/portrait?sourceDb=lx&promoteId=1&period=total", expected=7) elif channels.get("rows"): row = channels["rows"][0] get("/sw/promote/portrait?" + urllib.parse.urlencode({"sourceDb": row["sourceDb"], "promoteId": row["id"], "period": "today"})) get("/sw/promote/portrait?sourceDb=lx&promoteId=0", expected=7) get("/sw/promote/portrait?sourceDb=invalid&promoteId=1", expected=7) report = {"scope": "当前 sw7 登录会话,只读检查;完整累计、分层、日期与权限在隔离 TCP HTTP 中验证。", "checks": checks, "passed": all(item["passed"] for item in checks)} args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps(report, ensure_ascii=False)) if not report["passed"]: raise SystemExit(1) if __name__ == "__main__": main()