# script_with_exit_code.py
import sys
import logging

logging.basicConfig(level=logging.INFO)

def main():
    # 業(yè)務邏輯
    pass

if __name__ == "__main__":
    try:
        main()
        print("SUCCESS")
        sys.exit(0)   # 0 表示成功
    except Exception as e:
        logging.exception("任務執(zhí)行失敗")
        sys.exit(1)   # 非0表示失敗

import functools
import time
import logging
from typing import Callable, Any

logger = logging.getLogger(__name__)

def monitor_job(job_name: str, 
                alert_on_failure: bool = True,
                max_retries: int = 0,
                retry_delay: int = 5):
    """
    監(jiān)控任務執(zhí)行的裝飾器
    :param job_name: 任務名稱，用于標識
    :param alert_on_failure: 失敗時是否觸發(fā)告警
    :param max_retries: 失敗后自動重試次數(shù)
    :param retry_delay: 重試間隔（秒）
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            start_time = time.time()
            last_exception = None
            
            for attempt in range(max_retries + 1):
                try:
                    result = func(*args, **kwargs)
                    elapsed = time.time() - start_time
                    logger.info(f"任務[{job_name}] 執(zhí)行成功，耗時 {elapsed:.2f}秒")
                    
                    # 可選：成功時也可以發(fā)送通知（對于關(guān)鍵任務）
                    # if notify_on_success:
                    #     send_success_alert(job_name, elapsed)
                    
                    return result
                    
                except Exception as e:
                    last_exception = e
                    elapsed = time.time() - start_time
                    logger.error(f"任務[{job_name}] 執(zhí)行失敗 (嘗試 {attempt+1}/{max_retries+1}): {str(e)}")
                    
                    if attempt < max_retries:
                        time.sleep(retry_delay)
                    else:
                        # 重試耗盡，觸發(fā)告警
                        if alert_on_failure:
                            send_failure_alert(job_name, e, elapsed)
                        raise  # 重新拋出異常
            
            # 不會執(zhí)行到這里
            return None
        return wrapper
    return decorator

# 使用示例
@monitor_job("每日數(shù)據(jù)同步", alert_on_failure=True, max_retries=2, retry_delay=10)
def sync_data():
    # 業(yè)務邏輯，可能拋出異常
    pass

def sync_orders():
    orders = fetch_orders_from_api()
    if len(orders) == 0:
        # 這不是異常，但業(yè)務上屬于異常情況
        raise BusinessWarning("未獲取到任何訂單，可能API變更或權(quán)限失效")
    
    inserted = db.insert_many(orders)
    if inserted < len(orders) * 0.9:
        raise BusinessWarning(f"插入成功率過低: {inserted}/{len(orders)}")
    
    # 一切正常
    return {"total": len(orders), "inserted": inserted}

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
import traceback

def send_email_alert(subject: str, body: str, to_emails: list):
    """發(fā)送郵件告警"""
    smtp_server = "smtp.example.com"   # 替換為實際SMTP服務器
    smtp_port = 465                    # SSL端口
    sender_email = "alerts@yourdomain.com"
    sender_password = "your_password"
    
    msg = MIMEMultipart()
    msg['From'] = sender_email
    msg['To'] = ", ".join(to_emails)
    msg['Subject'] = Header(subject, 'utf-8')
    
    msg.attach(MIMEText(body, 'plain', 'utf-8'))
    
    try:
        with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
            server.login(sender_email, sender_password)
            server.sendmail(sender_email, to_emails, msg.as_string())
        print("郵件告警發(fā)送成功")
    except Exception as e:
        print(f"郵件發(fā)送失敗: {e}")

def send_failure_alert(job_name, exception, elapsed):
    subject = f"[告警] 任務失敗: {job_name}"
    body = f"""
