记录翻译历史

user_id = session.get("user_id")
    from app.kv_client import redis_get, redis_set, redis_incr
    entry = {
        "material_type": material_type,
        "target_lang": target_lang,
        "source_preview": source_text[:80],
        "translated_preview": translated[:80],
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }
    raw = redis_get(f"history:{user_id}")
    history = json.loads(raw) if raw else []
    history.append(entry)
    redis_set(f"history:{user_id}", json.dumps(history[-200:], ensure_ascii=False), ex=HISTORY_TTL)
    redis_incr(f"total:{user_id}")

return jsonify({"translatedText": translated})

app/auth.py — 认证

import re, hashlib, secrets, json from datetime import datetime, timezone from functools import wraps from flask import Blueprint, request, jsonify, session, render_template from app.kv_client import redis_get, redis_set, redis_incr from app.constants import PBKDF2_ITERATIONS, MIN_PASSWORD_LENGTH

auth_bp = Blueprint(“auth”, name)

PBKDF2_PREFIX = “pbkdf2:sha256:”

def _hash_password(password, salt=None):

if salt is None:
    salt = secrets.token_hex(16)
dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), PBKDF2_ITERATIONS)
return f"{PBKDF2_PREFIX}{PBKDF2_ITERATIONS}:{salt}:{dk.hex()}"

def _verify_password(stored, password):

if stored.startswith(PBKDF2_PREFIX):
    _, _, _, salt, dk_hex = stored.split(":", 4)
    iterations = int(stored.split(":")[2])
    dk = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), iterations)
    return dk.hex() == dk_hex
else:
    # 兼容旧版 SHA-256: "salt:hash"
    salt, _ = stored.split(":", 1)
    return f"{salt}:{hashlib.sha256((salt + password).encode()).hexdigest()}" == stored

def login_required(f):

@wraps(f)
def decorated(*args, **kwargs):
    if "user_id" not in session:
        return jsonify({"error": "请先登录"}), 401
    return f(*args, **kwargs)
return decorated

@auth_bp.route(“/login”) def login_page():

return render_template("login.html")

@auth_bp.route(“/api/auth/register”, methods=[“POST”]) def register():

data = request.get_json()
username = (data.get("username", "") or "").strip().lower()
password = (data.get("password", "") or "").strip()

if not re.match(r"^[a-z0-9_]{3,20}$", username):
    return jsonify({"error": "用户名需 3-20 位字母、数字或下划线"}), 400
if len(password) < MIN_PASSWORD_LENGTH:
    return jsonify({"error": f"密码至少{MIN_PASSWORD_LENGTH}位"}), 400

if redis_get(f"user:{username}"):
    return jsonify({"error": "该用户名已被注册"}), 409

redis_set(f"user:{username}", json.dumps({
    "username": username,
    "password": _hash_password(password),
    "created_at": datetime.now(timezone.utc).isoformat(),
}, ensure_ascii=False))

session["user_id"] = username
session["username"] = username
return jsonify({"success": True, "redirect": "/console"})

@auth_bp.route(“/api/auth/login”, methods=[“POST”]) @rate_limit(“login”) def login_api():

data = request.get_json()
username = (data.get("username", "") or "").strip().lower()
password = (data.get("password", "") or "").strip()

raw = redis_get(f"user:{username}")
if not raw:
    return jsonify({"error": "用户名或密码错误"}), 400

    return jsonify({"error": "用户名或密码错误"}), 400

# 旧 SHA-256 哈希自动升级为 PBKDF2
if not user.get("password", "").startswith(PBKDF2_PREFIX):
    user["password"] = _hash_password(password)
    redis_set(f"user:{username}", json.dumps(user, ensure_ascii=False))

session["user_id"] = username
session["username"] = username
return jsonify({"success": True, "redirect": "/console"})

app/rate_limit.py — 限流

import time, functools from flask import request, jsonify from app.kv_client import redis_incr, redis_set

