GithubBackup/codex-register
1
0
Fork 0
mirror of https://github.com/cnlimiter/codex-register.git synced 2026-05-16 09:37:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(accounts): 添加账号cookies存储及支付链接国家选择功能

Browse Source 
- 在账号详情页添加cookies编辑与保存功能，用于支付请求
- 支付页面新增国家选择下拉框，支持多国货币计费
- 优化无痕打开浏览器功能，支持注入账号cookies
- 更新数据库模型、API路由及前端界面
This commit is contained in:
cnlimiter
2026-03-17 13:59:00 +08:00
parent d1e37aff56
commit 036a66d72b
10 changed files with 274 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

231
src/core/payment.py
View File

@@ -23,62 +23,101 @@ def _build_proxies(proxy: Optional[str]) -> Optional[dict]:
return None
def generate_plus_link(account: Account, proxy: Optional[str] = None) -> str:
"""生成 Plus 支付链接"""
if not account.access_token:
raise ValueError("账号缺少 access_token")
headers = {
"Authorization": f"Bearer {account.access_token}",
"Content-Type": "application/json",
}
payload = {
"plan_type": "plus",
"checkout_ui_mode": "hosted",
"cancel_url": "https://chatgpt.com/",
"success_url": "https://chatgpt.com/",
}
resp = cffi_requests.post(
PAYMENT_CHECKOUT_URL,
headers=headers,
json=payload,
proxies=_build_proxies(proxy),
timeout=30,
impersonate="chrome110",
)
resp.raise_for_status()
data = resp.json()
if "url" in data:
return data["url"]
raise ValueError(data.get("detail", "API 未返回支付链接"))
_COUNTRY_CURRENCY_MAP = {
"SG": "SGD",
"US": "USD",
"TR": "TRY",
"JP": "JPY",
"HK": "HKD",
"GB": "GBP",
"EU": "EUR",
"AU": "AUD",
"CA": "CAD",
"IN": "INR",
"BR": "BRL",
"MX": "MXN",
}
def generate_team_link(
def _extract_oai_did(cookies_str: str) -> Optional[str]:
"""从 cookie 字符串中提取 oai-device-id"""
for part in cookies_str.split(";"):
part = part.strip()
if part.startswith("oai-did="):
return part[len("oai-did="):].strip()
return None
def _parse_cookie_str(cookies_str: str, domain: str) -> list:
"""将 'key=val; key2=val2' 格式解析为 Playwright cookie 列表"""
cookies = []
for part in cookies_str.split(";"):
part = part.strip()
if "=" not in part:
continue
name, _, value = part.partition("=")
cookies.append({
"name": name.strip(),
"value": value.strip(),
"domain": domain,
"path": "/",
})
return cookies
def _open_url_system_browser(url: str) -> bool:
"""回退方案：调用系统浏览器以无痕模式打开"""
platform = sys.platform
try:
if platform == "win32":
for browser, flag in [("chrome", "--incognito"), ("msedge", "--inprivate")]:
try:
subprocess.Popen(f'start {browser} {flag} "{url}"', shell=True)
return True
except Exception:
continue
elif platform == "darwin":
subprocess.Popen(["open", "-a", "Google Chrome", "--args", "--incognito", url])
return True
else:
for binary in ["google-chrome", "chromium-browser", "chromium"]:
try:
subprocess.Popen([binary, "--incognito", url])
return True
except FileNotFoundError:
continue
except Exception as e:
logger.warning(f"系统浏览器无痕打开失败: {e}")
return False
def generate_plus_link(
account: Account,
workspace_name: str = "MyTeam",
price_interval: str = "month",
seat_quantity: int = 5,
proxy: Optional[str] = None,
country: str = "SG",
) -> str:
"""生成 Team 支付链接"""
"""生成 Plus 支付链接（后端携带账号 cookie 发请求）"""
if not account.access_token:
raise ValueError("账号缺少 access_token")
currency = _COUNTRY_CURRENCY_MAP.get(country, "USD")
headers = {
"Authorization": f"Bearer {account.access_token}",
"Content-Type": "application/json",
"oai-language": "zh-CN",
}
if account.cookies:
headers["cookie"] = account.cookies
oai_did = _extract_oai_did(account.cookies)
if oai_did:
headers["oai-device-id"] = oai_did
payload = {
"plan_name": "chatgptteamplan",
"team_plan_data": {
"workspace_name": workspace_name,
"price_interval": price_interval,
"seat_quantity": seat_quantity,
},
"plan_name": "chatgptplusplan",
"billing_details": {"country": country, "currency": currency},
"promo_campaign": {
"promo_campaign_id": "team-1-month-free",
"is_coupon_from_query_param": True,
"promo_campaign_id": "plus-1-month-free",
"is_coupon_from_query_param": False,
},
"checkout_ui_mode": "custom",
}
@@ -98,36 +137,86 @@ def generate_team_link(
raise ValueError(data.get("detail", "API 未返回 checkout_session_id"))
def open_url_incognito(url: str) -> bool:
"""调用本机浏览器以无痕模式打开 URL"""
platform = sys.platform
def generate_team_link(
account: Account,
workspace_name: str = "MyTeam",
price_interval: str = "month",
seat_quantity: int = 5,
proxy: Optional[str] = None,
country: str = "SG",
) -> str:
"""生成 Team 支付链接（后端携带账号 cookie 发请求）"""
if not account.access_token:
raise ValueError("账号缺少 access_token")
currency = _COUNTRY_CURRENCY_MAP.get(country, "USD")
headers = {
"Authorization": f"Bearer {account.access_token}",
"Content-Type": "application/json",
"oai-language": "zh-CN",
}
if account.cookies:
headers["cookie"] = account.cookies
oai_did = _extract_oai_did(account.cookies)
if oai_did:
headers["oai-device-id"] = oai_did
payload = {
"plan_name": "chatgptteamplan",
"team_plan_data": {
"workspace_name": workspace_name,
"price_interval": price_interval,
"seat_quantity": seat_quantity,
},
"billing_details": {"country": country, "currency": currency},
"promo_campaign": {
"promo_campaign_id": "team-1-month-free",
"is_coupon_from_query_param": True,
},
"cancel_url": "https://chatgpt.com/#pricing",
"checkout_ui_mode": "custom",
}
resp = cffi_requests.post(
PAYMENT_CHECKOUT_URL,
headers=headers,
json=payload,
proxies=_build_proxies(proxy),
timeout=30,
impersonate="chrome110",
)
resp.raise_for_status()
data = resp.json()
if "checkout_session_id" in data:
return TEAM_CHECKOUT_BASE_URL + data["checkout_session_id"]
raise ValueError(data.get("detail", "API 未返回 checkout_session_id"))
def open_url_incognito(url: str, cookies_str: Optional[str] = None) -> bool:
"""用 Playwright 以无痕模式打开 URL，可注入 cookie"""
import threading
try:
if platform == "win32":
# 依次尝试 Chrome、Edge
for browser, flag in [("chrome", "--incognito"), ("msedge", "--inprivate")]:
try:
subprocess.Popen(
f'start {browser} {flag} "{url}"',
shell=True,
)
return True
except Exception:
continue
elif platform == "darwin":
subprocess.Popen(
["open", "-a", "Google Chrome", "--args", "--incognito", url]
)
return True
else:
for binary in ["google-chrome", "chromium-browser", "chromium"]:
try:
subprocess.Popen([binary, "--incognito", url])
return True
except FileNotFoundError:
continue
except Exception as e:
logger.warning(f"无痕打开浏览器失败: {e}")
return False
from playwright.sync_api import sync_playwright
except ImportError:
logger.warning("playwright 未安装，回退到系统浏览器")
return _open_url_system_browser(url)
def _launch():
try:
with sync_playwright() as p:
browser = p.chromium.launch(headless=False, args=["--incognito"])
ctx = browser.new_context()
if cookies_str:
ctx.add_cookies(_parse_cookie_str(cookies_str, "chatgpt.com"))
page = ctx.new_page()
page.goto(url)
# 保持窗口打开直到用户关闭
page.wait_for_timeout(300_000) # 最多等待 5 分钟
except Exception as e:
logger.warning(f"Playwright 无痕打开失败: {e}")
threading.Thread(target=_launch, daemon=True).start()
return True
def check_subscription_status(account: Account, proxy: Optional[str] = None) -> str:

1
src/database/models.py
View File

@@ -55,6 +55,7 @@ class Account(Base):
source = Column(String(20), default='register') # 'register' 或 'login'，区分账号来源
subscription_type = Column(String(20)) # None / 'plus' / 'team'
subscription_at = Column(DateTime) # 订阅开通时间
cookies = Column(Text) # 完整 cookie 字符串，用于支付请求
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

1
src/database/session.py
View File

@@ -97,6 +97,7 @@ class DatabaseSessionManager:
("accounts", "source", "VARCHAR(20) DEFAULT 'register'"),
("accounts", "subscription_type", "VARCHAR(20)"),
("accounts", "subscription_at", "DATETIME"),
("accounts", "cookies", "TEXT"),
]
with self.engine.connect() as conn:

17
src/web/routes/accounts.py
View File

@@ -39,6 +39,7 @@ class AccountResponse(BaseModel):
proxy_used: Optional[str] = None
cpa_uploaded: bool = False
cpa_uploaded_at: Optional[str] = None
cookies: Optional[str] = None
created_at: Optional[str] = None
updated_at: Optional[str] = None
@@ -56,6 +57,7 @@ class AccountUpdateRequest(BaseModel):
"""账号更新请求"""
status: Optional[str] = None
metadata: Optional[dict] = None
cookies: Optional[str] = None # 完整 cookie 字符串，用于支付请求
class BatchDeleteRequest(BaseModel):
@@ -116,6 +118,7 @@ def account_to_response(account: Account) -> AccountResponse:
proxy_used=account.proxy_used,
cpa_uploaded=account.cpa_uploaded or False,
cpa_uploaded_at=account.cpa_uploaded_at.isoformat() if account.cpa_uploaded_at else None,
cookies=account.cookies,
created_at=account.created_at.isoformat() if account.created_at else None,
updated_at=account.updated_at.isoformat() if account.updated_at else None,
)
@@ -216,10 +219,24 @@ async def update_account(account_id: int, request: AccountUpdateRequest):
current_metadata.update(request.metadata)
update_data["metadata"] = current_metadata
if request.cookies is not None:
# 留空则清空，非空则更新
update_data["cookies"] = request.cookies or None
account = crud.update_account(db, account_id, **update_data)
return account_to_response(account)
@router.get("/{account_id}/cookies")
async def get_account_cookies(account_id: int):
"""获取账号的 cookie 字符串（仅供支付使用）"""
with get_db() as db:
account = crud.get_account_by_id(db, account_id)
if not account:
raise HTTPException(status_code=404, detail="账号不存在")
return {"account_id": account_id, "cookies": account.cookies or ""}
@router.delete("/{account_id}")
async def delete_account(account_id: int):
"""删除单个账号"""