任務名稱: {job_name}
執(zhí)行耗時: {elapsed:.2f}秒
失敗時間: {time.strftime('%Y-%m-%d %H:%M:%S')}
異常類型: {type(exception).__name__}
異常信息: {str(exception)}
詳細堆棧:
{traceback.format_exc()}
    """
    send_email_alert(subject, body, ["ops@yourcompany.com", "oncall@yourcompany.com"])

import requests
import json
import time
import hmac
import hashlib
import base64
from urllib.parse import quote_plus

def send_dingtalk_alert(job_name, exception, elapsed, webhook_url, secret=None):
    """發(fā)送釘釘機器人告警（支持加簽）"""
    headers = {'Content-Type': 'application/json'}
    
    # 如果開啟了加簽，需要生成時間戳和簽名
    timestamp = str(round(time.time() * 1000))
    sign = ""
    if secret:
        string_to_sign = f"{timestamp}\n{secret}"
        hmac_code = hmac.new(
            secret.encode('utf-8'), 
            string_to_sign.encode('utf-8'), 
            digestmod=hashlib.sha256
        ).digest()
        sign = base64.b64encode(hmac_code).decode('utf-8')
        webhook_url = f"{webhook_url}&timestamp={timestamp}&sign={sign}"
    
    # 構(gòu)造消息內(nèi)容（Markdown格式）
    markdown_text = f"""## ?? 任務失敗告警
    - **任務名稱**: {job_name}
    - **執(zhí)行耗時**: {elapsed:.2f}秒
    - **異常類型**: {type(exception).__name__}
    - **異常信息**: {str(exception)}
    - **詳情**: 請查看完整日志
    """
    
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": f"任務失敗: {job_name}",
            "text": markdown_text
        },
        "at": {
            "atMobiles": ["13800000000"],  # 可選：@指定手機號
            "isAtAll": False
        }
    }
    
    response = requests.post(webhook_url, headers=headers, data=json.dumps(payload), timeout=5)
    if response.status_code != 200:
        print(f"釘釘告警發(fā)送失敗: {response.text}")

# 使用示例（webhook_url從環(huán)境變量獲?。?
# send_dingtalk_alert("數(shù)據(jù)同步", Exception("連接超時"), 30.5, webhook_url, secret)

def send_wecom_alert(job_name, exception, elapsed, webhook_url):
    """發(fā)送企業(yè)微信機器人告警"""
    headers = {'Content-Type': 'application/json'}
    
    # 企業(yè)微信markdown消息長度限制4096字節(jié)
    content = f"""## <font color="warning">任務失敗告警</font>
    > **任務名稱**: {job_name}
    > **執(zhí)行耗時**: {elapsed:.2f}秒
    > **異常類型**: {type(exception).__name__}
    > **異常信息**: {str(exception)}
    
    > 請及時處理：[查看日志](https://your-log-platform.com)
    """
    
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": content
        }
    }
    
    response = requests.post(webhook_url, headers=headers, json=payload, timeout=5)
    if response.status_code != 200:
        print(f"企業(yè)微信告警發(fā)送失敗: {response.text}")

from abc import ABC, abstractmethod

class AlertChannel(ABC):
    @abstractmethod
    def send(self, job_name: str, exception: Exception, elapsed: float):
        pass

class DingTalkChannel(AlertChannel):
    def __init__(self, webhook_url: str, secret: str = None):
        self.webhook_url = webhook_url
        self.secret = secret
    
    def send(self, job_name: str, exception: Exception, elapsed: float):
        # 調(diào)用上面的 send_dingtalk_alert 函數(shù)
        send_dingtalk_alert(job_name, exception, elapsed, self.webhook_url, self.secret)

class EmailChannel(AlertChannel):
    def __init__(self, smtp_config: dict, to_emails: list):
        self.smtp_config = smtp_config
        self.to_emails = to_emails
    
    def send(self, job_name: str, exception: Exception, elapsed: float):
        # 郵件發(fā)送邏輯
        pass

# 使用配置
channels = [
    DingTalkChannel(os.getenv("DINGTALK_WEBHOOK"), os.getenv("DINGTALK_SECRET")),
    EmailChannel(smtp_config, ["admin@example.com"])
]

def send_failure_alert(job_name, exception, elapsed):
    for channel in channels:
        try:
            channel.send(job_name, exception, elapsed)
        except Exception as e:
            logger.error(f"告警通道發(fā)送失敗: {e}")

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 告警通道配置（使用環(huán)境變量）
alert_channels = [
    DingTalkChannel(os.getenv("DINGTALK_WEBHOOK"), os.getenv("DINGTALK_SECRET")),
    # 可以再加郵件通道
]

def send_failure_alert(job_name, exception, elapsed):
    for ch in alert_channels:
        try:
            ch.send(job_name, exception, elapsed)
        except Exception as e:
            logger.error(f"告警發(fā)送失敗: {e}")

# 監(jiān)控裝飾器（復用之前的實現(xiàn)）
def monitored_job(job_name):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                logger.info(f"Job {job_name} succeeded in {time.time()-start:.2f}s")
                return result
            except Exception as e:
                elapsed = time.time() - start
                logger.exception(f"Job {job_name} failed: {e}")
                send_failure_alert(job_name, e, elapsed)
                raise
        return wrapper
    return decorator

# 實際業(yè)務任務
@monitored_job("daily_data_sync")
def daily_data_sync():
    # 業(yè)務邏輯
    pass

@monitored_job("cleanup_temp_files")
def cleanup_temp_files():
    pass

# APScheduler配置
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
    'default': ThreadPoolExecutor(max_workers=10)
}
scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, timezone='Asia/Shanghai')

# 添加任務
scheduler.add_job(daily_data_sync, 'cron', hour=2, minute=0, id='daily_sync')
scheduler.add_job(cleanup_temp_files, 'cron', hour=3, minute=0, id='cleanup')

scheduler.start()

# 在任務成功結(jié)束時寫入心跳
import redis
r = redis.Redis(host='localhost', port=6379, db=0)

def report_heartbeat(job_name):
    r.set(f"heartbeat:{job_name}", int(time.time()))

@monitored_job("daily_sync")
def daily_sync():
    # ... 業(yè)務邏輯
    report_heartbeat("daily_sync")

def check_heartbeats():
    expected_intervals = {
        "daily_sync": 86400,   # 期望每天一次
        "hourly_clean": 3600
    }
    now = time.time()
    for job_name, interval in expected_intervals.items():
        last_beat = r.get(f"heartbeat:{job_name}")
        if last_beat is None:
            send_alert(f"任務{(diào)job_name}從未執(zhí)行過")
        else:
            elapsed = now - int(last_beat)
            if elapsed > interval + 3600:  # 容忍1小時
                send_alert(f"任務{(diào)job_name}心跳超時，已{elapsed//3600}小時未執(zhí)行")

def send_escalated_alert(job_name, exception, elapsed):
    # 從Redis獲取連續(xù)失敗次數(shù)
    key = f"fail_count:{job_name}"
    fail_count = r.incr(key)
    r.expire(key, 3600)  # 1小時衰減
    
    if fail_count >= 3:
        # 發(fā)送嚴重告警（釘釘+短信）
        send_dingtalk_alert(job_name, exception, elapsed, webhook, at_mobiles=['manager'])
        send_sms_alert(...)
    elif fail_count >= 1:
        # 普通告警
        send_dingtalk_alert(job_name, exception, elapsed, webhook)

import json_logging
import logging

json_logging.init_non_web(enable_json=True)
logger = logging.getLogger(__name__)

# 之后的所有日志都會輸出JSON格式
logger.info("任務開始", extra={"job_name": "sync", "step": "fetch"})

import os

DINGTALK_WEBHOOK = os.getenv("DINGTALK_WEBHOOK")
if not DINGTALK_WEBHOOK:
    raise ValueError("請設(shè)置環(huán)境變量 DINGTALK_WEBHOOK")