Coverage for utils/docs/generate_api_docs.py: 0.00%

215 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2025-10-13 18:58 +0000

1#!/usr/bin/env python3 

2""" 

3生成分模块的API Markdown文档 

4从OpenAPI规范生成结构化的Markdown文档 

5""" 

6 

7import json 

8import sys 

9from collections import defaultdict 

10from pathlib import Path 

11from typing import Any, Dict, List 

12 

13# 模块分类映射(英文名称) 

14MODULE_MAPPING = { 

15 "authentication": ["login", "token", "auth"], 

16 "users": ["users", "user"], 

17 "brokers": ["brokers", "broker", "data-sources"], 

18 "trading": ["sessions", "orders", "positions"], 

19 "assets": ["assets", "balance"], 

20 "stocks": ["stocks", "stock"], 

21 "strategies": ["strategies", "strategy"], 

22 "backtest": ["backtest"], 

23 "risk": ["risk"], 

24 "settings": ["settings", "config"], 

25 "api-test": ["api-test"], 

26 "websocket": ["ws"], 

27 "trade-test": ["trade-test"], 

28} 

29 

30# 模块中文名称映射(用于显示) 

31MODULE_CN_NAMES = { 

32 "authentication": "认证", 

33 "users": "用户管理", 

34 "brokers": "券商管理", 

35 "trading": "交易管理", 

36 "assets": "资产管理", 

37 "stocks": "股票数据", 

38 "strategies": "策略管理", 

39 "backtest": "回测", 

40 "risk": "风险管理", 

41 "settings": "系统配置", 

42 "api-test": "API测试", 

43 "websocket": "WebSocket", 

44 "trade-test": "交易测试", 

45 "others": "其他", 

46} 

47 

48# HTTP方法的中文映射 

49METHOD_CN = { 

50 "get": "查询", 

51 "post": "创建", 

52 "put": "更新", 

53 "delete": "删除", 

54 "patch": "修改", 

55} 

56 

57# HTTP方法对应的emoji 

58METHOD_EMOJI = {"get": "🔍", "post": "➕", "put": "✏️", "delete": "🗑️", "patch": "📝"} 

59 

60 

61def categorize_path(path: str) -> str: 

62 """根据路径将端点分类到模块""" 

63 path_lower = path.lower() 

64 

65 for module, keywords in MODULE_MAPPING.items(): 

66 for keyword in keywords: 

67 if keyword in path_lower: 

68 return module 

69 

70 # 默认分类 

71 if "/sessions" in path_lower or "/orders" in path_lower: 

72 return "trading" 

73 return "others" 

74 

75 

76def format_parameter(param: Dict[str, Any]) -> str: 

77 """格式化参数信息""" 

78 name = param.get("name", "") 

79 param_type = param.get("schema", {}).get("type", "string") 

80 required = param.get("required", False) 

81 description = param.get("description", "") 

82 

83 required_badge = "**必填**" if required else "可选" 

84 

85 return f"- `{name}` ({param_type}) - {required_badge} - {description}" 

86 

87 

88def format_schema( 

89 schema: Dict[str, Any], schemas: Dict[str, Any], indent: int = 0 

90) -> str: 

91 """递归格式化schema为Markdown""" 

92 result = [] 

93 prefix = " " * indent 

94 

95 if "$ref" in schema: 

96 # 引用类型 

97 ref_name = schema["$ref"].split("/")[-1] 

98 if ref_name in schemas: 

99 return format_schema(schemas[ref_name], schemas, indent) 

100 return f"{prefix}- 引用: `{ref_name}`\n" 

101 

102 if "properties" in schema: 

103 properties = schema["properties"] 

104 required = schema.get("required", []) 

105 

106 for prop_name, prop_schema in properties.items(): 

107 is_required = "**必填**" if prop_name in required else "可选" 

108 prop_type = prop_schema.get("type", "object") 

109 description = prop_schema.get("description", "") 

110 

111 result.append( 

112 f"{prefix}- `{prop_name}` ({prop_type}) - {is_required} - {description}" 

113 ) 

114 

115 # 如果有子属性,递归处理 

116 if "properties" in prop_schema: 

117 result.append(format_schema(prop_schema, schemas, indent + 1)) 

118 elif "$ref" in prop_schema: 

119 ref_name = prop_schema["$ref"].split("/")[-1] 

120 result.append(f"{prefix} - 类型: `{ref_name}`") 

121 

122 return "\n".join(result) 

123 

124 

125def generate_endpoint_doc( 

126 path: str, method: str, operation: Dict[str, Any], schemas: Dict[str, Any] 

127) -> str: 

128 """生成单个端点的文档""" 

129 summary = operation.get("summary", "") 

130 description = operation.get("description", "") 

131 operation_id = operation.get("operationId", "") 

132 

133 method_upper = method.upper() 

134 emoji = METHOD_EMOJI.get(method, "📌") 

135 

136 doc = f"### {emoji} {summary}\n\n" 

137 

138 if description and description != summary: 

139 doc += f"{description}\n\n" 

140 

141 doc += f"**请求方式**: `{method_upper} {path}`\n\n" 

