对于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')
推荐方案
- 小项目:使用内置日志 + 文件记录
- 中型项目:使用 Sentry + 自定义通知
- 大型项目:Sentry + Datadog/自定义错误服务 + 多个通知渠道
选择时考虑:
- 成本:Sentry 有免费额度
- 隐私:自定义方案数据完全可控
- 复杂度:Sentry 几分钟即可集成
- 功能需求:是否需要堆栈跟踪、上下文信息等