From 410096547f3364ca4cc211f31bfc8d0fac5a682e Mon Sep 17 00:00:00 2001 From: DjangoAdmin Dev Date: Fri, 7 Aug 2026 19:39:28 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E6=B8=85=E7=90=86=20P5=20=E4=B8=B4?= =?UTF-8?q?=E6=97=B6=20codemod?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- _codemod_p5a.py | 132 ------------------------------------------------ 1 file changed, 132 deletions(-) delete mode 100644 _codemod_p5a.py diff --git a/_codemod_p5a.py b/_codemod_p5a.py deleted file mode 100644 index b046156..0000000 --- a/_codemod_p5a.py +++ /dev/null @@ -1,132 +0,0 @@ -# -*- 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()