# -*- coding: utf-8 -*- import argparse import copy import json import os import re from dataclasses import asdict, dataclass, field from difflib import SequenceMatcher from typing import Any, Callable, Dict, Iterable, List, Optional, Union COMMON_CN_KEYWORDS = [ "登录", "查询", "搜索", "筛选", "列表", "详情", "新增", "创建", "添加", "修改", "编辑", "更新", "删除", "导出", "导入", "上传", "下载", "提交", "审核", "确认", "保存", "分页", "分页查询", "性能", "响应时间", "超时", "返回", "结果", "成功", "失败", "校验", "重置", "启用", "停用", "开关", "状态", "排序", "标签", "分类", "关键字", "关键字搜索", "模糊搜索", ] METHOD_INTENT_KEYWORDS = { "GET": ["查询", "搜索", "筛选", "列表", "详情", "获取", "查看", "分页", "返回"], "POST": ["新增", "创建", "添加", "登录", "提交", "导入", "上传", "确认", "审核", "保存"], "PUT": ["修改", "编辑", "更新", "重置"], "PATCH": ["修改", "编辑", "更新", "状态", "启用", "停用", "开关"], "DELETE": ["删除", "移除"], } DEFAULT_PLACEHOLDER_URL = "/api/need-confirm" DEFAULT_PAGE_SIZE = 10 DEFAULT_RESPONSE_TIME_MS = 1000 @dataclass class FunctionalStep: stepNo: int action: str expectedResult: str = "" @dataclass class ApiContextItem: apiName: str = "" method: str = "" url: str = "" description: str = "" headers: Dict[str, Any] = field(default_factory=dict) queryParams: Dict[str, Any] = field(default_factory=dict) requestBody: Any = None responseExample: Any = None tags: List[str] = field(default_factory=list) @dataclass class FunctionalCaseInput: caseId: str = "" caseKey: str = "" caseName: str = "" projectName: str = "" moduleName: str = "" priority: str = "" preconditions: str = "" steps: List[FunctionalStep] = field(default_factory=list) expectedResults: List[str] = field(default_factory=list) apiContext: Optional[List[ApiContextItem]] = None generateOptions: Dict[str, Any] = field(default_factory=dict) extra: Dict[str, Any] = field(default_factory=dict) class FunctionalCaseParser: @classmethod def parse(cls, payload: Union[str, Dict[str, Any], FunctionalCaseInput]) -> FunctionalCaseInput: if isinstance(payload, FunctionalCaseInput): return payload if isinstance(payload, str): try: payload = json.loads(payload) except Exception: payload = { "caseName": "手工输入功能用例", "steps": [payload], "expectedResults": [], } if not isinstance(payload, dict): raise TypeError("functional case payload must be dict, json string or FunctionalCaseInput") steps = cls._normalize_steps(payload.get("steps") or payload.get("step") or []) expected_results = cls._normalize_text_list( payload.get("expectedResults") or payload.get("expectedResult") or payload.get("expectations") or [] ) api_context = cls._normalize_api_context(payload.get("apiContext") or payload.get("apis") or payload.get("apiInfo")) generate_options = payload.get("generateOptions") or payload.get("options") or {} known_keys = { "caseId", "caseKey", "caseName", "projectName", "moduleName", "priority", "preconditions", "steps", "step", "expectedResults", "expectedResult", "expectations", "apiContext", "apis", "apiInfo", "generateOptions", "options", } extra = {key: value for key, value in payload.items() if key not in known_keys} return FunctionalCaseInput( caseId=str(payload.get("caseId", "") or ""), caseKey=str(payload.get("caseKey", "") or ""), caseName=str(payload.get("caseName", "") or payload.get("name", "") or ""), projectName=str(payload.get("projectName", "") or ""), moduleName=str(payload.get("moduleName", "") or ""), priority=str(payload.get("priority", "") or ""), preconditions=str(payload.get("preconditions", "") or payload.get("precondition", "") or ""), steps=steps, expectedResults=expected_results, apiContext=api_context, generateOptions=generate_options if isinstance(generate_options, dict) else {}, extra=extra, ) @staticmethod def _normalize_steps(value: Any) -> List[FunctionalStep]: if not value: return [] if isinstance(value, str): raw_items = [item.strip() for item in re.split(r"[\n\r]+", value) if item.strip()] if len(raw_items) <= 1: raw_items = [item.strip() for item in re.split(r"(?<=[。；;])", value) if item.strip()] items = raw_items elif isinstance(value, list): items = value else: items = [value] normalized: List[FunctionalStep] = [] for index, item in enumerate(items, start=1): if isinstance(item, dict): step_no = int(item.get("stepNo") or item.get("step_no") or item.get("no") or index) action = str(item.get("action") or item.get("step") or item.get("content") or item.get("description") or "") expected_result = str(item.get("expectedResult") or item.get("expected") or "") else: raw_text = str(item).strip() step_no = index action = re.sub(r"^\s*\d+[\.、\)]\s*", "", raw_text) expected_result = "" if action: normalized.append(FunctionalStep(stepNo=step_no, action=action, expectedResult=expected_result)) return normalized @staticmethod def _normalize_text_list(value: Any) -> List[str]: if not value: return [] if isinstance(value, str): parts = [item.strip() for item in re.split(r"[\n\r]+", value) if item.strip()] if len(parts) <= 1: parts = [item.strip() for item in re.split(r"[。；;]\s*", value) if item.strip()] return [item for item in parts if item] if isinstance(value, list): result = [] for item in value: if isinstance(item, dict): text = str(item.get("text") or item.get("content") or item.get("expected") or "").strip() else: text = str(item).strip() if text: result.append(text) return result return [str(value).strip()] if str(value).strip() else [] @staticmethod def _normalize_api_context(value: Any) -> Optional[List[ApiContextItem]]: if not value: return None if isinstance(value, dict): if "apis" in value and isinstance(value["apis"], list): value = value["apis"] elif "items" in value and isinstance(value["items"], list): value = value["items"] else: value = [value] if isinstance(value, str): value = value.strip() if not value: return None try: parsed = json.loads(value) return FunctionalCaseParser._normalize_api_context(parsed) except Exception: return FunctionalCaseParser._parse_markdown_api_context(value) if not isinstance(value, list): return None apis: List[ApiContextItem] = [] for item in value: if isinstance(item, ApiContextItem): apis.append(item) continue if not isinstance(item, dict): continue headers = item.get("headers") or item.get("header") or {} query_params = item.get("queryParams") or item.get("query") or item.get("params") or {} tags = item.get("tags") or [] if isinstance(tags, str): tags = [tag.strip() for tag in re.split(r"[\s,，/|]+", tags) if tag.strip()] apis.append( ApiContextItem( apiName=str(item.get("apiName") or item.get("name") or ""), method=str(item.get("method") or item.get("httpMethod") or "").upper(), url=str(item.get("url") or item.get("path") or ""), description=str(item.get("description") or item.get("desc") or ""), headers=headers if isinstance(headers, dict) else {}, queryParams=query_params if isinstance(query_params, dict) else {}, requestBody=item.get("requestBody") if "requestBody" in item else item.get("body"), responseExample=item.get("responseExample") or item.get("response") or item.get("example"), tags=tags if isinstance(tags, list) else [], ) ) return apis or None @staticmethod def _parse_markdown_api_context(text: str) -> Optional[List[ApiContextItem]]: lines = text.splitlines() apis: List[ApiContextItem] = [] current: Dict[str, Any] = {} in_body = False body_lines: List[str] = [] def flush_current(): nonlocal current, body_lines, in_body if not current: return body_text = "\n".join(body_lines).strip() request_body = FunctionalCaseParser._safe_json_load(body_text) apis.append( ApiContextItem( apiName=current.get("apiName", ""), method=current.get("method", ""), url=current.get("url", ""), description=current.get("description", ""), headers=current.get("headers", {}), queryParams=current.get("queryParams", {}), requestBody=request_body if request_body is not None else body_text, responseExample=current.get("responseExample"), tags=current.get("tags", []), ) ) current = {} body_lines = [] in_body = False for line in lines: stripped = line.strip() if stripped.startswith("## "): flush_current() current["apiName"] = stripped[3:].strip() continue if stripped == "**接口URL**": continue if stripped.startswith("> ") and not current.get("url"): value = stripped[2:].strip() if value and value != "暂无参数": current["url"] = value continue if stripped == "**请求方式**": continue if stripped == "**请求Body参数**": in_body = True body_lines = [] continue if stripped == "**响应示例**": in_body = False continue if in_body: if stripped.startswith("```"): continue body_lines.append(line) flush_current() return apis or None @staticmethod def _safe_json_load(text: str) -> Any: if not text: return None cleaned = text.strip() if cleaned.startswith("```"): cleaned = re.sub(r"^```[a-zA-Z0-9_-]*\s*", "", cleaned) cleaned = re.sub(r"\s*```$", "", cleaned) try: return json.loads(cleaned) except Exception: return None class ApiMatcher: def match(self, functional_case: FunctionalCaseInput, candidates: List[ApiContextItem], top_k: int = 3) -> List[Dict[str, Any]]: matched: List[Dict[str, Any]] = [] keywords = self.extract_keywords(functional_case) intent = self.detect_intent(functional_case) for api in candidates: score = self._score_api(api, keywords, intent, functional_case) if score <= 0: continue matched.append( { "score": round(score, 4), "api": api, "reason": self._build_reason(api, keywords, intent), } ) matched.sort(key=lambda item: item["score"], reverse=True) return matched[:top_k] def extract_keywords(self, functional_case: FunctionalCaseInput) -> List[str]: texts = [functional_case.caseName, functional_case.moduleName, functional_case.preconditions] texts.extend(step.action for step in functional_case.steps) texts.extend(step.expectedResult for step in functional_case.steps if step.expectedResult) texts.extend(functional_case.expectedResults) texts.extend(str(value) for value in functional_case.extra.values() if value is not None) full_text = " \n ".join(texts) keywords: List[str] = [] for keyword in COMMON_CN_KEYWORDS: if keyword in full_text and keyword not in keywords: keywords.append(keyword) english_tokens = re.findall(r"[A-Za-z_][A-Za-z0-9_]{1,}", full_text) for token in english_tokens: if token not in keywords: keywords.append(token) generic_tokens = re.findall(r"[\u4e00-\u9fa5]{2,}", full_text) for token in generic_tokens: if len(token) <= 2: if token not in keywords: keywords.append(token) continue if token not in keywords: keywords.append(token) return self._dedupe(keywords) def detect_intent(self, functional_case: FunctionalCaseInput) -> str: text = self._all_text(functional_case) for method, keywords in METHOD_INTENT_KEYWORDS.items(): if any(keyword in text for keyword in keywords): return method return "GET" def _score_api(self, api: ApiContextItem, keywords: List[str], intent: str, functional_case: FunctionalCaseInput) -> float: score = 0.0 api_text = " ".join( [ api.apiName or "", api.method or "", api.url or "", api.description or "", " ".join(api.tags or []), " ".join(api.headers.keys()), " ".join(api.queryParams.keys()), ] ) if api.method and api.method.upper() == intent: score += 1.2 elif api.method and api.method.upper() in ("GET", "POST", "PUT", "PATCH", "DELETE"): score += 0.2 for keyword in keywords: if keyword and keyword in api_text: score += 0.8 for token in self._url_tokens(api.url): if token and any(keyword in token or token in keyword for keyword in keywords): score += 0.6 if functional_case.moduleName and functional_case.moduleName in api_text: score += 0.8 if functional_case.caseName and functional_case.caseName in api_text: score += 0.5 ratio = SequenceMatcher(None, self._all_text(functional_case), api_text).ratio() score += ratio * 0.8 return score @staticmethod def _build_reason(api: ApiContextItem, keywords: List[str], intent: str) -> str: matched_keywords = [keyword for keyword in keywords if keyword and keyword in (api.apiName + " " + api.description + " " + api.url)] parts = [] if api.method and api.method.upper() == intent: parts.append("方法匹配") if matched_keywords: parts.append("关键词命中：{}".format("、".join(matched_keywords[:5]))) return "; ".join(parts) or "模糊匹配" @staticmethod def _url_tokens(url: str) -> List[str]: if not url: return [] return [token for token in re.split(r"[/?&=_\-\.]+", url) if token] @staticmethod def _dedupe(items: Iterable[str]) -> List[str]: seen = set() result = [] for item in items: if item and item not in seen: seen.add(item) result.append(item) return result @staticmethod def _all_text(functional_case: FunctionalCaseInput) -> str: texts = [functional_case.caseName, functional_case.moduleName, functional_case.preconditions] texts.extend(step.action for step in functional_case.steps) texts.extend(step.expectedResult for step in functional_case.steps if step.expectedResult) texts.extend(functional_case.expectedResults) return " \n ".join(filter(None, texts)) class LLMApiCaseConverter: def convert(self, prompt: str, llm_client: Any = None) -> Optional[Dict[str, Any]]: if llm_client is None: return None raw = self._invoke_client(llm_client, prompt) if raw is None: return None if isinstance(raw, dict): return raw if hasattr(raw, "content"): raw = raw.content elif hasattr(raw, "text"): raw = raw.text elif hasattr(raw, "data"): raw = raw.data if isinstance(raw, dict): return raw if not isinstance(raw, str): raw = str(raw) return self._extract_json(raw) @staticmethod def _invoke_client(llm_client: Any, prompt: str) -> Any: if callable(llm_client): try: return llm_client(prompt=prompt) except TypeError: try: return llm_client(prompt) except TypeError: return llm_client({"prompt": prompt}) if hasattr(llm_client, "generate") and callable(llm_client.generate): return llm_client.generate(prompt) if hasattr(llm_client, "chat") and callable(llm_client.chat): return llm_client.chat(prompt) raise TypeError("llm_client must be callable or expose generate/chat") @staticmethod def _extract_json(text: str) -> Optional[Dict[str, Any]]: if not text: return None cleaned = text.strip() match = re.search(r"```json\s*(\{.*?\})\s*```", cleaned, flags=re.S) if match: cleaned = match.group(1) else: first = cleaned.find("{") last = cleaned.rfind("}") if first >= 0 and last > first: cleaned = cleaned[first : last + 1] try: parsed = json.loads(cleaned) return parsed if isinstance(parsed, dict) else {"data": parsed} except Exception: return None class FunctionalCaseToApiAutomationService: def __init__(self, matcher: Optional[ApiMatcher] = None): self.matcher = matcher or ApiMatcher() self.llm_converter = LLMApiCaseConverter() def convert(self, payload: Union[str, Dict[str, Any], FunctionalCaseInput], llm_client: Any = None) -> Dict[str, Any]: functional_case = FunctionalCaseParser.parse(payload) options = self._build_options(functional_case) candidates = functional_case.apiContext or [] matched = self.matcher.match(functional_case, candidates, top_k=int(options.get("topK", 3))) if candidates else [] llm_result = None if options.get("useLLM", True) and llm_client is not None: prompt = self.build_prompt(functional_case, matched, options) llm_result = self.llm_converter.convert(prompt, llm_client) if llm_result: return self._finalize_llm_result(functional_case, matched, llm_result, options) return self._build_rule_result(functional_case, matched, options) def build_prompt(self, functional_case: FunctionalCaseInput, matched: List[Dict[str, Any]], options: Dict[str, Any]) -> str: payload = { "functionalCase": asdict(functional_case), "matchedApis": [ { "score": item["score"], "reason": item["reason"], "api": asdict(item["api"]), } for item in matched ], "options": options, } return ( "你是接口自动化测试用例生成专家。\n" "请把功能测试用例转换成接口自动化测试用例。\n" "要求：\n" "1. 只输出严格 JSON。\n" "2. 如果给了候选接口信息，优先使用候选接口。\n" "3. 无法确认的接口信息写入 missingInfo。\n" "4. 需要包含 method、url、headers、queryParams、body、assertions、performanceAssertions。\n" "5. 如果涉及性能要求，生成 responseTime 断言。\n" "6. 如果涉及登录态，使用 ${token}。\n\n" "输入数据：\n" f"{json.dumps(payload, ensure_ascii=False, indent=2)}\n\n" "输出 JSON 结构：\n" "{\n" ' "caseId": "",\n' ' "caseKey": "",\n' ' "caseName": "",\n' ' "automationType": "api",\n' ' "convertStatus": "SUCCESS",\n' ' "apiTestCases": [],\n' ' "missingInfo": [],\n' ' "warnings": []\n' "}" ) def _build_options(self, functional_case: FunctionalCaseInput) -> Dict[str, Any]: options = copy.deepcopy(functional_case.generateOptions or {}) options.setdefault("useLLM", True) options.setdefault("outputFormat", "json") options.setdefault("targetFramework", "pytest") options.setdefault("allowMissingInfo", True) options.setdefault("topK", 3) return options def _build_rule_result(self, functional_case: FunctionalCaseInput, matched: List[Dict[str, Any]], options: Dict[str, Any]) -> Dict[str, Any]: api_test_cases: List[Dict[str, Any]] = [] missing_info: List[str] = [] warnings: List[str] = [] if matched: for index, item in enumerate(matched, start=1): api_test_cases.append(self._build_api_test_case(functional_case, item["api"], index, item["score"], options)) else: api_test_cases.append(self._build_fallback_test_case(functional_case, options)) missing_info.extend(self._collect_missing_info(functional_case)) warnings.append("未匹配到真实接口信息，已生成占位自动化用例") missing_info = self._dedupe_text(missing_info) warnings = self._dedupe_text(warnings) convert_status = "SUCCESS" if missing_info: convert_status = "SUCCESS_WITH_MISSING_INFO" if warnings and not matched: convert_status = "DRAFT" return { "caseId": functional_case.caseId, "caseKey": functional_case.caseKey, "caseName": functional_case.caseName, "projectName": functional_case.projectName, "moduleName": functional_case.moduleName, "automationType": "api", "convertStatus": convert_status, "apiTestCases": api_test_cases, "missingInfo": missing_info, "warnings": warnings, } def _finalize_llm_result(self, functional_case: FunctionalCaseInput, matched: List[Dict[str, Any]], llm_result: Dict[str, Any], options: Dict[str, Any]) -> Dict[str, Any]: result = copy.deepcopy(llm_result) result.setdefault("caseId", functional_case.caseId) result.setdefault("caseKey", functional_case.caseKey) result.setdefault("caseName", functional_case.caseName) result.setdefault("projectName", functional_case.projectName) result.setdefault("moduleName", functional_case.moduleName) result.setdefault("automationType", "api") result.setdefault("convertStatus", "SUCCESS") result.setdefault("apiTestCases", []) result.setdefault("missingInfo", []) result.setdefault("warnings", []) if not result.get("apiTestCases"): rule_result = self._build_rule_result(functional_case, matched, options) result["apiTestCases"] = rule_result["apiTestCases"] if not result.get("missingInfo"): result["missingInfo"] = rule_result["missingInfo"] if not result.get("warnings"): result["warnings"] = rule_result["warnings"] result["convertStatus"] = rule_result["convertStatus"] result["missingInfo"] = self._dedupe_text(result.get("missingInfo", [])) result["warnings"] = self._dedupe_text(result.get("warnings", [])) return result def _build_api_test_case(self, functional_case: FunctionalCaseInput, api: ApiContextItem, index: int, score: float, options: Dict[str, Any]) -> Dict[str, Any]: method = (api.method or self.matcher.detect_intent(functional_case)).upper() query_params = copy.deepcopy(api.queryParams or {}) body = copy.deepcopy(api.requestBody) headers = copy.deepcopy(api.headers or {}) if not headers: headers = {"Content-Type": "application/json"} if not any(key.lower() == "authorization" for key in headers): headers["Authorization"] = "Bearer ${token}" variables = self._build_variables(functional_case, api, query_params, body) query_params = self._fill_placeholders(query_params, variables) body = self._fill_placeholders(body, variables) assertions = self._build_assertions(functional_case, api, method) performance_assertions = self._build_performance_assertions(functional_case) step_name = api.apiName or functional_case.caseName or "接口自动化步骤" return { "stepNo": index, "name": step_name, "apiName": api.apiName or step_name, "method": method, "url": api.url or self._infer_placeholder_url(functional_case), "headers": headers, "queryParams": query_params, "body": body, "extractVariables": [], "assertions": assertions, "performanceAssertions": performance_assertions, "variables": variables, "matchedApi": { "apiName": api.apiName, "method": method, "url": api.url, "matchScore": round(score, 4), }, } def _build_fallback_test_case(self, functional_case: FunctionalCaseInput, options: Dict[str, Any]) -> Dict[str, Any]: method = self.matcher.detect_intent(functional_case) url = self._infer_placeholder_url(functional_case) variables = self._build_variables(functional_case, None, {}, None) return { "stepNo": 1, "name": functional_case.caseName or "功能用例转接口自动化", "apiName": functional_case.caseName or "待确认接口", "method": method, "url": url, "headers": { "Content-Type": "application/json", "Authorization": "Bearer ${token}", }, "queryParams": {}, "body": None, "extractVariables": [], "assertions": self._build_assertions(functional_case, None, method), "performanceAssertions": self._build_performance_assertions(functional_case), "variables": variables, "matchedApi": { "apiName": "", "method": method, "url": url, "matchScore": 0, }, } def _build_variables(self, functional_case: FunctionalCaseInput, api: Optional[ApiContextItem], query_params: Dict[str, Any], body: Any) -> Dict[str, Any]: variables: Dict[str, Any] = {} text = self.matcher._all_text(functional_case) for keyword in self.matcher.extract_keywords(functional_case): if keyword in ("搜索", "查询", "列表", "筛选", "关键字", "模糊搜索"): variables.setdefault("keyword", "测试") if keyword in ("分页", "分页查询"): variables.setdefault("page", 1) variables.setdefault("limit", DEFAULT_PAGE_SIZE) if keyword in ("登录",): variables.setdefault("token", "${token}") if not variables and text: variables["keyword"] = "测试" if query_params: for key in query_params.keys(): if key not in variables and any(token in key.lower() for token in ["keyword", "name", "id", "page", "limit", "status", "type"]): variables[key] = query_params[key] if isinstance(body, dict): for key in body.keys(): if key not in variables and any(token in key.lower() for token in ["keyword", "name", "id", "page", "limit", "status", "type"]): variables[key] = body[key] return variables def _build_assertions(self, functional_case: FunctionalCaseInput, api: Optional[ApiContextItem], method: str) -> List[Dict[str, Any]]: assertions = [ { "type": "statusCode", "actual": "$.code" if api or method != "DELETE" else "$.status", "operator": "==", "expected": 200, } ] text = self.matcher._all_text(functional_case) if api and isinstance(api.responseExample, dict): example = api.responseExample if "code" in example: assertions.append( { "type": "jsonPath", "actual": "$.code", "operator": "==", "expected": example.get("code", 0), } ) if "msg" in example: assertions.append( { "type": "jsonPath", "actual": "$.msg", "operator": "notEmpty", "expected": True, } ) if any(keyword in text for keyword in ["搜索", "查询", "列表", "筛选", "返回"]): assertions.append( { "type": "jsonPath", "actual": "$.data", "operator": "notEmpty", "expected": True, } ) return assertions def _build_performance_assertions(self, functional_case: FunctionalCaseInput) -> List[Dict[str, Any]]: text = self.matcher._all_text(functional_case) expected_ms = self._extract_response_time_ms(text) if expected_ms is None: return [] return [ { "type": "responseTime", "operator": "<=", "expected": expected_ms, "unit": "ms", } ] def _collect_missing_info(self, functional_case: FunctionalCaseInput) -> List[str]: missing = [] if not functional_case.steps: missing.append("缺少步骤信息") if not functional_case.caseName: missing.append("缺少用例名称") if not functional_case.apiContext: missing.append("未传入真实接口信息，已使用占位接口生成") if self._extract_response_time_ms(self.matcher._all_text(functional_case)) is None and any(k in self.matcher._all_text(functional_case) for k in ["性能", "响应时间", "超时"]): missing.append("性能阈值未明确，默认使用 1000ms") return missing @staticmethod def _extract_response_time_ms(text: str) -> Optional[int]: if not text: return None patterns = [ r"(\d+)\s*毫秒", r"(\d+)\s*ms", r"(\d+)\s*秒", r"不超过\s*(\d+)\s*秒", r"<=\s*(\d+)\s*秒", r"小于等于\s*(\d+)\s*秒", ] for pattern in patterns: match = re.search(pattern, text, flags=re.I) if match: value = int(match.group(1)) if "秒" in pattern: return value * 1000 return value if any(keyword in text for keyword in ["响应时间", "超时", "性能"]): return DEFAULT_RESPONSE_TIME_MS return None @staticmethod def _infer_placeholder_url(functional_case: FunctionalCaseInput) -> str: text = " ".join([functional_case.caseName, functional_case.moduleName] + [step.action for step in functional_case.steps]) if any(keyword in text for keyword in ["登录"]): return "/api/login" if any(keyword in text for keyword in ["搜索", "查询", "列表", "筛选"]): return "/api/list/query" if any(keyword in text for keyword in ["新增", "创建", "添加"]): return "/api/create" if any(keyword in text for keyword in ["修改", "编辑", "更新"]): return "/api/update" if any(keyword in text for keyword in ["删除", "移除"]): return "/api/delete" return DEFAULT_PLACEHOLDER_URL @staticmethod def _fill_placeholders(value: Any, variables: Dict[str, Any]) -> Any: if isinstance(value, dict): return {key: FunctionalCaseToApiAutomationService._fill_placeholders(item, variables) for key, item in value.items()} if isinstance(value, list): return [FunctionalCaseToApiAutomationService._fill_placeholders(item, variables) for item in value] if isinstance(value, str): result = value for key, variable in variables.items(): result = result.replace("${" + key + "}", str(variable)) return result return value @staticmethod def _dedupe_text(items: Iterable[str]) -> List[str]: seen = set() result = [] for item in items: text = str(item).strip() if not text or text in seen: continue seen.add(text) result.append(text) return result def convert_functional_case(payload: Union[str, Dict[str, Any], FunctionalCaseInput], llm_client: Any = None) -> Dict[str, Any]: return FunctionalCaseToApiAutomationService().convert(payload, llm_client=llm_client) def load_payload_from_file(file_path: str) -> Dict[str, Any]: with open(file_path, "r", encoding="utf-8") as file: return json.load(file) def main(): parser = argparse.ArgumentParser(description="功能测试用例转接口自动化用例") parser.add_argument("--input", type=str, help="输入JSON文件路径，未指定则从stdin读取") parser.add_argument("--output", type=str, help="输出JSON文件路径，未指定则打印到stdout") args = parser.parse_args() if args.input: payload = load_payload_from_file(args.input) else: payload = json.load(os.sys.stdin) result = convert_functional_case(payload) output = json.dumps(result, ensure_ascii=False, indent=2) if args.output: with open(args.output, "w", encoding="utf-8") as file: file.write(output) else: print(output) if __name__ == "__main__": main()