你的验证码是：
" + f"{code}
" + f"验证码 {current_app.config['EMAIL_CODE_MINUTES']} 分钟内有效，请勿向他人透露。
" + ), + ) + except (EmailConfigurationError, EmailDeliveryError): + db.session.delete(record) + db.session.commit() + return False + return True -def prediction_result_payload(artifacts, record: UploadRecord) -> dict: - image_url = url_for("static", filename=f"images/{artifacts.image_filename}") - return { - "original_filename": artifacts.original_filename, - "generated_at": format_app_datetime(record.upload_time), - "image_url": image_url, - "excel_url": url_for("main.download_file", record_id=record.id, file_type="prediction"), - "result_url": url_for("main.result_page"), - "sample_count": int(artifacts.sample_count), - "summary_rows": artifacts.summary_rows[:6], - "analysis_text": artifacts.analysis_text, - "model_version": artifacts.model_version, - } +def consume_email_code(email: str, purpose: str, code: str) -> bool: + record = ( + EmailVerificationCode.query.filter_by(email=email, purpose=purpose, used_at=None) + .order_by(EmailVerificationCode.created_at.desc()) + .first() + ) + if ( + record is None + or record.expires_at <= utc_now() + or record.attempts >= current_app.config["EMAIL_CODE_MAX_ATTEMPTS"] + ): + return False + record.attempts += 1 + if not secrets.compare_digest(record.code_hash, code_hash(email, purpose, code.strip())): + db.session.commit() + return False + record.used_at = utc_now() + db.session.commit() + return True + + +def trusted_device_for(user: User): + token = request.cookies.get(TRUSTED_DEVICE_COOKIE, "") + if not token: + return None + record = TrustedDevice.query.filter_by( + token_hash=hashlib.sha256(token.encode()).hexdigest(), + user_id=user.id, + ).first() + if record and record.expires_at > utc_now() and record.auth_version == user.auth_version: + record.last_used_at = utc_now() + db.session.commit() + return record + return None + + +def login_response(user: User, remember: bool, trust_device: bool = False): + login_user(user, remember=remember) + session["auth_version"] = user.auth_version + response = redirect(url_for("main.home")) + if trust_device: + token = secrets.token_urlsafe(32) + device = TrustedDevice( + user_id=user.id, + token_hash=hashlib.sha256(token.encode()).hexdigest(), + expires_at=utc_now() + + timedelta(days=current_app.config["TRUSTED_DEVICE_DAYS"]), + auth_version=user.auth_version, + ) + db.session.add(device) + db.session.commit() + response.set_cookie( + TRUSTED_DEVICE_COOKIE, + token, + max_age=current_app.config["TRUSTED_DEVICE_DAYS"] * 86400, + secure=current_app.config["SESSION_COOKIE_SECURE"], + httponly=True, + samesite="Lax", + ) + return response + + +def verification_page(title: str, purpose: str): + return render_template("verification.html", title=title, purpose=purpose, email=session.get("pending_email", "")) @bp.route("/") def index(): - if current_user.is_authenticated: - return redirect(url_for("main.home")) - return redirect(url_for("main.login")) + return redirect(url_for("main.home" if current_user.is_authenticated else "main.login")) @bp.route("/login", methods=["GET", "POST"]) def login(): if current_user.is_authenticated: return redirect(url_for("main.home")) - if request.method == "GET": return render_auth_template("login", captcha=refresh_captcha()) - - username = request.form.get("username", "").strip() - password = request.form.get("password", "") - if not captcha_is_valid(): - return render_auth_error("login", "验证码错误") - - user = User.query.filter_by(username=username).first() - if not user or not user.check_password(password): - return render_auth_error("login", "用户名或密码错误") - - login_user(user, remember=bool(request.form.get("remember"))) - return redirect(url_for("main.home")) + email, password = normal_email(request.form.get("email", "")), request.form.get("password", "") + if not captcha_is_valid(): return render_auth_error("login", "验证码错误") + user = User.query.filter_by(email=email).first() + if not user or not user.is_active_account or not user.check_password(password): + return render_auth_error("login", "邮箱或密码错误") + remember = bool(request.form.get("remember")) + if trusted_device_for(user): + return login_response(user, remember) + session.update(pending_email=email, pending_user_id=user.id, pending_remember=remember, pending_purpose="login") + if not issue_email_code(email, "login"): + return render_auth_error("login", "验证码发送失败，请稍后重试", 503) + return redirect(url_for("main.verify_email", purpose="login")) @bp.route("/register", methods=["GET", "POST"]) def register(): - if request.method == "GET": - return render_auth_template("register", captcha=refresh_captcha()) - - username = request.form.get("username", "").strip() - password = request.form.get("password", "") - - if not captcha_is_valid(): - return render_auth_error("register", "验证码错误") - if not registration_allowed(): - return render_auth_error("register", "当前未开放自助注册，请联系管理员。", 403) - - if not username: - return render_auth_error("register", "用户名不能为空") - if len(password) < 6: - return render_auth_error("register", "密码至少需要 6 位") - if User.query.filter_by(username=username).first(): - return render_auth_error("register", "用户名已存在") - - user = User(username=username, is_admin=False) - user.set_password(password) - db.session.add(user) - db.session.commit() - flash("注册成功，请登录", "info") - return render_auth_template("login", captcha=refresh_captcha()) + if request.method == "GET": return render_auth_template("register", captcha=refresh_captcha()) + username, email, password = request.form.get("username", "").strip(), normal_email(request.form.get("email", "")), request.form.get("password", "") + if not captcha_is_valid(): return render_auth_error("register", "验证码错误") + if not registration_allowed(): return render_auth_error("register", "当前未开放自助注册，请联系管理员。", 403) + if not username or len(username) > 100: return render_auth_error("register", "显示名不能为空且不能超过100个字符") + if not valid_email(email): return render_auth_error("register", "请输入有效的邮箱地址") + if not valid_password(password): return render_auth_error("register", "密码长度应为12至128位") + if User.query.filter((User.username == username) | (User.email == email)).first(): return render_auth_error("register", "显示名或邮箱已被使用") + user = User(username=username, email=email, is_admin=False, is_active_account=False) + user.set_password(password); db.session.add(user); db.session.commit() + session.update(pending_email=email, pending_user_id=user.id, pending_purpose="register") + if not issue_email_code(email, "register"): return render_auth_error("register", "验证码发送失败，请稍后重试", 503) + return redirect(url_for("main.verify_email", purpose="register")) -@bp.route("/password-reset/{user.username}，管理员已要求你重置密码。
请访问 找回密码，系统会将一次性验证码发送到本邮箱。
", + ) + except (EmailConfigurationError, EmailDeliveryError): + return jsonify({"error": "邮件发送失败"}), 503 + return jsonify({"message": "密码重置通知已发送至用户邮箱"}) @bp.route("/download/登录邮箱：{{ current_user.email }}
+ {% with messages=get_flashed_messages(with_categories=true) %}{% for c,m in messages %}{{ m }}
{% endfor %}{% endwith %} + + + + {% if session.get('password_change_verified') %}{% endif %} +管理系统注册状态、密码重置链接，查看所有用户的上传文件和预测结果。
-管理注册状态、密码重置和所有预测记录。
- 当前状态：{{ '已开放' if registration_allowed else '已关闭' }} -
-为普通用户生成一次性重置链接，旧链接会自动失效。
-向用户已验证邮箱发送重置通知。
+| 用户 | -创建时间 | -操作 | + +|||
|---|---|---|---|---|---|
| 显示名 | 邮箱 | 操作 | |||
| {{ user.username }} | -{{ format_datetime(user.created_at) }} | -- |
| 用户 | -原始文件 | -上传时间 | -下载 | -
|---|---|---|---|
| {{ record.user.username }} | -{{ record.original_filename }} | -{{ format_datetime(record.upload_time) }} | -- - | -
| 暂无上传记录 | -|||
| 用户 | 文件 | 时间 | |
| {{ record.user.username }} | {{ record.original_filename }} | {{ format_datetime(record.upload_time) }} | |