<think>我们将创建一个服务类来调用外部API获取题目详情。这个类将处理HTTP请求，并返回解析后的数据。 考虑到可能出现的错误，我们会加入异常处理。 由于授权信息（Authorization）是固定的，我们可以将其作为类属性或实例属性。另外，我们还需要处理其他请求头（如mode）和可能的参数（如question_changed）。 步骤： 1. 定义一个类，例如 ExternalEvaluationService。 2. 初始化方法中设置基础URL和授权令牌，以及其他可能需要的固定头信息。 3. 编写一个方法（如 get_evaluation_questions）来发送请求并获取响应。 4. 使用requests库发送GET请求，并传递必要的头信息。 5. 检查响应状态码和返回的JSON数据中的code字段（根据接口文档，code为0表示成功）。 6. 如果成功，返回data部分；否则，抛出异常或记录错误。 注意：由于接口使用了固定的授权令牌，我们将其作为常量。另外，我们可能需要根据环境（测试/生产）改变URL，但这里题目中给出的是测试URL。 代码示例： </think> 我将为您封装一个专门用于调用外部接口获取题目详情的服务类。这个类将处理认证、请求发送、错误处理和响应解析等逻辑。 ```python import requests import logging from typing import Dict, Any, List, Optional # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger("ExternalEvaluationService") class ExternalEvaluationService: def __init__(self, base_url: str, auth_token: str, mode: str = "1"): """ 初始化外部评估服务 :param base_url: API基础URL :param auth_token: 认证令牌 :param mode: 请求模式，默认为"1" """ self.base_url = base_url.rstrip('/') # 移除末尾斜杠 self.auth_token = auth_token self.mode = mode self.headers = { "Authorization": f"XDS {self.auth_token}", "mode": self.mode, "Content-Type": "application/json" } self.session = requests.Session() self.session.headers.update(self.headers) logger.info("ExternalEvaluationService 初始化完成") def get_evaluation_questions(self, question_changed: int = 0) -> Dict[str, Any]: """ 获取评估题目详情 :param question_changed: 题目是否已更改标志 (0-未更改, 1-已更改) :return: 包含题目详情的字典 """ endpoint = f"{self.base_url}/v3/pregnancy_health_plan/evaluation_questions" params = {"question_changed": question_changed} try: logger.info(f"请求评估题目: {endpoint}?question_changed={question_changed}") response = self.session.get(endpoint, params=params) response.raise_for_status() # 检查HTTP错误 data = response.json() # 验证响应格式 if data.get("code") != 0: error_msg = data.get("message", "未知错误") raise ExternalServiceException(f"API返回错误: {error_msg}") logger.info(f"成功获取评估题目，共 {data['data']['total']} 个问题") return data["data"] except requests.exceptions.RequestException as e: logger.error(f"请求失败: {str(e)}") raise ExternalServiceException(f"网络请求失败: {str(e)}") except ValueError as e: logger.error(f"JSON解析失败: {str(e)}") raise ExternalServiceException("响应数据解析失败") except KeyError as e: logger.error(f"响应格式错误，缺少必要字段: {str(e)}") raise ExternalServiceException("API响应格式不正确") def get_question_by_key(self, question_key: str, question_changed: int = 0) -> Optional[Dict[str, Any]]: """ 通过问题key获取特定问题详情 :param question_key: 问题key :param question_changed: 题目是否已更改标志 :return: 问题详情字典或None """ try: data = self.get_evaluation_questions(question_changed) for section in data["list"]: for question in section["ques_list"]: if question.get("question_key") == question_key: return question logger.warning(f"未找到问题: {question_key}") return None except ExternalServiceException: raise def get_all_questions(self, question_changed: int = 0) -> List[Dict[str, Any]]: """ 获取所有问题列表（平铺结构） :param question_changed: 题目是否已更改标志 :return: 所有问题的列表 """ try: data = self.get_evaluation_questions(question_changed) all_questions = [] for section in data["list"]: all_questions.extend(section["ques_list"]) return all_questions except ExternalServiceException: raise def get_question_options(self, question_key: str, question_changed: int = 0) -> List[Dict[str, str]]: """ 获取特定问题的所有选项 :param question_key: 问题key :param question_changed: 题目是否已更改标志 :return: 选项列表 """ question = self.get_question_by_key(question_key, question_changed) if not question: return [] options = [] for field in question["fields"]: if "options" in field: options.extend(field["options"]) return options class ExternalServiceException(Exception): """外部服务异常基类""" pass # 使用示例 if __name__ == "__main__": # 配置服务参数 BASE_URL = "https://test-gravidity.seeyouyima.com" AUTH_TOKEN = "7.-hRZ5vzOzxZ4PIZgoD-Ki2Bsp3F-AQhlPZbW74JZdEg" # 创建服务实例 evaluation_service = ExternalEvaluationService( base_url=BASE_URL, auth_token=AUTH_TOKEN ) try: # 获取所有题目 all_questions = evaluation_service.get_all_questions() print(f"共获取 {len(all_questions)} 个问题") # 获取特定问题 diet_question = evaluation_service.get_question_by_key("diet") if diet_question: print(f"饮食问题标题: {diet_question['title']}") # 获取选项 diet_options = evaluation_service.get_question_options("diet") print("饮食选项:") for option in diet_options: print(f" - {option['label']} ({option['value']})") # 获取完整题目结构 full_data = evaluation_service.get_evaluation_questions() print(f"评估ID: {full_data['evaluate_id']}") except ExternalServiceException as e: print(f"服务调用失败: {str(e)}") ``` ### 代码详细解释： #### 1. 类初始化 (`__init__` 方法) - 接收基础URL、认证令牌和模式参数 - 构建请求头信息（包含认证信息） - 创建持久化的Session对象提高性能 #### 2. 核心方法 `get_evaluation_questions` - 发送GET请求到题目接口 - 处理HTTP错误和响应验证 - 检查API返回的code是否为0（成功） - 返回data部分的数据 #### 3. 实用方法 - `get_question_by_key`: 通过问题key获取特定问题 - `get_all_questions`: 获取所有问题（平铺结构） - `get_question_options`: 获取特定问题的所有选项 #### 4. 错误处理 - 自定义 `ExternalServiceException` 异常类 - 捕获并处理网络请求错误、JSON解析错误和API格式错误 - 详细的日志记录 #### 5. 使用示例 - 演示如何初始化服务 - 调用不同方法获取数据 - 错误处理示例 ### 使用场景： 1. **获取完整题目结构**: ```python full_data = evaluation_service.get_evaluation_questions() print(f"评估ID: {full_data['evaluate_id']}") print(f"总题目数: {full_data['total']}") ``` 2. **获取特定问题**: ```python question = evaluation_service.get_question_by_key("any_disease") if question: print(f"问题标题: {question['title']}") print(f"问题类型: {question['fields'][0]['type']}") ``` 3. **获取问题选项**: ```python options = evaluation_service.get_question_options("diet") for option in options: print(f"{option['label']} -> {option['value']}") ``` 4. **获取所有问题（平铺结构）**: ```python all_questions = evaluation_service.get_all_questions() for idx, question in enumerate(all_questions, 1): print(f"{idx}. {question['title']} ({question.get('question_key', '无key')})") ``` ### 优势： 1. **封装性**：隐藏了HTTP请求的复杂性 2. **可重用性**：可在多个地方调用同一个服务实例 3. **错误处理**：统一的异常处理机制 4. **灵活性**：提供多种数据获取方式 5. **性能优化**：使用Session保持连接 6. **日志记录**：详细的运行日志便于调试 ### 注意事项： 1. 认证令牌需要妥善保管，避免泄露 2. 生产环境中应考虑添加重试机制 3. 对于高频调用，可添加缓存功能 4. 需要处理API版本变更的情况 这个服务类完全封装了与外部API的交互细节，使您能够更专注于业务逻辑的实现。