142 

143 # 路径参数 

144 path_params = [p for p in operation.get("parameters", []) if p.get("in") == "path"] 

145 if path_params: 

146 doc += "**路径参数**:\n\n" 

147 for param in path_params: 

148 doc += format_parameter(param) + "\n" 

149 doc += "\n" 

150 

151 # 查询参数 

152 query_params = [ 

153 p for p in operation.get("parameters", []) if p.get("in") == "query" 

154 ] 

155 if query_params: 

156 doc += "**查询参数**:\n\n" 

157 for param in query_params: 

158 doc += format_parameter(param) + "\n" 

159 doc += "\n" 

160 

161 # 请求体 

162 request_body = operation.get("requestBody") 

163 if request_body: 

164 content = request_body.get("content", {}) 

165 json_content = content.get("application/json", {}) 

166 schema = json_content.get("schema", {}) 

167 

168 doc += "**请求体**:\n\n" 

169 doc += "```json\n" 

170 if "$ref" in schema: 

171 ref_name = schema["$ref"].split("/")[-1] 

172 doc += f"// 参考模型: {ref_name}\n" 

173 if ref_name in schemas: 

174 doc += format_schema(schemas[ref_name], schemas) 

175 else: 

176 doc += format_schema(schema, schemas) 

177 doc += "\n```\n\n" 

178 

179 # 响应 

180 responses = operation.get("responses", {}) 

181 if "200" in responses or "201" in responses: 

182 response = responses.get("200") or responses.get("201") 

183 content = response.get("content", {}) 

184 json_content = content.get("application/json", {}) 

185 schema = json_content.get("schema", {}) 

186 

187 doc += "**响应**:\n\n" 

188 doc += "```json\n" 

189 if "$ref" in schema: 

190 ref_name = schema["$ref"].split("/")[-1] 

191 doc += f"// 参考模型: {ref_name}\n" 

192 if ref_name in schemas: 

193 doc += format_schema(schemas[ref_name], schemas) 

194 elif "items" in schema and "$ref" in schema["items"]: 

195 ref_name = schema["items"]["$ref"].split("/")[-1] 

196 doc += f"// 数组: {ref_name}[]\n" 

197 if ref_name in schemas: 

198 doc += format_schema(schemas[ref_name], schemas) 

199 else: 

200 doc += format_schema(schema, schemas) 

201 doc += "\n```\n\n" 

202 

203 # 错误响应 

204 error_responses = {k: v for k, v in responses.items() if k not in ["200", "201"]} 

205 if error_responses: 

206 doc += "**错误响应**:\n\n" 

207 for code, response in error_responses.items(): 

208 description = response.get("description", "") 

209 doc += f"- `{code}`: {description}\n" 

210 doc += "\n" 

211 

212 doc += "---\n\n" 

213 return doc 

214 

215 

216def generate_module_docs(openapi_spec: Dict[str, Any], output_dir: Path): 

217 """生成分模块的API文档""" 

218 paths = openapi_spec.get("paths", {}) 

219 schemas = openapi_spec.get("components", {}).get("schemas", {}) 

220 

221 # 按模块分组端点 

222 modules = defaultdict(list) 

223 

224 for path, methods in paths.items(): 

225 module = categorize_path(path) 

226 for method, operation in methods.items(): 

227 if method in ["get", "post", "put", "delete", "patch"]: 

228 modules[module].append( 

229 {"path": path, "method": method, "operation": operation} 

230 ) 

231 

232 # 为每个模块生成文档 

233 for module_name, endpoints in modules.items(): 

234 # 获取中文名称 

235 module_cn_name = MODULE_CN_NAMES.get(module_name, module_name) 

236 

237 # 创建模块目录(使用英文名) 

238 module_dir = output_dir / "v1" / module_name 

239 module_dir.mkdir(parents=True, exist_ok=True) 

240 

241 # 生成README.md 

242 readme_path = module_dir / "README.md" 

243 

244 with open(readme_path, "w", encoding="utf-8") as f: 

245 f.write(f"# {module_cn_name} API\n\n") 

246 f.write(f"**模块**: `{module_name}`\n\n") 

247 f.write(f"本模块包含 {len(endpoints)} 个API端点。\n\n") 

248 

249 # 端点概览 

250 f.write("## 📋 端点列表\n\n") 

251 for endpoint in endpoints: 

252 method = endpoint["method"].upper() 

253 path = endpoint["path"] 

254 summary = endpoint["operation"].get("summary", "") 

255 emoji = METHOD_EMOJI.get(endpoint["method"], "📌") 

256 f.write(f"- {emoji} `{method} {path}` - {summary}\n") 

257 f.write("\n---\n\n") 

258 

259 # 详细文档 

260 f.write("## 📖 详细说明\n\n") 

261 for endpoint in endpoints: 

262 doc = generate_endpoint_doc( 

263 endpoint["path"], endpoint["method"], endpoint["operation"], schemas 

264 ) 

265 f.write(doc) 

266 

267 print(f"✅ 生成模块文档: {module_cn_name} ({module_name}) -> {readme_path}") 

268 

