书接上回： https://www.v2ex.com/t/1218755#reply0
首先跟兄弟们道个歉，因为发现还是感情贴热度高，没法，蹭蹭热度.感谢，祝大家早日财务自由！
前两天闲来无事做了一个项目，初衷和目的上一章都有，可能就是觉得不甘心吧.期间也认识了一些新的人接触一些新的知识.
还是觉得 ai 是用来更好的服务人民的，我这个如果开源了，旅游业会发展跟快一点.起码不是跟我左右到处碰壁.万一有人真成了，街头店家都可以用，收外国的钱.我感觉这个意义更好.希望如果有人做起来也挺好的.也不知道行不行.
以下是代码的一些内容，有用的自取，感谢.
架构
Vercel Fluid Compute 单函数入口 │ ▼ api/index.py → app/init.py (Flask app factory) │ ├── Session: Flask session cookie, HTTPOnly, SameSite=Lax ├── 存储: Upstash Redis REST API → 内存 dict fallback ├── AI: DeepSeek API (deepseek-chat) ├── 前端: Jinja2 模板 + Tailwind CDN + Vanilla JS │ └── 8 个 Blueprint: translate /api/translate, /api/export-* auth /api/auth/, /login, /logout merchant /api/merchant/ gmb /api/gmb/* (Google OAuth) payment /api/payment/* (支付诊断) psb /api/psb/* (MRZ + 机构查询) beta /api/health, /api/admin/* guide /api/guide/checklist
源代码
api/index.py — Vercel 入口
from app import create_app
app = create_app()
app/init.py — Flask 工厂
import os from flask import Flask from dotenv import load_dotenv
load_dotenv(".env.local", override=False) load_dotenv(".env", override=False)
def create_app(): app = Flask(name, template_folder="templates")
app.secret_key = os.getenv("SESSION_SECRET", os.urandom(24).hex())
app.config["SESSION_COOKIE_HTTPONLY"] = True
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
from app.translate import translate_bp
from app.auth import auth_bp
from app.merchant import merchant_bp
from app.gmb import gmb_bp
from app.payment import payment_bp
from app.psb import psb_bp
from app.beta import beta_bp
from app.guide import guide_bp
app.register_blueprint(translate_bp)
app.register_blueprint(auth_bp)
app.register_blueprint(merchant_bp)
app.register_blueprint(gmb_bp)
app.register_blueprint(payment_bp)
app.register_blueprint(psb_bp)
app.register_blueprint(beta_bp)
app.register_blueprint(guide_bp)
@app.route("/")
def index():
from flask import render_template
return render_template("index.html")
@app.route("/dashboard")
def dashboard():
from flask import render_template
return render_template("dashboard.html")
@app.route("/tripadvisor")
def tripadvisor_guide():
from flask import render_template
return render_template("tripadvisor_guide.html")
@app.route("/psb")
def psb_guide():
from flask import render_template
return render_template("psb_guide.html")
@app.route("/gmb/guide")
def gmb_guide():
from flask import render_template
return render_template("gmb_guide.html")
@app.route("/health")
def health():
from flask import redirect
return redirect("/api/health")
return app
app/kv_client.py — 存储抽象层
import os, json, time, threading from urllib.request import Request, urlopen
def _strip_env(val): v = val.strip() if v and ord(v[0]) == 0xFEFF: # 去除 Upstash URL 的 BOM 前缀 v = v[1:] return v
_REST_URL = _strip_env(os.getenv("UPSTASH_REDIS_REST_URL", "")) _REST_TOKEN = _strip_env(os.getenv("UPSTASH_REDIS_REST_TOKEN", ""))
_REDIS_OK = None _MEMORY_STORE = {} _MEMORY_LOCK = threading.Lock()
def _redis_cmd(*args): if not (_REST_URL and _REST_TOKEN): return None try: body = json.dumps(args).encode("utf-8") req = Request(_REST_URL, data=body, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {_REST_TOKEN}", }) with urlopen(req, timeout=5) as r: resp = json.loads(r.read().decode("utf-8")) _REDIS_OK = True # FIXME: global mutable return resp.get("result") except Exception: _REDIS_OK = False return None
公开 API — Redis 优先，内存 fallback
def redis_get(key): if _REST_URL and _REST_TOKEN: result = _redis_cmd("GET", key) if result is not None or _REDIS_OK: return result with _MEMORY_LOCK: entry = _MEMORY_STORE.get(key) if entry and entry.get("exp") and time.time() > ent del _MEMORY_STORE[key] return None return entry.get("val") if entry else None
def redis_set(key, value, ex=None): if _REST_URL and _REST_TOKEN: if ex: result = _redis_cmd("SET", key, str(value), "EX", str(ex)) else: result = _redis_cmd("SET", key, str(value)) if result is not None or _REDIS_OK: return result with _MEMORY_LOCK: entry = {"val": str(value)} if ex: entry["exp"] = time.time() + int(ex) _MEMORY_STORE[key] = entry return True
def redis_incr(key): if _REST_URL and _REST_TOKEN: result = _redis_cmd("INCR", key) if result is not None: return result with _MEMORY_LOCK: val = redis_get(key) new_val = (int(val) + 1) if val is not None else 1 _MEMORY_STORE[key] = {"val": str(new_val)} return new_val
def redis_del(key): if _REST_URL and _REST_TOKEN: _redis_cmd("DEL", key) with _MEMORY_LOCK: _MEMORY_STORE.pop(key, None)
def redis_keys(pattern): if _REST_URL and _REST_TOKEN: keys, cursor = [], "0" while True: result = _redis_cmd("SCAN", cursor, "MATCH", pattern, "COUNT", "200") if not isinstance(result, list) or len(result) < 2: return [] cursor = str(result[0]) if result[0] is not None else "0" keys.extend(result[1] if isinstance(result[1], list) else []) if cursor == "0": break return keys import re escaped = re.escape(pattern).replace(r"*", ".*") compiled = re.compile("^" + escaped + "$") with _MEMORY_LOCK: return [k for k in _MEMORY_STORE if compiled.match(k)]
def is_redis_available(): if _REST_URL and _REST_TOKEN: return _redis_cmd("PING") == "PONG" return False
app/translate.py — AI 翻译（核心）
import os, json, sys from urllib.request import Request, urlopen from urllib.error import URLError from flask import Blueprint, request, jsonify, session
from app.rate_limit import rate_limit from app.auth import login_required from app.constants import HISTORY_MAX_ENTRIES, HISTORY_TTL
translate_bp = Blueprint("translate", name) PROMPTS = { "menu": { "en": "You are a professional restaurant menu translator. Translate Chinese menu items into natural, appetizing English...", "ja": "あなたはプロの料理メニュー翻訳者です...", "ko": "당신은 전문 레스토랑 메뉴 번역가입니다...", "ru": "Вы профессиональный переводчик меню ресторанов...", }, "checkin_guide": { ... }, "room_card": { ... }, "reg_card": { ... }, "emergency_card": { ... }, }
LANGUAGES = [ {"code": "en", "label": "English", "flag": "🇬🇧"}, {"code": "ja", "label": "日本語", "flag": "🇯🇵"}, {"code": "ko", "label": "한국어", "flag": "🇰🇷"}, {"code": "ru", "label": "Русский", "flag": "🇷🇺"}, ]
@translate_bp.route("/api/translate", methods=["POST"]) @rate_limit("translate", identifier_fn=lambda: session.get("user_id", "anon")) def translate(): data = request.get_json() source_text = data.get("sourceText", "") material_type = data.get("materialType", "") target_lang = data.get("targetLang", "")
api_key = os.getenv("DEEPSEEK_API_KEY", "").strip()
system_prompt = PROMPTS.get(material_type, {}).get(target_lang, "")
req_body = json.dumps({
"model": "deepseek-chat",
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": source_text},
],
"temperature": 0.3,
"max_tokens": 4096,
}).encode("utf-8")
req = Request("https://api.deepseek.com/v1/chat/completions",
data=req_body,
headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"})
with urlopen(req, timeout=30) as r:
body = json.loads(r.read().decode("utf-8"))
translated = body["choices"][0]["message"]["content"]