You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
203 lines
7.3 KiB
203 lines
7.3 KiB
"""
|
|
真实环境批量联调探测。
|
|
|
|
用途：拿真实 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()
|
|
|