flowchart LR
    A[前端发起导出] --> B[API 网关/入口服务]
    B --> C[参数校验 & 幂等键生成]
    C --> D[(Redis 队列<br/>LIST / Stream)]
    D --> E[Worker 消费者集群]
    E --> F[分片查询 & 流式写入]
    F --> G[对象存储<br/>MinIO / S3]
    G --> H[回调/WebSocket 推送下载链接]
    H --> A

# 伪代码：按会议 ID 范围分片
def build_shards(task_params, shard_size=5000):
    total = count_meetings(task_params)
    shards = []
    for offset in range(0, total, shard_size):
        shards.append({
            "task_id": task_params["task_id"],
            "shard_id": len(shards),
            "offset": offset,
            "limit": shard_size,
            "filters": task_params["filters"]
        })
    return shards

# ❌ 全量加载内存，百万行直接 OOM
rows = db.query_all(sql)
df = pd.DataFrame(rows)
df.to_excel(buffer)

# ✅ 生成器 + openpyxl write_only 模式 / csv 流式写入
def stream_export(shard_iter, tmp_path):
    with open(tmp_path, 'wb') as f:
        writer = csv.writer(f)
        writer.writerow(HEADER)  # 写表头
        for chunk in shard_iter:  # 分页游标迭代
            writer.writerows(chunk)

{
  "task_id": "export_20240520_001",
  "status": "RUNNING",
  "progress": {
    "total_shards": 12,
    "completed_shards": 7,
    "current_shard_rows": 3421,
    "total_rows": 120000,
    "eta_seconds": 38
  },
  "updated_at": "2024-05-20T10:15:32Z"
}

-- Redis Lua 脚本：原子性获取最高优先级可用任务
local function pop_task()
  for _, q in ipairs({'high','normal','low'}) do
    local msg = redis.call('XREADGROUP','GROUP','g1','c1','COUNT',1,'BLOCK',5000,'STREAMS',q,'>')
    if msg then return msg end
  end
  return nil
end

# K8s Deployment 示例：协程 Worker (IO型) + 多进程 Worker (CPU型) 分离部署
apiVersion: apps/v1
kind: Deployment
metadata:
  name: export-worker-io
spec:
  replicas: 10
  template:
    spec:
      containers:
      - name: worker
        image: export-worker:latest
        env:
        - name: WORKER_MODE
          value: "async_io"        # 入口识别模式
        - name: CONCURRENCY
          value: "100"
        resources:
          limits:
            cpu: "2000m"
            memory: "2Gi"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: export-worker-cpu
spec:
  replicas: 5
  template:
    spec:
      containers:
      - name: worker
        image: export-worker:latest
        env:
        - name: WORKER_MODE
          value: "multiprocess"    # 入口识别模式
        - name: PROCESSES
          value: "8"               # = CPU 核心数
        resources:
          limits:
            cpu: "8000m"
            memory: "8Gi"

# 入口服务：获取全局一致性快照点
def create_export_task(params):
    # 1. 开启只读事务，获取当前快照版本（MySQL/PostgreSQL/TiDB 均支持）
    with db.readonly_transaction() as txn:
        snapshot_ts = txn.get_snapshot_timestamp()  # MySQL: @@transaction_id / PG: txid_current_snapshot()
        task_id = gen_task_id()
        # 2. 写入任务元数据，绑定快照版本
        redis.hset(f"task:{task_id}", mapping={
            "snapshot_ts": snapshot_ts,
            "params": json.dumps(params),
            "status": "QUEUED"
        })
        # 3. 分片入队，携带 snapshot_ts
        for shard in build_shards(params):
            shard["snapshot_ts"] = snapshot_ts
            queue.xadd("queue:export", shard)
    return task_id

# Worker 分片消费：强制使用快照读
def process_shard(shard_data):
    snapshot_ts = shard_data["snapshot_ts"]
    # MySQL: SET TRANSACTION ISOLATION LEVEL REPEATABLE READ + START TRANSACTION READ ONLY
    # TiDB: SET @@tidb_snapshot = '{snapshot_ts}'
    # PG: SET TRANSACTION ISOLATION LEVEL REPEATABLE READ; -- 首条 SELECT 自动绑定快照
    with db.snapshot_read(snapshot_ts) as conn:
        rows = conn.execute(
            "SELECT * FROM meetings WHERE ... LIMIT %s OFFSET %s",
            (shard_data["limit"], shard_data["offset"])
        )
        stream_write_to_tmp(rows, shard_data["tmp_path"])

