Python脚本如何记录数据不一致日志

本文目录导读：

1. 使用标准库 logging（推荐）
2. 带统计功能的日志记录
3. 数据库比较日志
4. 带统计报表的日志系统
5. 最佳实践建议

我来介绍几种在Python中记录数据不一致日志的方法：

使用标准库 logging（推荐）

import logging
from datetime import datetime
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('data_inconsistency.log', encoding='utf-8'),
        logging.StreamHandler()  # 同时输出到控制台
    ]
)
logger = logging.getLogger(__name__)
class DataValidator:
    def check_data_consistency(self, data1, data2):
        """检查数据一致性"""
        inconsistencies = []
        # 检查字段不匹配
        if data1.get('id') != data2.get('id'):
            inconsistencies.append({
                'type': 'ID_MISMATCH',
                'field': 'id',
                'expected': data1.get('id'),
                'actual': data2.get('id')
            })
        # 检查数值不一致
        if data1.get('total_amount') != data2.get('total_amount'):
            inconsistencies.append({
                'type': 'AMOUNT_MISMATCH',
                'field': 'total_amount',
                'expected': data1.get('total_amount'),
                'actual': data2.get('total_amount'),
                'difference': data1.get('total_amount', 0) - data2.get('total_amount', 0)
            })
        # 记录不一致日志
        if inconsistencies:
            logger.warning(f"发现 {len(inconsistencies)} 条数据不一致")
            for issue in inconsistencies:
                logger.error(
                    f"数据不一致 - 类型: {issue['type']}, "
                    f"字段: {issue['field']}, "
                    f"期望值: {issue['expected']}, "
                    f"实际值: {issue['actual']}"
                )
        return inconsistencies
    def log_record_mismatch(self, record_type, source, record_id, details):
        """记录单条记录不一致"""
        logger.warning(
            f"记录不一致 - 类型: {record_type}, "
            f"源: {source}, "
            f"ID: {record_id}, "
            f"详情: {details}"
        )
# 使用示例
validator = DataValidator()
data_source_a = {'id': 1, 'total_amount': 100.50, 'status': 'active'}
data_source_b = {'id': 1, 'total_amount': 99.50, 'status': 'pending'}
inconsistencies = validator.check_data_consistency(data_source_a, data_source_b)

带统计功能的日志记录

import logging
from collections import Counter
from datetime import datetime
import json
class InconsistencyLogger:
    def __init__(self, log_file='data_inconsistency.log'):
        self.logger = self._setup_logger(log_file)
        self.stats = Counter()
        self.inconsistencies_history = []
    def _setup_logger(self, log_file):
        logger = logging.getLogger('DataInconsistency')
        logger.setLevel(logging.DEBUG)
        # 文件处理器 - 详细日志
        file_handler = logging.FileHandler(log_file, encoding='utf-8')
        file_handler.setLevel(logging.INFO)
        file_format = logging.Formatter('%(asctime)s | %(levelname)s | %(message)s')
        file_handler.setFormatter(file_format)
        # JSON 格式的专门日志
        json_handler = logging.FileHandler('inconsistency_export.json', encoding='utf-8')
        json_handler.setLevel(logging.WARNING)
        logger.addHandler(file_handler)
        logger.addHandler(json_handler)
        return logger
    def log_inconsistency(self, record_id, fields_issues, severity='WARNING'):
        """
        记录数据不一致
        Args:
            record_id: 记录标识
            fields_issues: 字段问题列表 [{'field': 'amount', 'expected': 100, 'actual': 99}, ...]
            severity: 严重级别 ERROR/WARNING/INFO
        """
        timestamp = datetime.now().isoformat()
        # 更新统计
        self.stats['total_inconsistencies'] += 1
        for issue in fields_issues:
            self.stats[f"field_{issue['field']}"] += 1
        # 创建日志记录
        log_entry = {
            'timestamp': timestamp,
            'record_id': record_id,
            'severity': severity,
            'issues': fields_issues,
            'total_issues': len(fields_issues)
        }
        # 根据严重级别记录
        if severity == 'ERROR':
            self.logger.error(json.dumps(log_entry, ensure_ascii=False))
        elif severity == 'WARNING':
            self.logger.warning(json.dumps(log_entry, ensure_ascii=False))
        else:
            self.logger.info(json.dumps(log_entry, ensure_ascii=False))
        # 保存到历史记录
        self.inconsistencies_history.append(log_entry)
        # 控制台输出（可选）
        print(f"[{timestamp}] {severity}: 记录 {record_id} 发现 {len(fields_issues)} 个问题")
    def log_comparison_result(self, source1_name, source2_name, mismatch_count, details):
        """记录源数据比较结果"""
        log_msg = (
            f"数据源比较: {source1_name} vs {source2_name} | "
            f"不匹配数: {mismatch_count} | "
            f"详情: {details}"
        )
        self.logger.info(log_msg)
        print(log_msg)
    def get_statistics(self):
        """获取不一致统计"""
        return dict(self.stats)
