Browse Source
实网联调（job.youthome.com.cn，学生号 19900009021）发现前端已接端点与后端契约不同构：
- 岗位 DTO：后端 title/company.name/薪资单位元，前端 positionName/companyName/K；
新增 mapBackendJobItem 统一转换（列表/收藏/详情三处共用），缺失字段留空不拿 mock 顶替
- 偏好 DTO：后端多选数组+英文 token，前端单值中文；读写双向映射，
salary/status 后端无字段不发送
- 评价汇总：totalReviews→total、ratingDistribution{rating5..}→distribution[]
- 分页：后端全站 0-based，前端按 1-based 传会丢第一页（/applications/me 首页恒空）；
新增 toBackendPage/toFrontendPage 集中转换
- 新增 scripts/audit_coverage.py（248 端点覆盖率审计）与 live_probe.py（实网批量探测）
tsc 0 错误；jest 518 全过
master
doublex001
4 weeks ago
8 changed files with 752 additions and 71 deletions
@ -0,0 +1,114 @@ |
|||
""" |
|||
前后端接口覆盖率审计。 |
|||
|
|||
用途：比对后端 OpenAPI（scripts/openapi.json）的全部操作与前端 src/api/*.ts 里 |
|||
真实发出的 request() 调用，得出「已接入 / 未接入」清单，按域分组统计。 |
|||
|
|||
匹配规则： |
|||
- 路径归一化：`{jobId}` 与模板串 `${jobId}` 一律折成 `{}`，避免占位符命名差异误判。 |
|||
- 方法：request 未显式传 method 时按 GET 计（http.ts 默认 GET）。 |
|||
|
|||
用法：python scripts/audit_coverage.py |
|||
""" |
|||
import json |
|||
import re |
|||
from collections import defaultdict |
|||
|
|||
# 权威端点源 = 后端给的 markdown 文档（scripts/openapi.json 是 9/28 的旧快照， |
|||
# 比 md 少 28 个端点，含 /auth/identity/select，拿它当基准会把已接接口误判成幽灵） |
|||
SPEC_MD = "docs/职来后端接口.md" |
|||
API_FILES = ["src/api/api.ts", "src/api/auth.ts", "src/api/http.ts"] |
|||
|
|||
HTTP_METHODS = {"get", "post", "put", "patch", "delete"} |
|||
|
|||
|
|||
def norm(path: str) -> str: |
|||
"""把 {id} / ${id} 统一折成 {}""" |
|||
p = re.sub(r"\$\{[^}]*\}", "{}", path) |
|||
p = re.sub(r"\{[^}]*\}", "{}", p) |
|||
if not p.startswith("/"): |
|||
p = "/" + p |
|||
return p.rstrip("/") or "/" |
|||
|
|||
|
|||
def load_spec(): |
|||
"""从后端 markdown 文档解析端点：`## GET /jobs/search（标题）` 行给出方法与路径""" |
|||
# md 结构是「## GET 中文标题」+ 下一行「GET /real/path」，标题行里的中文不是路径， |
|||
# 真正可请求的行就是行首 `METHOD /path` 那一行。 |
|||
ops = {} |
|||
entry_re = re.compile(r"^(GET|POST|PUT|PATCH|DELETE)\s+(/\S*)\s*$") |
|||
with open(SPEC_MD, encoding="utf-8") as f: |
|||
for ln in f: |
|||
m = entry_re.match(ln) |
|||
if not m: |
|||
continue |
|||
ops[(m.group(1).upper(), norm(m.group(2)))] = ln.strip() |
|||
return ops |
|||
|
|||
|
|||
def load_frontend(): |
|||
"""从 api.ts / auth.ts 抽取 request() 调用：返回 [(method, normpath, funcname, file, line)] |
|||
|
|||
注意：泛型参数常跨多行书写（如 request<{ accessToken: string;\n ... }>('/auth/login')）， |
|||
因此必须在「整文件文本」上匹配，逐行扫描会整片漏掉。 |
|||
""" |
|||
calls = [] |
|||
call_re = re.compile( |
|||
r"request\s*(?:<[^>]*>)?\s*\(\s*(?P<q>['\"`])(?P<path>[^'\"`]+)(?P=q)", |
|||
re.DOTALL, |
|||
) |
|||
func_re = re.compile( |
|||
r"^\s*(?:export\s+)?(?:async\s+)?function\s+(?P<name>\w+)\s*\(", re.MULTILINE |
|||
) |
|||
for file in API_FILES: |
|||
with open(file, encoding="utf-8") as f: |
|||
text = f.read() |
|||
# 函数名 → 起始偏移，用于把调用归属到所在函数 |
|||
func_starts = [(m.start(), m.group("name")) for m in func_re.finditer(text)] |
|||
for cm in call_re.finditer(text): |
|||
path = norm(cm.group("path")) |
|||
after = text[cm.end() : cm.end() + 200] |
|||
mm = re.search(r"method:\s*['\"](GET|POST|PUT|PATCH|DELETE)['\"]", after) |
|||
method = mm.group(1) if mm else "GET" |
|||
name = "(module)" |
|||
for start, fname in func_starts: |
|||
if start <= cm.start(): |
|||
name = fname |
|||
else: |
|||
break |
|||
line = text.count("\n", 0, cm.start()) + 1 |
|||
calls.append((method, path, name, file, line)) |
|||
return calls |
|||
|
|||
|
|||
def main(): |
|||
ops = load_spec() |
|||
calls = load_frontend() |
|||
used = {(m, p) for m, p, *_ in calls} |
|||
|
|||
covered = sorted(k for k in ops if k in used) |
|||
missing = sorted(k for k in ops if k not in used) |
|||
phantom = sorted({(m, p, fn, f, ln) for m, p, fn, f, ln in calls if (m, p) not in ops}) |
|||
|
|||
print(f"后端端点总数: {len(ops)}") |
|||
print(f"前端 request() 调用数: {len(calls)}（去重路径 {len(used)}）") |
|||
print(f"已接入: {len(covered)} 未接入: {len(missing)} 幽灵调用(后端无此端点): {len(phantom)}") |
|||
print() |
|||
|
|||
print("=== 幽灵调用（前端发了但后端 spec 没有，必改）===") |
|||
for m, p, fn, f, ln in phantom: |
|||
print(f" {m:6} {p:52} {fn} ({f}:{ln})") |
|||
print() |
|||
|
|||
by_domain = defaultdict(list) |
|||
for m, p in missing: |
|||
by_domain[p.strip("/").split("/")[0]].append((m, p)) |
|||
print("=== 未接入端点（按域）===") |
|||
for d in sorted(by_domain, key=lambda k: -len(by_domain[k])): |
|||
print(f"\n[{d}] {len(by_domain[d])} 个") |
|||
for m, p in sorted(by_domain[d]): |
|||
print(f" {m:6} {p}") |
|||
|
|||
|
|||
if __name__ == "__main__": |
|||
main() |
|||
@ -0,0 +1,203 @@ |
|||
""" |
|||
真实环境批量联调探测。 |
|||
|
|||
用途：拿真实 token 打后端 GET 端点，记录 HTTP 状态 / 业务 code / data 结构， |
|||
产出「前端已接但后端报错」与「后端已有但前端未接且能拿到真实字段」两类事实， |
|||
作为联调问题清单的依据。 |
|||
|
|||
占位符 id 通过先打列表端点动态取得（/jobs/search → jobId 等），取不到就跳过并标记 need-id。 |
|||
|
|||
用法：python scripts/live_probe.py |
|||
""" |
|||
import json |
|||
import re |
|||
import urllib.error |
|||
import urllib.request |
|||
from concurrent.futures import ThreadPoolExecutor |
|||
|
|||
BASE = "https://job.youthome.com.cn/api/v1" |
|||
PHONE = "19900009021" |
|||
PASSWORD = "ZhilaiTest!2026#A" |
|||
SPEC_MD = "docs/职来后端接口.md" |
|||
|
|||
ID_KEYS = ("id", "jobId", "applicationId", "conversationId", "interviewId", |
|||
"companyId", "candidateId", "reviewId", "notificationId", "taskId", |
|||
"attemptId", "offerId", "reportId", "blockId", "messageId", "userId", |
|||
"itemId", "cityCode", "calculationId", "replyId", "appealId", "knowledgeId") |
|||
|
|||
|
|||
def req(method, path, token=None, body=None, timeout=15): |
|||
url = BASE + path |
|||
data = json.dumps(body).encode() if body is not None else None |
|||
r = urllib.request.Request(url, data=data, method=method) |
|||
r.add_header("Content-Type", "application/json") |
|||
if token: |
|||
r.add_header("Authorization", "Bearer " + token) |
|||
try: |
|||
with urllib.request.urlopen(r, timeout=timeout) as resp: |
|||
raw = resp.read().decode(errors="replace") |
|||
status = resp.status |
|||
except urllib.error.HTTPError as e: |
|||
raw = e.read().decode(errors="replace") |
|||
status = e.code |
|||
except Exception as e: # 网络层 / 超时 |
|||
return {"http": None, "err": str(e), "code": None, "data": None} |
|||
try: |
|||
j = json.loads(raw) |
|||
except Exception: |
|||
return {"http": status, "err": raw[:120], "code": None, "data": None} |
|||
if isinstance(j, dict): |
|||
return {"http": status, "code": j.get("code"), "message": j.get("message"), |
|||
"data": j.get("data")} |
|||
return {"http": status, "code": None, "data": j} |
|||
|
|||
|
|||
def shape(value, depth=0): |
|||
"""递归描述返回结构（只到 2 层，够看字段命名）""" |
|||
if depth > 2: |
|||
return "..." |
|||
if isinstance(value, dict): |
|||
return {k: shape(v, depth + 1) for k, v in list(value.items())[:14]} |
|||
if isinstance(value, list): |
|||
if not value: |
|||
return "[]" |
|||
return [shape(value[0], depth + 1)] |
|||
return type(value).__name__ |
|||
|
|||
|
|||
def load_endpoints(method_filter=None): |
|||
out = [] |
|||
entry_re = re.compile(r"^(GET|POST|PUT|PATCH|DELETE)\s+(/\S*)\s*$") |
|||
with open(SPEC_MD, encoding="utf-8") as f: |
|||
for ln in f: |
|||
m = entry_re.match(ln) |
|||
if not m: |
|||
continue |
|||
if method_filter and m.group(1) not in method_filter: |
|||
continue |
|||
out.append((m.group(1), m.group(2))) |
|||
return sorted(set(out)) |
|||
|
|||
|
|||
def pick_ids(token): |
|||
"""从列表端点抓真实 id，供带占位符的端点使用""" |
|||
ids = {} |
|||
plan = [ |
|||
("/jobs/search?page=1&pageSize=1", "jobId"), |
|||
("/applications/me?page=1&pageSize=1", "applicationId"), |
|||
("/conversations", "conversationId"), |
|||
("/notifications?page=1&pageSize=1", "notificationId"), |
|||
("/offers?page=1&pageSize=1", "offerId"), |
|||
("/assessments/me", "attemptId"), |
|||
("/reports/me?page=1&pageSize=1", "reportId"), |
|||
("/blocks/me", "blockId"), |
|||
("/companies/me", "companyId"), |
|||
("/interviews/me?page=1&pageSize=1", "interviewId"), |
|||
] |
|||
for path, key in plan: |
|||
res = req("GET", path, token) |
|||
data = res.get("data") |
|||
if isinstance(data, dict): |
|||
for listkey in ("items", "records", "list", "content"): |
|||
if isinstance(data.get(listkey), list): |
|||
data = data[listkey] |
|||
break |
|||
if isinstance(data, list) and data: |
|||
first = data[0] |
|||
if isinstance(first, dict): |
|||
ids[key] = first.get("id") or first.get(key) |
|||
elif isinstance(data, dict): |
|||
ids[key] = data.get("id") |
|||
if ids.get(key): |
|||
ids[key.split("Id")[0] + "Id"] = ids[key] |
|||
# jobId 兜底：/jobs 列表 |
|||
if not ids.get("jobId"): |
|||
res = req("GET", "/jobs?page=1&pageSize=1", token) |
|||
d = res.get("data") |
|||
if isinstance(d, dict): |
|||
for k in ("items", "records", "list", "content"): |
|||
if isinstance(d.get(k), list): |
|||
d = d[k] |
|||
break |
|||
if isinstance(d, list) and d and isinstance(d[0], dict): |
|||
ids["jobId"] = d[0].get("id") |
|||
return ids |
|||
|
|||
|
|||
def fill(path, ids): |
|||
"""把 {xxx} 占位符换成真实 id；换不掉返回 None""" |
|||
def sub(m): |
|||
name = m.group(1) |
|||
for k in (name, name + "Id"): |
|||
if ids.get(k): |
|||
return str(ids[k]) |
|||
# 退而求其次：任意同名前缀 |
|||
base = name.replace("Id", "") |
|||
for k, v in ids.items(): |
|||
if k.replace("Id", "") == base and v: |
|||
return str(v) |
|||
return None |
|||
filled = re.sub(r"\{([^}]+)\}", lambda m: sub(m) or "{" + m.group(1) + "}", path) |
|||
return None if "{" in filled else filled |
|||
|
|||
|
|||
def main(): |
|||
login = req("POST", "/auth/password/login", body={"phone": PHONE, "password": PASSWORD}) |
|||
token = (login.get("data") or {}).get("accessToken") |
|||
if not token: |
|||
print("登录失败:", login) |
|||
return |
|||
print(f"登录成功 role={(login.get('data') or {}).get('role')} userId={(login.get('data') or {}).get('userId')}") |
|||
|
|||
ids = pick_ids(token) |
|||
print("抓到的 id:", {k: v for k, v in ids.items() if v}) |
|||
|
|||
endpoints = load_endpoints({"GET"}) |
|||
targets = [] |
|||
skipped = [] |
|||
for method, path in endpoints: |
|||
if "{" in path: |
|||
filled = fill(path, ids) |
|||
if filled: |
|||
targets.append((method, path, filled)) |
|||
else: |
|||
skipped.append(path) |
|||
else: |
|||
targets.append((method, path, path)) |
|||
|
|||
print(f"\nGET 端点 {len(endpoints)} 个：可探测 {len(targets)}，缺 id 跳过 {len(skipped)}") |
|||
|
|||
results = [] |
|||
|
|||
def probe(item): |
|||
method, raw, filled = item |
|||
res = req(method, filled, token) |
|||
return { |
|||
"method": method, |
|||
"path": raw, |
|||
"url": filled, |
|||
"http": res.get("http"), |
|||
"code": res.get("code"), |
|||
"message": res.get("message"), |
|||
"err": res.get("err"), |
|||
"shape": shape(res.get("data")) if res.get("http") == 200 else None, |
|||
} |
|||
|
|||
with ThreadPoolExecutor(max_workers=8) as ex: |
|||
results = list(ex.map(probe, targets)) |
|||
|
|||
ok = [r for r in results if r["http"] == 200 and r["code"] in (200, 0, None)] |
|||
bad = [r for r in results if r not in ok] |
|||
print(f"\n探测完成：正常 {len(ok)} / 异常 {len(bad)}") |
|||
print("\n=== 异常端点（HTTP != 200 或 code 非 200）===") |
|||
for r in sorted(bad, key=lambda x: x["path"]): |
|||
print(f" {r['path']:58} http={r['http']} code={r['code']} msg={(r.get('message') or r.get('err') or '')[:60]}") |
|||
|
|||
with open(".workbuddy/tmp/live_probe.json", "w", encoding="utf-8") as f: |
|||
json.dump({"ids": ids, "results": results, "skipped": skipped}, f, |
|||
ensure_ascii=False, indent=2) |
|||
print("\n完整结果: .workbuddy/tmp/live_probe.json") |
|||
|
|||
|
|||
if __name__ == "__main__": |
|||
main() |
|||
Loading…
Reference in new issue