""" 禅道 AI 通道客户端（生产环境） 封装三个接口： submit 接口一：评估结果提交 POST /zt-story-expand/saveOrUpdate batch-add-tasks 接口二：AI批量创建任务 POST /zt-task/aiBatchAdd upload 接口三：文件上传并绑定 POST /common/uploadBind (multipart) 用法示例： python zentao_client.py submit --story-id 9130 --number-units 3 --b 2.2 --ft 1.4 --ga 0.55 --w 5.1 --status finished python zentao_client.py batch-add-tasks --story-id 9130 --tasks-json "[{\"name\":\"后端接口开发\",\"type\":\"devel\"}]" python zentao_client.py upload --file 报告.md --object-type aiCodeReview --object-id 9130 --review-result pass 环境变量覆盖： ZENTAO_AI_TOKEN 访问令牌（默认使用内置的 ai 账号永久令牌） ZENTAO_BASE_URL 服务地址（默认生产环境） """ import argparse import json import os import sys import requests BASE_URL = os.environ.get("ZENTAO_BASE_URL", "https://itsm.sino-assist.com/zentao") # ai 账号永久令牌（来自《禅道AI通道接口文档_v1.0》，注意保管、勿外泄） DEFAULT_TOKEN = ("eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9." "eyJhY2NvdW50IjoiYWkiLCJwYXNzd29yZCI6ImUxMGFkYzM5NDliYTU5YWJiZTU2ZTA1N2YyMGY4ODNlIiwidXNlclR5cGUiOjN9." "d43vA9pN_dhwwfeO0zg3_AaX76vEMXRYFZSc6AkYF7c") TOKEN = os.environ.get("ZENTAO_AI_TOKEN", DEFAULT_TOKEN) TIMEOUT = 30 def _headers(): # 直接放 token 原文，不要加 "Bearer " 前缀 return {"Authorization": TOKEN} def _print_result(resp): data = resp.json() msg = data.get("message", "") if isinstance(msg, str): try: msg = msg.encode("latin-1").decode("utf-8") except Exception: pass print(f"HTTP {resp.status_code} | code={data.get('code')} | message={msg}") if data.get("data") is not None: print(json.dumps(data["data"], ensure_ascii=False, indent=2)) return data def submit(args): """接口一：评估结果提交（同一需求重复提交为覆盖更新，幂等）""" payload = { "storyId": args.story_id, "numberUnits": args.number_units, "unitBusinessComplexity": str(args.b), "technicalComplexityCoefficient": str(args.ft), "aiEfficiencyCoefficient": str(args.ga), "evaluationTime": args.w, "workloadIndex": str(args.w), "requirementStatus": args.status, } if args.completion_degree is not None: payload["requirementCompletionDegree"] = args.completion_degree acceptance = args.acceptance_criteria if args.acceptance_criteria_file: with open(args.acceptance_criteria_file, encoding="utf-8") as f: acceptance = f.read() if acceptance: payload["acceptanceCriteria"] = acceptance for attr, key in [("product_person", "productPerson"), ("develop_person", "developPerson"), ("test_person", "testPerson")]: value = getattr(args, attr) if value is not None: payload[key] = value resp = requests.post(f"{BASE_URL}/zt-story-expand/saveOrUpdate", json=payload, headers=_headers(), timeout=TIMEOUT) return _print_result(resp) def batch_add_tasks(args): """接口二：AI 批量创建任务（整批拒绝校验；同名同类型任务自动跳过防重）""" tasks = json.loads(args.tasks_json) if not isinstance(tasks, list) or not tasks: print("错误：--tasks-json 必须是非空 JSON 数组", file=sys.stderr) sys.exit(2) payload = {"storyId": args.story_id, "tasks": tasks} resp = requests.post(f"{BASE_URL}/zt-task/aiBatchAdd", json=payload, headers=_headers(), timeout=TIMEOUT) return _print_result(resp) def upload(args): """接口三：文件上传并绑定（multipart/form-data）""" form = {"objectType": args.object_type, "objectId": str(args.object_id)} if args.title: form["title"] = args.title if args.review_result: form["reviewResult"] = args.review_result with open(args.file, "rb") as f: files = {"file": (os.path.basename(args.file), f)} resp = requests.post(f"{BASE_URL}/common/uploadBind", data=form, files=files, headers=_headers(), timeout=TIMEOUT) return _print_result(resp) def main(): parser = argparse.ArgumentParser(description="禅道 AI 通道客户端（生产环境）") sub = parser.add_subparsers(dest="command", required=True) p = sub.add_parser("submit", help="接口一：评估结果提交 saveOrUpdate") p.add_argument("--story-id", type=int, required=True, help="需求ID") p.add_argument("--number-units", type=int, required=True, help="功能单元数量 S") p.add_argument("--b", type=float, required=True, help="单元业务复杂度 B") p.add_argument("--ft", type=float, required=True, help="技术复杂度系数 F(T)") p.add_argument("--ga", type=float, required=True, help="AI效率系数 G(A)") p.add_argument("--w", type=float, required=True, help="评估工时/工作量指数 W（人日）") p.add_argument("--status", required=True, choices=["inProgress", "finished"], help="完成状态；finished 将锁定记录并结算当月工作量") p.add_argument("--completion-degree", default=None, help='完成度 "0"~"100"（finished 时服务端自动置 100）') p.add_argument("--acceptance-criteria", default=None, help="验收标准（Markdown 文本）") p.add_argument("--acceptance-criteria-file", default=None, help="验收标准 Markdown 文件路径（优先于 --acceptance-criteria）") p.add_argument("--product-person", default=None, help="产品人员（中文名）") p.add_argument("--develop-person", default=None, help="开发人员（中文名）") p.add_argument("--test-person", default=None, help="测试人员（中文名）") p.set_defaults(func=submit) p = sub.add_parser("batch-add-tasks", help="接口二：AI 批量创建任务 aiBatchAdd") p.add_argument("--story-id", type=int, required=True, help="需求ID") p.add_argument("--tasks-json", required=True, help='任务列表 JSON 数组，如 [{"name":"后端接口开发","type":"devel",' '"desc":"实现 xx 接口","assignedTo":"zhangsan","aiEvaluationTime":8,' '"planStartDate":"2026-08-10","deadline":"2026-08-11"}]；' 'type 仅支持 devel/test；desc 为可选任务描述') p.set_defaults(func=batch_add_tasks) p = sub.add_parser("upload", help="接口三：文件上传并绑定 uploadBind") p.add_argument("--file", required=True, help="上传文件路径") p.add_argument("--object-type", required=True, choices=["story", "aiCodeReview", "aiWorkLog", "aiDocUpdate", "testCase", "testReport", "testReportSubmit", "testOther", "meeting"], help="业务对象类型") p.add_argument("--object-id", type=int, required=True, help="业务对象 ID（需求 ID 或会议 ID）") p.add_argument("--title", default=None, help="文件标题（默认取原始文件名，支持中文）") p.add_argument("--review-result", default=None, choices=["pass", "reject"], help="仅 object-type=aiCodeReview 时有效：pass=审查通过 / reject=审查不通过") p.set_defaults(func=upload) args = parser.parse_args() args.func(args) if __name__ == "__main__": main()