# 配置驱动，热加载，不改代码
MASK_RULES = {
    "phone": {"type": "regex", "pattern": r"(d{3})d{4}(d{4})", "replace": r"1****2"},
    "id_card": {"type": "func", "handler": "mask_id_card"},  # 保留前6后4
    "amount": {"type": "condition", "rule": "role != 'CFO' -> '***'"},
}

def apply_mask(row, user_role, column_level_map):
    masked = {}
    for col, val in row.items():
        level = column_level_map.get(col, "L1")
        if level == "L4": 
            masked[col] = "[REDACTED]"
        elif level == "L3":
            rule = MASK_RULES.get(col)
            masked[col] = mask_engine.apply(rule, val, context={"role": user_role}) if rule else val
        else:
            masked[col] = val
    return masked

# .github/workflows/export-compliance.yml
jobs:
  compliance-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: 敏感字段扫描
        run: |
          python tools/scan_sensitive_fields.py --config config/export_fields.yaml --fail-on-leak
      - name: 脱敏规则覆盖率检查
        run: |
          python tools/check_mask_coverage.py --threshold 1.0  # L3/L4 字段 100% 覆盖
      - name: 审计日志字段完整性测试
        run: |
          pytest tests/test_audit_log_schema.py -v
      - name: 签名 URL 最小权限验证
        run: |
          python tools/verify_presigned_policy.py --max-ttl 86400 --actions GetObject

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Iterator, Dict, Any

@dataclass
class ExportContext:
    task_id: str
    shard_id: int
    snapshot_ts: str
    user_role: str
    column_meta: List[Dict]  # [{name, type, level, format, width}]

class Exporter(ABC):
    """统一导出器接口：流式写入、分片感知、进度回调"""
    
    @abstractmethod
    def write_header(self, ctx: ExportContext, columns: List[str]): ...
    
    @abstractmethod
    def write_rows(self, ctx: ExportContext, rows: Iterator[Dict[str, Any]]): ...
    
    @abstractmethod
    def finalize(self, ctx: ExportContext) -> bytes:  # 返回分片二进制或路径
        ...

# 注册表模式，配置化驱动
EXPORTER_REGISTRY = {
    "xlsx": XlsxStreamExporter,      # openpyxl write_only + 临时文件
    "csv": CsvStreamExporter,        # csv.writer + gzip 压缩流
    "pdf": PdfTableExporter,         # reportlab / weasyprint (HTML转PDF)
    "jsonl": JsonLinesExporter,      # 大数据/下游计算专用
    "zip": ZipArchiveExporter,       # 多sheet/多文件打包
}

def get_exporter(format: str) -> Exporter:
    return EXPORTER_REGISTRY[format]()

# templates/meeting_weekly.yaml
meta:
  name: "周例会纪要导出"
  formats: [xlsx, pdf]
  default_format: xlsx
sheets:
  - name: "会议列表"
    query: "meeting_list_shard"  # 对应 SQL 模板名
    columns:
      - {field: meeting_id, header: "会议ID", width: 12, level: L1}
      - {field: title, header: "主题", width: 40, level: L2, wrap: true}
      - {field: start_time, header: "开始时间", width: 20, format: "yyyy-mm-dd hh:mm", level: L1}
      - {field: attendees, header: "参会人", width: 30, level: L2, transform: "join_names"}
      - {field: action_items, header: "行动项", width: 50, level: L3, mask: true}
    styles:
      header: {bold: true, bg: "1F4E79", font_color: "FFFFFF", freeze: true}
      row_alt: {bg: "D6E4F0"}  # 斑马纹
  - name: "统计汇总"
    query: "meeting_stats"
    pivot: true  # Worker 端透视计算

flowchart LR
    A[读取分片数据] --> B[字段级脱敏/格式化]
    B --> C[模板解析: 列顺序/宽度/样式/公式]
    C --> D{格式分发}
    D -->|XLSX| E[openpyxl 流式写入]
    D -->|CSV| F[csv.writer + gzip]
    D -->|PDF| G[HTML模板 -> WeasyPrint]
    D -->|JSONL| H[json.dumps 行写入]
    E & F & G & H --> I[分片临时文件落盘]
    I --> J[上报进度/校验行数/校验和]

# 1. 定义 Queue 为 ScaledObject 触发源
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: export-worker-scaler
spec:
  scaleTargetRef:
    name: export-worker  # Deployment
  pollingInterval: 15
  cooldownPeriod: 300
  minReplicaCount: 0      # 无任务缩容到 0，极致省钱
  maxReplicaCount: 100
  triggers:
  - type: redis
    metadata:
      address: redis-master:6379
      listName: queue:export  # 或 Stream Key
      listLength: "10"        # 每 10 个积压任务扩 1 个 Pod
      enableTLS: "true"
  - type: cpu
    metadata:
      type: Utilization
      value: "70"

