ClaudeandClaude
dbcce79536
重构前保存现有 apps/ 结构基线，供后续各阶段回滚。 Co-Authored-By: Claude <noreply@anthropic.com>
122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
import logging
|
|
|
|
from sqlalchemy import and_
|
|
|
|
from apps.constants.constants import JOB_LOG_STATUS_LIST
|
|
from apps.constants.message import PAGE_SIZE
|
|
from apps.models.job import Job
|
|
from apps.models.job_log import JobLog
|
|
from extensions import get_db_session
|
|
from extensions.sqlalchemy import SessionLocal
|
|
from utils import R
|
|
from utils.request import like_contains, parse_id_list
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ============================================================
|
|
# 日志写入（供调度器调用，可在请求上下文外执行）
|
|
# ============================================================
|
|
|
|
def record_job_log(job_id, start_time, end_time, status, exception_info=None, job_message=None):
|
|
"""
|
|
记录任务执行日志
|
|
|
|
使用独立会话写入，调度器线程等无请求上下文场景也可调用。
|
|
"""
|
|
session = SessionLocal()
|
|
try:
|
|
job = session.query(Job).filter(Job.id == job_id, Job.is_delete == 0).first()
|
|
consume_time = 0
|
|
if end_time and start_time:
|
|
consume_time = int((end_time - start_time).total_seconds() * 1000)
|
|
log = JobLog(
|
|
job_id=job_id,
|
|
job_name=job.job_name if job else '',
|
|
job_group=job.job_group if job else None,
|
|
job_trigger=job.job_trigger if job else None,
|
|
cron_expression=job.cron_expression if job else None,
|
|
job_message=job_message,
|
|
status=status,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
consume_time=consume_time,
|
|
exception_info=exception_info,
|
|
)
|
|
session.add(log)
|
|
session.commit()
|
|
return log
|
|
except Exception as e:
|
|
session.rollback()
|
|
logger.error(f"记录任务执行日志异常: {str(e)}")
|
|
return None
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
# ============================================================
|
|
# 日志查询/删除
|
|
# ============================================================
|
|
|
|
def get_job_log_page(request):
|
|
db = get_db_session()
|
|
# 页码
|
|
page = int(request.query_params.get("pageNo", 1))
|
|
# 每页数
|
|
limit = int(request.query_params.get("pageSize", PAGE_SIZE))
|
|
# 实例化查询对象
|
|
query = db.query(JobLog).filter(JobLog.del_flag == 0)
|
|
# 任务名称
|
|
job_name = request.query_params.get('jobName')
|
|
if job_name:
|
|
query = query.filter(JobLog.job_name.like(like_contains(job_name), escape='\\'))
|
|
# 任务分组
|
|
job_group = request.query_params.get('jobGroup')
|
|
if job_group:
|
|
query = query.filter(JobLog.job_group == job_group)
|
|
# 执行状态
|
|
status = request.query_params.get('status')
|
|
if status:
|
|
query = query.filter(JobLog.status == status)
|
|
# 排序
|
|
query = query.order_by(JobLog.id.desc())
|
|
# 记录总数
|
|
count = query.count()
|
|
# 分页查询
|
|
log_list = query.limit(limit).offset((page - 1) * limit).all()
|
|
# 若当前页超出末页（数据被并发删除），自动回退到第一页
|
|
if page > 1 and not log_list:
|
|
page = 1
|
|
count = query.count()
|
|
log_list = query.limit(limit).offset(0).all()
|
|
# 实例化结果
|
|
result = []
|
|
for item in log_list:
|
|
data = item.to_dict()
|
|
data['status_name'] = JOB_LOG_STATUS_LIST.get(item.status)
|
|
result.append(data)
|
|
return R.ok(data=result, count=count)
|
|
|
|
|
|
# 根据ID查询日志详情
|
|
def get_job_log_detail(job_log_id):
|
|
db = get_db_session()
|
|
log = db.query(JobLog).filter(and_(JobLog.id == job_log_id, JobLog.del_flag == 0)).first()
|
|
if not log:
|
|
return None
|
|
data = log.to_dict()
|
|
data['status_name'] = JOB_LOG_STATUS_LIST.get(log.status)
|
|
return data
|
|
|
|
|
|
# 删除任务日志（软删除）
|
|
def delete_job_log(ids_str):
|
|
id_list = parse_id_list(ids_str)
|
|
if not id_list:
|
|
return R.failed("记录ID不存在")
|
|
db = get_db_session()
|
|
result = db.query(JobLog).filter(JobLog.id.in_(id_list), JobLog.del_flag == 0).update(
|
|
{"del_flag": 1}, synchronize_session=False
|
|
)
|
|
return R.ok(msg="本次共删除{0}条数据".format(result))
|