我在__init__文件中配置，可以写入日志文件，在蓝图中，日志就无法写入日志文件，这是怎么回事？

Question

我在__init__文件中配置，可以写入日志文件，在蓝图中，日志就无法写入日志文件，这是怎么回事？
这是__init__文件
import logging
import os
from logging.handlers import RotatingFileHandler
from flask import Flask
from flask_login import LoginManager
from flask_session import Session
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate, upgrade
from flask_redis import FlaskRedis
from .Config import Config

db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
login_manager.login_view = 'auth.login'
redis_store = FlaskRedis()


def create_app():
    # 创建Flask对象
    app = Flask(__name__)
    app.config.from_object(Config)
    app.logger.info("Flask app created with config: %s", Config)

    # 初始化配置
    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)
    redis_store.init_app(app)

    # 初始化 Flask-Session
    Session(app)

    # 配置日志
    # 定义日志文件路径
    log_file = os.path.join(app.config['LOG_FOLDER'], 'log.txt')
    # 确保日志文件所在目录存在，如果不存在则创建，exist_ok=True表示如果目录已存在则不抛出异常
    os.makedirs(os.path.dirname(log_file), exist_ok=True)
    # 创建一个RotatingFileHandler实例，指定日志文件路径，最大文件大小为100MB，保留最近10个日志文件作为备份
    file_handler = RotatingFileHandler(log_file, maxBytes=1024 * 1024 * 100, backupCount=10)
    # 设置日志记录级别为DEBUG，意味着将记录DEBUG级别及以上的日志信息
    file_handler.setLevel(logging.DEBUG)
    # 创建一个日志格式器，指定日志信息的输出格式
    formatter = logging.Formatter('%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]')
    # 将日志格式器设置到日志处理器上，以便日志信息按照指定格式输出
    file_handler.setFormatter(formatter)
    # 确保日志处理器添加到应用的日志记录器上
    app.logger.addHandler(file_handler)

    # 确保文件存在
    os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)

    # 注册蓝图
    from .auth import auth as auth_blueprint
    from .main import main as main_blueprint
    app.register_blueprint(auth_blueprint)
    app.register_blueprint(main_blueprint, url_prefix='/main')
    from .employee import employee as employee_blueprint
    app.register_blueprint(employee_blueprint, url_prefix='/employee')
    from .admin import admin as admin_blueprint
    app.register_blueprint(admin_blueprint, url_prefix='/admin')
    from .profile import profile as profile_blueprint
    app.register_blueprint(profile_blueprint, url_prefix='/profile')

    # 全局异常处理
    @app.errorhandler(Exception)
    def handle_exception(error):
        app.logger.error("Unhandled exception occurred: %s", error)
        # 返回更详细的错误信息
        return f"An error occurred: {str(error)}. Please try again later.", 500

    with app.app_context():
        try:
            # 检查 migrations 文件夹是否存在，如果不存在则初始化
            migrations_path = os.path.join(os.getcwd(), 'migrations')
            if not os.path.exists(migrations_path):
                from flask_migrate import init
                init(directory=migrations_path)
                app.logger.info("Migrations folder initialized at: %s", migrations_path)
            # 应用迁移
            upgrade()
            app.logger.info("Database migrations applied")
        except Exception as e:
            app.logger.error("Error during database migration: %s", e)
            raise

    return app

这是蓝图文件
from datetime import timedelta
from werkzeug.exceptions import InternalServerError
from .Captcha import Captcha
from flask import current_app, Blueprint, render_template, redirect, url_for, flash, request, make_response
from flask_login import logout_user, login_user, current_user
from .forms import LoginForm, RegisterForm
from .models import User
from . import db, redis_store


auth = Blueprint('auth', __name__)


# 获取当前路由名称的辅助函数
def get_current_route_name():
    return request.endpoint


@auth.context_processor
def inject_current_route():
    # 将当前路由名称注入到所有模板中
    return dict(current_route=get_current_route_name())


# 首页
@auth.route('/')
def index():
    return render_template('home.html')


# 登陆页
@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if current_user.is_authenticated:
        current_app.logger.info("User %s is already logged in", current_user.username)
        return redirect(url_for('main.home'))
    if form.validate_on_submit():
        try:
            user = User.query.filter_by(username=form.username.data).first()
            if user and user.check_password(form.password.data):
                remember = form.remember.data
                login_user(user, remember=remember, duration=timedelta(days=1) if remember else None)
                next_page = request.args.get('next')
                if not next_page or not next_page.startswith('/'):
                    next_page = None
                current_app.logger.info("User %s logged in", user.username)  # 登录成功日志
                flash("登录成功！", 'success')
                if user.role == 'Admin':
                    return redirect(url_for('admin.dashboard'))
                elif user.role == 'Employee':
                    return redirect(url_for('employee.dashboard'))
                else:
                    return redirect(next_page or url_for('main.home'))
            else:
                current_app.logger.warning("Failed login attempt for username: %s", form.username.data)  # 登录失败日志
                flash("用户名或密码错误", "danger")
        except Exception as e:
            current_app.logger.error("Error during login: %s", e)  # 系统错误日志
            db.session.rollback()
            flash("系统错误，请稍后再试", "danger")

    return render_template('login.html', form=form)


