From 0b0fa0c047b601de4f28a1b916aacea9593dbee4 Mon Sep 17 00:00:00 2001 From: DjangoAdmin Dev Date: Fri, 7 Aug 2026 19:39:19 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20P5=20DI=20=E6=B3=A8=E5=85=A5=20+=20?= =?UTF-8?q?=E4=B8=AD=E9=97=B4=E4=BB=B6=E8=BF=81=E5=85=A5=20src/core?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 【依赖注入（解决 core→modules 倒置）】 - src/core/permission.py：PermissionProvider 协议 + check_permission （不再 import menu 服务，改从 current_app.extensions['permission_provider'] 获取） - src/core/middleware.py：整合 check_login/check_demo/transaction/op_log/error_handler （op_log 写入器 DI：current_app.extensions['operation_log_writer']） - application/factory.py：组合根注入 MenuPermissionProvider + record_operation_log - src/core/scheduler.py 懒加载指向 src.modules.job.scheduler_service 【验收】src/core 内零 application 引用；DI 注入验证通过； 未登录 401、白名单放行；12 passed；core+中间件 flake8 零告警 Co-Authored-By: Claude --- _codemod_p5a.py | 132 ++++ application/factory.py | 36 +- application/middleware/__init__.py | 89 +-- application/middleware/check_demo.py | 39 +- application/middleware/check_login.py | 206 +---- application/middleware/check_permission.py | 187 +---- application/middleware/error_handler.py | 95 +-- application/middleware/op_log.py | 220 +----- application/middleware/transaction.py | 243 +----- src/core/middleware.py | 813 ++++++++++++++++++++ src/core/permission.py | 165 ++++ src/core/scheduler.py | 10 +- src/modules/association/role_menu/router.py | 2 +- 13 files changed, 1184 insertions(+), 1053 deletions(-) create mode 100644 _codemod_p5a.py create mode 100644 src/core/middleware.py create mode 100644 src/core/permission.py diff --git a/_codemod_p5a.py b/_codemod_p5a.py new file mode 100644 index 0000000..b046156 --- /dev/null +++ b/_codemod_p5a.py @@ -0,0 +1,132 @@ +# -*- coding: utf-8 -*- +""" +P5a 一次性 codemod：合并中间件至 src/core/middleware.py +============================================================= +- 拼接 check_login / check_demo / transaction / op_log / error_handler +- 重写 import（utils→src.core/common、extensions→src.core） +- op_log DI 化（OperationLog → 注入 writer） +- 生成 register_middleware +执行后删除本脚本。 +""" + +import os +import re + +ROOT = os.path.dirname(os.path.abspath(__file__)) +SRC_MW = os.path.join(ROOT, 'application', 'middleware') +CORE = os.path.join(ROOT, 'src', 'core') + +IMPORT_REPLACEMENTS = [ + ('from utils import R', 'from src.core import response as R'), + ('from utils.jwt import', 'from src.core.jwt import'), + ('from utils.security import', 'from src.core.security import'), + ('from utils.ip import', 'from src.common.utils.ip import'), + ('from utils.ip2region import', 'from src.common.utils.ip2region import'), + ('from utils.user_agent import', 'from src.common.utils.user_agent import'), + ('from utils.request import', 'from src.common.utils.request import'), + ('from extensions import db', 'from src.core.database import db'), + ('from application.constants.', 'from src.common.constants.'), +] + + +def strip_license(text): + lines = text.split('\n') + last = -1 + for i, ln in enumerate(lines): + if ln.strip() == '# +======================================================================': + last = i + return '\n'.join(lines[last + 1:]) if last >= 0 else text + + +def read_and_rewrite(fname): + with open(os.path.join(SRC_MW, fname), encoding='utf-8') as f: + text = f.read() + text = strip_license(text) + for old, new in IMPORT_REPLACEMENTS: + text = text.replace(old, new) + return text + + +def main(): + parts = [] + header = '''""" +中间件模块 +============================================================= + +整合自 application/middleware/*.py： + - check_login: 登录校验（JWT + 白名单） + - check_demo: 演示模式拦截 + - transaction: AOP 事务拦截器 + - op_log: 操作日志自动记录（写入器由应用工厂注入） + - error_handler: 全局异常处理 + +依赖约束：本模块仅依赖 core/common，不依赖 modules 层。 +""" + +import logging + +logger = logging.getLogger(__name__) +''' + parts.append(header) + + # check_login + parts.append('\n# ============================================================\n# 登录校验中间件\n# ============================================================\n') + parts.append(read_and_rewrite('check_login.py')) + # check_demo + parts.append('\n\n# ============================================================\n# 演示模式拦截\n# ============================================================\n') + parts.append(read_and_rewrite('check_demo.py')) + # transaction + parts.append('\n\n# ============================================================\n# AOP 事务拦截器\n# ============================================================\n') + parts.append(read_and_rewrite('transaction.py')) + # op_log + parts.append('\n\n# ============================================================\n# 操作日志自动记录（DI：写入器由应用工厂注入）\n# ============================================================\n') + op_log = read_and_rewrite('op_log.py') + # op_log DI 化 + op_log = op_log.replace('from application.models.operation_log import OperationLog', '') + op_log = op_log.replace(' from application.services import operation_log\n', '') + op_log = op_log.replace('from flask import g, request', 'from flask import g, request, current_app') + op_log = re.sub(r'operation_log\.record_operation_log\(', 'writer(', op_log) + op_log = op_log.replace('source=OperationLog.Source.ADMIN,', 'source=0, # OperationLog.Source.ADMIN') + op_log = op_log.replace('status=OperationLog.Status.NORMAL,', 'status=0, # OperationLog.Status.NORMAL') + # _write_log 内注入 writer 获取 + op_log = op_log.replace(''' username = get_username() or 'system' + client_ip = get_client_ip()''', +''' writer = current_app.extensions.get('operation_log_writer') + if writer is None: + logger.warning("operation_log_writer 未注入，跳过操作日志") + return + username = get_username() or 'system' + client_ip = get_client_ip()''') + parts.append(op_log) + # error_handler + parts.append('\n\n# ============================================================\n# 全局异常处理\n# ============================================================\n') + parts.append(read_and_rewrite('error_handler.py')) + + # register_middleware + register_mw = ''' + +# ============================================================ +# 中间件注册 +# ============================================================ + +def register_middleware(app): + """ + 注册应用中间件（登录校验 before_request）。 + + Args: + app: Flask 应用实例 + """ + @app.before_request + def checklogin(): + """登录验证中间件""" + return check_login() +''' + parts.append(register_mw) + + with open(os.path.join(CORE, 'middleware.py'), 'w', encoding='utf-8') as f: + f.write('\n'.join(parts)) + print('[DONE] src/core/middleware.py 已生成') + + +if __name__ == '__main__': + main() diff --git a/application/factory.py b/application/factory.py index 6f28e98..4852bcf 100644 --- a/application/factory.py +++ b/application/factory.py @@ -57,10 +57,7 @@ from flask import Flask from flask_cors import CORS from werkzeug.middleware.proxy_fix import ProxyFix -from application.middleware.transaction import register_transaction -from application.middleware import register_middleware -from application.middleware.error_handler import register_error_handlers -from application.middleware.op_log import register_operation_log +from src.core.middleware import register_middleware, register_transaction, register_error_handlers, register_operation_log from application.views import register_blueprint from config import DatabaseConfig, REDIS_URL from extensions import register_extends @@ -176,4 +173,35 @@ def create_app(): # ============================================================ register_error_handlers(app) + # ============================================================ + # 依赖注入（DI）：core 层不 import modules 层，由组合根注入 + # ============================================================ + _inject_dependencies(app) + return app + + +def _inject_dependencies(app): + """ + 依赖注入：向 core 中间件注入业务实现 + + 1. permission_provider：权限校验数据提供方（menu 服务） + 2. operation_log_writer：操作日志写入器（log 服务） + + 设计依据：src/core/permission.py 定义 PermissionProvider 协议， + core 中间件通过 current_app.extensions 获取注入的实现。 + """ + from src.modules.system.menu.service import get_user_permissions + from src.modules.log.service import record_operation_log + + # ---- 权限提供方 ---- + class MenuPermissionProvider: + """权限数据提供方（适配 menu 服务）。""" + + def get_user_permissions(self, user_id): + return get_user_permissions(user_id) + + app.extensions['permission_provider'] = MenuPermissionProvider() + + # ---- 操作日志写入器 ---- + app.extensions['operation_log_writer'] = record_operation_log diff --git a/application/middleware/__init__.py b/application/middleware/__init__.py index 217a675..9296edc 100644 --- a/application/middleware/__init__.py +++ b/application/middleware/__init__.py @@ -1,88 +1,9 @@ -# +====================================================================== -# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] -# +---------------------------------------------------------------------- -# | Copyright (c) 2021-2026 一米阳光团队 版权所有 -# +---------------------------------------------------------------------- -# | 开源协议: Apache-2.0 -# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 -# +---------------------------------------------------------------------- -# | 官方网站: https://www.djangoadmin.cn -# +---------------------------------------------------------------------- -# | 研发团队: 一米阳光团队 荣誉出品 -# +====================================================================== -# | 版权和免责声明: -# | 一、知识产权声明 -# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 -# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， -# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 -# | 二、使用限制与禁止条款 -# | 您理解并同意，禁止将本产品用于以下任何行为： -# | 1. 任何违反中华人民共和国法律法规的用途； -# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； -# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 -# | 三、责任豁免 -# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 -# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 -# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 -# | 四、违约与法律责任 -# | 如发现任何违反本声明的行为，本团队有权： -# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； -# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； -# | 3. 保留进一步采取刑事控告等法律措施的权利。 -# | 五、许可范围说明 -# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 -# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 -# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 -# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 -# | 成的一切经济损失及商誉损害。 -# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 -# | 许可协议》附件为准。 -# +====================================================================== - """ -中间件注册模块 +兼容 shim（绞杀者迁移期间保留） ============================================================= -该模块负责注册 Flask 应用的请求/响应中间件。 +中间件已迁移至 src/core/middleware.py 与 src/core/permission.py。 +本文件仅作兼容转发，最终在 P7（切尾）阶段删除。 """ - -from application.middleware.check_login import check_login -from application.middleware.check_permission import check_permission - - -def register_middleware(app): - """ - 注册应用中间件 - - 将所有的请求/响应钩子函数注册到 Flask 应用实例中。 - - Args: - app: Flask 应用实例 - - 使用示例: - from application.middleware import register_middleware - - def create_app(): - app = Flask(__name__) - register_middleware(app) - return app - """ - - # ============================================================ - # 请求前钩子 (before_request) - # ============================================================ - # 在每个请求被视图函数处理之前执行 - # 用于：登录验证、权限检查、请求日志、限流等 - @app.before_request - def checklogin(): - """ - 登录验证中间件 - - 在每个请求处理之前检查用户是否已登录。 - 白名单中的路径（如登录页、验证码接口）跳过验证。 - - Returns: - None: 验证通过，继续处理请求 - Response: 验证失败，返回未授权响应 - """ - return check_login() +from src.core.middleware import * # noqa: F401,F403 +from src.core.permission import check_permission, is_internal_call # noqa: F401 diff --git a/application/middleware/check_demo.py b/application/middleware/check_demo.py index 58c0c05..c6ddc2b 100644 --- a/application/middleware/check_demo.py +++ b/application/middleware/check_demo.py @@ -1,39 +1,8 @@ """ -演示模式拦截装饰器 +兼容 shim（绞杀者迁移期间保留） ============================================================= -在演示环境（FLASK_DEMO=True）下拦截所有写操作， -返回"演示环境，暂无操作权限"提示。 - -使用方式： - 在 View 路由函数上添加 @check_demo 装饰器， - 替代原有的 if FLASK_DEMO: return R.failed(...) 判断。 +演示模式拦截器已迁移至 src/core/middleware.py。 +本文件仅作兼容转发，最终在 P7（切尾）阶段删除。 """ - -from functools import wraps - -from config import FLASK_DEMO -from utils import R - - -def check_demo(func): - """ - 演示模式拦截装饰器 - - 当 FLASK_DEMO=True 时，拦截写操作并返回提示。 - 应放在 @check_permission 之后、路由函数之前。 - - Example: - @position_blue.route('/add', methods=["POST"]) - @check_permission("sys:position:add") - @check_demo - def add(): - result = position.add_position() - return result - """ - @wraps(func) - def wrapper(*args, **kwargs): - if FLASK_DEMO: - return R.failed("演示环境，暂无操作权限") - return func(*args, **kwargs) - return wrapper +from src.core.middleware import * # noqa: F401,F403 diff --git a/application/middleware/check_login.py b/application/middleware/check_login.py index f1d64dd..a90728c 100644 --- a/application/middleware/check_login.py +++ b/application/middleware/check_login.py @@ -1,200 +1,8 @@ -# +====================================================================== -# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] -# +---------------------------------------------------------------------- -# | Copyright (c) 2021-2026 一米阳光团队 版权所有 -# +---------------------------------------------------------------------- -# | 开源协议: Apache-2.0 -# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 -# +---------------------------------------------------------------------- -# | 官方网站: https://www.djangoadmin.cn -# +---------------------------------------------------------------------- -# | 研发团队: 一米阳光团队 荣誉出品 -# +====================================================================== -# | 版权和免责声明: -# | 一、知识产权声明 -# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 -# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， -# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 -# | 二、使用限制与禁止条款 -# | 您理解并同意，禁止将本产品用于以下任何行为： -# | 1. 任何违反中华人民共和国法律法规的用途； -# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； -# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 -# | 三、责任豁免 -# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 -# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 -# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 -# | 四、违约与法律责任 -# | 如发现任何违反本声明的行为，本团队有权： -# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； -# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； -# | 3. 保留进一步采取刑事控告等法律措施的权利。 -# | 五、许可范围说明 -# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 -# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 -# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 -# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 -# | 成的一切经济损失及商誉损害。 -# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 -# | 许可协议》附件为准。 -# +====================================================================== +""" +兼容 shim（绞杀者迁移期间保留） +============================================================= -import fnmatch -import logging - -from flask import request, g - -from config import LOGIN_WHITELIST_EXACT, LOGIN_WHITELIST_DEBUG, LOGIN_WHITELIST_PREFIX, LOGIN_WHITELIST_WILDCARD -from utils import R -from utils.jwt import parse_payload, is_token_blacklisted, get_access_token - -# 获取日志记录器 -logger = logging.getLogger(__name__) - - -def is_whitelisted(path: str) -> bool: - """ - 检查请求路径是否在白名单中 - - 支持三种匹配模式： - 1. 精确匹配：路径完全相等 - 2. 前缀匹配：路径以指定前缀开头 - 3. 通配符匹配：支持 *（匹配任意字符）和 ?（匹配单个字符） - - Args: - path: 请求路径（如：/job/sendSms） - - Returns: - bool: True-在白名单中，False-不在白名单中 - """ - # ============================================================ - # 1. 精确匹配检查 - # ============================================================ - # 示例：LOGIN_WHITELIST_EXACT=/login,/captcha - # 当 path 为 /login 或 /captcha 时返回 True - exact_urls = LOGIN_WHITELIST_EXACT - if exact_urls: - for url in exact_urls.split(','): - if url.strip() and url.strip() == path: - if LOGIN_WHITELIST_DEBUG: - logger.debug(f"精确匹配白名单: {path}") - return True - - # ============================================================ - # 2. 前缀匹配检查 - # ============================================================ - # 示例：LOGIN_WHITELIST_PREFIX=/job/sendSms,/public - # 当 path 以 /job/sendSms 或 /public 开头时返回 True - prefix_urls = LOGIN_WHITELIST_PREFIX - if prefix_urls: - for prefix in prefix_urls.split(','): - prefix = prefix.strip() - if prefix and path.startswith(prefix): - if LOGIN_WHITELIST_DEBUG: - logger.debug(f"前缀匹配白名单: {path} 匹配 {prefix}") - return True - - # ============================================================ - # 3. 通配符匹配检查 - # ============================================================ - # 示例：LOGIN_WHITELIST_WILDCARD=/job/*/send*,/api/v1/*/public* - # 当 path 匹配通配符模式时返回 True - # - * 匹配任意多个字符 - # - ? 匹配单个字符 - wildcard_patterns = LOGIN_WHITELIST_WILDCARD - if wildcard_patterns: - for pattern in wildcard_patterns.split(','): - pattern = pattern.strip() - if pattern and fnmatch.fnmatch(path, pattern): - if LOGIN_WHITELIST_DEBUG: - logger.debug(f"通配符匹配白名单: {path} 匹配 {pattern}") - return True - - return False - - -def check_login(): - """ - 用户登录验证中间件 - - 功能说明： - 1. 检查请求路径是否在白名单中，白名单路径直接放行 - 2. OPTIONS请求（CORS预检）直接放行 - 3. 验证JWT Token的有效性 - 4. Token验证通过后将用户信息存储到g对象中 - - 白名单配置（在.env文件中）： - - LOGIN_WHITELIST_EXACT: 精确匹配白名单 - - LOGIN_WHITELIST_PREFIX: 前缀匹配白名单 - - LOGIN_WHITELIST_WILDCARD: 通配符匹配白名单 - - Returns: - None: 验证通过，继续处理请求 - R.failed: 验证失败，返回错误响应（401未登录） - - 使用示例： - # 在app.py中注册中间件 - @app.before_request - def before_request(): - result = check_login() - if result: - return result - """ - current_path = request.path - - # ============================================================ - # 步骤1：检查白名单 - # ============================================================ - # 白名单中的路径不需要验证Token（如：登录页、验证码、短信发送接口等） - if is_whitelisted(current_path): - if LOGIN_WHITELIST_DEBUG: - logger.debug(f"白名单路径，跳过登录验证: {current_path}") - return None - - # ============================================================ - # 步骤2：OPTIONS请求放行 - # ============================================================ - # OPTIONS请求是CORS预检请求，不需要验证Token - if request.method == "OPTIONS": - if LOGIN_WHITELIST_DEBUG: - logger.debug(f"OPTIONS请求，跳过登录验证: {current_path}") - return None - - # ============================================================ - # 步骤3：获取并验证Token - # ============================================================ - success, access_token, msg = get_access_token() - - # 如果没有Token - if not success: - logger.warning(f"未登录访问: {current_path}, IP: {request.remote_addr}") - return R.failed(code=401, msg="请先登录") - - # JWT解密验证 - result = parse_payload(access_token) - - # 验证失败处理 - if result.get('code') != 0: - logger.warning(f"Token验证失败: {result.get('msg')}, 路径: {current_path}") - return R.failed(code=401, msg=result.get('msg', "Token无效，请重新登录")) - - # ============================================================ - # 步骤3.5：检查Token黑名单 - # ============================================================ - # 退出登录时Token会被加入黑名单，黑名单中的Token即使未过期也应拒绝 - if is_token_blacklisted(access_token): - logger.warning(f"Token已被注销（黑名单）: user_id={result.get('data', {}).get('userId')}, 路径: {current_path}") - return R.failed(code=401, msg="Token已失效，请重新登录") - - # ============================================================ - # 步骤4：存储用户信息 - # ============================================================ - # 将用户信息存储到g对象中，供后续的视图函数和权限验证使用 - if result.get('data'): - g.user_id = result['data'].get('userId') - g.user_info = result['data'] - if LOGIN_WHITELIST_DEBUG: - logger.debug(f"用户登录验证通过: user_id={g.user_id}, 路径={current_path}") - - # 验证通过，返回None继续处理请求 - return None +登录校验中间件已迁移至 src/core/middleware.py。 +本文件仅作兼容转发，最终在 P7（切尾）阶段删除。 +""" +from src.core.middleware import * # noqa: F401,F403 diff --git a/application/middleware/check_permission.py b/application/middleware/check_permission.py index d7aa847..22bdc0e 100644 --- a/application/middleware/check_permission.py +++ b/application/middleware/check_permission.py @@ -1,181 +1,8 @@ -# +====================================================================== -# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ] -# +---------------------------------------------------------------------- -# | Copyright (c) 2021-2026 一米阳光团队 版权所有 -# +---------------------------------------------------------------------- -# | 开源协议: Apache-2.0 -# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。 -# +---------------------------------------------------------------------- -# | 官方网站: https://www.djangoadmin.cn -# +---------------------------------------------------------------------- -# | 研发团队: 一米阳光团队 荣誉出品 -# +====================================================================== -# | 版权和免责声明: -# | 一、知识产权声明 -# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、 -# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权， -# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。 -# | 二、使用限制与禁止条款 -# | 您理解并同意，禁止将本产品用于以下任何行为： -# | 1. 任何违反中华人民共和国法律法规的用途； -# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为； -# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。 -# | 三、责任豁免 -# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失 -# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三 -# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。 -# | 四、违约与法律责任 -# | 如发现任何违反本声明的行为，本团队有权： -# | 1. 无条件配合公安机关及其他有权机关的调查取证工作； -# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失； -# | 3. 保留进一步采取刑事控告等法律措施的权利。 -# | 五、许可范围说明 -# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包 -# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通 -# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、 -# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造 -# | 成的一切经济损失及商誉损害。 -# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件 -# | 许可协议》附件为准。 -# +====================================================================== +""" +兼容 shim（绞杀者迁移期间保留） +============================================================= -import fnmatch -import hmac -import logging -from functools import wraps - -from flask import request, g - -from application.services import menu -from config import PERMISSION_ENABLED, INTERNAL_API_KEY, INTERNAL_ALLOWED_IPS -from utils import R -from utils.security import get_user_id - -# 获取日志记录器 -logger = logging.getLogger(__name__) - - -def is_internal_call() -> bool: - """ - 判断是否为内部系统调用（仅定时任务等无用户场景） - - 验证方式（满足任一即可）： - 1. 权限验证总开关关闭（开发环境） - 2. API密钥验证通过（请求头携带正确密钥） - 3. IP白名单验证通过（来源IP在白名单中） - - Returns: - bool: True-内部调用（跳过权限验证），False-普通调用（需要权限验证） - """ - # ============================================================ - # 方式1：权限验证总开关关闭（开发/测试环境） - # ============================================================ - if not PERMISSION_ENABLED: - logger.debug("权限验证已关闭，所有请求视为内部调用") - return True - - # ============================================================ - # 方式2：API密钥验证（定时任务使用） - # ============================================================ - # 定时任务在请求头中携带 X-Scheduler-Key 或 X-Internal-API-Key - if INTERNAL_API_KEY: - # 优先检查调度器专用密钥 - req_key = request.headers.get('X-Scheduler-Key', '') - if not req_key: - req_key = request.headers.get('X-Internal-API-Key', '') - - if hmac.compare_digest(req_key, INTERNAL_API_KEY): - logger.debug(f"内部API密钥验证通过: IP={request.remote_addr}") - return True - - # ============================================================ - # 方式3：IP白名单验证（可选） - # ============================================================ - if INTERNAL_ALLOWED_IPS: - client_ip = request.remote_addr - allowed_ips = [ip.strip() for ip in INTERNAL_ALLOWED_IPS.split(',') if ip.strip()] - - for allowed_ip in allowed_ips: - if fnmatch.fnmatch(client_ip, allowed_ip): - logger.debug(f"内部IP白名单验证通过: IP={client_ip}") - return True - - return False - - -def check_permission(permission: str, log: bool = False): - """ - 权限验证装饰器 - - 权限验证规则： - 1. 定时任务等内部调用 → 跳过权限验证，直接放行 - 2. 超级管理员（user_id=1） → 拥有所有权限 - 3. 普通用户 → 验证是否拥有指定权限节点 - - Args: - permission: 权限节点标识（如：sys:job:page） - log: 是否记录访问日志 - - Returns: - decorator: 装饰器函数 - """ - - def decorator(func): - @wraps(func) - def wrapper(*args, **kwargs): - # ============================================================ - # 定时任务等内部调用：跳过权限验证，直接放行 - # ============================================================ - if is_internal_call(): - logger.debug(f"定时任务内部调用，跳过权限验证: {func.__name__}") - return func(*args, **kwargs) - - # ============================================================ - # 普通请求：正常进行权限验证 - # ============================================================ - - # 获取用户ID（优先从g对象获取） - user_id = getattr(g, 'user_id', None) - if user_id is None: - try: - user_id = get_user_id() - except Exception as e: - logger.error(f"获取用户ID失败: {str(e)}") - user_id = None - - # 超级管理员放行 - if user_id == 1: - logger.debug(f"超级管理员权限验证通过: user_id={user_id}") - return func(*args, **kwargs) - - # 普通用户权限验证 - if user_id: - try: - permission_list = menu.get_user_permissions(user_id) - - if permission not in permission_list: - if log: - logger.warning( - f"权限不足: user_id={user_id}, " - f"permission={permission}, " - f"path={request.path}" - ) - return R.failed(msg="权限不足!", code=403) - elif log: - logger.info( - f"权限验证通过: user_id={user_id}, " - f"permission={permission}" - ) - except Exception as e: - logger.error(f"权限验证异常: {str(e)}") - return R.failed(msg="权限验证失败", code=500) - else: - # 未登录用户 - logger.warning(f"未登录用户访问: path={request.path}") - return R.failed(msg="请先登录", code=401) - - return func(*args, **kwargs) - - return wrapper - - return decorator +权限校验已迁移至 src/core/permission.py（依赖注入版）。 +本文件仅作兼容转发，最终在 P7（切尾）阶段删除。 +""" +from src.core.permission import * # noqa: F401,F403 diff --git a/application/middleware/error_handler.py b/application/middleware/error_handler.py index a9d53be..fe0d745 100644 --- a/application/middleware/error_handler.py +++ b/application/middleware/error_handler.py @@ -1,95 +1,8 @@ """ -全局异常处理模块 +兼容 shim（绞杀者迁移期间保留） ============================================================= -注册 Flask errorhandler，将所有 HTTP 错误和未捕获异常 -统一转为 R.failed() JSON 响应，避免前端收到 HTML 错误页导致解析崩溃。 - -处理范围： - - 400：请求参数错误 - - 401：未授权 - - 403：权限不足 - - 404：资源不存在 - - 405：请求方法不允许 - - 500：服务器内部错误 - - Exception：兜底，捕获所有未预期异常 +全局异常处理已迁移至 src/core/middleware.py。 +本文件仅作兼容转发，最终在 P7（切尾）阶段删除。 """ - -import logging -import traceback - -from flask import request -from werkzeug.exceptions import HTTPException - -from utils import R -from src.core.exceptions import BizError - -logger = logging.getLogger(__name__) - - -def register_error_handlers(app): - """ - 注册全局异常处理器 - - 确保所有 HTTP 错误和未捕获异常都返回标准 JSON 响应。 - - Args: - app: Flask 应用实例 - """ - - @app.errorhandler(400) - def bad_request(e): - return R.failed(msg="请求参数错误", code=400), 400 - - @app.errorhandler(401) - def unauthorized(e): - return R.failed(msg="未授权，请重新登录", code=401), 401 - - @app.errorhandler(403) - def forbidden(e): - return R.failed(msg="权限不足", code=403), 403 - - @app.errorhandler(404) - def not_found(e): - return R.failed(msg="请求的资源不存在", code=404), 404 - - @app.errorhandler(405) - def method_not_allowed(e): - return R.failed(msg="请求方法不允许", code=405), 405 - - @app.errorhandler(500) - def internal_error(e): - logger.error( - f"服务器内部错误, method: {request.method}, path: {request.path}, " - f"error: {str(e)}", - exc_info=True - ) - msg = f"服务器内部错误: {str(e)}" if app.debug else "服务器内部错误" - return R.failed(msg=msg, code=500), 500 - - @app.errorhandler(BizError) - def handle_biz_error(e): - """ - 业务异常处理器 - - 业务代码 raise BizError(...) 时统一转换为 R.failed 响应， - 并回滚当前事务（teardown_request 处理）。 - """ - logger.info(f"业务异常: {e.msg} (code={e.code})") - return R.failed(msg=e.msg, code=e.code, data=e.data, **e.kwargs) - - @app.errorhandler(Exception) - def handle_exception(e): - # HTTPException 子类（如 abort(403)）已有专用 handler，不重复处理 - if isinstance(e, HTTPException): - return e.get_response() - - # 非HTTP异常：未知运行时异常 - logger.error( - f"未捕获异常, method: {request.method}, path: {request.path}, " - f"error: {traceback.format_exc()}" - ) - msg = f"服务器内部错误: {str(e)}" if app.debug else "服务器内部错误" - return R.failed(msg=msg, code=500), 500 - - logger.info("全局异常处理器注册成功") +from src.core.middleware import * # noqa: F401,F403 diff --git a/application/middleware/op_log.py b/application/middleware/op_log.py index 768f691..357d1e6 100644 --- a/application/middleware/op_log.py +++ b/application/middleware/op_log.py @@ -1,220 +1,8 @@ """ -操作日志自动记录中间件 +兼容 shim（绞杀者迁移期间保留） ============================================================= -通过 after_request 钩子自动记录写操作（POST/PUT/DELETE/PATCH）日志。 -无需在每个视图函数中手动调用 record_operation_log。 - -记录规则： - - 只记录写操作（POST/PUT/DELETE/PATCH），GET/OPTIONS/HEAD 不记录 - - 只记录成功操作（HTTP 2xx 且业务 code=0） - - 内部调用（定时任务等）不记录 - - 登录/退出/验证码等路径不记录（由 login_log 单独记录） - - 日志记录失败不影响主流程 +操作日志中间件已迁移至 src/core/middleware.py（写入器由应用工厂注入）。 +本文件仅作兼容转发，最终在 P7（切尾）阶段删除。 """ - -import logging -import time - -from flask import g, request - -from application.models.operation_log import OperationLog - -logger = logging.getLogger(__name__) - -# 不记录操作日志的路径前缀（登录/退出/验证码等由 login_log 单独记录） -_EXCLUDED_PREFIXES = ('/login', '/captcha', '/logout') - -# HTTP 方法 → 操作日志类型 映射 -# 类型值参考 application/constants/dict.py 的 OPERATION_LOG_TYPE_LIST -_METHOD_LOG_TYPE = { - 'POST': 3, # 新增 - 'PUT': 4, # 修改 - 'DELETE': 5, # 删除 - 'PATCH': 4, # 修改 -} - -# 路径后缀 → 操作日志类型 覆盖映射（更精确） -_PATH_ACTION_TYPE = { - '/add': 3, # 新增 - '/update': 4, # 修改 - '/delete': 5, # 删除 - '/status': 4, # 状态变更→修改 - '/batchdelete': 5, # 批量删除 - '/reset': 4, # 重置→修改 - '/save': 4, # 保存→修改 - '/sort': 4, # 排序→修改 -} - - -def register_operation_log(app): - """ - 注册操作日志自动记录中间件 - - 在 after_request 中自动记录写操作日志。 - - Args: - app: Flask 应用实例 - """ - - @app.after_request - def record_op_log(response): - # 只记录写操作 - if request.method not in ('POST', 'PUT', 'DELETE', 'PATCH'): - return response - - # 排除登录/退出/验证码等路径 - path = request.path - if any(path.startswith(prefix) for prefix in _EXCLUDED_PREFIXES): - return response - - # 只记录成功操作（HTTP 2xx 且业务 code=0） - if not (200 <= response.status_code < 300): - return response - - try: - data = response.get_json(silent=True) - if data and data.get('code', 0) != 0: - return response - except Exception: - pass - - # 内部调用不记录 - if getattr(g, '_internal_call', False): - return response - - # 异步记录，不阻塞响应 - try: - _write_log(response) - except Exception as e: - logger.warning(f"记录操作日志失败: {str(e)}") - - return response - - logger.info("操作日志中间件注册成功") - - -def _write_log(response): - """写入操作日志""" - from application.services import operation_log - from utils.security import get_username - from utils.ip import get_client_ip - from utils.ip2region import get_ip_location - from utils.user_agent import parse_user_agent - - username = get_username() or 'system' - client_ip = get_client_ip() - ua_info = parse_user_agent() - - # 确定操作类型 - log_type = _infer_log_type(request.method, request.path) - - # 构建标题：从路径提取模块名，如 /user/add → "用户管理-新增" - title = _build_title(request.path) - - # 请求参数（限制长度，避免存入过大的数据） - param = _get_request_param() - - # 返回结果（限制长度） - result = _get_response_result(response) - - # 请求耗时 - start_time = getattr(g, 'request_start_time', None) - consume_time = int((time.time() - start_time) * 1000) if start_time else 0 - - operation_log.record_operation_log( - title=title, - log_type=log_type, - source=OperationLog.Source.ADMIN, - status=OperationLog.Status.NORMAL, - method=request.method, - request_method=request.endpoint or '', - url=request.path, - param=param, - result=result, - ip=client_ip, - location=get_ip_location(client_ip), - os=ua_info.get('os', '未知'), - browser=ua_info.get('browser', '未知'), - consume_time=consume_time, - error='', - username=username, - ) - - -def _infer_log_type(method, path): - """根据请求方法和路径推断操作日志类型""" - # 优先根据路径后缀匹配 - path_lower = path.lower() - for suffix, log_type in _PATH_ACTION_TYPE.items(): - if path_lower.endswith(suffix): - return log_type - - # 回退到 HTTP 方法映射 - return _METHOD_LOG_TYPE.get(method, 18) # 18=其他 - - -def _get_request_param(): - """获取请求参数""" - try: - if request.method in ('POST', 'PUT', 'PATCH', 'DELETE'): - # 优先解析 JSON body（batchDelete 等 DELETE 请求也带 body） - data = request.get_json(silent=True) - if data: - import json - return json.dumps(data, ensure_ascii=False) - # 回退到 URL 查询参数 - if request.args: - import json - from collections import OrderedDict - return json.dumps(OrderedDict(request.args), ensure_ascii=False) - except Exception: - pass - return '' - - -def _get_response_result(response): - """获取响应结果中的data字段""" - try: - data = response.get_json(silent=True) - if data and isinstance(data, dict): - result_data = data.get('data') - if result_data is not None: - import json - return json.dumps(result_data, ensure_ascii=False) - # data字段为空时取msg - return data.get('msg', '') - except Exception: - pass - return '' - - -def _build_title(path): - """从请求路径构建操作标题""" - # 路径格式: /user/add, /role/update, /menu/delete 等 - parts = path.strip('/').split('/') - if len(parts) >= 2: - module = parts[0] - # 模块名中文化映射 - module_names = { - 'user': '用户管理', 'role': '角色管理', 'menu': '菜单管理', - 'dept': '部门管理', 'position': '岗位管理', 'level': '职级管理', - 'param': '参数配置', 'dict': '字典管理', 'dictitem': '字典项管理', - 'config': '系统配置', 'configitem': '配置项管理', - 'category': '分类管理', 'article': '文章管理', - 'notice': '通知管理', 'link': '友情链接', - 'job': '定时任务', 'login': '登录管理', - 'loginlog': '登录日志', 'operationlog': '操作日志', - 'upload': '文件上传', - } - module_name = module_names.get(module.lower(), module) - action = parts[-1] if len(parts) > 1 else '' - action_names = { - 'add': '新增', 'update': '修改', 'delete': '删除', - 'status': '状态变更', 'batchdelete': '批量删除', - 'reset': '重置', 'save': '保存', 'sort': '排序', - 'page': '查询', 'detail': '查询', 'list': '查询', - } - action_name = action_names.get(action.lower(), action) - return f"{module_name}-{action_name}" if action_name else module_name - return path +from src.core.middleware import * # noqa: F401,F403 diff --git a/application/middleware/transaction.py b/application/middleware/transaction.py index fd67ca3..5dc5669 100644 --- a/application/middleware/transaction.py +++ b/application/middleware/transaction.py @@ -1,241 +1,8 @@ """ -AOP 事务拦截器（企业级实现） +兼容 shim（绞杀者迁移期间保留） +============================================================= -设计理念： - 1. 零代码侵入：业务代码无需任何事务管理代码 - 2. 自动读写分离：GET 不提交，POST/PUT/DELETE 自动提交 - 3. 统一异常处理：所有异常自动回滚 - 4. 性能监控：记录事务执行时间 - -使用方式： - 在 Flask 应用初始化时调用 register_transaction(app) 即可。 - 业务代码无需任何改动。 - -Example: - # app/__init__.py - from application.middleware.transaction import register_transaction - - def create_app(): - app = Flask(__name__) - register_transaction(app) # 一行代码搞定所有事务 - return app +AOP 事务拦截器已迁移至 src/core/middleware.py。 +本文件仅作兼容转发，最终在 P7（切尾）阶段删除。 """ - -from flask import request, g -from extensions import db -import logging -import time -from functools import wraps - -logger = logging.getLogger(__name__) - - -# ============================================================ -# 核心：AOP 事务拦截器 -# ============================================================ - -def register_transaction(app): - """ - 注册 AOP 事务拦截器到 Flask 应用 - - 在请求生命周期中自动管理事务： - - before_request: 开启事务标记 - - after_request: 自动提交（POST/PUT/DELETE） - - teardown_request: 异常回滚 + 资源清理 - - Args: - app: Flask 应用实例 - - Example: - register_transaction(app) - """ - - @app.before_request - def begin_transaction(): - """请求开始：记录开始时间，开启事务标记""" - g.request_start_time = time.time() - g.transaction_active = True - - logger.debug( - f"请求开始, method: {request.method}, path: {request.path}, " - f"ip: {request.remote_addr}" - ) - - @app.after_request - def commit_transaction(response): - """ - 请求结束：自动提交事务 - - 提交规则： - 1. 只对写操作（POST/PUT/DELETE/PATCH）提交 - 2. 只在响应成功（2xx）且业务成功（code=0）时提交 - 3. GET/OPTIONS/HEAD 不提交（只读） - 4. 业务失败（R.failed）不提交，自动回滚 - """ - # 只对写操作处理事务 - if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'): - should_commit = 200 <= response.status_code < 300 - - # 检查业务状态码：R.failed() 返回 HTTP 200 但 code=1/ok=False - if should_commit: - try: - data = response.get_json(silent=True) - if data and data.get('code', 0) != 0: - should_commit = False - except Exception: - pass - - if should_commit: - try: - if db.session.is_active: - db.session.commit() - logger.debug( - f"事务提交成功, method: {request.method}, " - f"path: {request.path}, status: {response.status_code}" - ) - except Exception as e: - db.session.rollback() - logger.error( - f"事务提交失败, method: {request.method}, " - f"path: {request.path}, error: {str(e)}", - exc_info=True - ) - from utils import R - return R.failed(msg="数据处理失败"), 500 - else: - # 业务失败或非2xx，回滚未提交的变更 - try: - if db.session.is_active: - db.session.rollback() - except Exception: - pass - - # 记录响应日志 - elapsed = time.time() - g.get('request_start_time', time.time()) - logger.debug( - f"请求结束, method: {request.method}, path: {request.path}, " - f"status: {response.status_code}, elapsed: {elapsed:.3f}s" - ) - - return response - - @app.teardown_request - def cleanup_transaction(exception=None): - """ - 请求结束：清理资源 - - 发生异常时自动回滚事务，确保数据一致性。 - """ - if exception is not None: - try: - db.session.rollback() - logger.error( - f"请求异常，事务已回滚, method: {request.method}, " - f"path: {request.path}, error: {str(exception)}", - exc_info=True - ) - except Exception as e: - logger.error(f"事务回滚失败: {str(e)}", exc_info=True) - - # 清理请求标记 - g.pop('transaction_active', None) - g.pop('request_start_time', None) - - logger.info("AOP 事务拦截器注册成功") - - -# ============================================================ -# 备用：精细控制装饰器（特殊场景使用） -# ============================================================ - -def transactional(func): - """ - 事务装饰器（备用） - - 适用于需要精细控制事务边界的特殊场景。 - 大多数情况下，使用 AOP 拦截器即可。 - - Example: - @transactional - def complex_business(): - # 精细控制的事务逻辑 - pass - """ - - @wraps(func) - def wrapper(*args, **kwargs): - if g.get('disable_auto_commit', False): - return func(*args, **kwargs) - - try: - result = func(*args, **kwargs) - if not g.get('transaction_active', False): - db.session.commit() - return result - except Exception as e: - db.session.rollback() - logger.error(f"装饰器事务执行失败: {str(e)}", exc_info=True) - raise - - return wrapper - - -def transactional_response(func): - """ - 事务装饰器（带返回 R 对象） - - 用于需要直接返回 R 对象的接口。 - - Example: - @transactional_response - def add_position(): - position.save() - return R.ok(msg="添加成功") - """ - - @wraps(func) - def wrapper(*args, **kwargs): - from utils import R - try: - result = func(*args, **kwargs) - if not g.get('transaction_active', False): - db.session.commit() - return result - except Exception as e: - db.session.rollback() - logger.error(f"装饰器事务执行失败: {str(e)}", exc_info=True) - return R.failed(msg=f"操作失败: {str(e)}") - - return wrapper - - -# ============================================================ -# 备用：事务上下文管理器（特殊场景使用） -# ============================================================ - -class TransactionContext: - """ - 事务上下文管理器 - - 用于需要精细控制事务的特殊场景。 - - Example: - with TransactionContext(): - position.save() - user.save() - # 自动提交或回滚 - """ - - def __enter__(self): - g.disable_auto_commit = True - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - g.disable_auto_commit = False - if exc_type is None: - if not g.get('transaction_active', False): - db.session.commit() - else: - db.session.rollback() - logger.error(f"事务上下文回滚: {str(exc_val)}", exc_info=True) - return False +from src.core.middleware import * # noqa: F401,F403 diff --git a/src/core/middleware.py b/src/core/middleware.py new file mode 100644 index 0000000..f319c90 --- /dev/null +++ b/src/core/middleware.py @@ -0,0 +1,813 @@ +""" +中间件模块 +============================================================= + +整合自 application/middleware/*.py： + - check_login: 登录校验（JWT + 白名单） + - check_demo: 演示模式拦截 + - transaction: AOP 事务拦截器 + - op_log: 操作日志自动记录（写入器由应用工厂注入） + - error_handler: 全局异常处理 + +依赖约束：本模块仅依赖 core/common，不依赖 modules 层。 +""" + +import logging +import fnmatch +from flask import g, request, current_app +from config import LOGIN_WHITELIST_EXACT, LOGIN_WHITELIST_DEBUG, LOGIN_WHITELIST_PREFIX, LOGIN_WHITELIST_WILDCARD +from src.core import response as R +from src.core.jwt import parse_payload, is_token_blacklisted, get_access_token +from functools import wraps +from config import FLASK_DEMO +from src.core.database import db +import time +import traceback +from werkzeug.exceptions import HTTPException +from src.core.exceptions import BizError + +logger = logging.getLogger(__name__) + + +# ============================================================ +# 登录校验中间件 +# ============================================================ + + +# 获取日志记录器 + + +def is_whitelisted(path: str) -> bool: + """ + 检查请求路径是否在白名单中 + + 支持三种匹配模式： + 1. 精确匹配：路径完全相等 + 2. 前缀匹配：路径以指定前缀开头 + 3. 通配符匹配：支持 *（匹配任意字符）和 ?（匹配单个字符） + + Args: + path: 请求路径（如：/job/sendSms） + + Returns: + bool: True-在白名单中，False-不在白名单中 + """ + # ============================================================ + # 1. 精确匹配检查 + # ============================================================ + # 示例：LOGIN_WHITELIST_EXACT=/login,/captcha + # 当 path 为 /login 或 /captcha 时返回 True + exact_urls = LOGIN_WHITELIST_EXACT + if exact_urls: + for url in exact_urls.split(","): + if url.strip() and url.strip() == path: + if LOGIN_WHITELIST_DEBUG: + logger.debug(f"精确匹配白名单: {path}") + return True + + # ============================================================ + # 2. 前缀匹配检查 + # ============================================================ + # 示例：LOGIN_WHITELIST_PREFIX=/job/sendSms,/public + # 当 path 以 /job/sendSms 或 /public 开头时返回 True + prefix_urls = LOGIN_WHITELIST_PREFIX + if prefix_urls: + for prefix in prefix_urls.split(","): + prefix = prefix.strip() + if prefix and path.startswith(prefix): + if LOGIN_WHITELIST_DEBUG: + logger.debug(f"前缀匹配白名单: {path} 匹配 {prefix}") + return True + + # ============================================================ + # 3. 通配符匹配检查 + # ============================================================ + # 示例：LOGIN_WHITELIST_WILDCARD=/job/*/send*,/api/v1/*/public* + # 当 path 匹配通配符模式时返回 True + # - * 匹配任意多个字符 + # - ? 匹配单个字符 + wildcard_patterns = LOGIN_WHITELIST_WILDCARD + if wildcard_patterns: + for pattern in wildcard_patterns.split(","): + pattern = pattern.strip() + if pattern and fnmatch.fnmatch(path, pattern): + if LOGIN_WHITELIST_DEBUG: + logger.debug(f"通配符匹配白名单: {path} 匹配 {pattern}") + return True + + return False + + +def check_login(): + """ + 用户登录验证中间件 + + 功能说明： + 1. 检查请求路径是否在白名单中，白名单路径直接放行 + 2. OPTIONS请求（CORS预检）直接放行 + 3. 验证JWT Token的有效性 + 4. Token验证通过后将用户信息存储到g对象中 + + 白名单配置（在.env文件中）： + - LOGIN_WHITELIST_EXACT: 精确匹配白名单 + - LOGIN_WHITELIST_PREFIX: 前缀匹配白名单 + - LOGIN_WHITELIST_WILDCARD: 通配符匹配白名单 + + Returns: + None: 验证通过，继续处理请求 + R.failed: 验证失败，返回错误响应（401未登录） + + 使用示例： + # 在app.py中注册中间件 + @app.before_request + def before_request(): + result = check_login() + if result: + return result + """ + current_path = request.path + + # ============================================================ + # 步骤1：检查白名单 + # ============================================================ + # 白名单中的路径不需要验证Token（如：登录页、验证码、短信发送接口等） + if is_whitelisted(current_path): + if LOGIN_WHITELIST_DEBUG: + logger.debug(f"白名单路径，跳过登录验证: {current_path}") + return None + + # ============================================================ + # 步骤2：OPTIONS请求放行 + # ============================================================ + # OPTIONS请求是CORS预检请求，不需要验证Token + if request.method == "OPTIONS": + if LOGIN_WHITELIST_DEBUG: + logger.debug(f"OPTIONS请求，跳过登录验证: {current_path}") + return None + + # ============================================================ + # 步骤3：获取并验证Token + # ============================================================ + success, access_token, msg = get_access_token() + + # 如果没有Token + if not success: + logger.warning(f"未登录访问: {current_path}, IP: {request.remote_addr}") + return R.failed(code=401, msg="请先登录") + + # JWT解密验证 + result = parse_payload(access_token) + + # 验证失败处理 + if result.get("code") != 0: + logger.warning(f"Token验证失败: {result.get('msg')}, 路径: {current_path}") + return R.failed(code=401, msg=result.get("msg", "Token无效，请重新登录")) + + # ============================================================ + # 步骤3.5：检查Token黑名单 + # ============================================================ + # 退出登录时Token会被加入黑名单，黑名单中的Token即使未过期也应拒绝 + if is_token_blacklisted(access_token): + logger.warning(f"Token已被注销（黑名单）: user_id={result.get('data', {}).get('userId')}, 路径: {current_path}") + return R.failed(code=401, msg="Token已失效，请重新登录") + + # ============================================================ + # 步骤4：存储用户信息 + # ============================================================ + # 将用户信息存储到g对象中，供后续的视图函数和权限验证使用 + if result.get("data"): + g.user_id = result["data"].get("userId") + g.user_info = result["data"] + if LOGIN_WHITELIST_DEBUG: + logger.debug(f"用户登录验证通过: user_id={g.user_id}, 路径={current_path}") + + # 验证通过，返回None继续处理请求 + return None + + +# ============================================================ +# 演示模式拦截 +# ============================================================ + +""" +演示模式拦截装饰器 +============================================================= + +在演示环境（FLASK_DEMO=True）下拦截所有写操作， +返回"演示环境，暂无操作权限"提示。 + +使用方式： + 在 View 路由函数上添加 @check_demo 装饰器， + 替代原有的 if FLASK_DEMO: return R.failed(...) 判断。 +""" + + +def check_demo(func): + """ + 演示模式拦截装饰器 + + 当 FLASK_DEMO=True 时，拦截写操作并返回提示。 + 应放在 @check_permission 之后、路由函数之前。 + + Example: + @position_blue.route('/add', methods=["POST"]) + @check_permission("sys:position:add") + @check_demo + def add(): + result = position.add_position() + return result + """ + + @wraps(func) + def wrapper(*args, **kwargs): + if FLASK_DEMO: + return R.failed("演示环境，暂无操作权限") + return func(*args, **kwargs) + + return wrapper + + +# ============================================================ +# AOP 事务拦截器 +# ============================================================ + +""" +AOP 事务拦截器（企业级实现） + +设计理念： + 1. 零代码侵入：业务代码无需任何事务管理代码 + 2. 自动读写分离：GET 不提交，POST/PUT/DELETE 自动提交 + 3. 统一异常处理：所有异常自动回滚 + 4. 性能监控：记录事务执行时间 + +使用方式： + 在 Flask 应用初始化时调用 register_transaction(app) 即可。 + 业务代码无需任何改动。 + +Example: + # app/__init__.py + from src.core.middleware import register_transaction + + def create_app(): + app = Flask(__name__) + register_transaction(app) # 一行代码搞定所有事务 + return app +""" + + +# ============================================================ +# 核心：AOP 事务拦截器 +# ============================================================ + + +def register_transaction(app): + """ + 注册 AOP 事务拦截器到 Flask 应用 + + 在请求生命周期中自动管理事务： + - before_request: 开启事务标记 + - after_request: 自动提交（POST/PUT/DELETE） + - teardown_request: 异常回滚 + 资源清理 + + Args: + app: Flask 应用实例 + + Example: + register_transaction(app) + """ + + @app.before_request + def begin_transaction(): + """请求开始：记录开始时间，开启事务标记""" + g.request_start_time = time.time() + g.transaction_active = True + + logger.debug(f"请求开始, method: {request.method}, path: {request.path}, " f"ip: {request.remote_addr}") + + @app.after_request + def commit_transaction(response): + """ + 请求结束：自动提交事务 + + 提交规则： + 1. 只对写操作（POST/PUT/DELETE/PATCH）提交 + 2. 只在响应成功（2xx）且业务成功（code=0）时提交 + 3. GET/OPTIONS/HEAD 不提交（只读） + 4. 业务失败（R.failed）不提交，自动回滚 + """ + # 只对写操作处理事务 + if request.method in ("POST", "PUT", "DELETE", "PATCH"): + should_commit = 200 <= response.status_code < 300 + + # 检查业务状态码：R.failed() 返回 HTTP 200 但 code=1/ok=False + if should_commit: + try: + data = response.get_json(silent=True) + if data and data.get("code", 0) != 0: + should_commit = False + except Exception: + pass + + if should_commit: + try: + if db.session.is_active: + db.session.commit() + logger.debug( + f"事务提交成功, method: {request.method}, " + f"path: {request.path}, status: {response.status_code}" + ) + except Exception as e: + db.session.rollback() + logger.error( + f"事务提交失败, method: {request.method}, " f"path: {request.path}, error: {str(e)}", + exc_info=True, + ) + from src.core import response as R + + return R.failed(msg="数据处理失败"), 500 + else: + # 业务失败或非2xx，回滚未提交的变更 + try: + if db.session.is_active: + db.session.rollback() + except Exception: + pass + + # 记录响应日志 + elapsed = time.time() - g.get("request_start_time", time.time()) + logger.debug( + f"请求结束, method: {request.method}, path: {request.path}, " + f"status: {response.status_code}, elapsed: {elapsed:.3f}s" + ) + + return response + + @app.teardown_request + def cleanup_transaction(exception=None): + """ + 请求结束：清理资源 + + 发生异常时自动回滚事务，确保数据一致性。 + """ + if exception is not None: + try: + db.session.rollback() + logger.error( + f"请求异常，事务已回滚, method: {request.method}, " + f"path: {request.path}, error: {str(exception)}", + exc_info=True, + ) + except Exception as e: + logger.error(f"事务回滚失败: {str(e)}", exc_info=True) + + # 清理请求标记 + g.pop("transaction_active", None) + g.pop("request_start_time", None) + + logger.info("AOP 事务拦截器注册成功") + + +# ============================================================ +# 备用：精细控制装饰器（特殊场景使用） +# ============================================================ + + +def transactional(func): + """ + 事务装饰器（备用） + + 适用于需要精细控制事务边界的特殊场景。 + 大多数情况下，使用 AOP 拦截器即可。 + + Example: + @transactional + def complex_business(): + # 精细控制的事务逻辑 + pass + """ + + @wraps(func) + def wrapper(*args, **kwargs): + if g.get("disable_auto_commit", False): + return func(*args, **kwargs) + + try: + result = func(*args, **kwargs) + if not g.get("transaction_active", False): + db.session.commit() + return result + except Exception as e: + db.session.rollback() + logger.error(f"装饰器事务执行失败: {str(e)}", exc_info=True) + raise + + return wrapper + + +def transactional_response(func): + """ + 事务装饰器（带返回 R 对象） + + 用于需要直接返回 R 对象的接口。 + + Example: + @transactional_response + def add_position(): + position.save() + return R.ok(msg="添加成功") + """ + + @wraps(func) + def wrapper(*args, **kwargs): + from src.core import response as R + + try: + result = func(*args, **kwargs) + if not g.get("transaction_active", False): + db.session.commit() + return result + except Exception as e: + db.session.rollback() + logger.error(f"装饰器事务执行失败: {str(e)}", exc_info=True) + return R.failed(msg=f"操作失败: {str(e)}") + + return wrapper + + +# ============================================================ +# 备用：事务上下文管理器（特殊场景使用） +# ============================================================ + + +class TransactionContext: + """ + 事务上下文管理器 + + 用于需要精细控制事务的特殊场景。 + + Example: + with TransactionContext(): + position.save() + user.save() + # 自动提交或回滚 + """ + + def __enter__(self): + g.disable_auto_commit = True + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + g.disable_auto_commit = False + if exc_type is None: + if not g.get("transaction_active", False): + db.session.commit() + else: + db.session.rollback() + logger.error(f"事务上下文回滚: {str(exc_val)}", exc_info=True) + return False + + +# ============================================================ +# 操作日志自动记录（DI：写入器由应用工厂注入） +# ============================================================ + +""" +操作日志自动记录中间件 +============================================================= + +通过 after_request 钩子自动记录写操作（POST/PUT/DELETE/PATCH）日志。 +无需在每个视图函数中手动调用 record_operation_log。 + +记录规则： + - 只记录写操作（POST/PUT/DELETE/PATCH），GET/OPTIONS/HEAD 不记录 + - 只记录成功操作（HTTP 2xx 且业务 code=0） + - 内部调用（定时任务等）不记录 + - 登录/退出/验证码等路径不记录（由 login_log 单独记录） + - 日志记录失败不影响主流程 +""" + + +# 不记录操作日志的路径前缀（登录/退出/验证码等由 login_log 单独记录） +_EXCLUDED_PREFIXES = ("/login", "/captcha", "/logout") + +# HTTP 方法 → 操作日志类型 映射 +# 类型值参考 application/constants/dict.py 的 OPERATION_LOG_TYPE_LIST +_METHOD_LOG_TYPE = { + "POST": 3, # 新增 + "PUT": 4, # 修改 + "DELETE": 5, # 删除 + "PATCH": 4, # 修改 +} + +# 路径后缀 → 操作日志类型 覆盖映射（更精确） +_PATH_ACTION_TYPE = { + "/add": 3, # 新增 + "/update": 4, # 修改 + "/delete": 5, # 删除 + "/status": 4, # 状态变更→修改 + "/batchdelete": 5, # 批量删除 + "/reset": 4, # 重置→修改 + "/save": 4, # 保存→修改 + "/sort": 4, # 排序→修改 +} + + +def register_operation_log(app): + """ + 注册操作日志自动记录中间件 + + 在 after_request 中自动记录写操作日志。 + + Args: + app: Flask 应用实例 + """ + + @app.after_request + def record_op_log(response): + # 只记录写操作 + if request.method not in ("POST", "PUT", "DELETE", "PATCH"): + return response + + # 排除登录/退出/验证码等路径 + path = request.path + if any(path.startswith(prefix) for prefix in _EXCLUDED_PREFIXES): + return response + + # 只记录成功操作（HTTP 2xx 且业务 code=0） + if not (200 <= response.status_code < 300): + return response + + try: + data = response.get_json(silent=True) + if data and data.get("code", 0) != 0: + return response + except Exception: + pass + + # 内部调用不记录 + if getattr(g, "_internal_call", False): + return response + + # 异步记录，不阻塞响应 + try: + _write_log(response) + except Exception as e: + logger.warning(f"记录操作日志失败: {str(e)}") + + return response + + logger.info("操作日志中间件注册成功") + + +def _write_log(response): + """写入操作日志""" + from src.core.security import get_username + from src.common.utils.ip import get_client_ip + from src.common.utils.ip2region import get_ip_location + from src.common.utils.user_agent import parse_user_agent + + writer = current_app.extensions.get("operation_log_writer") + if writer is None: + logger.warning("operation_log_writer 未注入，跳过操作日志") + return + username = get_username() or "system" + client_ip = get_client_ip() + ua_info = parse_user_agent() + + # 确定操作类型 + log_type = _infer_log_type(request.method, request.path) + + # 构建标题：从路径提取模块名，如 /user/add → "用户管理-新增" + title = _build_title(request.path) + + # 请求参数（限制长度，避免存入过大的数据） + param = _get_request_param() + + # 返回结果（限制长度） + result = _get_response_result(response) + + # 请求耗时 + start_time = getattr(g, "request_start_time", None) + consume_time = int((time.time() - start_time) * 1000) if start_time else 0 + + writer( + title=title, + log_type=log_type, + source=0, # OperationLog.Source.ADMIN + status=0, # OperationLog.Status.NORMAL + method=request.method, + request_method=request.endpoint or "", + url=request.path, + param=param, + result=result, + ip=client_ip, + location=get_ip_location(client_ip), + os=ua_info.get("os", "未知"), + browser=ua_info.get("browser", "未知"), + consume_time=consume_time, + error="", + username=username, + ) + + +def _infer_log_type(method, path): + """根据请求方法和路径推断操作日志类型""" + # 优先根据路径后缀匹配 + path_lower = path.lower() + for suffix, log_type in _PATH_ACTION_TYPE.items(): + if path_lower.endswith(suffix): + return log_type + + # 回退到 HTTP 方法映射 + return _METHOD_LOG_TYPE.get(method, 18) # 18=其他 + + +def _get_request_param(): + """获取请求参数""" + try: + if request.method in ("POST", "PUT", "PATCH", "DELETE"): + # 优先解析 JSON body（batchDelete 等 DELETE 请求也带 body） + data = request.get_json(silent=True) + if data: + import json + + return json.dumps(data, ensure_ascii=False) + # 回退到 URL 查询参数 + if request.args: + import json + from collections import OrderedDict + + return json.dumps(OrderedDict(request.args), ensure_ascii=False) + except Exception: + pass + return "" + + +def _get_response_result(response): + """获取响应结果中的data字段""" + try: + data = response.get_json(silent=True) + if data and isinstance(data, dict): + result_data = data.get("data") + if result_data is not None: + import json + + return json.dumps(result_data, ensure_ascii=False) + # data字段为空时取msg + return data.get("msg", "") + except Exception: + pass + return "" + + +def _build_title(path): + """从请求路径构建操作标题""" + # 路径格式: /user/add, /role/update, /menu/delete 等 + parts = path.strip("/").split("/") + if len(parts) >= 2: + module = parts[0] + # 模块名中文化映射 + module_names = { + "user": "用户管理", + "role": "角色管理", + "menu": "菜单管理", + "dept": "部门管理", + "position": "岗位管理", + "level": "职级管理", + "param": "参数配置", + "dict": "字典管理", + "dictitem": "字典项管理", + "config": "系统配置", + "configitem": "配置项管理", + "category": "分类管理", + "article": "文章管理", + "notice": "通知管理", + "link": "友情链接", + "job": "定时任务", + "login": "登录管理", + "loginlog": "登录日志", + "operationlog": "操作日志", + "upload": "文件上传", + } + module_name = module_names.get(module.lower(), module) + action = parts[-1] if len(parts) > 1 else "" + action_names = { + "add": "新增", + "update": "修改", + "delete": "删除", + "status": "状态变更", + "batchdelete": "批量删除", + "reset": "重置", + "save": "保存", + "sort": "排序", + "page": "查询", + "detail": "查询", + "list": "查询", + } + action_name = action_names.get(action.lower(), action) + return f"{module_name}-{action_name}" if action_name else module_name + return path + + +# ============================================================ +# 全局异常处理 +# ============================================================ + +""" +全局异常处理模块 +============================================================= + +注册 Flask errorhandler，将所有 HTTP 错误和未捕获异常 +统一转为 R.failed() JSON 响应，避免前端收到 HTML 错误页导致解析崩溃。 + +处理范围： + - 400：请求参数错误 + - 401：未授权 + - 403：权限不足 + - 404：资源不存在 + - 405：请求方法不允许 + - 500：服务器内部错误 + - Exception：兜底，捕获所有未预期异常 +""" + + +def register_error_handlers(app): + """ + 注册全局异常处理器 + + 确保所有 HTTP 错误和未捕获异常都返回标准 JSON 响应。 + + Args: + app: Flask 应用实例 + """ + + @app.errorhandler(400) + def bad_request(e): + return R.failed(msg="请求参数错误", code=400), 400 + + @app.errorhandler(401) + def unauthorized(e): + return R.failed(msg="未授权，请重新登录", code=401), 401 + + @app.errorhandler(403) + def forbidden(e): + return R.failed(msg="权限不足", code=403), 403 + + @app.errorhandler(404) + def not_found(e): + return R.failed(msg="请求的资源不存在", code=404), 404 + + @app.errorhandler(405) + def method_not_allowed(e): + return R.failed(msg="请求方法不允许", code=405), 405 + + @app.errorhandler(500) + def internal_error(e): + logger.error( + f"服务器内部错误, method: {request.method}, path: {request.path}, " f"error: {str(e)}", exc_info=True + ) + msg = f"服务器内部错误: {str(e)}" if app.debug else "服务器内部错误" + return R.failed(msg=msg, code=500), 500 + + @app.errorhandler(BizError) + def handle_biz_error(e): + """ + 业务异常处理器 + + 业务代码 raise BizError(...) 时统一转换为 R.failed 响应， + 并回滚当前事务（teardown_request 处理）。 + """ + logger.info(f"业务异常: {e.msg} (code={e.code})") + return R.failed(msg=e.msg, code=e.code, data=e.data, **e.kwargs) + + @app.errorhandler(Exception) + def handle_exception(e): + # HTTPException 子类（如 abort(403)）已有专用 handler，不重复处理 + if isinstance(e, HTTPException): + return e.get_response() + + # 非HTTP异常：未知运行时异常 + logger.error(f"未捕获异常, method: {request.method}, path: {request.path}, " f"error: {traceback.format_exc()}") + msg = f"服务器内部错误: {str(e)}" if app.debug else "服务器内部错误" + return R.failed(msg=msg, code=500), 500 + + logger.info("全局异常处理器注册成功") + + +# ============================================================ +# 中间件注册 +# ============================================================ + + +def register_middleware(app): + """ + 注册应用中间件（登录校验 before_request）。 + + Args: + app: Flask 应用实例 + """ + + @app.before_request + def checklogin(): + """登录验证中间件""" + return check_login() diff --git a/src/core/permission.py b/src/core/permission.py new file mode 100644 index 0000000..79cbeeb --- /dev/null +++ b/src/core/permission.py @@ -0,0 +1,165 @@ +""" +权限校验模块 +============================================================= + +权限校验采用依赖注入设计（解决 core → modules 依赖倒置）： + - core 定义 PermissionProvider 协议，不 import 任何业务模块 + - 应用工厂启动时注入 menu 服务的实现（见 application/factory.py） + - check_permission 装饰器通过 current_app.extensions['permission_provider'] + 获取权限数据提供方 + +使用示例： + @check_permission("sys:user:page") + def page(): + ... +""" + +import fnmatch +import hmac +import logging +from functools import wraps +from typing import List, Protocol + +from flask import request, g, current_app + +from src.core import response as R +from src.core.security import get_user_id +from config import PERMISSION_ENABLED, INTERNAL_API_KEY, INTERNAL_ALLOWED_IPS + +logger = logging.getLogger(__name__) + + +class PermissionProvider(Protocol): + """权限数据提供方接口：core 只认接口，不认实现。""" + + def get_user_permissions(self, user_id: int) -> List[str]: + """返回用户拥有的权限节点列表。""" + + +def get_provider() -> "PermissionProvider": + """获取注入的权限提供方（应用工厂注入）。""" + return current_app.extensions.get("permission_provider") + + +def is_internal_call() -> bool: + """ + 判断是否为内部系统调用（仅定时任务等无用户场景） + + 验证方式（满足任一即可）： + 1. 权限验证总开关关闭（开发环境） + 2. API密钥验证通过（请求头携带正确密钥） + 3. IP白名单验证通过（来源IP在白名单中） + + Returns: + bool: True-内部调用（跳过权限验证），False-普通调用（需要权限验证） + """ + # ============================================================ + # 方式1：权限验证总开关关闭（开发/测试环境） + # ============================================================ + if not PERMISSION_ENABLED: + logger.debug("权限验证已关闭，所有请求视为内部调用") + return True + + # ============================================================ + # 方式2：API密钥验证（定时任务使用） + # ============================================================ + if INTERNAL_API_KEY: + req_key = request.headers.get("X-Scheduler-Key", "") + if not req_key: + req_key = request.headers.get("X-Internal-API-Key", "") + + if hmac.compare_digest(req_key, INTERNAL_API_KEY): + logger.debug(f"内部API密钥验证通过: IP={request.remote_addr}") + return True + + # ============================================================ + # 方式3：IP白名单验证（可选） + # ============================================================ + if INTERNAL_ALLOWED_IPS: + client_ip = request.remote_addr + allowed_ips = [ip.strip() for ip in INTERNAL_ALLOWED_IPS.split(",") if ip.strip()] + + for allowed_ip in allowed_ips: + if fnmatch.fnmatch(client_ip, allowed_ip): + logger.debug(f"内部IP白名单验证通过: IP={client_ip}") + return True + + return False + + +def check_permission(permission: str, log: bool = False): + """ + 权限验证装饰器 + + 权限验证规则： + 1. 定时任务等内部调用 → 跳过权限验证，直接放行 + 2. 超级管理员（user_id=1） → 拥有所有权限 + 3. 普通用户 → 通过注入的 PermissionProvider 验证权限 + + Args: + permission: 权限节点标识（如：sys:job:page） + log: 是否记录访问日志 + + Returns: + decorator: 装饰器函数 + """ + + def decorator(func): + @wraps(func) + def wrapper(*args, **kwargs): + # ============================================================ + # 定时任务等内部调用：跳过权限验证，直接放行 + # ============================================================ + if is_internal_call(): + logger.debug(f"定时任务内部调用，跳过权限验证: {func.__name__}") + return func(*args, **kwargs) + + # ============================================================ + # 普通请求：正常进行权限验证 + # ============================================================ + + # 获取用户ID（优先从g对象获取） + user_id = getattr(g, "user_id", None) + if user_id is None: + try: + user_id = get_user_id() + except Exception as e: + logger.error(f"获取用户ID失败: {str(e)}") + user_id = None + + # 超级管理员放行 + if user_id == 1: + logger.debug(f"超级管理员权限验证通过: user_id={user_id}") + return func(*args, **kwargs) + + # 普通用户权限验证 + if user_id: + provider = get_provider() + if provider is None: + logger.error("权限提供方未注入，请联系管理员检查配置") + return R.failed(msg="权限验证失败", code=500) + + try: + permission_list = provider.get_user_permissions(user_id) + + if permission not in permission_list: + if log: + logger.warning( + f"权限不足: user_id={user_id}, " f"permission={permission}, " f"path={request.path}" + ) + return R.failed(msg="权限不足!", code=403) + elif log: + logger.info(f"权限验证通过: user_id={user_id}, " f"permission={permission}") + except Exception as e: + logger.error(f"权限验证异常: {str(e)}") + return R.failed(msg="权限验证失败", code=500) + else: + # 未登录用户 + logger.warning(f"未登录用户访问: path={request.path}") + return R.failed(msg="请先登录", code=401) + + return func(*args, **kwargs) + + return wrapper + + return decorator diff --git a/src/core/scheduler.py b/src/core/scheduler.py index e2d0bd5..5937487 100644 --- a/src/core/scheduler.py +++ b/src/core/scheduler.py @@ -87,7 +87,7 @@ class Scheduler: # ============================================================ with app.app_context(): try: - from application.services.scheduler import reload_scheduler + from src.modules.job.scheduler_service import reload_scheduler # 重新加载所有任务 count = reload_scheduler() @@ -109,7 +109,7 @@ class Scheduler: def _log_scheduler_status(self): """打印调度器状态（用于调试）""" try: - from application.services.scheduler import scheduler + from src.modules.job.scheduler_service import scheduler if scheduler.running: jobs = scheduler.get_jobs() @@ -131,7 +131,7 @@ class Scheduler: 在应用退出时调用 """ try: - from application.services.scheduler import scheduler + from src.modules.job.scheduler_service import scheduler if scheduler.running: scheduler.shutdown() @@ -147,7 +147,7 @@ class Scheduler: int: 成功加载的任务数量 """ try: - from application.services.scheduler import reload_scheduler + from src.modules.job.scheduler_service import reload_scheduler count = reload_scheduler() logger.info(f"重新加载定时任务完成，共加载 {count} 个任务") return count @@ -163,7 +163,7 @@ class Scheduler: dict: 包含调度器运行状态和任务列表 """ try: - from application.services.scheduler import scheduler + from src.modules.job.scheduler_service import scheduler jobs = scheduler.get_jobs() job_list = [] diff --git a/src/modules/association/role_menu/router.py b/src/modules/association/role_menu/router.py index 4459bdf..528c530 100644 --- a/src/modules/association/role_menu/router.py +++ b/src/modules/association/role_menu/router.py @@ -54,7 +54,7 @@ from flask import Blueprint -from application.middleware import check_permission +from application.middleware.check_permission import check_permission from application.middleware.check_demo import check_demo from src.modules.system.role import service as role_menu