OrgTJWater/pipeline-lifetime
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: standardize UTC timestamp handling

Browse Source
This commit is contained in:
jiang
2026-07-08 10:07:08 +08:00
parent 22dd364405
commit 8d347cfbc2
8 changed files with 77 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/__init__.py
+2 -3
View File
@@ -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"])
app/models.py
+5 -6
View File
@@ -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:
app/prediction.py
+2 -2
View File
@@ -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)
app/routes.py
+7 -7
View File
@@ -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}
)
app/time_utils.py
+10 -1
View File
@@ -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)
templates/admin.html
+2 -2
View File
@@ -72,7 +72,7 @@
{% for user in password_reset_users %}
<tr class="hover:bg-slate-50">
<td class="px-4 py-3 font-semibold">{{ user.username }}</td>
<td class="px-4 py-3 text-textSub">{{ user.created_at.strftime('%Y-%m-%d %H:%M:%S') if user.created_at else '-' }}</td>
<td class="px-4 py-3 text-textSub">{{ format_datetime(user.created_at) }}</td>
<td class="px-4 py-3">
<form method="post" action="{{ url_for('main.create_password_reset_link', user_id=user.id) }}" data-reset-link-form class="inline-flex">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
@@ -117,7 +117,7 @@
<tr class="hover:bg-slate-50">
<td class="px-5 py-4 font-semibold">{{ record.user.username }}</td>
<td class="max-w-[420px] truncate px-5 py-4">{{ record.original_filename }}</td>
<td class="px-5 py-4 text-textSub">{{ record.upload_time.strftime('%Y-%m-%d %H:%M:%S') }}</td>
<td class="px-5 py-4 text-textSub">{{ format_datetime(record.upload_time) }}</td>
<td class="px-5 py-4">
<div class="flex flex-wrap gap-3">
<a class="font-bold text-primary" href="{{ url_for('main.download_file', record_id=record.id, file_type='original') }}">原始文件</a>
templates/history.html
+1 -1
View File
@@ -31,7 +31,7 @@
<div class="min-w-0 overflow-hidden">
<div class="truncate text-sm font-bold">{{ record.original_filename }}</div>
<div class="mt-1 flex flex-wrap items-center gap-2 text-xs text-textSub">
<span>{{ record.upload_time.strftime('%Y-%m-%d %H:%M:%S') }}</span>
<span>{{ format_datetime(record.upload_time) }}</span>
<span class="hidden sm:inline">·</span>
<span>记录 #{{ record.id }}</span>
</div>
tests/test_auth_registration.py
+48 -2
View File
@@ -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()