# 注册页
@auth.route('/register', methods=['GET', "POST"])
def register():
    form = RegisterForm()
    if form.validate_on_submit():
        try:
            username = form.username.data
            email = form.email.data
            password = form.password.data
            user = User(username=username, email=email, password=password)
            db.session.add(user)
            db.session.commit()
            current_app.logger.info("User %s registered successfully", username)  # 注册成功日志
            flash("注册成功！", "success")
            return redirect(url_for('auth.login'))
        except Exception as e:
            current_app.logger.error("Error during registration: %s", e)  # 注册失败日志
            db.session.rollback()
            flash("注册失败，请稍后再试", "danger")
    return render_template('register.html', form=form)


# 退出页
@auth.route('/logout')
def logout():
    logout_user()
    return redirect(url_for('main.home'))


@auth.route('/captcha/<image_data_id>')
def get_captcha(image_data_id):
    try:
        # 生成新的验证码
        captcha = Captcha()
        image_data, text = captcha.generate()
        current_app.logger.info("Captcha generated for image_data_id: %s", image_data_id)  # 验证码生成日志
        # 存储验证码文本到Redis
        redis_store.setex("image_data_%s" % image_data_id, timedelta(minutes=1), text)
        # 构建HTTP响应
        resp = make_response(image_data.getvalue())
        resp.headers['Content-Type'] = 'image/png'
        # 确保图像数据不为空
        if not image_data.getvalue():
            raise ValueError("Generated empty image data")
        return resp

    except Exception as e:
        current_app.logger.error("Failed to generate captcha: %s", e)  # 验证码生成失败日志
        raise InternalServerError("Failed to generate captcha") from e

Answer 1

以下回复引自免费微信小程序：皆我百晓生。 如问题解决，还请采纳。如仍有疑问，请回复留言或私信博主，以便获得更精准的帮助。

在__INIT__.PY文件中配置日志时，如果使用了PYTHON的内置LOGGING模块，则可以通过设置日志级别和输出格式来控制日志信息的输出方式。然而，如果你在蓝图中没有使用LOGGING模块，并且你希望在蓝图中使用日志记录，那么你可能需要创建一个自定义的日志记录器类，然后在这个类中添加日志记录逻辑。

例如：

CLASS MYLOGGER:
    DEF __INIT__(SELF):
        SELF.LOGGER = LOGGING.GETLOGGER('MY_LOGGER')
        SELF.HANDLER = LOGGING.STREAMHANDLER()
        FORMATTER = LOGGING.FORMATTER('%(ASCTIME)S - %(NAME)S - %(LEVELNAME)S - %(MESSAGE)S')
        HANDLER = LOGGING.FILEHANDLER('/VAR/LOG/MY_LOGGER.LOG', MODE='W')
        HANDLER.SETFORMATTER(FORMATTER)
        SELF.LOGGER.ADDHANDLER(HANDLER)
        SELF.LOGGER.SETLEVEL(LOGGING.DEBUG)

    DEF LOG(SELF, MESSAGE):
        SELF.LOGGER.INFO(MESSAGE)

# 使用蓝图的方式：
FROM FLASK IMPORT BLUEPRINT, RENDER_TEMPLATE, REQUEST
FROM APP.MODELS IMPORT DB, USER, POST

BP = BLUEPRINT('MY_BLUEPRINT', __NAME__)

@BP.ROUTE('/')
DEF INDEX():
    USER = USER.QUERY.FILTER_BY(USERNAME='TEST_USER').FIRST_OR_404()  # 查询用户，如果没有返回NONE
    POSTS = POST.QUERY.ALL()  # 批量查询所有POST对象
    RETURN RENDER_TEMPLATE('INDEX.HTML', USER=USER, POSTS=POSTS)  # 返回渲染后的INDEX.HTML页面，将USER和POSTS对象传递给它

IF __NAME__ == '__MAIN__':
    APP.RUN(DEBUG=TRUE)

在这种情况下，你可以通过以下代码来处理你的需求：

FROM FLASK IMPORT FLASK, RENDER_TEMPLATE, REQUEST
FROM FLASK_SQLALCHEMY IMPORT SQLALCHEMY

APP = FLASK(__NAME__)
DB.INIT_APP(APP)
APP.CONFIG['SQLALCHEMY_DATABASE_URI'] = 'SQLITE:////TMP/TEST.DB'  # 这里是你的数据库文件路径
APP.CONFIG['SQLALCHEMY_TRACK_MODIFICATIONS'] = FALSE  # 这里是是否跟踪修改

DB.CREATE_ALL()

@APP.ROUTE('/', METHODS=['GET'])
DEF INDEX():
    USER = USER.QUERY.FILTER_BY(USERNAME='TEST_USER').FIRST_OR_404()  # 查询用户，如果没有返回NONE
    POSTS = POST.QUERY.ALL()  # 批量查询所有POST对象
    RETURN RENDER_TEMPLATE('INDEX.HTML', USER=USER, POSTS=POSTS)

IF __NAME__ == '__MAIN__':
    APP.RUN(DEBUG=TRUE)

这样，你就可以在你的应用中配置日志并访问日志文件了。请注意，这只是一个基本示例，实际的实现可能会根据你的具体需求进行调整。