# 使用示例
inconsistency_logger = InconsistencyLogger()
# 模拟数据不一致
issues = [
    {'field': 'amount', 'expected': 100.50, 'actual': 99.50, 'difference': 1.00},
    {'field': 'status', 'expected': 'active', 'actual': 'inactive'}
]
inconsistency_logger.log_inconsistency('RECORD_001', issues, 'ERROR')
inconsistency_logger.log_comparison_result('DatabaseA', 'DatabaseB', 5, '总记录数不匹配')
# 获取统计
stats = inconsistency_logger.get_statistics()
print(f"不一致统计: {stats}")

数据库比较日志

import logging
import pandas as pd
from datetime import datetime
class DatabaseComparisonLogger:
    def __init__(self, source_db_name, target_db_name):
        self.logger = self._setup_logger()
        self.source_db = source_db_name
        self.target_db = target_db_name
    def _setup_logger(self):
        logger = logging.getLogger('DBComparison')
        logger.setLevel(logging.DEBUG)
        # 主日志文件
        handler = logging.FileHandler('db_comparison.log', encoding='utf-8')
        handler.setLevel(logging.INFO)
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        # 错误日志单独文件
        error_handler = logging.FileHandler('db_errors.log', encoding='utf-8')
        error_handler.setLevel(logging.ERROR)
        error_handler.setFormatter(formatter)
        logger.addHandler(error_handler)
        return logger
    def compare_record_count(self, source_count, target_count):
        """比较记录数"""
        if source_count != target_count:
            self.logger.error(
                f"记录数不一致 - {self.source_db}: {source_count}, "
                f"{self.target_db}: {target_count}"
            )
            return False
        else:
            self.logger.info(
                f"记录数一致 - 共 {source_count} 条"
            )
            return True
    def compare_dataframes(self, df_source, df_target, key_column='id'):
        """比较两个DataFrame"""
        mismatches = []
        # 找出差异
        merged = pd.merge(df_source, df_target, 
                         on=key_column, how='outer', 
                         suffixes=('_source', '_target'), 
                         indicator=True)
        # 记录缺失记录
        missing_in_target = merged[merged['_merge'] == 'left_only']
        missing_in_source = merged[merged['_merge'] == 'right_only']
        if len(missing_in_target) > 0:
            self.logger.warning(
                f"在 {self.target_db} 中缺失 {len(missing_in_target)} 条记录"
            )
            for _, row in missing_in_target.iterrows():
                self.logger.warning(
                    f"缺失记录 - ID: {row[key_column]}, "
                    f"来源: {self.source_db}"
                )
        if len(missing_in_source) > 0:
            self.logger.warning(
                f"在 {self.source_db} 中缺失 {len(missing_in_source)} 条记录"
            )
        # 比较字段值
        common_records = merged[merged['_merge'] == 'both']
        for col in df_source.columns:
            if col == key_column:
                continue
            source_col = f"{col}_source"
            target_col = f"{col}_target"
            if source_col in common_records.columns and target_col in common_records.columns:
                diff = common_records[common_records[source_col] != common_records[target_col]]
                if len(diff) > 0:
                    mismatches.append({
                        'field': col,
                        'count': len(diff),
                        'records': diff[[key_column, source_col, target_col]].to_dict('records')
                    })
                    self.logger.warning(
                        f"字段 {col} 有 {len(diff)} 条不一致"
                    )
        return mismatches