-- ❌ OFFSET 1000000 LIMIT 5000  -> 全表扫描 + filesort
-- ✅ 游标分页 (Keyset Pagination) / Seek Method
SELECT * FROM meetings 
WHERE tenant_id = ? AND create_time < ?  -- 上一页最后一条的 create_time
ORDER BY create_time DESC, id DESC
LIMIT 5000;

# gunicorn/uvicorn 启动前注入
import gc
# 1. 调大阈值，减少年轻代扫描频率 (对象存活率高的流式场景)
gc.set_threshold(1000, 20, 20)  # 默认 700, 10, 10
# 2. 禁用自动 GC，手动在分片间隙触发 (确定性停顿)
gc.disable()

async def process_shard(shard):
    try:
        # ... 流式写入 ...
        pass
    finally:
        # 分片完成显式回收，避免累积
        gc.collect()  

# KEDA 外挂自定义 Scaler (Python Flask)
@app.route("/metrics")
def custom_metrics():
    queue_len = redis.xlen("queue:export")
    # 1. 基础排队论：目标等待时间 < 60s
    # Worker 吞吐 ≈ 5 tasks/s/pod (实测基线)
    desired_replicas = max(1, math.ceil(queue_len / (5 * 60)))
    
    # 2. 业务日历修正：周一早高峰 / 财务结账日 预扩容
    if is_peak_hour() or is_fiscal_period():
        desired_replicas = max(desired_replicas, PRE_WARM_REPLICAS)
    
    # 3. 成本护栏：单日成本超阈值锁定上限
    if daily_cost > DAILY_BUDGET * 0.9:
        desired_replicas = min(desired_replicas, COST_CAP_REPLICAS)
    
    return jsonify({"desiredReplicas": desired_replicas})

# 运维控制器：每 30s 巡检一次
def reconcile():
    # 1. 僵尸任务检测：RUNNING > 2h 无进度更新
    stuck_tasks = redis.zrangebyscore("task:heartbeat", 0, now - 7200)
    for task_id in stuck_tasks:
        handle_stuck_task(task_id)  # 标记 FAILED, 释放分片锁, 触发补偿入队
    
    # 2. 孤儿分片清理：临时文件 > 24h 未合并
    orphan_files = list_oss_prefix("tmp/export/", older_than=86400)
    batch_delete(orphan_files)
    
    # 3. Worker 健康度：心跳超时 60s -> 驱逐 Pod, 任务重入队
    dead_workers = [w for w in workers if w.last_heartbeat < now - 60]
    for w in dead_workers:
        k8s.delete_pod(w.pod_name)
        requeue_unacked_shards(w.worker_id)
    
    # 4. 死信队列自动重试：连续失败 < 3 次且错误可重试
    retry_dlq_tasks(max_retry=3, retryable_errors=["timeout", "deadlock", "network"])

export-service/
├── cmd/
│   ├── api-server/          # 入口服务
│   ├── worker/              # 消费者 (支持 async_io / multiprocess 双模式)
│   └── merger/              # Serverless 合并函数
├── internal/
│   ├── queue/               # Redis Stream / RabbitMQ / Kafka 适配器
│   ├── storage/             # OSS / MinIO / S3 统一封装 (分片上传/预签名)
│   ├── exporter/            # 统一抽象 + XLSX/CSV/PDF/JSONL 实现
│   ├── mask/                # 动态脱敏引擎 + 规则热加载
│   ├── snapshot/            # MVCC 快照读封装 (MySQL/PG/TiDB 方言)
│   ├── audit/               # 审计日志写入 (异步批量 -> ClickHouse)
│   └── metrics/             # Prometheus 指标定义 + 自定义 KEDA Scaler
├── configs/
│   ├── templates/           # YAML 模板库 (Git 管理、热加载)
│   ├── mask_rules.yaml      # 脱敏规则
│   └── queue_priority.yaml  # 优先级/租户配额
├── deploy/
│   ├── k8s/                 # Deployment / HPA / KEDA / ServiceMonitor
│   ├── helm/                # Helm Chart (values.yaml 环境化)
│   └── terraform/           # 云资源 (Redis/MQ/OSS/监控)
├── tests/
│   ├── integration/         # 契约测试、混沌测试
│   ├── load/                # k6 / Locust 压测脚本
│   └── compliance/          # 合规自动化扫描
├── docs/
│   ├── ARCHITECTURE.md      # 架构决策记录 (ADR)
│   ├── RUNBOOK.md           # 故障处理 SOP
│   └── API.md               # OpenAPI 3.0 规范
├── go.mod / pyproject.toml
├── Dockerfile.multiarch
└── Makefile                 # build/test/lint/release 自动化