Python全套实战项目：从脚本小子到系统构建者

# 你的工具叫 mytools
mytools json pretty data.json          # 格式化JSON文件
mytools file split large.log 100       # 拆分大文件（按行数）
mytools time calc "2026-07-01" "2026-12-31"  # 计算日期差
mytools net ping 192.168.1.1 -c 5      # 带超时的ping探测
mytools encrypt md5 myfile.txt         # 计算文件MD5

# 配置驱动的爬虫
crawler:
  name: "news_crawler"
  start_urls:
    - "https://example.com/news"
  parse_rules:
    title: "h1.article-title"
    content: "div.article-content"
    time: "span.publish-time"
  output:
    format: "jsonl"
    path: "./data/news/"
  concurrency: 10
  retry: 3
  delay: 1.0

# 数据清洗
raw_data → 去重 → 缺失值处理 → 异常值检测 → 格式统一 → 清洗后数据

# 数据分析
清洗后数据 → 聚合统计 → 相关性分析 → 趋势检测 → 分析报告

# 可视化
分析结果 → Matplotlib/Seaborn图表 → 自动化报告（HTML/PDF）

# 用户认证
POST /api/auth/login   → JWT Token
POST /api/auth/refresh → 刷新Token

# 爬虫任务管理（结合项目二）
POST /api/crawler/start  → 启动爬虫任务（异步执行）
GET  /api/crawler/status/{task_id} → 查询任务状态
GET  /api/crawler/data/{task_id} → 获取爬取结果

# 数据分析（结合项目三）
POST /api/analysis/run → 运行分析流水线
GET  /api/analysis/report/{run_id} → 获取分析报告

# 系统
GET  /health → 健康检查
GET  /metrics → Prometheus指标

-- 用户表
users (id, username, password_hash, email, created_at)

-- 爬虫任务表
crawler_tasks (id, name, status, config, result_path, created_at, finished_at)

-- 分析任务表
analysis_tasks (id, name, status, input_path, report_path, created_at, finished_at)

-- 审计日志
audit_logs (id, user_id, action, timestamp, ip)

# config/alerts.yaml
rules:
  - name: "CPU过载预警"
    metric: "cpu.percent"
    condition: "> 80"
    duration: "5m"
    severity: "warning"
    receivers: ["devops@company.com"]

  - name: "API错误率突增"
    metric: "api.error_rate"
    condition: "> 5%"
    duration: "1m"
    severity: "critical"
    receivers: ["oncall@company.com"]

# 工作流定义
workflow:
  name: "daily_etl_pipeline"
  schedule: "0 2 * * *"  # 每天凌晨2点执行
  tasks:
    - id: "extract"
      type: "python_script"
      script: "extract_data.py"
      retry: 3
    
    - id: "transform"
      type: "python_script"
      script: "transform_data.py"
      depends_on: ["extract"]  # 依赖上一任务
      retry: 2
    
    - id: "load"
      type: "http_request"
      url: "http://data-api/load"
      method: "POST"
      depends_on: ["transform"]
      timeout: 300
    
    - id: "report"
      type: "send_email"
      depends_on: ["load"]
      config:
        to: ["team@company.com"]
        subject: "ETL执行完成"

# gateway/config.yaml
routes:
  - path: "/api/v1/crawler/*"
    target: "http://crawler-service:8000"
    auth_required: true
    rate_limit: "100/minute"
  
  - path: "/api/v1/analysis/*"
    target: "http://analysis-service:8001"
    auth_required: true
    rate_limit: "50/minute"
  
  - path: "/api/v1/public/*"
    target: "http://public-service:8002"
    auth_required: false
    rate_limit: "1000/minute"

auth:
  jwt_secret: "${JWT_SECRET}"
  token_expire: 3600

项目一（CLI工具集）
        ↓ 提升工程化规范
项目二（异步爬虫系统）
        ↓ 产出数据
项目三（数据分析流水线）
        ↓ 能力封装
项目四（FastAPI后端服务）
        ↓ 增加运维视角
项目五（监控告警系统）
        ↓ 增加编排能力
项目六（工作流引擎）
        ↓ 增加架构能力
项目七（API网关）

# mytools/cli.py
import click
import json
import sys
from pathlib import Path

@click.group()
def cli():
    """我的工具箱 - 让日常开发更高效"""
    pass

@cli.command()
@click.argument('file', type=click.Path(exists=True))
@click.option('--indent', default=2, help='缩进空格数')
def json_pretty(file, indent):
    """格式化JSON文件，使其更易读"""
    try:
        data = json.loads(Path(file).read_text(encoding='utf-8'))
        print(json.dumps(data, ensure_ascii=False, indent=indent))
    except Exception as e:
        click.echo(f"❌ 错误: {e}", err=True)
        sys.exit(1)

@cli.command()
@click.argument('file', type=click.Path(exists=True))
def md5(file):
    """计算文件的MD5哈希值"""
    import hashlib
    hash_md5 = hashlib.md5()
    with open(file, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            hash_md5.update(chunk)
    click.echo(f"MD5: {hash_md5.hexdigest()}")

if __name__ == '__main__':
    cli()

pip install click
python mytools/cli.py json_pretty data.json
python mytools/cli.py md5 myfile.txt