Python项目错误怎么自动上报

wen python案例 23

对于Python项目错误自动上报,以下是几种常用方案:

Python项目错误怎么自动上报

使用第三方错误追踪服务

Sentry(最推荐)

# 安装
pip install sentry-sdk
# 配置
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
from sentry_sdk.integrations.django import DjangoIntegration
sentry_sdk.init(
    dsn="https://your-dsn@sentry.io/project-id",
    integrations=[
        # 根据你的框架选择
        FlaskIntegration(),
        DjangoIntegration(),
    ],
    traces_sample_rate=1.0,
    environment="production"
)
# 自动捕获所有未处理异常
# 也可以手动上报
try:
    raise ValueError("测试错误")
except Exception as e:
    sentry_sdk.capture_exception(e)

Datadog APM

from ddtrace import tracer, patch_all
patch_all()
@tracer.wrap()
def my_function():
    # 自动追踪错误
    pass

自行实现错误上报系统

基础错误处理器

import sys
import traceback
import requests
import logging
from functools import wraps
class ErrorReporter:
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url
    def report(self, exception, context=None):
        """上报错误到服务器"""
        error_info = {
            'error_type': type(exception).__name__,
            'error_message': str(exception),
            'traceback': traceback.format_exc(),
            'context': context or {},
        }
        try:
            requests.post(
                self.webhook_url,
                json=error_info,
                timeout=5
            )
        except Exception as e:
            # 避免上报过程中的错误影响主程序
            logging.error(f"Failed to report error: {e}")
# 全局异常处理器
def global_exception_handler(exc_type, exc_value, exc_traceback):
    reporter = ErrorReporter("https://your-server.com/error-report")
    reporter.report(exc_value)
sys.excepthook = global_exception_handler

装饰器方式

def auto_report_errors(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            reporter = ErrorReporter("your-webhook-url")
            reporter.report(e, {
                'function': func.__name__,
                'args': args,
                'kwargs': kwargs
            })
            raise  # 或返回错误响应
    return wrapper
# 使用
@auto_report_errors
def risk_function():
    # 可能出错的代码
    pass

集成日志系统

自定义日志处理器

import logging
import json
from datetime import datetime
class ErrorReportHandler(logging.Handler):
    def __init__(self, webhook_url):
        super().__init__(logging.ERROR)
        self.webhook_url = webhook_url
    def emit(self, record):
        error_data = {
            'timestamp': datetime.now().isoformat(),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno,
        }
        if record.exc_info:
            error_data['traceback'] = self.format(record)
        self._send(error_data)
    def _send(self, data):
        # 发送到你的错误收集服务
        import requests
        try:
            requests.post(self.webhook_url, json=data, timeout=3)
        except:
            pass
# 配置
logger = logging.getLogger(__name__)
handler = ErrorReportHandler("your-webhook-url")
logger.addHandler(handler)
# 使用
try:
    risky_operation()
except Exception as e:
    logger.error("Operation failed", exc_info=True)

集成消息通知

发送到企业微信/钉钉/Slack

import requests
import json
def send_to_wechat(webhook_url, error_info):
    """发送到企业微信"""
    data = {
        "msgtype": "markdown",
        "markdown": {
            "content": f"""
### 🚨 错误告警
- **类型**: {error_info['type']}
- **消息**: {error_info['message']}
- **位置**: {error_info['location']}
- **时间**: {error_info['time']}
```python
{error_info['traceback'][:2000]}
        """
    }
}
requests.post(webhook_url, json=data)

集成到错误处理器中

class ErrorNotifier: def init(self, webhook_url, notification_type='wechat'): self.webhook_url = webhook_url self.notification_type = notification_type

def notify(self, error_info):
    if self.notification_type == 'wechat':
        send_to_wechat(self.webhook_url, error_info)
    elif self.notification_type == 'dingtalk':
        send_to_dingtalk(self.webhook_url, error_info)
    # ... 其他通知方式

## 5. **完整的生产环境方案**
```python
import os
import sys
import logging
from functools import wraps
class ProductionErrorHandler:
    def __init__(self, app_name, environment):
        self.app_name = app_name
        self.environment = environment
        self.setup_error_reporting()
    def setup_error_reporting(self):
        # 1. 配置 Sentry
        if os.getenv('SENTRY_DSN'):
            import sentry_sdk
            sentry_sdk.init(
                dsn=os.getenv('SENTRY_DSN'),
                environment=self.environment,
            )
        # 2. 配置本地日志
        logging.basicConfig(
            level=logging.ERROR,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(f'{self.app_name}_errors.log'),
                logging.StreamHandler()
            ]
        )
        # 3. 设置全局异常处理
        sys.excepthook = self.global_exception_handler
    def global_exception_handler(self, exc_type, exc_value, exc_traceback):
        # 记录到日志
        logging.error(
            "Uncaught exception",
            exc_info=(exc_type, exc_value, exc_traceback)
        )
        # 上报到 Sentry
        if os.getenv('SENTRY_DSN'):
            import sentry_sdk
            sentry_sdk.capture_exception(exc_value)
        # 发送通知(可选)
        if os.getenv('ERROR_WEBHOOK'):
            self.send_notification(exc_type, exc_value)
    @classmethod
    def auto_report(cls, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logging.error(f"Error in {func.__name__}", exc_info=True)
                if os.getenv('SENTRY_DSN'):
                    import sentry_sdk
                    sentry_sdk.capture_exception(e)
                raise
        return wrapper
# 使用
error_handler = ProductionErrorHandler('myapp', 'production')
@ProductionErrorHandler.auto_report
def critical_function():
    # 自动跟踪和上报错误
    pass

轻量级替代方案

如果不想使用外部服务,可以考虑:

邮件通知

import smtplib
from email.mime.text import MIMEText
def send_error_email(error_info):
    msg = MIMEText(json.dumps(error_info, indent=2))
    msg['Subject'] = f"Error Report - {error_info['type']}"
    msg['From'] = "errors@yourdomain.com"
    msg['To'] = "admin@yourdomain.com"
    with smtplib.SMTP('smtp.gmail.com', 587) as server:
        server.login("user", "password")
        server.send_message(msg)

文件记录

import json
from datetime import datetime
class FileErrorLogger:
    def __init__(self, log_file='errors.json'):
        self.log_file = log_file
    def log_error(self, error_info):
        error_info['timestamp'] = datetime.now().isoformat()
        with open(self.log_file, 'a') as f:
            f.write(json.dumps(error_info) + '\n')

推荐方案

  1. 小项目:使用内置日志 + 文件记录
  2. 中型项目:使用 Sentry + 自定义通知
  3. 大型项目:Sentry + Datadog/自定义错误服务 + 多个通知渠道

选择时考虑:

  • 成本:Sentry 有免费额度
  • 隐私:自定义方案数据完全可控
  • 复杂度:Sentry 几分钟即可集成
  • 功能需求:是否需要堆栈跟踪、上下文信息等

抱歉,评论功能暂时关闭!