From 8d347cfbc2d85832a50e8d2c214ac518c246f52f Mon Sep 17 00:00:00 2001 From: Huarch Date: Wed, 8 Jul 2026 10:07:08 +0800 Subject: [PATCH] fix: standardize UTC timestamp handling --- app/__init__.py | 5 ++-- app/models.py | 11 ++++---- app/prediction.py | 4 +-- app/routes.py | 14 ++++----- app/time_utils.py | 11 +++++++- templates/admin.html | 4 +-- templates/history.html | 2 +- tests/test_auth_registration.py | 50 +++++++++++++++++++++++++++++++-- 8 files changed, 77 insertions(+), 24 deletions(-) diff --git a/app/__init__.py b/app/__init__.py index 271f3bc..92027f2 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -1,7 +1,6 @@ from __future__ import annotations import logging -from datetime import datetime from flask import Flask, abort, jsonify, request @@ -10,7 +9,7 @@ from .extensions import db, login_manager from .models import AppSetting, User from .prediction import FEATURES, load_model from .security import csrf_token, validate_csrf_token -from .time_utils import format_datetime_for_timezone +from .time_utils import current_year_for_timezone, format_datetime_for_timezone def create_app(config_object: type[Config] = Config, *, load_model_on_start: bool = True) -> Flask: @@ -90,7 +89,7 @@ def register_app_hooks(app: Flask) -> None: @app.context_processor def inject_helpers(): def now_year() -> int: - return datetime.now().year + return current_year_for_timezone(app.config["APP_TIMEZONE"]) def format_datetime(value) -> str: return format_datetime_for_timezone(value, app.config["APP_TIMEZONE"]) diff --git a/app/models.py b/app/models.py index 8a614f9..a701adc 100644 --- a/app/models.py +++ b/app/models.py @@ -1,11 +1,10 @@ from __future__ import annotations -from datetime import datetime - from flask_login import UserMixin from werkzeug.security import check_password_hash, generate_password_hash from .extensions import db +from .time_utils import utc_now class User(UserMixin, db.Model): @@ -15,7 +14,7 @@ class User(UserMixin, db.Model): username = db.Column(db.String(100), unique=True, nullable=False) password_hash = db.Column(db.String(255), nullable=False) is_admin = db.Column(db.Boolean, default=False, nullable=False) - created_at = db.Column(db.DateTime, default=datetime.utcnow) + created_at = db.Column(db.DateTime, default=utc_now) def set_password(self, password: str) -> None: self.password_hash = generate_password_hash(password) @@ -33,7 +32,7 @@ class PasswordResetToken(db.Model): token_hash = db.Column(db.String(64), unique=True, nullable=False, index=True) expires_at = db.Column(db.DateTime, nullable=False) used_at = db.Column(db.DateTime) - created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False) + created_at = db.Column(db.DateTime, default=utc_now, nullable=False) user = db.relationship( "User", @@ -52,7 +51,7 @@ class UploadRecord(db.Model): saved_path = db.Column(db.String(500), nullable=False) prediction_path = db.Column(db.String(500), nullable=False) image_path = db.Column(db.String(500), nullable=False) - upload_time = db.Column(db.DateTime, default=datetime.utcnow) + upload_time = db.Column(db.DateTime, default=utc_now) user = db.relationship("User", backref=db.backref("uploads", lazy=True)) @@ -63,7 +62,7 @@ class AppSetting(db.Model): id = db.Column(db.Integer, primary_key=True) key = db.Column(db.String(100), unique=True, nullable=False) value = db.Column(db.String(255), nullable=False) - updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + updated_at = db.Column(db.DateTime, default=utc_now, onupdate=utc_now) @classmethod def get_bool(cls, key: str, default: bool = False) -> bool: diff --git a/app/prediction.py b/app/prediction.py index adbd6e2..19aad27 100644 --- a/app/prediction.py +++ b/app/prediction.py @@ -4,7 +4,6 @@ import logging import os import uuid from dataclasses import dataclass -from datetime import datetime from pathlib import Path from typing import Any @@ -20,6 +19,7 @@ from werkzeug.datastructures import FileStorage from werkzeug.utils import secure_filename from .config import IMAGE_DIR, UPLOAD_DIR +from .time_utils import utc_now CHINESE_FONT_CANDIDATES = [ "Noto Sans CJK SC", @@ -386,7 +386,7 @@ def run_prediction(uploaded: FileStorage, user_id: int, model) -> PredictionArti if not original_filename: raise PredictionError("未选择文件") - timestamp = datetime.now().strftime("%Y%m%d%H%M%S") + timestamp = utc_now().strftime("%Y%m%d%H%M%S") run_id = f"{timestamp}_{uuid.uuid4().hex[:8]}" saved_filename, suffix = secure_upload_name(original_filename, run_id) diff --git a/app/routes.py b/app/routes.py index 3b4be7d..20a3ebe 100644 --- a/app/routes.py +++ b/app/routes.py @@ -3,7 +3,7 @@ from __future__ import annotations import os import hashlib import secrets -from datetime import datetime, timedelta +from datetime import timedelta from flask import ( Blueprint, @@ -26,7 +26,7 @@ from .extensions import db from .models import AppSetting, PasswordResetToken, UploadRecord, User from .prediction import PredictionError, run_prediction from .security import new_captcha -from .time_utils import format_datetime_for_timezone +from .time_utils import format_datetime_for_timezone, utc_now bp = Blueprint("main", __name__) REFERENCE_PDF_NAME = "20260630标准文本——供水管道健康状态与剩余寿命评估技术导则.pdf" @@ -60,7 +60,7 @@ def password_reset_token_hash(token: str) -> str: def password_reset_expiry() -> datetime: minutes = max(int(current_app.config["PASSWORD_RESET_TOKEN_MINUTES"]), 1) - return datetime.utcnow() + timedelta(minutes=minutes) + return utc_now() + timedelta(minutes=minutes) def format_app_datetime(value: datetime | None) -> str: @@ -74,7 +74,7 @@ def active_password_reset_token(token: str) -> PasswordResetToken | None: if ( reset_token is None or reset_token.used_at is not None - or reset_token.expires_at <= datetime.utcnow() + or reset_token.expires_at <= utc_now() ): return None return reset_token @@ -137,7 +137,7 @@ def prediction_result_payload(artifacts, record: UploadRecord) -> dict: ) return { "original_filename": artifacts.original_filename, - "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "generated_at": format_app_datetime(record.upload_time), "image_url": image_url, "importance_url": importance_url, "excel_url": url_for("main.download_file", record_id=record.id, file_type="prediction"), @@ -238,7 +238,7 @@ def password_reset(token: str): ), 400 reset_token.user.set_password(password) - reset_token.used_at = datetime.utcnow() + reset_token.used_at = utc_now() db.session.commit() flash("密码已重置，请使用新密码登录", "info") return render_auth_template("login", captcha=refresh_captcha()) @@ -326,7 +326,7 @@ def create_password_reset_link(user_id: int): if user.is_admin: return jsonify({"error": "管理员账号不支持通过此入口重置密码"}), 403 - now = datetime.utcnow() + now = utc_now() PasswordResetToken.query.filter_by(user_id=user.id, used_at=None).update( {"used_at": now} ) diff --git a/app/time_utils.py b/app/time_utils.py index 063874e..aadf5df 100644 --- a/app/time_utils.py +++ b/app/time_utils.py @@ -5,6 +5,15 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError DEFAULT_TIMEZONE = "Asia/Shanghai" +DATETIME_DISPLAY_FORMAT = "%Y-%m-%d %H:%M:%S" + + +def utc_now() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def current_year_for_timezone(timezone_name: str | None) -> int: + return datetime.now(timezone.utc).astimezone(timezone_for_name(timezone_name)).year def timezone_for_name(name: str | None) -> ZoneInfo: @@ -19,4 +28,4 @@ def format_datetime_for_timezone(value: datetime | None, timezone_name: str | No return "-" if value.tzinfo is None: value = value.replace(tzinfo=timezone.utc) - return value.astimezone(timezone_for_name(timezone_name)).strftime("%Y-%m-%d %H:%M:%S") + return value.astimezone(timezone_for_name(timezone_name)).strftime(DATETIME_DISPLAY_FORMAT) diff --git a/templates/admin.html b/templates/admin.html index ed421f7..a2fb58a 100644 --- a/templates/admin.html +++ b/templates/admin.html @@ -72,7 +72,7 @@ {% for user in password_reset_users %} {{ user.username }} - {{ user.created_at.strftime('%Y-%m-%d %H:%M:%S') if user.created_at else '-' }} + {{ format_datetime(user.created_at) }}
@@ -117,7 +117,7 @@ {{ record.user.username }} {{ record.original_filename }} - {{ record.upload_time.strftime('%Y-%m-%d %H:%M:%S') }} + {{ format_datetime(record.upload_time) }}
原始文件 diff --git a/templates/history.html b/templates/history.html index cc27c5e..c0a31c0 100644 --- a/templates/history.html +++ b/templates/history.html @@ -31,7 +31,7 @@
{{ record.original_filename }}
- {{ record.upload_time.strftime('%Y-%m-%d %H:%M:%S') }} + {{ format_datetime(record.upload_time) }} 记录 #{{ record.id }}
diff --git a/tests/test_auth_registration.py b/tests/test_auth_registration.py index 9d516d5..366f489 100644 --- a/tests/test_auth_registration.py +++ b/tests/test_auth_registration.py @@ -10,16 +10,24 @@ from app import create_app from app.config import Config from app.extensions import db from app.models import AppSetting, PasswordResetToken, UploadRecord, User +from app.time_utils import utc_now class RegistrationRoutesTest(unittest.TestCase): - def create_test_app(self, temp_dir: str, *, allow_registration: bool): + def create_test_app( + self, + temp_dir: str, + *, + allow_registration: bool, + app_timezone: str = "Asia/Shanghai", + ): class TestConfig(Config): TESTING = True SECRET_KEY = "test-secret" SECRET_KEY_GENERATED = False SQLALCHEMY_DATABASE_URI = f"sqlite:///{temp_dir}/test.db" ALLOW_REGISTRATION = allow_registration + APP_TIMEZONE = app_timezone ADMIN_PASSWORD = None return create_app(TestConfig, load_model_on_start=False) @@ -422,7 +430,7 @@ class RegistrationRoutesTest(unittest.TestCase): with app.app_context(): reset_token = PasswordResetToken.query.one() - reset_token.expires_at = datetime.utcnow() - timedelta(minutes=1) + reset_token.expires_at = utc_now() - timedelta(minutes=1) db.session.commit() client = app.test_client() @@ -512,6 +520,22 @@ class RegistrationRoutesTest(unittest.TestCase): self.assertIn("alice-file-01.xlsx", second_page) self.assertIn("alice-file-00.xlsx", second_page) + def test_history_page_displays_utc_upload_time_in_configured_timezone(self) -> None: + with TemporaryDirectory() as temp_dir: + app = self.create_test_app( + temp_dir, + allow_registration=False, + app_timezone="America/New_York", + ) + self.create_user(app, "alice", "secret123") + self.add_upload_records(app, "alice", 1) + client = app.test_client() + self.login(client, "alice", "secret123") + + html = client.get("/history").get_data(as_text=True) + + self.assertIn("2026-01-01 07:00:00", html) + def test_admin_page_paginates_upload_records(self) -> None: with TemporaryDirectory() as temp_dir: app = self.create_test_app(temp_dir, allow_registration=False) @@ -531,6 +555,28 @@ class RegistrationRoutesTest(unittest.TestCase): self.assertIn("alice-file-01.xlsx", second_page) self.assertIn("alice-file-00.xlsx", second_page) + def test_admin_page_displays_utc_times_in_configured_timezone(self) -> None: + with TemporaryDirectory() as temp_dir: + app = self.create_test_app( + temp_dir, + allow_registration=False, + app_timezone="America/New_York", + ) + self.create_user(app, "admin", "secret123", is_admin=True) + self.create_user(app, "alice", "secret123") + with app.app_context(): + alice = User.query.filter_by(username="alice").one() + alice.created_at = datetime(2026, 1, 1, 0, 0, 0) + db.session.commit() + self.add_upload_records(app, "alice", 1) + client = app.test_client() + self.login(client, "admin", "secret123") + + html = client.get("/admin").get_data(as_text=True) + + self.assertIn("2025-12-31 19:00:00", html) + self.assertIn("2026-01-01 07:00:00", html) + if __name__ == "__main__": unittest.main()