chore: 清理 P5 临时 codemod
This commit is contained in:
DjangoAdmin Dev
-132
@@ -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()
|
|
||||||
Reference in New Issue
Block a user