mydreamjob/scripts/transform_api.py

import json, re

with open("scripts/openapi.json", encoding="utf-8") as f:
    spec = json.load(f)
real_paths = set()
for p, methods in spec.get("paths", {}).items():
    for m in methods:
        real_paths.add(p)
real_base = set()
for p in real_paths:
    seg = [s for s in p.split("/") if s and "{" not in s and "$" not in s]
    if seg:
        real_base.add("/" + seg[0])

with open("src/api/api.ts", encoding="utf-8") as f:
    raw = f.read()
lines = raw.split("\n")  # keep as list, we will rejoin

func_re = re.compile(r"^(\s*)(?:export\s+)?(?:async\s+)?function\s+(\w+)\s*\(")
static_req = re.compile(r"request(?:<[^>]*>)?\(\s*['\"]([^'\"]+)['\"]")
tmpl_req = re.compile(r"request(?:<[^>]*>)?\(\s*`([^`]*)`")

# find function start indices
starts = []
for i, ln in enumerate(lines):
    m = func_re.match(ln)
    if m:
        starts.append((m.group(2), i))
func_ranges = []
for idx, (name, s) in enumerate(starts):
    e = starts[idx+1][1] if idx+1 < len(starts) else len(lines)
    func_ranges.append((name, s, e))

PATH_FIX_STATIC = {
    "fetchConversations": ("'/messages/conversations'", "'/conversations'"),
    "fetchNotifications": ("'/messages/notifications'", "'/notifications'"),
    "fetchInterviews": ("'/messages/interviews'", "'/interviews/me'"),
    "fetchJobPreference": ("'/profile/me/job-preference'", "'/profile/me/preferences'"),
    "saveJobPreference": ("'/profile/me/job-preference'", "'/profile/me/preferences'"),
    "fetchResume": ("'/resume'", "'/profile/me/resume'"),
    "fetchFavorites": ("'/profile/me/favorites'", "'/jobs/favorites'"),
}
JOBPROGRESS_OLD = "'/tools/job-progress', { method: 'GET', auth: false }"
JOBPROGRESS_NEW = "'/profile/me/career-progress', { method: 'GET' }"
TEMPLATE_FIX = {
    "fetchRankData": ("/jobs/rank", "/jobs/${jobId}/competition-ranking"),
    "fetchCandidateDetail": ("/enterprise/candidates/${candidateId}", "/company/candidates/${candidateId}"),
    "fetchInterviewScore": ("/enterprise/interviews/${interviewId}/score", "/interviews/${interviewId}/score"),
    "fetchInterviewScoreDraft": ("/enterprise/interviews/${interviewId}/score/draft", "/interviews/${interviewId}/score"),
    "fetchConversation": ("/chat/conversations/${id}", "/conversations/${id}"),
}

def get_info(sl):
    static = []; tmpl = []
    hmi = any("if (isMockMode())" in l for l in sl)
    hnmi = any("if (!isMockMode())" in l for l in sl)
    for l in sl:
        m = static_req.search(l)
        if m: static.append(m.group(1))
        t = tmpl_req.search(l)
        if t: tmpl.append(t.group(1))
    return static, tmpl, hmi, hnmi

def is_phantom(p):
    if p in real_paths: return False
    segs = [s for s in p.split("/") if s and "{" not in s and "$" not in s]
    base = "/" + segs[0] if segs else ""
    return base not in real_base

DAI_RE = re.compile(r'【待确认[^】]*】')

def strip_dai(line):
    if DAI_RE.search(line):
        return DAI_RE.sub('', line).rstrip() + "\n"
    return line

changed = {"pathfix": [], "revert": [], "strip_only": []}
new_full = []

def find_close(sl, open_i):
    depth = 0
    for j in range(open_i, len(sl)):
        depth += sl[j].count("{") - sl[j].count("}")
        if depth == 0 and j > open_i:
            return j
    return None

# preserve leading lines before the first function (imports, top-level consts, banners)
new_full = list(lines[0:func_ranges[0][1]])

for name, s, e in func_ranges:
    sl = lines[s:e]
    static, tmpl, hmi, hnmi = get_info(sl)
    in_openapi = any(p in real_paths for p in static) or any(not is_phantom(t) for t in tmpl)
    dai = any("【待确认】" in l for l in sl)
    is_pathfix = name in PATH_FIX_STATIC or name in TEMPLATE_FIX or name == "fetchJobProgress"
    phantom = (dai or any(is_phantom(p) for p in static) or any(is_phantom(t) for t in tmpl)) and not in_openapi

    if is_pathfix:
        out = []
        for ln in sl:
            if name in PATH_FIX_STATIC:
                o, n = PATH_FIX_STATIC[name]
                if o in ln: ln = ln.replace(o, n)
                ou = o.strip("'\"")
                nu = n.strip("'\"")
                if ou in ln: ln = ln.replace(ou, nu)
            elif name == "fetchJobProgress":
                if JOBPROGRESS_OLD in ln: ln = ln.replace(JOBPROGRESS_OLD, JOBPROGRESS_NEW)
                if "/tools/job-progress" in ln: ln = ln.replace("/tools/job-progress", "/profile/me/career-progress")
            elif name in TEMPLATE_FIX:
                o, n = TEMPLATE_FIX[name]
                if o in ln: ln = ln.replace(o, n)
                ou = o.replace("`", "")
                nu = n.replace("`", "")
                if ou in ln: ln = ln.replace(ou, nu)
            out.append(strip_dai(ln))
        new_full.extend(out)
        changed["pathfix"].append(name)
        continue

    if not phantom:
        if dai:
            new_full.extend(strip_dai(ln) for ln in sl)
        else:
            new_full.extend(sl)
        continue

    # REVERT (phantom -> always mock)
    if hnmi:
        out = []
        i = 0
        while i < len(sl):
            if "if (!isMockMode())" in sl[i]:
                ci = find_close(sl, i)
                i = (ci + 1) if ci is not None else i + 1
                continue
            out.append(strip_dai(sl[i]))
            i += 1
        new_full.extend(out)
        changed["revert"].append(name)
    else:
        # pattern A: collapse if(isMockMode()){...} and drop trailing return request
        out = []
        i = 0
        collapsed = False
        while i < len(sl):
            if "if (isMockMode())" in sl[i]:
                ci = find_close(sl, i)
                if ci is not None:
                    inner = sl[i+1:ci]
                    inner_out = []
                    for il in inner:
                        if "return request" in il:
                            continue
                        if il.lstrip().startswith("//") and "【待确认】" in il:
                            continue
                        inner_out.append(il)
                    inner_out = [l[2:] if l.startswith("    ") else l for l in inner_out]
                    out.append("  // 后端未提供 OpenAPI 契约：保持 mock（翻真实开关也不触网），待后端就绪再接真实分支")
                    out.extend(inner_out)
                    i = ci + 1
                    collapsed = True
                    continue
            if "return request" in sl[i]:
                i += 1
                continue
            out.append(strip_dai(sl[i]))
            i += 1
        if not collapsed:
            # dai but no isMockMode block and no real branch -> just strip
            changed["strip_only"].append(name)
        else:
            changed["revert"].append(name)
        new_full.extend(out)

# preserve trailing lines after the last function
new_full.extend(lines[func_ranges[-1][2]:])

with open("src/api/api.ts", "w", encoding="utf-8") as f:
    f.write("\n".join(new_full))

print("PATHFIX:", len(changed["pathfix"]))
print("REVERT:", len(changed["revert"]))
print("STRIP_ONLY:", len(changed["strip_only"]), changed["strip_only"])
print("--- revert ---")
for n in changed["revert"]:
    print("  ", n)