22
src/web/routes/payment.py
View File

@@ -38,10 +38,12 @@ class GenerateLinkRequest(BaseModel):
seat_quantity: int = 5
proxy: Optional[str] = None
auto_open: bool = False # 生成后是否自动无痕打开
country: str = "SG" # 计费国家，决定货币 # 生成后是否自动无痕打开
class OpenIncognitoRequest(BaseModel):
url: str
account_id: Optional[int] = None # 可选，用于注入账号 cookie
class MarkSubscriptionRequest(BaseModel):
@@ -83,7 +85,7 @@ def generate_payment_link(request: GenerateLinkRequest):
try:
if request.plan_type == "plus":
link = generate_plus_link(account, proxy)
link = generate_plus_link(account, proxy, country=request.country)
elif request.plan_type == "team":
link = generate_team_link(
account,
@@ -91,6 +93,7 @@ def generate_payment_link(request: GenerateLinkRequest):
price_interval=request.price_interval,
seat_quantity=request.seat_quantity,
proxy=proxy,
country=request.country,
)
else:
raise HTTPException(status_code=400, detail="plan_type 必须为 plus 或 team")
@@ -102,7 +105,8 @@ def generate_payment_link(request: GenerateLinkRequest):
opened = False
if request.auto_open and link:
opened = open_url_incognito(link)
cookies_str = account.cookies if account else None
opened = open_url_incognito(link, cookies_str)
return {
"success": True,
@@ -114,13 +118,21 @@ def generate_payment_link(request: GenerateLinkRequest):
@router.post("/open-incognito")
def open_browser_incognito(request: OpenIncognitoRequest):
"""后端命令行以无痕模式打开指定 URL"""
"""后端以无痕模式打开指定 URL，可注入账号 cookie"""
if not request.url:
raise HTTPException(status_code=400, detail="URL 不能为空")
success = open_url_incognito(request.url)
cookies_str = None
if request.account_id:
with get_db() as db:
account = db.query(Account).filter(Account.id == request.account_id).first()
if account:
cookies_str = account.cookies
success = open_url_incognito(request.url, cookies_str)
if success:
return {"success": True, "message": "已在无痕模式打开浏览器"}
return {"success": False, "message": "未找到可用的 Chrome/Edge，请手动复制链接"}
return {"success": False, "message": "未找到可用的浏览器，请手动复制链接"}
# ============== 订阅状态 ==============