LIMITS = {

"login":                {"window": 60,  "max": 10},
"translate":            {"window": 60,  "max": 30},
"generate_description": {"window": 300, "max": 5},

def _get_ip():

real_ip = request.headers.get("X-Real-IP", "")
if real_ip:
    return real_ip.strip()
forwarded = request.headers.get("X-Forwarded-For", "")
if forwarded:
    return [p.strip() for p in forwarded.split(",")][-1]
return request.remote_addr or "127.0.0.1"

def rate_limit(prefix, identifier_fn=None):

def decorator(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        identifier = identifier_fn() if identifier_fn else _get_ip()
        if not identifier:
            identifier = _get_ip()

        limit_config = LIMITS.get(prefix, {"window": 60, "max": 30})
        window = limit_config["window"]
        max_req = limit_config["max"]
        block = int(time.time() / window)
        key = f"rate_limit:{prefix}:{identifier}:{block}"

        count = redis_incr(key)
        if isinstance(count, int) and count == 1:
            redis_set(key, "1", ex=window * 2)

        if isinstance(count, int) and count > max_req:
            return jsonify({
                "error": f"请求过于频繁，请 {window} 秒后重试",
                "retry_after": window,
            }), 429

        return fn(*args, **kwargs)
    return wrapper
return decorator

app/payment.py — 支付诊断引擎（节选核心算法）

3 个方案各有评分规则矩阵

SOLUTIONS = {

"airwallex": {
    "name": "Airwallex 空中云汇",
    "score_rules": {
        "monthly_volume": {"low": 1, "medium": 2, "high": 3},
        "customer_type":  {"mixed": 2, "foreign": 3},
        "tech_level":     {"basic": -1, "intermediate": 1, "advanced": 2},
        ...
    },
},
"lianlian": { ... },
"bank_pos": { ... },

}

def diagnose_payment(answers):

"""对 3 个方案各自打分，按分数降序排列，第一名标记 recommended"""
results = []
for slug, sol in SOLUTIONS.items():
    score = 0
    reasons = []
    for qid, answer in answers.items():
        pts = sol["score_rules"].get(qid, {}).get(answer, 0)
        score += pts
        reason_key = f"{qid}={answer}"
        if reason_key in REASON_TEMPLATES.get(slug, {}):
            reasons.append(REASON_TEMPLATES[slug][reason_key])
    results.append({"slug": slug, "name": sol["name"], "score
    results[0]["recommended"] = True
return results

app/psb.py — MRZ 护照解析（节选）

def parse_mrz_td3(line1, line2):

"""解析标准护照 MRZ (2×44 字符), 符合 ICAO 9303"""
issuing_state = line1[2:5].replace("<", "").strip()
nationality    = line2[10:13].replace("<", "").strip()
doc_number     = line2[0:9].replace("<", "").strip()
dob_raw        = line2[13:19]
sex            = line2[20]
expiry_raw     = line2[21:27]

# 解析姓名
name_part = line1[5:44]
surnames, given_names = [], []
if "<<" in name_part:
    surname_part, given_part = name_part.split("<<", 1)
    surnames = [s for s in surname_part.split("<") if s]
    given_names = [s for s in given_part.split("<") if s]

def fmt_date(raw):
    yy = int(raw[:2])
    century = "19" if yy >= 70 else "20"  # ICAO 标准: YY≥70→19xx
    return f"{century}{raw[:2]}-{raw[2:4]}-{raw[4:6]}"

    "passport_number": doc_number,
    "nationality": COUNTRY_MAP.get(nationality, nationality),
    "dob": fmt_date(dob_raw),
    "sex": "男" if sex == "M" else "女" if sex == "F" else sex,
    "expiry": fmt_date(expiry_raw),
}, None

def parse_mrz(line1, line2):

line1, line2 = line1.strip().upper(), line2.strip().upper()
if len(line1) == 44 and len(line2) == 44:
    return parse_mrz_td3(line1, line2)   # 标准护照
if len(line1) == 36 and len(line2) == 36:
    return parse_mrz_td2(line1, line2)   # 护照卡
return None, f"不支持 MRZ 格式 (行 1={len(line1)}, 行 2={len(line2)})"

app/gmb.py — Google 商家 OAuth （节选）

@gmb_bp.route(“/gmb/connect”) @login_required def gmb_connect():

"""生成 Google OAuth URL, state 存在 Redis (10 分钟 TTL)"""
state = os.urandom(16).hex()
redis_set(f"oauth_state:{state}", session["user_id"], ex=600)
params = urlencode({
    "client_id": GMB_CLIENT_ID,
    "redirect_uri": GMB_REDIRECT_URI,
    "response_type": "code",
    "scope": "https://www.googleapis.com/auth/business.manage",
    "access_type": "offline",
    "prompt": "consent",
    "state": state,
})
return redirect(f"https://accounts.google.com/o/oauth2/v2/auth?{params}")

@gmb_bp.route(“/gmb/callback”) def gmb_callback():

"""Google 回调: 验证 state → 换 token → 发现/创建 GMB 账号 → 存 Redis"""
code = request.args.get("code")
state = request.args.get("state")

stored_user = redis_get(f"oauth_state:{state}")
redis_del(f"oauth_state:{state}")
if not stored_user:
    return render_template("gmb_status.html", error="授权会话已过期")

# 用 code 换 access_token + refresh_token
body = urlencode({ "code": code, "client_id": ..., "grant_type": "authorization_code" })
# ... HTTP POST 到 https://oauth2.googleapis.com/token ...

# 存储 token
redis_set(f"gmb_token:{user_id}", json.dumps({
    "access_token": access_token,
    "refresh_token": refresh_token,
    "expires_at": time.time() + expires_in - 60,
    "account_id": account_id,
}))
return redirect("/gmb/form")

app/image_export.py — 卡片图片渲染

from PIL import Image, ImageDraw, ImageFont

CARD_W = 800 _FONT_CACHE = {}

def generate_card_image(text, lang, material_type, lang_label):

"""用 PIL 渲染翻译卡片为 PNG, 蓝顶栏 + 语言标签 + 正文 + 灰底页脚"""
f_body = _font(20)
# 计算正文高度 → 确定卡片总高度
body_text_height = _text_height(text, f_body, CARD_W - 80)
card_h = 56 + 24 + 20 + body_text_height + 24 + 32

img = Image.new("RGB", (CARD_W, int(card_h)), "#ffffff")
draw = ImageDraw.Draw(img)

# 顶部蓝条 + 物料类型标题
draw.rectangle([(0, 0), (CARD_W, 56)], fill="#2563eb")
# 语言标签 (右上角蓝色圆角徽章)
# 正文区域 (自动换行)
# 底部灰条 + "国际宾客支持 · 多语种物料生成"

buf = io.BytesIO()

img.save(buf, format="PNG", optimize=True)
buf.seek(0)
return buf

def generate_zip_of_cards(results, material_type, lang_label_map):

"""多语种结果打包为一个 ZIP"""
zip_buf = io.BytesIO()
with zipfile.ZipFile(zip_buf, "w", zipfile.ZIP_DEFLATED) as zf:
    for r in results:
        img_buf = generate_card_image(r["text"], r["lang"], material_type, ...)
        zf.writestr(f"{r['lang']}.png", img_buf.read())
zip_buf.seek(0)
return zip_buf

app/merchant.py — 三档订阅套餐

SUBSCRIPTION_PLANS = {

"free":     {"name": "免费版", "price": 0,   "translations_per_day": 20,   "languages": ["en","ja","ko","ru"]},
"pro":      {"name": "专业版", "price": 29,  "translations_per_day": 200,  "languages": ["en","ja","ko","ru","fr","de","es","ar"]},
"business": {"name": "企业版", "price": 99,  "translations_per_day": 1000, "languages": ["en","ja","ko","ru","fr","de","es","ar","th","vi","it","pt"]},

}

app/constants.py

DEFAULT_DAILY_LIMIT = 20 RATE_LIMIT_TTL = 86400 PROFILE_TTL = 7776000 # 90 天 FEEDBACK_TTL = 7776000 HISTORY_TTL = 7776000 PAY_ORDER_TTL = 2592000 # 30 天 OAUTH_STATE_TTL = 600 # 10 分钟 HISTORY_MAX_ENTRIES = 200 MAX_ERROR_LOG = 100 PBKDF2_ITERATIONS = 600000 MIN_PASSWORD_LENGTH = 8

app/static/app.js — 共享前端工具

function msg(type, text) { /* 浮动 toast 提示, 3.5 秒自动消失 / } function escapeHtml(str) { / XSS 防护 / } function requireAuth(callback) { / 检查 /api/auth/me, 未登录跳转 /login, 防重复跳转 */ }

关键设计决策

• INCR-first 速率限制：先原子递增再检查，避免 TOCTOU 竞态
• IP 提取：Vercel 上信任 X-Real-IP ，其次取 X-Forwarded-For 最后一跳
• BOM 前缀：Upstash URL 在 Vercel 环境变量面板粘贴时可能带 （ U+FEFF ），kv_client.py 自动剥离
• 密码哈希：PBKDF2-SHA256 60 万次迭代；旧 SHA-256 格式的账户登录时自动升级
• load_dotenv 顺序：先 .env.local 再 .env ，均为 override=False （前者优先）
• 无 JS 框架：前端纯 Vanilla JS ，所有模板内联
  

  PeiXyJ · 2026-06-10 16:43

  👍 0 #4

  这人在干什么啊?

  

  SURA907 · 2026-06-10 16:43

  👍 0 #5

  有管理吗？这得封号吧

  

  so2back · 2026-06-10 16:43

  👍 0 #6

  这边建议封号处理

  

  loveshuyuan · 2026-06-10 16:43

  👍 0 #7

  什么鬼玩意儿 @Livid

  

  minibear2021 · 2026-06-10 17:08

  👍 0 #8

  这是走火入魔了？

  

  Nasdaq · 2026-06-10 17:08

  👍 0 #9

  此处省略问候你个人小作文一百字～

  

  darksword21 · 2026-06-10 17:08

  👍 0 #10

  让 AI 服务你妈的尸体，傻逼东西