-- 啟用通用查詢?nèi)罩荆ㄓ绊懶阅?，建議臨時(shí)使用）
SET GLOBAL general_log = 'ON';
SET GLOBAL general_log_file = '/var/log/mysql/mysql-general.log';

-- 啟用二進(jìn)制日志
SET GLOBAL log_bin = ON;

mysqlbinlog /var/lib/mysql/mysql-bin.000001 | grep "UPDATE statistics_test"

INSTALL PLUGIN server_audit SONAME 'server_audit.so';
SET GLOBAL server_audit_logging = ON;
SET GLOBAL server_audit_events = 'QUERY';
SET GLOBAL server_audit_file_path = '/var/log/mysql/audit.log';

log_statement = 'mod'  # 記錄所有修改數(shù)據(jù)的SQL
log_hostname = on      # 記錄客戶端主機(jī)名
log_line_prefix = '%t %h %u %d '  # 時(shí)間、IP、用戶、數(shù)據(jù)庫

sudo systemctl restart postgresql

-- 創(chuàng)建審計(jì)表
CREATE TABLE statistics_test_audit (
    id INT AUTO_INCREMENT PRIMARY KEY,
    change_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    db_user VARCHAR(100),
    client_ip VARCHAR(50),
    action VARCHAR(10),  -- 'UPDATE'
    old_data JSON,
    new_data JSON
);

-- 創(chuàng)建觸發(fā)器
DELIMITER //
CREATE TRIGGER tr_statistics_test_update
AFTER UPDATE ON statistics_test
FOR EACH ROW
BEGIN
    INSERT INTO statistics_test_audit (db_user, client_ip, action, old_data, new_data)
    VALUES (
        CURRENT_USER(),
        SUBSTRING_INDEX(USER(), '@', -1),  -- 提取客戶端IP
        'UPDATE',
        JSON_OBJECT('id', OLD.id, 'value', OLD.value),
        JSON_OBJECT('id', NEW.id, 'value', NEW.value)
    );
END//
DELIMITER ;

-- 創(chuàng)建審計(jì)表
CREATE TABLE statistics_test_audit (
    id SERIAL PRIMARY KEY,
    change_time TIMESTAMP DEFAULT NOW(),
    db_user TEXT,
    client_ip TEXT,
    action TEXT,
    old_data JSONB,
    new_data JSONB
);

-- 創(chuàng)建觸發(fā)器函數(shù)
CREATE OR REPLACE FUNCTION log_statistics_test_update()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO statistics_test_audit (db_user, client_ip, action, old_data, new_data)
    VALUES (
        current_user,
        inet_client_addr()::TEXT,
        'UPDATE',
        jsonb_build_object('id', OLD.id, 'value', OLD.value),
        jsonb_build_object('id', NEW.id, 'value', NEW.value)
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- 綁定觸發(fā)器
CREATE TRIGGER tr_statistics_test_update
AFTER UPDATE ON statistics_test
FOR EACH ROW EXECUTE FUNCTION log_statistics_test_update();

-- 安裝審計(jì)插件
INSTALL PLUGIN audit_log SONAME 'audit_log.so';

-- 配置審計(jì)規(guī)則
SET GLOBAL audit_log_policy = 'ALL';

from flask import Flask, request
import logging
from datetime import datetime

app = Flask(__name__)

# 配置日志
logging.basicConfig(
    filename='db_updates.log',
    level=logging.INFO,
    format='%(asctime)s - %(client_ip)s - %(message)s'
)

def log_db_update(user, table, action, data):
    client_ip = request.remote_addr
    logging.info(
        f"User={user}, Table={table}, Action={action}, Data={data}",
        extra={'client_ip': client_ip}
    )

@app.route('/update_stats', methods=['POST'])
def update_stats():
    data = request.json
    # 執(zhí)行數(shù)據(jù)庫更新
    log_db_update("api_user", "statistics_test", "UPDATE", data)
    return {"status": "success"}

mysql.query contains "UPDATE statistics_test"