DjangoAdmin_Flask_EleVue/_codemod_p5a.py

# -*- 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()