Web测试专属浏览器框架Playwright，全部api接口录制&mock
在web测试过程中可以通过设置代理抓包的方式记录浏览器的所有api接口请求，但是如果想要mock指定接口的返回操作起来还是比较麻烦的，我们可以借助Playwright 脚本打开浏览器并接管浏览器的请求来实现。
原理
使用Playwright 打开浏览器后监听指定规则的api 请求，然后全部写入excel 文件，从而实现api接口请求全录制功能。然后写入的excel文件可以直接复制（字段基本一样）到api_rules.xlsx 这个规则模版，检测到规则模版变化后会自动生效新规则，如果想要mock 一个接口可以先在浏览器里面操作产生excel接口记录，然后直接复制到规则模版里面修改成你想要mock的响应数据即可！
这种mock的主要用处是给测试以及前端开发人员定制“制作”返回数据，比如要查看某个“状态”的订单数据，除了可以直接修改测试数据库，也可以直接写入规则模版想改什么数据就改什么数据，避免产生不必要的脏数据。甚至在后端接口没开发出来之前你可以自己把预期的接口数据类型进行定义，让后端根据你提供的数据定义来做接口！
实现方式
感谢AI 助力生产力的提高，我的想法已经直接让AI 落地开发完成了。这个时代很多琐碎的工作已经不再受限于个人能力，有想法、有创造力才更重要！下面附上成品源码。
API Recorder 程序功能说明书：
# API Recorder Excel 使用说明
## 功能概述
基于 Playwright 的 API 记录和 Mock 工具，支持从 Excel 读取规则，实时记录请求/响应到 Excel。
## 核心特性
- **Excel 规则管理**：从 `api_rules.xlsx` 读取 Mock 规则
- **实时记录**：API 请求/响应自动写入按日期命名的 Excel 文件
- **失败重试**：写入失败自动重试，确保数据不丢失
- **异常保护**：程序崩溃时未写入数据保存到独立文件
- **控制台请求捕获**：支持捕获浏览器控制台 fetch 请求
## 快速开始
### 1. 安装依赖
```bash
pip install playwright openpyxl
playwright install
```
### 2. 首次运行
```bash
python api_recorder_excel.py
```
首次运行会自动创建 `api_rules.xlsx` 示例文件。
### 3. 配置规则
编辑 `api_rules.xlsx`：
| 字段 | 说明 | 示例 |
|------|------|------|
| url_pattern | URL匹配正则 | `/api/user/\d+` |
| method | HTTP方法 | GET/POST/* |
| mock_response | Mock响应JSON | `{"code":200}` |
| status_code | 状态码 | 200 |
| delay | 延迟(秒) | 0.5 |
| enabled | 是否启用(1/0) | 1 |
| priority | 优先级(越小越优先) | 1 |
| description | 描述 | Mock用户信息 |
| request_params | 请求URL参数 | `id=123&type=1` |
| request_body | 请求Body | `{"name":"test"}` |
### 4. 使用流程
```
运行程序 → 输入网址 → 浏览器操作 → Ctrl+C结束 → 查看记录
```
## 输出文件
| 文件 | 说明 |
|------|------|
| `api_records_YYYYMMDD.xlsx` | API记录文件（按日期命名，如 `api_records_20260429.xlsx`） |
| `failed_records.json` | 失败记录缓存(临时) |
| `api_records_failed_*.xlsx` | 异常退出时未写入数据 |
| `api_records_failed_*.json` | 备选保存格式 |
## Excel 字段说明
### api_records_YYYYMMDD.xlsx 字段
```
url_pattern - URL模式(如 /api/user/\d+)
method - HTTP方法
mock_response - 响应数据(JSON)
status_code - 状态码
delay - 延迟
enabled - 是否启用(默认0)
priority - 优先级(默认999)
description - 描述
request_params - URL参数(key1=value1&key2=value2)
request_body - 请求Body(JSON)
```
## 规则匹配逻辑
1. 相同 `url_pattern` 按 `priority` 排序，取最小值
2. 只处理 `enabled=1` 的规则
3. 匹配规则后返回 Mock 数据，否则转发真实请求
## 数据保护机制
```
写入失败 → 加入失败队列 → 下次请求重试 → 重试3次仍失败
↓
程序退出时 → 尝试最后写入 → 仍失败则保存到新Excel(带随机文件名)
```
## 控制台 Fetch 请求说明
浏览器控制台发起的 `fetch` 请求有以下限制：
| 数据 | 是否可捕获 | 说明 |
|------|-----------|------|
| 请求 URL、Method、Headers | ✅ 可以 | 完全支持 |
| 请求 Body | ✅ 可以 | 正常记录到 Excel |
| 响应 Body | ⚠️ 受限 | 可能显示 `"_note": "控制台fetch请求的响应无法通过Playwright获取"` |
**原因**：浏览器消费响应后，Playwright 无法再次读取。这是浏览器安全机制限制，非程序问题。
## 注意事项
- 修改 `api_rules.xlsx` 后自动生效(5秒检测间隔)
- 记录的 `enabled` 默认为 0，需手动改为 1 才能作为规则使用
- 动态ID会自动替换为 `\d+` 通配符
- 每天生成新的记录文件，避免单文件数据过多
程序源码：
"""
Playwright API 记录器和 Mock 工具 - Excel 规则版
功能：
1. 从 Excel 模板读取 Mock 规则
2. 支持启用/禁用字段
3. 相同 url_pattern 取第一个或优先级最小的
4. 实时重新加载规则（修改Excel后自动生效）
5. API请求和响应实时写入Excel，格式与规则模板一致
6. 失败记录重试机制，确保数据不丢失
7. 程序异常退出时保存未写入的失败记录
8. 记录请求参数和Body
Excel 模板字段：
- url_pattern: URL匹配模式（正则表达式）
- method: HTTP方法，默认为 *
- mock_response: Mock响应JSON字符串
- status_code: 状态码，默认为 200
- delay: 延迟秒数，默认为 0
- enabled: 是否启用，1/0 或 true/false
- priority: 优先级，数字越小优先级越高，默认为 999
- description: 规则描述
- request_params: 请求URL参数
- request_body: 请求Body参数
使用方法：
1. 创建 api_rules.xlsx 文件
2. 运行: python api_recorder_excel.py
3. 按 Ctrl+C 结束程序
4. 查看 api_records.xlsx 获取记录的API数据，可直接复制到规则中使用
"""
try:
from playwright.sync_api import sync_playwright, Route
except ImportError:
print("错误: 未安装 Playwright 库", flush=True)
print("请运行: pip install playwright && playwright install", flush=True)
exit(1)
try:
import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
except ImportError:
print("错误: 未安装 openpyxl 库", flush=True)
print("请运行: pip install openpyxl", flush=True)
exit(1)
import json
import sys
import signal
import re
import time
import os
import threading
import uuid
import atexit
from datetime import datetime
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from urllib.parse import urlparse, parse_qs
@dataclass
class MockRule:
"""Mock 规则数据类"""
url_pattern: str
method: str = "*"
mock_response: Optional[Dict] = None
status_code: int = 200
delay: float = 0
enabled: bool = True
priority: int = 999
description: str = ""
compiled_pattern: Any = field(default=None, repr=False)
def __post_init__(self):
if self.compiled_pattern is None and self.url_pattern:
try:
self.compiled_pattern = re.compile(self.url_pattern)
except re.error as e:
print(f"[警告] 正则表达式编译失败: {self.url_pattern}, 错误: {e}", flush=True)
self.compiled_pattern = None
def matches(self, url: str, method: str) -> bool:
"""检查请求是否匹配此规则"""
if not self.enabled:
return False
if self.method != "*" and self.method.upper() != method.upper():
return False
if self.compiled_pattern is None:
return False
return bool(self.compiled_pattern.search(url))
class ExcelRuleLoader:
"""Excel 规则加载器"""
def __init__(self, excel_path: str = "api_rules.xlsx"):
self.excel_path = excel_path
self.last_modified = 0
self.rules: List[MockRule] = []
self._url_pattern_rules: Dict[str, List[MockRule]] = {} # 按 url_pattern 分组
def load_rules(self) -> List[MockRule]:
"""从 Excel 加载规则"""
if not os.path.exists(self.excel_path):
print(f"[警告] Excel 文件不存在: {self.excel_path}", flush=True)
return []
try:
# 检查文件是否修改
current_modified = os.path.getmtime(self.excel_path)
if current_modified == self.last_modified and self.rules:
return self.rules
self.last_modified = current_modified
# 加载 Excel
wb = openpyxl.load_workbook(self.excel_path)
ws = wb.active
# 获取表头
headers = [cell.value.strip().lower() if cell.value else "" for cell in ws[1]]
# 字段映射
field_map = {
'url_pattern': ['url_pattern', 'url pattern', 'url', 'pattern', 'url匹配'],
'method': ['method', '方法', 'http_method'],
'mock_response': ['mock_response', 'mock response', 'response', '响应', 'mock'],
'status_code': ['status_code', 'status code', 'status', '状态码'],
'delay': ['delay', '延迟', '延时'],
'enabled': ['enabled', 'enable', '启用', '是否启用', 'active'],
'priority': ['priority', '优先级', 'priority_level'],
'description': ['description', 'desc', '描述', '说明'],
'request_params': ['request_params', 'params', 'url_params', '请求参数'],
'request_body': ['request_body', 'body', '请求体']
}
# 找到列索引
col_indices = {}
for field_name, possible_names in field_map.items():
for idx, header in enumerate(headers):
if header in possible_names:
col_indices[field_name] = idx
break
if 'url_pattern' not in col_indices:
print("[错误] Excel 中未找到 url_pattern 列", flush=True)
return []
# 读取数据
raw_rules = []
for row_idx, row in enumerate(ws.iter_rows(min_row=2), start=2):
try:
url_pattern = self._get_cell_value(row[col_indices.get('url_pattern')]) if 'url_pattern' in col_indices else None
if not url_pattern:
continue
rule = {
'url_pattern': url_pattern,
'method': self._get_cell_value(row[col_indices.get('method')]) if 'method' in col_indices else '*',
'mock_response': self._parse_json(self._get_cell_value(row[col_indices.get('mock_response')])) if 'mock_response' in col_indices else None,
'status_code': self._parse_int(self._get_cell_value(row[col_indices.get('status_code')]), 200),
'delay': self._parse_float(self._get_cell_value(row[col_indices.get('delay')]), 0),
'enabled': self._parse_bool(self._get_cell_value(row[col_indices.get('enabled')]), True),
'priority': self._parse_int(self._get_cell_value(row[col_indices.get('priority')]), 999),
'description': self._get_cell_value(row[col_indices.get('description')]) if 'description' in col_indices else '',
'request_params': self._get_cell_value(row[col_indices.get('request_params')]) if 'request_params' in col_indices else '',
'request_body': self._get_cell_value(row[col_indices.get('request_body')]) if 'request_body' in col_indices else ''
}
raw_rules.append(rule)
except Exception as e:
print(f"[警告] 读取第 {row_idx} 行时出错: {e}", flush=True)
continue
wb.close()
# 按 url_pattern 分组并选择规则
self._url_pattern_rules = {}
for rule_data in raw_rules:
url_pattern = rule_data['url_pattern']
if url_pattern not in self._url_pattern_rules:
self._url_pattern_rules[url_pattern] = []
self._url_pattern_rules[url_pattern].append(rule_data)
# 处理相同 url_pattern 的规则
selected_rules = []
for url_pattern, rules_list in self._url_pattern_rules.items():
# 过滤启用的规则
enabled_rules = [r for r in rules_list if r['enabled']]
if not enabled_rules:
continue
# 按优先级排序，取第一个
enabled_rules.sort(key=lambda x: x['priority'])
selected = enabled_rules[0]
selected_rules.append(MockRule(
url_pattern=selected['url_pattern'],
method=selected['method'],
mock_response=selected['mock_response'],
status_code=selected['status_code'],
delay=selected['delay'],
enabled=selected['enabled'],
priority=selected['priority'],
description=selected['description']
))
self.rules = selected_rules
print(f"[加载] 从 Excel 加载了 {len(self.rules)} 条规则", flush=True)
# 显示加载的规则
for i, rule in enumerate(self.rules, 1):
status = "✓" if rule.enabled else "✗"
print(f" {i}. [{status}] {rule.url_pattern} [{rule.method}] (优先级:{rule.priority})", flush=True)
return self.rules
except Exception as e:
print(f"[错误] 加载 Excel 失败: {e}", flush=True)
return self.rules if self.rules else []
def _get_cell_value(self, cell) -> str:
"""获取单元格值"""
if cell is None or cell.value is None:
return ""
value = str(cell.value).strip()
# 处理 Excel 中的布尔值
if value.lower() in ('true', '1', 'yes', '是'):
return 'true'
if value.lower() in ('false', '0', 'no', '否'):
return 'false'
return value
def _parse_json(self, value: str) -> Optional[Dict]:
"""解析 JSON 字符串"""
if not value:
return None
try:
return json.loads(value)
except:
# 尝试作为字符串返回
return {"message": value}
def _parse_int(self, value: str, default: int) -> int:
"""解析整数"""
if not value:
return default
try:
return int(float(value))
except:
return default
def _parse_float(self, value: str, default: float) -> float:
"""解析浮点数"""
if not value:
return default
try:
return float(value)
except:
return default
def _parse_bool(self, value: str, default: bool) -> bool:
"""解析布尔值"""
if not value:
return default
value = value.lower()
return value in ('true', '1', 'yes', '是', '启用', 'enabled')
class FailedRecord:
"""失败记录数据类"""
def __init__(self, url: str, method: str, status_code: int, response_body: Any,
post_data: Any = None, request_params: str = "", request_body: str = ""):
self.url = url
self.method = method
self.status_code = status_code
self.response_body = response_body
self.post_data = post_data
self.request_params = request_params
self.request_body = request_body
self.fail_time = datetime.now().isoformat()
self.retry_count = 0
def to_dict(self) -> Dict:
"""转换为字典"""
return {
'url': self.url,
'method': self.method,
'status_code': self.status_code,
'response_body': self.response_body,
'post_data': self.post_data,
'request_params': self.request_params,
'request_body': self.request_body,
'fail_time': self.fail_time,
'retry_count': self.retry_count
}
@classmethod
def from_dict(cls, data: Dict) -> 'FailedRecord':
"""从字典创建"""
record = cls(
url=data['url'],
method=data['method'],
status_code=data['status_code'],
response_body=data['response_body'],
post_data=data.get('post_data'),
request_params=data.get('request_params', ''),
request_body=data.get('request_body', '')
)
record.fail_time = data.get('fail_time', datetime.now().isoformat())
record.retry_count = data.get('retry_count', 0)
return record
class ExcelRecorder:
"""Excel API 记录器 - 带失败重试机制"""
# 与规则模板一致的表头，增加请求参数和Body字段
HEADERS = ["url_pattern", "method", "mock_response", "status_code", "delay", "enabled", "priority", "description", "request_params", "request_body"]
def __init__(self, excel_path: str = "api_records.xlsx", failed_records_path: str = "failed_records.json"):
self.excel_path = excel_path
self.failed_records_path = failed_records_path
self.lock = threading.Lock()
self.record_count = 0
self.failed_records: List[FailedRecord] = [] # 失败记录队列
self.max_retry = 3 # 最大重试次数
# 加载历史失败记录
self._load_failed_records()
# 初始化 Excel
self._init_excel()
# 注册退出处理
atexit.register(self._on_exit)
def _load_failed_records(self):
"""加载历史失败记录"""
try:
if os.path.exists(self.failed_records_path):
with open(self.failed_records_path, 'r', encoding='utf-8') as f:
data = json.load(f)
for item in data:
self.failed_records.append(FailedRecord.from_dict(item))
print(f"[恢复] 加载了 {len(self.failed_records)} 条历史失败记录", flush=True)
# 立即尝试写入
self._retry_failed_records()
except Exception as e:
print(f"[警告] 加载历史失败记录失败: {e}", flush=True)
def _save_failed_records(self):
"""保存失败记录到文件"""
try:
data = [r.to_dict() for r in self.failed_records]
with open(self.failed_records_path, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"[错误] 保存失败记录失败: {e}", flush=True)
def _on_exit(self):
"""程序退出时的处理"""
if self.failed_records:
print(f"\n[退出] 还有 {len(self.failed_records)} 条记录未写入，正在保存...", flush=True)
# 先尝试最后一次写入
self._retry_failed_records()
# 如果还有失败的，保存到新的 Excel
if self.failed_records:
self._save_failed_to_new_excel()
# 清理失败记录文件
if os.path.exists(self.failed_records_path):
try:
os.remove(self.failed_records_path)
except:
pass
def _save_failed_to_new_excel(self):
"""将未写入的失败记录保存到新的 Excel 文件"""
try:
# 生成带随机数的文件名
random_suffix = str(uuid.uuid4())[:8]
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
new_filename = f"api_records_failed_{timestamp}_{random_suffix}.xlsx"
# 创建新工作簿
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "失败记录"
# 写入表头
ws.append(self.HEADERS)
# 设置表头样式
header_fill = PatternFill(start_color="C00000", end_color="C00000", fill_type="solid")
header_font = Font(bold=True, color="FFFFFF")
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal="center", vertical="center")
# 写入失败记录
for failed_record in self.failed_records:
row_data = self._prepare_row_data(
failed_record.url,
failed_record.method,
failed_record.status_code,
failed_record.response_body,
failed_record.post_data,
failed_record.request_params,
failed_record.request_body
)
ws.append(row_data)
# 设置列宽
ws.column_dimensions['A'].width = 40 # url_pattern
ws.column_dimensions['B'].width = 10 # method
ws.column_dimensions['C'].width = 50 # mock_response
ws.column_dimensions['D'].width = 12 # status_code
ws.column_dimensions['E'].width = 8 # delay
ws.column_dimensions['F'].width = 10 # enabled
ws.column_dimensions['G'].width = 10 # priority
ws.column_dimensions['H'].width = 25 # description
ws.column_dimensions['I'].width = 40 # request_params
ws.column_dimensions['J'].width = 50 # request_body
wb.save(new_filename)
wb.close()
print(f"✓ 未写入记录已保存到: {new_filename}", flush=True)
print(f" 共 {len(self.failed_records)} 条记录", flush=True)
except Exception as e:
print(f"✗ 保存失败记录到 Excel 失败: {e}", flush=True)
# 最后的备选：保存为 JSON
self._save_failed_as_json()
def _save_failed_as_json(self):
"""将失败记录保存为 JSON（最后备选）"""
try:
random_suffix = str(uuid.uuid4())[:8]
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
json_filename = f"api_records_failed_{timestamp}_{random_suffix}.json"
data = {
'save_time': datetime.now().isoformat(),
'count': len(self.failed_records),
'records': [r.to_dict() for r in self.failed_records]
}
with open(json_filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"✓ 未写入记录已保存为 JSON: {json_filename}", flush=True)
except Exception as e:
print(f"✗ 保存失败记录为 JSON 也失败了: {e}", flush=True)
def _init_excel(self):
"""初始化 Excel 文件"""
try:
if os.path.exists(self.excel_path):
# 文件已存在，读取现有记录数
wb = openpyxl.load_workbook(self.excel_path)
ws = wb.active
self.record_count = ws.max_row - 1
wb.close()
print(f"[记录器] 已加载现有记录文件，当前记录数: {self.record_count}", flush=True)
else:
# 创建新文件
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "API记录"
# 写入表头
ws.append(self.HEADERS)
# 设置表头样式
header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
header_font = Font(bold=True, color="FFFFFF")
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal="center", vertical="center")
# 设置列宽
ws.column_dimensions['A'].width = 40 # url_pattern
ws.column_dimensions['B'].width = 10 # method
ws.column_dimensions['C'].width = 50 # mock_response
ws.column_dimensions['D'].width = 12 # status_code
ws.column_dimensions['E'].width = 8 # delay
ws.column_dimensions['F'].width = 10 # enabled
ws.column_dimensions['G'].width = 10 # priority
ws.column_dimensions['H'].width = 25 # description
ws.column_dimensions['I'].width = 40 # request_params
ws.column_dimensions['J'].width = 50 # request_body
wb.save(self.excel_path)
wb.close()
print(f"[记录器] 已创建新的记录文件: {self.excel_path}", flush=True)
except Exception as e:
print(f"[错误] 初始化 Excel 记录器失败: {e}", flush=True)
def _extract_url_params(self, url: str) -> str:
"""提取 URL 查询参数"""
try:
parsed = urlparse(url)
if not parsed.query:
return ""
# 保持 key1=value1&key2=value2 格式
return parsed.query
except:
return ""
def _format_request_body(self, post_data: Any) -> str:
"""格式化请求 Body"""
if not post_data:
return ""
try:
# 如果是 JSON 字符串，尝试格式化
if isinstance(post_data, str):
try:
parsed = json.loads(post_data)
return json.dumps(parsed, ensure_ascii=False, separators=(',', ':'))
except:
return post_data[:500]
elif isinstance(post_data, dict):
return json.dumps(post_data, ensure_ascii=False, separators=(',', ':'))
else:
return str(post_data)[:500]
except:
return str(post_data)[:500]
def _prepare_row_data(self, url: str, method: str, status_code: int, response_body: Any,
post_data: Any = None, request_params: str = "", request_body: str = "") -> List:
"""准备行数据"""
# 提取 url_pattern（简化版）
url_pattern = self._extract_url_pattern(url)
# 格式化响应为 JSON 字符串
if isinstance(response_body, dict):
mock_response = json.dumps(response_body, ensure_ascii=False, separators=(',', ':'))
elif isinstance(response_body, str):
try:
# 尝试解析并重新格式化
parsed = json.loads(response_body)
mock_response = json.dumps(parsed, ensure_ascii=False, separators=(',', ':'))
except:
mock_response = response_body[:500] # 限制长度
else:
mock_response = str(response_body)[:500]
# 生成描述
description = f"记录于 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
# 准备行数据（与规则模板一致）
return [
url_pattern, # url_pattern
method, # method
mock_response, # mock_response
status_code, # status_code
0, # delay
0, # enabled (默认禁用，需要手动启用)
999, # priority
description, # description
request_params, # request_params
request_body # request_body
]
def _write_to_excel(self, row_data: List) -> bool:
"""写入单条记录到 Excel"""
try:
wb = openpyxl.load_workbook(self.excel_path)
ws = wb.active
ws.append(row_data)
# 设置新行的样式
row_num = ws.max_row
for col_num in range(1, len(self.HEADERS) + 1):
cell = ws.cell(row=row_num, column=col_num)
cell.alignment = Alignment(vertical="center", wrap_text=True)
# 为 mock_response 和 request_body 列设置特定样式
if col_num in [3, 10]: # mock_response 和 request_body 列
cell.font = Font(size=9)
# 设置行高
ws.row_dimensions[row_num].height = 30
wb.save(self.excel_path)
wb.close()
return True
except Exception as e:
print(f"[错误] 写入 Excel 失败: {e}", flush=True)
return False
def _retry_failed_records(self) -> int:
"""重试写入失败记录，返回成功写入的数量"""
if not self.failed_records:
return 0
success_count = 0
still_failed = []
print(f"[重试] 尝试写入 {len(self.failed_records)} 条历史失败记录...", flush=True)
for failed_record in self.failed_records:
if failed_record.retry_count >= self.max_retry:
# 超过最大重试次数，保留
still_failed.append(failed_record)
continue
failed_record.retry_count += 1
row_data = self._prepare_row_data(
failed_record.url,
failed_record.method,
failed_record.status_code,
failed_record.response_body,
failed_record.post_data,
failed_record.request_params,
failed_record.request_body
)
if self._write_to_excel(row_data):
success_count += 1
self.record_count += 1
else:
still_failed.append(failed_record)
self.failed_records = still_failed
if success_count > 0:
print(f"[重试] 成功写入 {success_count} 条，还有 {len(still_failed)} 条失败", flush=True)
# 保存当前的失败记录
self._save_failed_records()
return success_count
def add_record(self, url: str, method: str, status_code: int, response_body: Any, post_data: Any = None):
"""添加一条 API 记录（带失败重试机制）"""
with self.lock:
# 先尝试重试历史失败记录
if self.failed_records:
self._retry_failed_records()
# 提取请求参数和Body
request_params = self._extract_url_params(url)
request_body = self._format_request_body(post_data)
# 准备当前记录数据
row_data = self._prepare_row_data(url, method, status_code, response_body, post_data, request_params, request_body)
# 尝试写入当前记录
if self._write_to_excel(row_data):
self.record_count += 1
print(f"[记录] 已保存到 Excel ({self.record_count}): {method} {self._extract_url_pattern(url)}", flush=True)
else:
# 写入失败，添加到失败队列
failed_record = FailedRecord(url, method, status_code, response_body, post_data, request_params, request_body)
self.failed_records.append(failed_record)
self._save_failed_records()
print(f"[失败] 记录写入失败，已加入重试队列 (当前队列: {len(self.failed_records)} 条): {method} {url}", flush=True)
def _extract_url_pattern(self, url: str) -> str:
"""从完整 URL 提取模式"""
# 移除协议和域名部分
pattern = url
# 移除 http:// 或 https://
if '://' in pattern:
pattern = pattern.split('://', 1)[1]
# 移除域名部分，保留路径
if '/' in pattern:
pattern = '/' + pattern.split('/', 1)[1]
# 移除查询参数
if '?' in pattern:
pattern = pattern.split('?', 1)[0]
# 将动态 ID 替换为通配符
# 例如: /api/user/123 -> /api/user/\d+
pattern = re.sub(r'/\d+', r'/\\d+', pattern)
return pattern
def get_stats(self) -> Dict:
"""获取统计信息"""
return {
'record_count': self.record_count,
'failed_count': len(self.failed_records),
'failed_records_path': self.failed_records_path if self.failed_records else None
}
class APIRecorderExcel:
"""API 记录器 - Excel 规则版"""
def __init__(self, rules_excel: str = "api_rules.xlsx", records_excel: str = "api_records.xlsx", listen_new_pages: bool = False):
self.rules_excel = rules_excel
self.records_excel = records_excel
self.rule_loader = ExcelRuleLoader(rules_excel)
self.excel_recorder = ExcelRecorder(records_excel)
self.rules: List[MockRule] = []
self.request_count = 0
self.response_count = 0
self.mock_count = 0
self.running = True
self.last_rule_reload = 0
# 是否监听新标签页
self.listen_new_pages = listen_new_pages
# 请求缓存：用于存储所有请求的body（包括控制台fetch）
self._request_cache: Dict[str, Any] = {}
self._cache_lock = threading.Lock()
# 标记已通过route处理的请求（避免重复记录）
self._routed_requests: set = set()
self._route_lock = threading.Lock()
def setup_page_listeners(self, page):
"""为指定页面设置监听器"""
# 监听所有请求（用于捕获body，包括控制台fetch）
page.on("request", lambda request: self.handle_request(request))
# 设置路由拦截（用于mock）
page.route("**/*", lambda route: self.handle_route(route))
# 监听响应
page.on("response", lambda response: self.handle_response(response))
print(f"[监听] 已为页面设置 API 监听器: {page.url[:60]}...", flush=True)
def handle_new_page(self, page):
"""处理新打开的页面"""
print(f"\n[新页面] 检测到新标签页: {page.url[:60]}...", flush=True)
self.setup_page_listeners(page)
def reload_rules(self):
"""重新加载规则"""
current_time = time.time()
# 每 5 秒检查一次文件修改
if current_time - self.last_rule_reload >= 5:
new_rules = self.rule_loader.load_rules()
if new_rules:
self.rules = new_rules
self.last_rule_reload = current_time
def _get_request_body(self, request) -> Any:
"""获取请求 Body，尝试多种方式"""
try:
# 方式1: 直接获取 post_data
post_data = request.post_data
if post_data:
return post_data
# 方式2: 获取 post_data_buffer
try:
buffer = request.post_data_buffer
if buffer:
return buffer.decode('utf-8')
except:
pass
# 方式3: 获取 post_data_json
try:
json_data = request.post_data_json
if json_data:
return json_data
except:
pass
return None
except Exception as e:
print(f"[警告] 获取请求Body失败: {e}", flush=True)
return None
def _get_cache_key(self, url: str, method: str) -> str:
"""生成请求缓存的key"""
return f"{method}:{url}"
def handle_request(self, request):
"""处理请求 - 用于捕获所有请求的body（包括控制台fetch）"""
url = request.url
method = request.method
# 只处理 API 请求
if not self._is_api_request(url):
return
# 获取请求body（多种方式尝试）
post_data = self._get_request_body(request)
# 缓存请求信息
cache_key = self._get_cache_key(url, method)
with self._cache_lock:
self._request_cache[cache_key] = {
'post_data': post_data,
'headers': dict(request.headers),
'timestamp': time.time()
}
if post_data:
print(f"[请求捕获] {method} {url}", flush=True)
print(f" Body: {str(post_data)[:100]}...", flush=True)
def handle_route(self, route):
"""处理路由请求"""
request = route.request
url = request.url
method = request.method
post_data = self._get_request_body(request)
headers = request.headers
# 只处理 API 请求
if not self._is_api_request(url):
route.continue_()
return
self.request_count += 1
# 标记此请求已通过route处理
cache_key = self._get_cache_key(url, method)
with self._route_lock:
self._routed_requests.add(cache_key)
# 检查是否需要重新加载规则
self.reload_rules()
# 查找匹配的规则
matched_rule = None
for rule in self.rules:
if rule.matches(url, method):
matched_rule = rule
break
if matched_rule:
# 应用 Mock 规则
if matched_rule.delay > 0:
time.sleep(matched_rule.delay)
self.mock_count += 1
print(f"\n[MOCK] {method} {url}", flush=True)
if matched_rule.description:
print(f" 描述: {matched_rule.description}", flush=True)
if post_data:
print(f" Body: {str(post_data)[:200]}", flush=True)
try:
response_body = matched_rule.mock_response if matched_rule.mock_response else {"mocked": True}
response_text = json.dumps(response_body, ensure_ascii=False)
route.fulfill(
status=matched_rule.status_code,
content_type="application/json",
body=response_text
)
# 记录 Mock 响应到 Excel
self.excel_recorder.add_record(
url=url,
method=method,
status_code=matched_rule.status_code,
response_body=response_body,
post_data=post_data
)
# 清理缓存和标记
with self._cache_lock:
if cache_key in self._request_cache:
del self._request_cache[cache_key]
with self._route_lock:
self._routed_requests.discard(cache_key)
return
except Exception as e:
print(f" ✗ Mock 失败: {e}", flush=True)
# 没有匹配规则，继续正常请求
print(f"\n[请求] {method} {url}", flush=True)
if post_data:
data_str = str(post_data)
print(f" Body: {data_str[:200]}..." if len(data_str) > 200 else f" Body: {post_data}", flush=True)
route.continue_()
def handle_response(self, response):
"""处理响应"""
url = response.url
status = response.status
if not self._is_api_request(url):
return
self.response_count += 1
# 获取请求信息
request = response.request
method = request.method if request else "GET"
# 获取响应体 - 带错误处理
body_json = None
# 方式1: 尝试获取响应体
try:
# 检查响应是否可用
if response.ok or status > 0:
body = response.text()
if body:
try:
body_json = json.loads(body)
except:
body_json = {"response_text": body[:500]}
else:
body_json = {"_note": "响应体为空"}
else:
body_json = {"_note": f"响应状态: {status}"}
except Exception as e:
error_msg = str(e)
# 针对控制台fetch的特殊处理
if "Target page, context or browser has been closed" in error_msg:
body_json = {"_note": "控制台fetch请求的响应无法通过Playwright获取（浏览器限制）"}
elif "Response body is unavailable for redirect" in error_msg:
body_json = {"_note": "重定向响应，无响应体"}
else:
body_json = {"_note": f"无法获取响应体: {error_msg}"}
# 获取请求Body（从缓存中获取，包括控制台fetch请求）
post_data = None
cache_key = self._get_cache_key(url, method)
# 从缓存获取（handle_request中设置，包含所有请求的body）
with self._cache_lock:
if cache_key in self._request_cache:
cached_data = self._request_cache[cache_key]
post_data = cached_data.get('post_data')
# 获取后删除缓存（避免内存泄漏）
del self._request_cache[cache_key]
if post_data:
print(f"[调试] 从缓存获取到Body: {str(post_data)[:100]}...", flush=True)
# 备用：尝试直接从 request 获取
if not post_data and request:
post_data = self._get_request_body(request)
if post_data:
print(f"[调试] 从request直接获取到Body: {str(post_data)[:100]}...", flush=True)
# 记录到 Excel
self.excel_recorder.add_record(
url=url,
method=method,
status_code=status,
response_body=body_json,
post_data=post_data
)
# 调试信息
body_preview = str(post_data)[:100] if post_data else "无"
print(f"[响应] {method} {status} {url}", flush=True)
print(f" Body: {body_preview}...", flush=True)
def _is_api_request(self, url: str) -> bool:
"""判断是否为 API 请求"""
api_patterns = [
r"/api/",
r"/ajax/",
r"/rest/",
r"/graphql",
r"\.json$",
r"/v\d+/",
r"/service/"
]
return any(re.search(pattern, url) for pattern in api_patterns)
def show_stats(self):
"""显示统计信息"""
recorder_stats = self.excel_recorder.get_stats()
print("\n" + "=" * 70, flush=True)
print("统计信息", flush=True)
print("=" * 70, flush=True)
print(f"总请求数: {self.request_count}", flush=True)
print(f"总响应数: {self.response_count}", flush=True)
print(f"Mock 请求: {self.mock_count}", flush=True)
print(f"成功记录: {recorder_stats['record_count']}", flush=True)
print(f"待重试记录: {recorder_stats['failed_count']}", flush=True)
print(f"记录文件: {self.records_excel}", flush=True)
print(f"规则文件: {self.rules_excel}", flush=True)
if recorder_stats['failed_records_path']:
print(f"失败记录缓存: {recorder_stats['failed_records_path']}", flush=True)
print(flush=True)
def create_example_rules_excel(excel_path: str = "api_rules.xlsx"):
"""创建示例规则 Excel 文件"""
try:
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "Mock规则"
# 表头（与记录器一致）
headers = ["url_pattern", "method", "mock_response", "status_code", "delay", "enabled", "priority", "description", "request_params", "request_body"]
ws.append(headers)
# 设置表头样式
header_fill = PatternFill(start_color="70AD47", end_color="70AD47", fill_type="solid")
header_font = Font(bold=True, color="FFFFFF")
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal="center", vertical="center")
# 示例数据
examples = [
[
r"/api/user/info",
"GET",
'{"code":200,"message":"success","data":{"id":123,"name":"Mock用户"}}',
200,
0.5,
1,
1,
"Mock用户信息",
"",
""
],
[
r"/api/auth/login",
"POST",
'{"code":200,"message":"登录成功","data":{"token":"mock_token"}}',
200,
1.0,
1,
2,
"Mock登录接口",
"",
'{"username":"test","password":"123456"}'
],
[
r"/api/user/info", # 相同的 url_pattern，但优先级较低，不会被使用
"GET",
'{"code":200,"data":{"id":999,"name":"低优先级"}}',
200,
0,
1,
10,
"低优先级规则（不会被使用）",
"",
""
],
[
r"/api/error",
"*",
'{"code":500,"message":"服务器错误"}',
500,
0,
0, # 禁用
1,
"禁用状态的错误Mock",
"",
""
]
]
for row in examples:
ws.append(row)
# 调整列宽
ws.column_dimensions['A'].width = 40
ws.column_dimensions['B'].width = 10
ws.column_dimensions['C'].width = 50
ws.column_dimensions['D'].width = 12
ws.column_dimensions['E'].width = 8
ws.column_dimensions['F'].width = 10
ws.column_dimensions['G'].width = 10
ws.column_dimensions['H'].width = 25
ws.column_dimensions['I'].width = 40
ws.column_dimensions['J'].width = 50
wb.save(excel_path)
print(f"✓ 已创建示例规则 Excel 文件: {excel_path}", flush=True)
wb.close()
return True
except Exception as e:
print(f"✗ 创建示例 Excel 失败: {e}", flush=True)
return False
def main():
"""主函数"""
if hasattr(sys.stdout, 'reconfigure'):
sys.stdout.reconfigure(line_buffering=True)
rules_excel = "api_rules.xlsx"
# 生成带日期的记录文件名，格式: api_records_20250429.xlsx
today_str = datetime.now().strftime('%Y%m%d')
records_excel = f"api_records_{today_str}.xlsx"
print("=" * 70, flush=True)
print("Playwright API 记录器和 Mock 工具 - Excel 版", flush=True)
print("=" * 70, flush=True)
# 检查规则 Excel 文件是否存在
if not os.path.exists(rules_excel):
print(f"\n[提示] 未找到 {rules_excel}，正在创建示例文件...", flush=True)
create_example_rules_excel(rules_excel)
print("\n请编辑 api_rules.xlsx 文件添加您的规则，然后重新运行程序。", flush=True)
return
# 询问是否监听新标签页
print("\n是否启用新标签页 API 监听? (y/n, 直接回车默认 n): ", end="", flush=True)
try:
listen_choice = input().strip().lower()
listen_new_pages = listen_choice in ('y', 'yes', '1', 'true', '是')
except:
listen_new_pages = False
if listen_new_pages:
print("[配置] 已启用新标签页监听模式", flush=True)
else:
print("[配置] 仅监听首个标签页", flush=True)
# 创建记录器
recorder = APIRecorderExcel(rules_excel=rules_excel, records_excel=records_excel, listen_new_pages=listen_new_pages)
# 初始加载规则
recorder.rules = recorder.rule_loader.load_rules()
print("\n请输入要访问的网址 (直接回车使用默认): ", end="", flush=True)
try:
url = input().strip()
if not url:
url = "https://www.example.com"
except:
url = "https://www.example.com"
print(f"\n准备启动浏览器，访问: {url}", flush=True)
print(f"[提示] 所有 API 请求和响应将实时记录到 {records_excel}", flush=True)
print("[提示] 记录的格式与规则模板一致，可直接复制到规则中使用", flush=True)
print("[提示] 写入失败时会自动重试，确保数据不丢失", flush=True)
print("=" * 70, flush=True)
# 设置信号处理
def signal_handler(sig, frame):
print("\n\n[停止] 正在保存...", flush=True)
recorder.running = False
signal.signal(signal.SIGINT, signal_handler)
with sync_playwright() as p:
browser = p.chromium.launch(
headless=False,
args=['--disable-web-security']
)
context = browser.new_context(viewport={'width': 1280, 'height': 800})
page = context.new_page()
# 为首个页面设置监听器
recorder.setup_page_listeners(page)
# 如果启用新标签页监听，监听新页面创建事件
if recorder.listen_new_pages:
context.on("page", lambda new_page: recorder.handle_new_page(new_page))
print("[监听] 已启用新标签页自动监听", flush=True)
try:
print(f"\n[启动] 正在打开 {url}...", flush=True)
page.goto(url, wait_until="networkidle", timeout=60000)
print("\n[状态] 浏览器已打开，请手动操作...", flush=True)
print("[提示] 按 Ctrl+C 结束程序\n", flush=True)
# 保持程序运行
while recorder.running:
try:
page.wait_for_timeout(100)
except:
break
except KeyboardInterrupt:
print("\n\n[停止] 检测到 Ctrl+C", flush=True)
except Exception as e:
print(f"\n[错误] 发生异常: {e}", flush=True)
finally:
recorder.show_stats()
print("\n[关闭] 正在关闭浏览器...", flush=True)
try:
browser.close()
except:
pass
print("\n" + "=" * 70, flush=True)
print("程序已结束", flush=True)
print(f"✓ API 记录已保存到: {records_excel}", flush=True)
print("=" * 70, flush=True)
if __name__ == "__main__":
main()
再附上另一个api测试程序（会自动使用上面录制保存的excel 测试）
这个程序可以配合上面的程序操作，比如你录制了一个新建商品的接口，那么在这个API 接口测试工具里面你就可以先在左边选择对应接口，然后在右边直接修改请求body里面的数据，直接发送请求就行（注意接口的Host配置 要先改成后端接口的host哦~）
如果上面的API Recorder 程序运行时选择了监听新标签，那么如果你在Playwright打开的浏览器里面打开localhost:5000 这个网址发送的请求也会被记录/mock 下来！当然你也可以使用你日常使用的正常浏览器打开这样就不会被记录了，但是可能测试接口的时候会有跨域问题（API Recorder 这个程序打开的浏览器是配了允许跨域请求的）
"""
API 接口测试 Web 工具
功能：
1. 读取当天 API 记录 Excel 文件
2. 展示接口列表
3. 支持 Host 配置
4. 点击接口进行请求测试
使用方法：
1. 先运行 api_recorder_excel.py 记录接口
2. 运行本程序: python api_tester_web.py
3. 在浏览器中打开 http://localhost:5000
4. 配置 Host，选择接口进行测试
"""
from flask import Flask, render_template, jsonify, request
import openpyxl
import os
from datetime import datetime
import json
import requests
app = Flask(__name__)
# 全局配置
DEFAULT_HOST = "http://localhost:8080"
current_host = DEFAULT_HOST
def get_today_excel_path():
"""获取当天的 Excel 文件路径"""
today_str = datetime.now().strftime('%Y%m%d')
excel_path = f"api_records_{today_str}.xlsx"
return excel_path
def load_api_records(excel_path=None):
"""从 Excel 加载 API 记录"""
if excel_path is None:
excel_path = get_today_excel_path()
records = []
if not os.path.exists(excel_path):
return records
try:
wb = openpyxl.load_workbook(excel_path)
ws = wb.active
# 获取表头
headers = [cell.value for cell in ws[1]]
# 读取数据
for row in ws.iter_rows(min_row=2):
record = {}
for idx, cell in enumerate(row):
if idx < len(headers):
header = headers[idx]
value = cell.value
# 处理 mock_response 字段，尝试解析为 JSON
if header == 'mock_response' and value:
try:
record[header] = json.loads(value)
except:
record[header] = value
else:
record[header] = value
# 只添加有 url_pattern 的记录
if record.get('url_pattern'):
records.append(record)
wb.close()
except Exception as e:
print(f"[错误] 读取 Excel 失败: {e}")
return records
def parse_url_pattern(url_pattern, host):
"""将 URL 模式转换为可请求的 URL"""
# 如果 url_pattern 已经是完整 URL，直接返回
if url_pattern.startswith('http://') or url_pattern.startswith('https://'):
return url_pattern
# 确保 url_pattern 以 / 开头
if not url_pattern.startswith('/'):
url_pattern = '/' + url_pattern
# 替换正则通配符为实际值（简化处理）
# 例如: /api/user/\d+ -> /api/user/123
import re
url = url_pattern
url = re.sub(r'\\d\+', '1', url) # \d+ -> 1
url = re.sub(r'\\w\+', 'abc', url) # \w+ -> abc
url = re.sub(r'\[.*?\]', '', url) # 移除 [...]
url = re.sub(r'\.', '', url) # 移除 .
url = re.sub(r'\*', '', url) # 移除 *
url = re.sub(r'\+', '', url) # 移除 +
url = re.sub(r'\?', '', url) # 移除 ?
url = re.sub(r'\\', '', url) # 移除 \
# 组合成完整 URL
full_url = host.rstrip('/') + url
return full_url
@app.route('/')
def index():
"""首页"""
return render_template('api_tester.html')
@app.route('/api/config', methods=['GET', 'POST'])
def config():
"""获取/设置 Host 配置"""
global current_host
if request.method == 'POST':
data = request.get_json()
new_host = data.get('host', '').strip()
if new_host:
# 确保 host 以 http:// 或 https:// 开头
if not new_host.startswith(('http://', 'https://')):
new_host = 'http://' + new_host
current_host = new_host
return jsonify({'success': True, 'host': current_host})
return jsonify({'success': False, 'error': 'Invalid host'})
return jsonify({'host': current_host})
@app.route('/api/records')
def get_records():
"""获取 API 记录列表"""
records = load_api_records()
# 添加生成的测试 URL
for record in records:
url_pattern = record.get('url_pattern', '')
if url_pattern:
record['test_url'] = parse_url_pattern(url_pattern, current_host)
return jsonify(records)
@app.route('/api/test', methods=['POST'])
def test_api():
"""测试 API 接口"""
data = request.get_json()
url = data.get('url', '')
method = data.get('method', 'GET').upper()
headers = data.get('headers', {})
body = data.get('body', None)
if not url:
return jsonify({'success': False, 'error': 'URL is required'})
try:
# 设置默认 headers
default_headers = {
'Content-Type': 'application/json',
'Accept': 'application/json'
}
default_headers.update(headers)
# 发送请求
start_time = datetime.now()
if method == 'GET':
response = requests.get(url, headers=default_headers, timeout=30)
elif method == 'POST':
json_body = json.loads(body) if body else None
response = requests.post(url, headers=default_headers, json=json_body, timeout=30)
elif method == 'PUT':
json_body = json.loads(body) if body else None
response = requests.put(url, headers=default_headers, json=json_body, timeout=30)
elif method == 'DELETE':
response = requests.delete(url, headers=default_headers, timeout=30)
elif method == 'PATCH':
json_body = json.loads(body) if body else None
response = requests.patch(url, headers=default_headers, json=json_body, timeout=30)
else:
return jsonify({'success': False, 'error': f'Unsupported method: {method}'})
end_time = datetime.now()
duration = (end_time - start_time).total_seconds() * 1000 # 毫秒
# 解析响应
try:
response_body = response.json()
except:
response_body = response.text
return jsonify({
'success': True,
'status_code': response.status_code,
'duration': round(duration, 2),
'headers': dict(response.headers),
'body': response_body,
'url': response.url
})
except requests.exceptions.ConnectionError as e:
return jsonify({'success': False, 'error': f'Connection error: {str(e)}'})
except requests.exceptions.Timeout:
return jsonify({'success': False, 'error': 'Request timeout'})
except Exception as e:
return jsonify({'success': False, 'error': str(e)})
@app.route('/api/history', methods=['GET', 'POST'])
def history():
"""获取/保存测试历史"""
history_file = 'api_test_history.json'
if request.method == 'POST':
# 保存测试历史
data = request.get_json()
history_data = []
if os.path.exists(history_file):
try:
with open(history_file, 'r', encoding='utf-8') as f:
history_data = json.load(f)
except:
history_data = []
# 添加时间戳
data['timestamp'] = datetime.now().isoformat()
history_data.append(data)
# 只保留最近 100 条
history_data = history_data[-100:]
with open(history_file, 'w', encoding='utf-8') as f:
json.dump(history_data, f, ensure_ascii=False, indent=2)
return jsonify({'success': True})
# 获取测试历史
if os.path.exists(history_file):
try:
with open(history_file, 'r', encoding='utf-8') as f:
history_data = json.load(f)
return jsonify(history_data[-50:]) # 返回最近 50 条
except:
pass
return jsonify([])
def create_template():
"""创建 HTML 模板"""
template_dir = os.path.join(os.path.dirname(__file__), 'templates')
if not os.path.exists(template_dir):
os.makedirs(template_dir)
template_path = os.path.join(template_dir, 'api_tester.html')
# 删除旧模板（如果存在），确保使用最新版本
if os.path.exists(template_path):
try:
os.remove(template_path)
except:
pass
html_content = '''<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>API 接口测试工具</title>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: #f5f7fa;
color: #333;
}
.header {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 20px;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
}
.header h1 {
font-size: 24px;
margin-bottom: 15px;
}
.host-config {
display: flex;
align-items: center;
gap: 10px;
flex-wrap: wrap;
}
.host-config label {
font-weight: 500;
}
.host-config input {
padding: 8px 12px;
border: none;
border-radius: 4px;
font-size: 14px;
min-width: 300px;
}
.host-config button {
padding: 8px 20px;
background: #48bb78;
color: white;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 14px;
transition: background 0.3s;
}
.host-config button:hover {
background: #38a169;
}
.container {
max-width: 1400px;
margin: 0 auto;
padding: 20px;
display: grid;
grid-template-columns: 1fr 1fr;
gap: 20px;
}
@media (max-width: 1024px) {
.container {
grid-template-columns: 1fr;
}
}
.panel {
background: white;
border-radius: 8px;
box-shadow: 0 2px 8px rgba(0,0,0,0.08);
overflow: hidden;
}
.panel-header {
background: #f8fafc;
padding: 15px 20px;
border-bottom: 1px solid #e2e8f0;
font-weight: 600;
font-size: 16px;
display: flex;
justify-content: space-between;
align-items: center;
}
.panel-body {
padding: 20px;
max-height: 600px;
overflow-y: auto;
}
.api-list {
list-style: none;
}
.api-item {
border: 1px solid #e2e8f0;
border-radius: 6px;
margin-bottom: 10px;
padding: 12px;
cursor: pointer;
transition: all 0.2s;
position: relative;
}
.api-item:hover {
border-color: #667eea;
box-shadow: 0 2px 8px rgba(102, 126, 234, 0.15);
}
.api-item.active {
border-color: #667eea;
background: #f7fafc;
}
.api-method {
display: inline-block;
padding: 2px 8px;
border-radius: 3px;
font-size: 12px;
font-weight: 600;
margin-right: 8px;
}
.api-method.GET { background: #48bb78; color: white; }
.api-method.POST { background: #4299e1; color: white; }
.api-method.PUT { background: #ed8936; color: white; }
.api-method.DELETE { background: #f56565; color: white; }
.api-method.PATCH { background: #9f7aea; color: white; }
.api-method.* { background: #718096; color: white; }
.api-url {
font-family: monospace;
font-size: 13px;
color: #4a5568;
word-break: break-all;
}
.api-desc {
font-size: 12px;
color: #718096;
margin-top: 4px;
}
.test-form {
display: flex;
flex-direction: column;
gap: 15px;
}
.form-group {
display: flex;
flex-direction: column;
gap: 5px;
}
.form-group label {
font-weight: 500;
font-size: 14px;
color: #4a5568;
}
.form-group input,
.form-group select,
.form-group textarea {
padding: 10px;
border: 1px solid #e2e8f0;
border-radius: 4px;
font-size: 14px;
font-family: inherit;
}
.form-group input:focus,
.form-group select:focus,
.form-group textarea:focus {
outline: none;
border-color: #667eea;
}
.form-group textarea {
min-height: 150px;
resize: vertical;
font-family: monospace;
}
.form-row {
display: grid;
grid-template-columns: 120px 1fr;
gap: 10px;
align-items: center;
}
.btn-primary {
padding: 12px 24px;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 14px;
font-weight: 500;
transition: opacity 0.3s;
}
.btn-primary:hover {
opacity: 0.9;
}
.btn-primary:disabled {
opacity: 0.6;
cursor: not-allowed;
}
.response-panel {
margin-top: 20px;
border-top: 1px solid #e2e8f0;
padding-top: 20px;
}
.response-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 10px;
}
.status-badge {
padding: 4px 12px;
border-radius: 12px;
font-size: 12px;
font-weight: 600;
}
.status-success {
background: #c6f6d5;
color: #22543d;
}
.status-error {
background: #fed7d7;
color: #742a2a;
}
.response-body {
background: #f7fafc;
border: 1px solid #e2e8f0;
border-radius: 4px;
padding: 15px;
font-family: monospace;
font-size: 13px;
white-space: pre-wrap;
word-break: break-all;
max-height: 400px;
overflow-y: auto;
}
.empty-state {
text-align: center;
padding: 40px;
color: #a0aec0;
}
.loading {
display: inline-block;
width: 20px;
height: 20px;
border: 2px solid #e2e8f0;
border-top-color: #667eea;
border-radius: 50%;
animation: spin 0.8s linear infinite;
}
@keyframes spin {
to { transform: rotate(360deg); }
}
.stats {
font-size: 12px;
color: #718096;
}
.filter-box {
margin-bottom: 15px;
}
.filter-box input {
width: 100%;
padding: 10px;
border: 1px solid #e2e8f0;
border-radius: 4px;
font-size: 14px;
}
</style>
</head>
<body>
<div class="header">
<h1>🔧 API 接口测试工具</h1>
<div class="host-config">
<label>Host 配置:</label>
<input type="text" id="hostInput" placeholder="http://localhost:8080" value="">
<button onclick="saveHost()">保存配置</button>
<span id="hostStatus" style="font-size: 12px;"></span>
</div>
</div>
<div class="container">
<!-- 左侧：接口列表 -->
<div class="panel">
<div class="panel-header">
<span>📋 接口列表</span>
<span class="stats" id="apiCount">加载中...</span>
</div>
<div class="panel-body">
<div class="filter-box">
<input type="text" id="filterInput" placeholder="🔍 搜索接口..." oninput="filterApis()">
</div>
<ul class="api-list" id="apiList">
<li class="empty-state">加载中...</li>
</ul>
</div>
</div>
<!-- 右侧：测试面板 -->
<div class="panel">
<div class="panel-header">
<span>🚀 接口测试</span>
</div>
<div class="panel-body">
<div id="testPanel">
<div class="empty-state">
<p>👈 请从左侧选择一个接口进行测试</p>
</div>
</div>
</div>
</div>
</div>
<script>
let apiRecords = [];
let currentHost = '';
let selectedApi = null;
const HOST_STORAGE_KEY = 'api_tester_host';
const DEFAULT_HOST = window.location.origin; // 默认使用当前访问的 host
// 页面加载时初始化
document.addEventListener('DOMContentLoaded', function() {
loadConfig();
loadApiRecords();
});
// 从 localStorage 加载 Host
function loadHostFromStorage() {
try {
const savedHost = localStorage.getItem(HOST_STORAGE_KEY);
if (savedHost) {
return savedHost;
}
} catch (e) {
console.error('读取 localStorage 失败:', e);
}
return null;
}
// 保存 Host 到 localStorage
function saveHostToStorage(host) {
try {
localStorage.setItem(HOST_STORAGE_KEY, host);
} catch (e) {
console.error('保存到 localStorage 失败:', e);
}
}
// 加载配置
async function loadConfig() {
// 首先尝试从 localStorage 读取
let host = loadHostFromStorage();
// 如果 localStorage 没有，使用后端配置的默认值
if (!host) {
try {
const response = await fetch('/api/config');
const data = await response.json();
host = data.host;
} catch (error) {
console.error('加载后端配置失败:', error);
}
}
// 如果还是没有，使用当前访问的 host
if (!host) {
host = DEFAULT_HOST;
}
currentHost = host;
document.getElementById('hostInput').value = currentHost;
}
// 保存 Host 配置
async function saveHost() {
const hostInput = document.getElementById('hostInput');
const host = hostInput.value.trim();
if (!host) {
showStatus('请输入有效的 Host', 'error');
return;
}
// 确保 host 格式正确
let formattedHost = host;
if (!formattedHost.startsWith('http://') && !formattedHost.startsWith('https://')) {
formattedHost = 'http://' + formattedHost;
}
// 保存到 localStorage
saveHostToStorage(formattedHost);
// 更新当前 host
currentHost = formattedHost;
// 同步到后端（可选，保持兼容性）
try {
await fetch('/api/config', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ host: formattedHost })
});
} catch (error) {
console.error('同步到后端失败:', error);
}
showStatus('✓ 配置已保存', 'success');
// 刷新接口列表中的测试 URL
loadApiRecords();
}
// 显示状态信息
function showStatus(message, type) {
const statusEl = document.getElementById('hostStatus');
statusEl.textContent = message;
statusEl.style.color = type === 'success' ? '#48bb78' : '#f56565';
setTimeout(() => {
statusEl.textContent = '';
}, 3000);
}
// 加载 API 记录
async function loadApiRecords() {
try {
const response = await fetch('/api/records');
apiRecords = await response.json();
renderApiList(apiRecords);
} catch (error) {
console.error('加载 API 记录失败:', error);
document.getElementById('apiList').innerHTML =
'<li class="empty-state">加载失败，请检查 Excel 文件是否存在</li>';
}
}
// 渲染接口列表
function renderApiList(records) {
const listEl = document.getElementById('apiList');
const countEl = document.getElementById('apiCount');
countEl.textContent = `共 ${records.length} 个接口`;
if (records.length === 0) {
listEl.innerHTML = '<li class="empty-state">暂无接口记录</li>';
return;
}
listEl.innerHTML = records.map((record, index) => {
const method = (record.method || '*').toUpperCase();
const url = record.url_pattern || '';
const desc = record.description || '';
const testUrl = record.test_url || '';
return `
<li class="api-item ${selectedApi === index ? 'active' : ''}"
onclick="selectApi(${index})"
data-url="${url.toLowerCase()}">
<span class="api-method ${method}">${method}</span>
<span class="api-url">${escapeHtml(url)}</span>
${desc ? `<div class="api-desc">${escapeHtml(desc)}</div>` : ''}
</li>
`;
}).join('');
}
// 搜索过滤
function filterApis() {
const keyword = document.getElementById('filterInput').value.toLowerCase();
const filtered = apiRecords.filter(record => {
const url = (record.url_pattern || '').toLowerCase();
const method = (record.method || '').toLowerCase();
const desc = (record.description || '').toLowerCase();
return url.includes(keyword) || method.includes(keyword) || desc.includes(keyword);
});
renderApiList(filtered);
}
// 选择接口
function selectApi(index) {
selectedApi = index;
const record = apiRecords[index];
// 更新列表样式
document.querySelectorAll('.api-item').forEach((el, i) => {
el.classList.toggle('active', i === index);
});
// 渲染测试表单
renderTestForm(record);
}
// 渲染测试表单
function renderTestForm(record) {
const method = (record.method || 'GET').toUpperCase();
const testUrl = record.test_url || '';
const requestBody = record.request_body || '';
// 尝试格式化请求 body
let formattedBody = '';
if (requestBody) {
try {
const parsed = JSON.parse(requestBody);
formattedBody = JSON.stringify(parsed, null, 2);
} catch {
formattedBody = requestBody;
}
}
const panelEl = document.getElementById('testPanel');
panelEl.innerHTML = `
<div class="test-form">
<div class="form-row">
<label>请求方法</label>
<select id="testMethod">
<option value="GET" ${method === 'GET' ? 'selected' : ''}>GET</option>
<option value="POST" ${method === 'POST' ? 'selected' : ''}>POST</option>
<option value="PUT" ${method === 'PUT' ? 'selected' : ''}>PUT</option>
<option value="DELETE" ${method === 'DELETE' ? 'selected' : ''}>DELETE</option>
<option value="PATCH" ${method === 'PATCH' ? 'selected' : ''}>PATCH</option>
</select>
</div>
<div class="form-group">
<label>请求 URL</label>
<input type="text" id="testUrl" value="${escapeHtml(testUrl)}" placeholder="http://...">
</div>
<div class="form-group">
<label>请求 Headers (JSON 格式)</label>
<textarea id="testHeaders" placeholder='{"Authorization": "Bearer xxx"}'>{
"Content-Type": "application/json"
}</textarea>
</div>
<div class="form-group">
<label>请求 Body (JSON 格式)</label>
<textarea id="testBody" placeholder="{}">${escapeHtml(formattedBody)}</textarea>
</div>
<div class="btn-group" style="display: flex; gap: 10px;">
<button class="btn-primary" onclick="sendRequestBackend()" id="sendBtnBackend" style="flex: 1;">
发送请求（后端）
</button>
<button class="btn-primary" onclick="sendRequestFrontend()" id="sendBtnFrontend" style="flex: 1; background: linear-gradient(135deg, #48bb78 0%, #38a169 100%);">
发送请求（前端）
</button>
</div>
<div id="responsePanel" style="display: none;">
<div class="response-header">
<span class="status-badge" id="statusBadge"></span>
<span id="durationBadge"></span>
<span id="requestModeBadge" style="font-size: 12px; color: #718096; margin-left: 10px;"></span>
</div>
<div class="response-body" id="responseBody"></div>
</div>
</div>
`;
}
// 获取请求参数
function getRequestParams(useCurrentHost = false) {
let url = document.getElementById('testUrl').value.trim();
const method = document.getElementById('testMethod').value;
const headersText = document.getElementById('testHeaders').value.trim();
const bodyText = document.getElementById('testBody').value.trim();
if (!url) {
alert('请输入请求 URL');
return null;
}
// 如果使用当前设置的 Host，替换 URL 中的 Host 部分
if (useCurrentHost && currentHost) {
try {
const urlObj = new URL(url);
// 提取路径和查询参数
const path = urlObj.pathname + urlObj.search;
// 使用当前设置的 Host 构建新 URL
url = currentHost.replace(/\/$/, '') + path;
} catch (e) {
// URL 解析失败，可能是相对路径，直接拼接
if (!url.startsWith('http')) {
url = currentHost.replace(/\/$/, '') + (url.startsWith('/') ? url : '/' + url);
}
}
}
// 解析 headers
let headers = {};
if (headersText) {
try {
headers = JSON.parse(headersText);
} catch {
alert('Headers 格式错误，请使用 JSON 格式');
return null;
}
}
return { url, method, headers, bodyText };
}
// 显示响应结果
function displayResponse(success, statusCode, duration, body, mode) {
const panel = document.getElementById('responsePanel');
const badge = document.getElementById('statusBadge');
const durationEl = document.getElementById('durationBadge');
const modeEl = document.getElementById('requestModeBadge');
const bodyEl = document.getElementById('responseBody');
panel.style.display = 'block';
if (success) {
badge.className = 'status-badge status-success';
badge.textContent = `${statusCode} OK`;
} else {
badge.className = 'status-badge status-error';
badge.textContent = statusCode ? `${statusCode} Error` : 'Error';
}
durationEl.textContent = duration ? `⏱️ ${duration}ms` : '';
modeEl.textContent = mode === 'frontend' ? '(前端请求)' : '(后端请求)';
if (typeof body === 'object') {
bodyEl.textContent = JSON.stringify(body, null, 2);
} else {
bodyEl.textContent = body || 'No response body';
}
}
// 后端发送请求
async function sendRequestBackend() {
const params = getRequestParams(true); // 使用当前设置的 Host
if (!params) return;
const { url, method, headers, bodyText } = params;
const btn = document.getElementById('sendBtnBackend');
btn.disabled = true;
btn.innerHTML = '<span class="loading"></span> 请求中...';
const startTime = Date.now();
try {
const response = await fetch('/api/test', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
url: url,
method: method,
headers: headers,
body: bodyText
})
});
const data = await response.json();
const duration = Date.now() - startTime;
if (data.success) {
displayResponse(true, data.status_code, duration, data.body, 'backend');
} else {
displayResponse(false, null, duration, data.error || 'Unknown error', 'backend');
}
// 保存测试历史
saveTestHistory({
url: url,
method: method,
success: data.success,
status_code: data.status_code,
mode: 'backend'
});
} catch (error) {
displayResponse(false, null, Date.now() - startTime, error.message, 'backend');
} finally {
btn.disabled = false;
btn.textContent = '发送请求（后端）';
}
}
// 前端发送请求
async function sendRequestFrontend() {
const params = getRequestParams(true); // 使用当前设置的 Host
if (!params) return;
const { url, method, headers, bodyText } = params;
const btn = document.getElementById('sendBtnFrontend');
btn.disabled = true;
btn.innerHTML = '<span class="loading"></span> 请求中...';
const startTime = Date.now();
try {
// 构建 fetch 选项
const fetchOptions = {
method: method,
headers: headers,
credentials: 'include' // 携带 cookie
};
// 添加 body（非 GET 请求）
if (method !== 'GET' && bodyText) {
// 如果 Content-Type 是 application/json，解析 body 为 JSON
const contentType = headers['Content-Type'] || headers['content-type'] || '';
if (contentType.includes('application/json')) {
try {
fetchOptions.body = JSON.stringify(JSON.parse(bodyText));
} catch {
fetchOptions.body = bodyText;
}
} else {
fetchOptions.body = bodyText;
}
}
const response = await fetch(url, fetchOptions);
const duration = Date.now() - startTime;
// 解析响应 body
let responseBody;
const responseContentType = response.headers.get('content-type') || '';
if (responseContentType.includes('application/json')) {
responseBody = await response.json();
} else {
responseBody = await response.text();
}
displayResponse(response.ok, response.status, duration, responseBody, 'frontend');
// 保存测试历史
saveTestHistory({
url: url,
method: method,
success: response.ok,
status_code: response.status,
mode: 'frontend'
});
} catch (error) {
const duration = Date.now() - startTime;
let errorMessage = error.message;
// 处理常见的跨域错误
if (error.name === 'TypeError' && errorMessage.includes('Failed to fetch')) {
errorMessage = '请求失败，可能是跨域问题或网络错误\\n' +
'1. 确保目标服务器允许跨域请求\\n' +
'2. 或者使用【后端发送请求】模式';
}
displayResponse(false, null, duration, errorMessage, 'frontend');
} finally {
btn.disabled = false;
btn.textContent = '发送请求（前端）';
}
}
// 保存测试历史
async function saveTestHistory(data) {
try {
await fetch('/api/history', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(data)
});
} catch (error) {
console.error('保存历史失败:', error);
}
}
// HTML 转义
function escapeHtml(text) {
if (!text) return '';
const div = document.createElement('div');
div.textContent = text;
return div.innerHTML;
}
</script>
</body>
</html>'''
with open(template_path, 'w', encoding='utf-8') as f:
f.write(html_content)
print(f"[创建] HTML 模板已创建: {template_path}")
if __name__ == '__main__':
# 创建模板文件
create_template()
print("=" * 70)
print("API 接口测试 Web 工具")
print("=" * 70)
print(f"[启动] 正在启动服务...")
print(f"[访问] 请在浏览器中打开: http://localhost:5000")
print(f"[提示] 确保已运行 api_recorder_excel.py 生成了当天的记录文件")
print("=" * 70)
# 运行 Flask 应用
app.run(host='0.0.0.0', port=5000, debug=True)
发表评论