File size: 4,886 Bytes
89419ea | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | from __future__ import annotations
from typing import Any, Callable, Dict, Optional
POST_CREATE_RESUME_SOURCE = "post_create_continue"
POST_CREATE_REENTERED_LOGIN_SOURCE = "post_create_continue_reentered_login"
def is_post_create_continue_url(url: Optional[str]) -> bool:
candidate = str(url or "").strip().lower()
if not candidate:
return False
return any(
token in candidate
for token in (
"/add-phone",
"/sign-in-with-chatgpt/",
"/consent",
"/workspace/select",
"/organization/select",
)
)
def extract_post_create_continue_url(data: Any) -> Optional[str]:
if isinstance(data, dict):
for key, value in data.items():
if isinstance(value, str):
candidate = value.strip()
key_text = str(key).strip().lower()
if key_text == "type" and candidate.lower() in {"add_phone", "add-phone"}:
return "https://auth.openai.com/add-phone"
if key_text in {"continue_url", "url", "location", "redirect_url"} and is_post_create_continue_url(candidate):
return candidate
if is_post_create_continue_url(candidate):
return candidate
candidate = extract_post_create_continue_url(value)
if candidate:
return candidate
elif isinstance(data, list):
for item in data:
candidate = extract_post_create_continue_url(item)
if candidate:
return candidate
return None
def get_locked_post_create_continue_url(workspace_context: Dict[str, Any]) -> str:
return str(workspace_context.get("post_create_continue_url") or "").strip()
def lock_post_create_continue_url(
workspace_context: Dict[str, Any],
candidate: Optional[str],
*,
source_name: str,
log_fn: Callable[[str], None],
sanitize_url: Callable[[Optional[str]], str],
) -> bool:
candidate_text = str(candidate or "").strip()
if not is_post_create_continue_url(candidate_text):
return False
current_locked = get_locked_post_create_continue_url(workspace_context)
current_resume = str(workspace_context.get("resume_url") or "").strip()
if current_locked == candidate_text and current_resume == candidate_text:
workspace_context["post_create_continue_source"] = source_name
return True
if current_resume and current_resume != candidate_text:
log_fn(
"post_create_continue_overrode_cached_resume: "
f"current={sanitize_url(current_resume) or '-'}; "
f"new={sanitize_url(candidate_text)}"
)
if current_locked and current_locked != candidate_text:
log_fn(
"post_create_continue_upgraded: "
f"current={sanitize_url(current_locked)}; "
f"new={sanitize_url(candidate_text)}"
)
else:
log_fn(f"post_create_continue_locked: {sanitize_url(candidate_text)}")
workspace_context["post_create_continue_url"] = candidate_text
workspace_context["post_create_continue_source"] = source_name
workspace_context["continue_url"] = candidate_text
workspace_context["resume_url"] = candidate_text
workspace_context["resume_source"] = POST_CREATE_RESUME_SOURCE
return True
def should_preserve_locked_post_create_continue(
workspace_context: Dict[str, Any],
candidate: Optional[str],
) -> bool:
current_locked = get_locked_post_create_continue_url(workspace_context)
if not current_locked:
return False
current_resume = str(workspace_context.get("resume_url") or "").strip()
candidate_text = str(candidate or "").strip()
if not candidate_text or candidate_text == current_resume:
return False
if current_resume != current_locked:
return False
return not is_post_create_continue_url(candidate_text)
def build_post_create_failure_details(
*,
locked_url: Optional[str],
terminal_url: Optional[str],
resolution_error: Optional[str],
reentered_login: bool,
) -> Dict[str, Optional[str]]:
if reentered_login:
error_message = "账号已创建，但 post-create 续跑重新进入登录页"
summary_reason = POST_CREATE_REENTERED_LOGIN_SOURCE
else:
error_message = "账号已创建，但 post-create 续跑失败"
summary_reason = "post_create_continue_failed"
return {
"error_message": error_message,
"resume_source": summary_reason,
"summary_reason": summary_reason,
"workspace_resolution_source": POST_CREATE_RESUME_SOURCE,
"locked_url": str(locked_url or "").strip() or None,
"terminal_url": str(terminal_url or "").strip() or None,
"resolution_error": str(resolution_error or "").strip() or None,
}
|