邮件日志邮箱地址哈希替换脚本实现与技术求助

先上修改后的完整脚本，直接能用：

import hashlib
import re
import time
import os
from stat import ST_INODE

# 自己改这里的路径，适配你的环境
SOURCE_LOG = '/var/log/mail.log'  # 原始邮件日志的路径
DESENSITIZED_LOG = '/var/log/mail_desensitized.log'  # 脱敏后存到这个文件
STATE_FILE = '/var/log/mail_desensitized.state'  # 记录上次读到哪的小文件
CHECK_INTERVAL = 60  # 每60秒检查一次新日志

def extract_root_domain(email):
    parts = email.split('@')
    if len(parts) != 2:
        return email  # 不是正经邮箱，直接返回
    domain = parts[1]
    domain_parts = domain.split('.')
    # 处理co.uk这种多级顶级域名，直接取最后两段，不够的话就全返回
    if len(domain_parts) >= 2:
        return '.'.join(domain_parts[-2:])
    return domain

def desensitize_email(email):
    if '@' not in email:
        return email
    username, domain = email.split('@', 1)
    # 用户名取前两位，太短的话就全用
    prefix = username[:2] if len(username) >=2 else username
    # 哈希取前5位，保证唯一性又不会太长
    hashed_part = hashlib.sha256(email.encode()).hexdigest()[:5]
    root_domain = extract_root_domain(email)
    return f"{prefix}_{hashed_part}@{root_domain}"

def process_new_lines(file, start_pos, inode):
    """只处理上次没读过的新日志行"""
    file.seek(start_pos)
    new_lines = file.readlines()
    if not new_lines:
        return start_pos, inode
    
    # 把处理后的行追加到脱敏日志里
    with open(DESENSITIZED_LOG, 'a') as des_file:
        for line in new_lines:
            # 两种匹配：被"<包裹的邮箱，以及直接裸写的邮箱，适配多数日志格式
            emails = re.findall(r'(?:["<])(\S+@\S+)(?:[">])|(\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b)', line)
            # 把匹配到的邮箱整理出来，去掉空值
            matched_emails = [email for group in emails for email in group if email]
            for email in matched_emails:
                des_email = desensitize_email(email)
                line = line.replace(email, des_email)
            des_file.write(line)
    
    # 更新读到的位置
    new_pos = file.tell()
    # 检查文件inode变了没——比如logrotate把旧日志切走了，新日志是个新文件
    new_inode = os.fstat(file.fileno()).st_ino
    if new_inode != inode:
        # 文件换了，重新从开头读
        file.seek(0)
        new_pos = 0
        new_inode = os.fstat(file.fileno()).st_ino
    return new_pos, new_inode

def load_state():
    """读取上次记录的位置和inode"""
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, 'r') as f:
            try:
                pos, inode = map(int, f.readline().split())
                return pos, inode
            except:
                pass
    return 0, 0

def save_state(pos, inode):
    """把当前读到的位置和inode存起来，下次接着来"""
    with open(STATE_FILE, 'w') as f:
        f.write(f"{pos} {inode}")

def monitor_logs():
    start_pos, current_inode = load_state()
    while True:
        try:
            with open(SOURCE_LOG, 'r') as source_file:
                file_inode = os.fstat(source_file.fileno()).st_ino
                if file_inode != current_inode:
                    # 日志被轮转了，重置位置
                    start_pos = 0
                    current_inode = file_inode
                
                start_pos, current_inode = process_new_lines(source_file, start_pos, current_inode)
                save_state(start_pos, current_inode)
        except FileNotFoundError:
            print(f"警告：没找到日志文件 {SOURCE_LOG}，{CHECK_INTERVAL}秒后再试")
        except Exception as e:
            print(f"处理日志出错了：{str(e)}")
        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    # 先创建脱敏日志文件（不存在的话）
    if not os.path.exists(DESENSITIZED_LOG):
        open(DESENSITIZED_LOG, 'w').close()
    monitor_logs()

核心调整说明 #

• 绝不碰原日志文件：mail.log一般是系统服务在写，直接改会搞乱日志，甚至导致服务出问题。这个脚本生成独立的脱敏日志，原文件一动不动，安全。
• 只处理新内容：用字节位置+inode跟踪进度，比原脚本的行数靠谱多了——日志轮转、截断都不怕，不会重复处理老内容，也不会漏新内容。
• 适配更多日志格式：原脚本只能匹配被</"包着的邮箱，现在加了裸邮箱的匹配，大部分日志场景都能覆盖。
• 稳得一批的错误处理：日志文件找不到、读错了都会提示，不会直接崩掉，适合长期后台跑。

用起来的步骤 #

1. 把脚本里的路径改成你实际的日志路径，比如SOURCE_LOG如果是/root/mail.log就改过去
2. 给脚本加执行权限：chmod +x desensitize_mail_log.py
3. 后台跑：nohup python3 desensitize_mail_log.py &，想开机自启就整个systemd服务，搜一下教程就能搞定

原脚本的坑我都填上了 #

• 原脚本直接读写同一个文件，会导致日志重复写，文件越变越大
• 按行数跟踪进度，日志一转轮就乱套，要么漏内容要么重复处理
• 没处理无效邮箱，遇到格式不对的会报错崩掉
• 没错误处理，有点意外就停了

内容的提问来源于stack exchange，提问作者Tirth Joshi