# 使用示例
comparison_logger = DatabaseComparisonLogger('DB_Main', 'DB_Backup')
# 模拟记录数比较
comparison_logger.compare_record_count(1000, 998)
# 模拟DataFrame比较
import pandas as pd
df1 = pd.DataFrame({'id': [1, 2, 3], 'value': [100, 200, 300]})
df2 = pd.DataFrame({'id': [1, 2, 4], 'value': [100, 250, 400]})
mismatches = comparison_logger.compare_dataframes(df1, df2)

带统计报表的日志系统

import logging
import json
from datetime import datetime, timedelta
from collections import defaultdict
class AdvancedInconsistencyTracker:
    def __init__(self):
        self.logger = self._setup_logger()
        self.daily_stats = defaultdict(lambda: defaultdict(int))
        self.alert_threshold = 10  # 触发告警的不一致数量
    def _setup_logger(self):
        logger = logging.getLogger('AdvancedTracker')
        logger.setLevel(logging.DEBUG)
        # 主日志
        main_handler = logging.FileHandler('inconsistency_tracker.log')
        main_handler.setLevel(logging.INFO)
        logger.addHandler(main_handler)
        # 详细日志
        detail_handler = logging.FileHandler('inconsistency_details.log')
        detail_handler.setLevel(logging.DEBUG)
        logger.addHandler(detail_handler)
        return logger
    def log_with_context(self, inconsistency_type, context, details):
        """
        记录带上下文的不一致信息
        Args:
            inconsistency_type: 不一致类型
            context: 上下文信息（数据源、表名等）
            details: 详细信息
        """
        timestamp = datetime.now()
        date_key = timestamp.strftime('%Y-%m-%d')
        # 更新统计
        self.daily_stats[date_key][inconsistency_type] += 1
        self.daily_stats[date_key]['total'] += 1
        # 记录日志
        log_entry = {
            'timestamp': timestamp.isoformat(),
            'type': inconsistency_type,
            'context': context,
            'details': details,
            'daily_count': self.daily_stats[date_key][inconsistency_type]
        }
        self.logger.info(json.dumps(log_entry, ensure_ascii=False))
        # 检查是否需要告警
        if self.daily_stats[date_key]['total'] >= self.alert_threshold:
            self._send_alert(date_key)
    def _send_alert(self, date_key):
        """发送告警"""
        alert_msg = (
            f"[ALERT] 数据不一致数量已达到 {self.alert_threshold}!"
            f" 日期: {date_key}, "
            f"总不一致: {self.daily_stats[date_key]['total']}"
        )
        self.logger.warning(alert_msg)
        print(f"⚠️ {alert_msg}")
    def generate_report(self, days=7):
        """生成统计报告"""
        report = []
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        current = start_date
        while current <= end_date:
            date_key = current.strftime('%Y-%m-%d')
            stats = self.daily_stats.get(date_key, {'total': 0})
            report.append({
                'date': date_key,
                'total_inconsistencies': stats['total'],
                'types': {k: v for k, v in stats.items() if k != 'total'}
            })
            current += timedelta(days=1)
        # 保存报告
        report_file = f'inconsistency_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json'
        with open(report_file, 'w', encoding='utf-8') as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        self.logger.info(f"报告已保存到: {report_file}")
        return report
# 使用示例
tracker = AdvancedInconsistencyTracker()
# 模拟不同场景的不一致
inconsistencies = [
    ('FIELD_MISMATCH', {'source': 'order_db', 'table': 'orders'}, 
     {'field': 'amount', 'expected': 100, 'actual': 99}),
    ('MISSING_RECORD', {'source': 'user_db', 'table': 'users'}, 
     {'id': '12345', 'expected': 'present', 'actual': 'missing'}),
    ('RECORD_DUPLICATE', {'source': 'product_db', 'table': 'products'}, 
     {'id': 'P001', 'duplicate_count': 2})
]
for inc_type, context, details in inconsistencies:
    tracker.log_with_context(inc_type, context, details)
# 生成报告
report = tracker.generate_report(days=7)

最佳实践建议

1. 分级日志：使用不同级别记录不同严重程度的不一致
2. 结构化数据：使用JSON格式便于后续分析
3. 自动统计：记录不一致的数量和类型统计
4. 告警机制：当不一致超过阈值时自动告警
5. 定期清理：定期归档和清理旧日志

选择哪种方式取决于你的具体需求,建议在项目初期就建立完善的数据不一致日志系统。