如何写一个脚本自动上报状态

本文目录导读：

1. 基础版 - Shell脚本 + Cron任务
2. Python版 - 更灵活
3. 带重试机制的可靠版本
4. 使用Docker部署
5. 使用建议

我来帮你写一个自动上报状态的脚本,根据不同场景，提供几种常见实现方案：

基础版 - Shell脚本 + Cron任务

#!/bin/bash
# auto_report.sh
# 配置参数
API_URL="https://your-api.com/report"
SERVER_NAME=$(hostname)
STATUS="running"
TIMESTAMP=$(date '+%Y-%m-%d %H:%M:%S')
# 收集系统状态
CPU_USAGE=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)
MEM_USAGE=$(free | grep Mem | awk '{print $3/$2 * 100.0}')
DISK_USAGE=$(df -h / | awk 'NR==2 {print $5}' | cut -d'%' -f1)
# 构建JSON数据
JSON_DATA=$(cat <<EOF
{
    "server": "$SERVER_NAME",
    "status": "$STATUS",
    "timestamp": "$TIMESTAMP",
    "metrics": {
        "cpu": $CPU_USAGE,
        "memory": $MEM_USAGE,
        "disk": $DISK_USAGE
    }
}
EOF
)
# 发送请求
curl -X POST "$API_URL" \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer YOUR_TOKEN" \
    -d "$JSON_DATA"
# 检查结果
if [ $? -eq 0 ]; then
    echo "$(date): Status reported successfully"
else
    echo "$(date): Failed to report status"
fi

设置Cron定时任务：

# 每分钟执行一次
* * * * * /path/to/auto_report.sh
# 每5分钟执行一次
*/5 * * * * /path/to/auto_report.sh
# 每天8点执行
0 8 * * * /path/to/auto_report.sh

Python版 - 更灵活

# auto_report.py
import requests
import json
import psutil
import socket
import datetime
import time
import logging
class StatusReporter:
    def __init__(self, config):
        self.api_url = config.get('api_url')
        self.token = config.get('token')
        self.interval = config.get('interval', 60)  # 默认60秒
        self.server_name = socket.gethostname()
        # 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
    def collect_metrics(self):
        """收集系统指标"""
        return {
            'server': self.server_name,
            'status': 'running',
            'timestamp': datetime.datetime.now().isoformat(),
            'metrics': {
                'cpu_percent': psutil.cpu_percent(interval=1),
                'memory_percent': psutil.virtual_memory().percent,
                'disk_percent': psutil.disk_usage('/').percent,
                'network_connections': len(psutil.net_connections())
            }
        }
    def report_status(self):
        """上报状态"""
        try:
            data = self.collect_metrics()
            response = requests.post(
                self.api_url,
                json=data,
                headers={
                    'Authorization': f'Bearer {self.token}',
                    'Content-Type': 'application/json'
                },
                timeout=10
            )
            if response.status_code == 200:
                logging.info(f"Status reported: {response.json()}")
                return True
            else:
                logging.error(f"Report failed: {response.status_code}")
                return False
        except requests.exceptions.RequestException as e:
            logging.error(f"Request failed: {e}")
            return False
        except Exception as e:
            logging.error(f"Unexpected error: {e}")
            return False
    def run_continuous(self):
        """持续运行"""
        logging.info(f"Starting continuous reporting every {self.interval} seconds")
        while True:
            self.report_status()
            time.sleep(self.interval)
# 使用示例
if __name__ == "__main__":
    config = {
        'api_url': 'https://your-api.com/report',
        'token': 'your-secret-token',
        'interval': 300  # 5分钟
    }
    reporter = StatusReporter(config)
    # 单次上报
    # reporter.report_status()
    # 持续运行
    reporter.run_continuous()

带重试机制的可靠版本

# reliable_reporter.py
import requests
import json
import time
from retry import retry
from datetime import datetime
import hashlib
class ReliableReporter:
    def __init__(self, config):
        self.config = config
        self.failed_attempts = 0
        self.max_retries = config.get('max_retries', 3)
    @retry(tries=3, delay=2, backoff=2)
    def send_with_retry(self, data):
        """带重试的发送"""
        response = requests.post(
            self.config['api_url'],
            json=data,
            headers=self.get_headers(data),
            timeout=5
        )
        response.raise_for_status()
        return response
    def get_headers(self, data):
        """生成请求头"""
        timestamp = str(int(time.time()))
        signature = hashlib.sha256(
            f"{json.dumps(data)}{self.config['secret']}{timestamp}".encode()
        ).hexdigest()
        return {
            'Content-Type': 'application/json',
            'X-Timestamp': timestamp,
            'X-Signature': signature,
            'Authorization': f'Bearer {self.config["token"]}'
        }
    def report_with_circuit_breaker(self):
        """带熔断机制的上报"""
        if self.failed_attempts >= self.config.get('max_failures', 5):
            # 进入熔断状态
            print("Circuit breaker opened, waiting...")
            time.sleep(self.config.get('cooldown', 300))
            self.failed_attempts = 0
        try:
            data = self.collect_status()
            self.send_with_retry(data)
            self.failed_attempts = 0
            return True
        except Exception as e:
            self.failed_attempts += 1
            print(f"Report failed ({self.failed_attempts}/{self.config.get('max_failures', 5)}): {e}")
            # 保存失败记录
            self.save_failed_report(data)
            return False
    def save_failed_report(self, data):
        """保存失败的上报"""
        with open('failed_reports.json', 'a') as f:
            record = {
                'timestamp': datetime.now().isoformat(),
                'data': data,
                'failed_count': self.failed_attempts
            }
            f.write(json.dumps(record) + '\n')
    def retry_failed_reports(self):
        """重试失败的上报"""
        try:
            with open('failed_reports.json', 'r') as f:
                for line in f:
                    record = json.loads(line)
                    try:
                        response = requests.post(
                            self.config['api_url'],
                            json=record['data'],
                            headers=self.get_headers(record['data']),
                            timeout=5
                        )
                        if response.status_code == 200:
                            print(f"Retry success for {record['timestamp']}")
                    except:
                        print(f"Retry failed for {record['timestamp']}")
        except FileNotFoundError:
            pass
# 配置示例
config = {
    'api_url': 'https://your-api.com/report',
    'token': 'your-token',
    'secret': 'your-secret',
    'max_retries': 3,
    'max_failures': 5,
    'cooldown': 300
}
reporter = ReliableReporter(config)
# 定时执行（使用schedule库）
# pip install schedule
import schedule
def job():
    reporter.report_with_circuit_breaker()
# 每5分钟执行一次
schedule.every(5).minutes.do(job)
while True:
    schedule.run_pending()
    time.sleep(1)

使用Docker部署

# Dockerfile
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "auto_report.py"]

# docker-compose.yml
version: '3'
services:
  reporter:
    build: .
    environment:
      - API_URL=https://your-api.com/report
      - TOKEN=your-token
      - INTERVAL=300
    restart: always

使用建议

1. 选择方案：

   • 简单场景：Shell + Cron
   • 复杂场景：Python全功能版
   • 高可靠性：带重试和熔断版本

2. 安全考虑：

   • 使用环境变量存储敏感信息
   • 添加签名验证
   • 使用HTTPS传输

3. 监控和告警：

   • 添加失败告警
   • 记录日志
   • 监控上报成功率

需要根据你的具体需求进行调整,比如API格式、认证方式、上报频率等。