""" 真实环境批量联调探测。 用途：拿真实 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()