269 return modules 

270 

271 

272def generate_api_overview( 

273 modules: Dict[str, List], output_dir: Path, info: Dict[str, Any] 

274): 

275 """生成API概览文档""" 

276 readme_path = output_dir / "README.md" 

277 

278 with open(readme_path, "w", encoding="utf-8") as f: 

279 f.write("# API 文档\n\n") 

280 f.write(f"**版本**: {info.get('version', '1.0.0')}\n\n") 

281 f.write(f"{info.get('description', '')}\n\n") 

282 

283 f.write("## 🗂️ 模块组织\n\n") 

284 f.write("API按功能模块组织,每个模块都有独立的文档目录:\n\n") 

285 

286 # 模块统计 

287 total_endpoints = sum(len(endpoints) for endpoints in modules.values()) 

288 

289 for module_name, endpoints in sorted(modules.items()): 

290 # 获取中文名称 

291 module_cn_name = MODULE_CN_NAMES.get(module_name, module_name) 

292 

293 f.write(f"### 📁 [{module_cn_name}](./v1/{module_name}/README.md)\n\n") 

294 f.write( 

295 f"**模块**: `{module_name}` | **端点数量**: {len(endpoints)} 个\n\n" 

296 ) 

297 

298 # 列出主要端点 

299 for endpoint in endpoints[:5]: # 只显示前5个 

300 method = endpoint["method"].upper() 

301 path = endpoint["path"] 

302 summary = endpoint["operation"].get("summary", "") 

303 emoji = METHOD_EMOJI.get(endpoint["method"], "📌") 

304 f.write(f"- {emoji} `{method} {path}` - {summary}\n") 

305 

306 if len(endpoints) > 5: 

307 f.write(f"- ... 还有 {len(endpoints) - 5} 个端点\n") 

308 f.write("\n") 

309 

310 f.write(f"\n**总计**: {len(modules)} 个模块, {total_endpoints} 个端点\n\n") 

311 

312 f.write("---\n\n") 

313 f.write("## 🚀 快速开始\n\n") 

314 f.write("### 认证\n\n") 

315 f.write("所有API请求都需要JWT认证(除了登录端点)。\n\n") 

316 f.write("```bash\n") 

317 f.write("# 1. 登录获取token\n") 

318 f.write('curl -X POST "http://localhost:8000/api/v1/login" \\\n') 

319 f.write(' -H "Content-Type: application/json" \\\n') 

320 f.write( 

321 ' -d \'{"username": "your_username", "password": "your_password"}\'\n\n' 

322 ) 

323 f.write("# 2. 使用token访问API\n") 

324 f.write('curl -X GET "http://localhost:8000/api/v1/..." \\\n') 

325 f.write(' -H "Authorization: Bearer YOUR_JWT_TOKEN"\n') 

326 f.write("```\n\n") 

327 

328 f.write("### 基础URL\n\n") 

329 f.write("- **开发环境**: `http://localhost:8000`\n") 

330 f.write("- **API版本**: `v1`\n") 

331 f.write("- **基础路径**: `/api/v1`\n\n") 

332 

333 f.write("---\n\n") 

334 f.write("## 📚 相关文档\n\n") 

335 f.write("- [OpenAPI规范](./openapi.json) - 完整的API规范\n") 

336 f.write("- [开发指南](../development.md) - 后端开发指南\n") 

337 f.write("- [架构文档](../architecture/) - 系统架构文档\n\n") 

338 

339 f.write("---\n\n") 

340 f.write("**生成时间**: 由 `scripts/generate_docs.sh` 自动生成\n") 

341 

342 print(f"✅ 生成API概览: {readme_path}") 

343 

344 

345def main(): 

346 """主函数""" 

347 # 检查参数 

348 if len(sys.argv) < 2: 

349 print("❌ 用法: python generate_api_docs.py <openapi.json路径> [输出目录]") 

350 sys.exit(1) 

351 

352 openapi_file = Path(sys.argv[1]) 

353 output_dir = Path(sys.argv[2]) if len(sys.argv) > 2 else Path("docs/api") 

354 

355 if not openapi_file.exists(): 

356 print(f"❌ OpenAPI文件不存在: {openapi_file}") 

357 sys.exit(1) 

358 

359 # 读取OpenAPI规范 

360 print(f"📖 读取OpenAPI规范: {openapi_file}") 

361 with open(openapi_file, "r", encoding="utf-8") as f: 

362 openapi_spec = json.load(f) 

363 

364 # 生成分模块文档 

365 print("📝 生成分模块API文档...") 

366 modules = generate_module_docs(openapi_spec, output_dir) 

367 

368 # 生成概览文档 

369 print("📝 生成API概览...") 

370 info = openapi_spec.get("info", {}) 

371 generate_api_overview(modules, output_dir, info) 

372 

373 print(f"\n✅ API文档生成完成!") 

374 print(f"📁 输出目录: {output_dir}") 

375 print(f"📋 生成了 {len(modules)} 个模块的文档") 

376 print(f"📖 查看概览: {output_dir}/README.md") 

377 

378 

379if __name__ == "__main__": 

380 main()