Python遍历日志文件提取指定行生成列表 实现SQL安全写入防注入

日志解析逻辑 #

日志中作业记录以Job ID = 为起始标记，后续连续跟7个固定字段，遇到下一个作业起始标记、分隔线或日志结束时，单条记录组装完成。头尾的系统运行信息、分隔线直接跳过即可。
解析代码如下：

from typing import List, Dict

def parse_job_log(log_path: str) -> List[Dict]:
    # 按日志中出现顺序定义需要提取的字段
    TARGET_FIELDS = [
        "Job ID", "Owner", "Job Status", "Description",
        "Creation Time", "Project ID", "Project", "Job Duration"
    ]
    records = []
    current_record = None

    with open(log_path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.rstrip("\n").strip()
            # 跳过空行、分隔线
            if not line or line.startswith("=") or line.startswith("#"):
                continue
            # 匹配到新作业的起始标记
            if line.startswith("Job ID = "):
                # 先把已组装完成的上一条记录存入结果
                if current_record and len(current_record) == len(TARGET_FIELDS):
                    records.append(current_record)
                current_record = {}
                # 仅按第一个等号拆分，避免字段值中包含等号导致解析错误
                key, value = line.split(" = ", 1)
                current_record[key] = value
                continue
            # 组装当前作业的其他字段
            if current_record and " = " in line:
                key, value = line.split(" = ", 1)
                if key in TARGET_FIELDS:
                    current_record[key] = value
        # 处理文件末尾的最后一条记录
        if current_record and len(current_record) == len(TARGET_FIELDS):
            records.append(current_record)

    # 数值类型字段转换，方便后续入库计算
    for item in records:
        item["Job ID"] = int(item["Job ID"])
        item["Job Duration"] = int(item["Job Duration"])
    return records

运行上述函数传入日志文件路径，即可得到结构化的作业记录列表，输出格式和需求中给出的样例完全一致。

SQL注入防范方案 #

绝对禁止使用字符串拼接的方式组装SQL插入语句，所有正规数据库驱动都提供参数化查询能力，由驱动自动完成特殊字符转义，从根源阻断SQL注入风险。
以SQLite为例的安全写入代码如下，MySQL、PostgreSQL等其他数据库逻辑完全一致，仅占位符根据驱动要求调整即可：

import sqlite3

def write_records_to_db(records: List[Dict], db_path: str):
    # 建表语句
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS job_info (
        job_id INTEGER PRIMARY KEY,
        owner TEXT,
        job_status TEXT,
        description TEXT,
        creation_time TEXT,
        project_id TEXT,
        project TEXT,
        job_duration INTEGER
    )
    """
    # 插入语句使用占位符，不直接拼接值
    insert_sql = """
    INSERT OR REPLACE INTO job_info
    (job_id, owner, job_status, description, creation_time, project_id, project, job_duration)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    """
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute(create_table_sql)
    for record in records:
        # 参数按顺序传入元组，驱动自动处理转义
        params = (
            record["Job ID"],
            record["Owner"],
            record["Job Status"],
            record["Description"],
            record["Creation Time"],
            record["Project ID"],
            record["Project"],
            record["Job Duration"]
        )
        cursor.execute(insert_sql, params)
    conn.commit()
    conn.close()

使用示例 #

if __name__ == "__main__":
    # 解析日志
    job_list = parse_job_log("job_log.txt")
    # 打印验证解析结果
    for job in job_list:
        print(tuple(f"{k} = {v}" for k, v in job.items()))
    # 安全写入数据库
    write_records_to_db(job_list, "job_data.db")

注意事项 #

• 字段拆分使用split(" = ", 1)限制仅拆分一次，兼容字段值中包含等号的场景
• 参数化查询会将传入的所有值当做纯字符串/数值处理，即使日志内容中包含注入恶意语句，也不会被数据库执行
• 代码自动跳过日志中非作业记录的无关行，无需提前对原始日志做裁剪处理

内容的提问来源于stack exchange，提问作者Sean Conor Patton