【存储工程】数据生命周期管理

# 分析对象访问频率随数据年龄的衰减
import json
from collections import defaultdict

def analyze_access_decay(access_log_path: str) -> dict:
    """
    分析访问日志，计算不同年龄段数据的平均日访问次数。
    access_log 格式：每行一个 JSON，包含 object_key, timestamp, created_at 字段。
    """
    age_buckets = defaultdict(lambda: {"access_count": 0, "object_count": 0})

    with open(access_log_path, "r") as f:
        for line in f:
            record = json.loads(line)
            age_days = (record["timestamp"] - record["created_at"]) / 86400
            bucket = categorize_age(age_days)
            age_buckets[bucket]["access_count"] += 1
            age_buckets[bucket]["object_count"] += 1

    result = {}
    for bucket, stats in sorted(age_buckets.items()):
        avg_daily = stats["access_count"] / max(stats["object_count"], 1)
        result[bucket] = round(avg_daily, 4)

    return result


def categorize_age(age_days: float) -> str:
    """将数据年龄分到预定义的桶中。"""
    if age_days <= 1:
        return "0-1d"
    elif age_days <= 7:
        return "1-7d"
    elif age_days <= 30:
        return "7-30d"
    elif age_days <= 90:
        return "30-90d"
    elif age_days <= 180:
        return "90-180d"
    elif age_days <= 365:
        return "180-365d"
    else:
        return "365d+"


def recommend_tiers(decay_data: dict, hot_threshold: float = 10.0,
                     warm_threshold: float = 1.0,
                     cold_threshold: float = 0.01) -> dict:
    """
    根据访问频率阈值，推荐各年龄段的存储层级。
    hot_threshold: 日均访问次数 >= 该值，归为热层
    warm_threshold: 日均访问次数 >= 该值且 < hot，归为温层
    cold_threshold: 日均访问次数 >= 该值且 < warm，归为冷层
    其余归为归档层。
    """
    recommendations = {}
    for bucket, avg_daily in decay_data.items():
        if avg_daily >= hot_threshold:
            tier = "hot"
        elif avg_daily >= warm_threshold:
            tier = "warm"
        elif avg_daily >= cold_threshold:
            tier = "cold"
        else:
            tier = "archive"
        recommendations[bucket] = {"avg_daily_access": avg_daily, "tier": tier}

    return recommendations

# 查看当前挂载选项
mount | grep /data

# relatime（默认）：仅当 atime 早于 mtime/ctime 时才更新 atime
# 大多数场景下足够用，性能影响小
mount -o remount,relatime /data

# strictatime：每次访问都更新 atime，精确但性能差
mount -o remount,strictatime /data

# noatime：完全不更新 atime，性能最好但无法追踪访问
mount -o remount,noatime /data

# lazytime：atime 更新缓存在内存中，定期刷盘
mount -o remount,lazytime /data

#!/bin/bash
# scan_cold_files.sh
# 扫描指定目录下超过 N 天未访问的文件，输出文件路径和大小

DATA_DIR="${1:-/data}"
COLD_DAYS="${2:-90}"
OUTPUT_FILE="${3:-cold_files_report.csv}"

echo "path,size_bytes,last_access,age_days" > "$OUTPUT_FILE"

find "$DATA_DIR" -type f -atime +"$COLD_DAYS" -printf '%p,%s,%A@,%T@\n' | \
while IFS=',' read -r path size atime mtime; do
    now=$(date +%s)
    age_days=$(( (now - ${atime%.*}) / 86400 ))
    echo "$path,$size,$atime,$age_days"
done >> "$OUTPUT_FILE"

total_files=$(tail -n +2 "$OUTPUT_FILE" | wc -l)
total_size=$(tail -n +2 "$OUTPUT_FILE" | awk -F',' '{sum+=$2} END {print sum}')

echo "扫描完成："
echo "  冷文件数量：$total_files"
echo "  冷文件总大小：$(numfmt --to=iec "$total_size")"
echo "  报告路径：$OUTPUT_FILE"

{
  "Id": "daily-inventory",
  "InventoryConfiguration": {
    "Destination": {
      "S3BucketDestination": {
        "AccountId": "123456789012",
        "Bucket": "arn:aws:s3:::inventory-bucket",
        "Format": "CSV",
        "Prefix": "inventory"
      }
    },
    "IsEnabled": true,
    "Id": "daily-inventory",
    "IncludedObjectVersions": "Current",
    "OptionalFields": [
      "Size",
      "LastModifiedDate",
      "StorageClass",
      "ETag"
    ],
    "Schedule": {
      "Frequency": "Daily"
    }
  }
}

# 从 S3 访问日志中统计各对象的 GET 请求次数（最近 30 天）
aws s3 cp s3://log-bucket/access-logs/ ./logs/ --recursive \
    --exclude "*" --include "*.log"

cat ./logs/*.log | \
    awk '$6 ~ /GET/ {print $8}' | \
    sort | uniq -c | sort -rn | \
    head -100 > top_100_hot_objects.txt

# 输出格式：访问次数 对象路径
# 示例：
#   154892 /images/product/banner.jpg
#    23451 /api/config.json
#      342 /reports/2024-Q1.pdf
#        1 /archives/2019-backup.tar.gz

# 基于时间的分层策略配置示例
tiering_policy:
  name: "default-time-based"
  rules:
    - name: "hot-to-warm"
      condition:
        age_days: 30
        source_tier: "hot"
      action:
        target_tier: "warm"

    - name: "warm-to-cold"
      condition:
        age_days: 90
        source_tier: "warm"
      action:
        target_tier: "cold"

    - name: "cold-to-archive"
      condition:
        age_days: 365
        source_tier: "cold"
      action:
        target_tier: "archive"

    - name: "archive-to-delete"
      condition:
        age_days: 2555    # 7 年，满足合规要求后销毁
        source_tier: "archive"
      action:
        type: "delete"
        secure_delete: true

# 基于访问频率的分层决策引擎
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional


@dataclass
class ObjectMetadata:
    key: str
    size_bytes: int
    created_at: datetime
    last_accessed: datetime
    current_tier: str
    access_count_7d: int
    access_count_30d: int
    access_count_90d: int


@dataclass
class TieringDecision:
    object_key: str
    current_tier: str
    target_tier: str
    reason: str


class FrequencyBasedTieringEngine:
    """基于访问频率的分层决策引擎。"""

    def __init__(self, config: dict):
        self.hot_threshold_7d = config.get("hot_threshold_7d", 100)
        self.warm_threshold_30d = config.get("warm_threshold_30d", 10)
        self.cold_threshold_90d = config.get("cold_threshold_90d", 1)
        self.min_age_for_demotion = timedelta(
            days=config.get("min_age_for_demotion_days", 7)
        )
        self.promotion_cooldown = timedelta(
            hours=config.get("promotion_cooldown_hours", 24)
        )

    def evaluate(self, obj: ObjectMetadata) -> Optional[TieringDecision]:
        """评估单个对象的分层决策。"""
        target = self._calculate_target_tier(obj)

        if target == obj.current_tier:
            return None

        if self._is_promotion(obj.current_tier, target):
            return TieringDecision(
                object_key=obj.key,
                current_tier=obj.current_tier,
                target_tier=target,
                reason=f"访问频率升高：7d={obj.access_count_7d}, "
                       f"30d={obj.access_count_30d}",
            )

        age = datetime.utcnow() - obj.created_at
        if age < self.min_age_for_demotion:
            return None

        return TieringDecision(
            object_key=obj.key,
            current_tier=obj.current_tier,
            target_tier=target,
            reason=f"访问频率下降：90d={obj.access_count_90d}",
        )

    def _calculate_target_tier(self, obj: ObjectMetadata) -> str:
        if obj.access_count_7d >= self.hot_threshold_7d:
            return "hot"
        elif obj.access_count_30d >= self.warm_threshold_30d:
            return "warm"
        elif obj.access_count_90d >= self.cold_threshold_90d:
            return "cold"
        else:
            return "archive"

    @staticmethod
    def _is_promotion(current: str, target: str) -> bool:
        tier_order = {"archive": 0, "cold": 1, "warm": 2, "hot": 3}
        return tier_order.get(target, 0) > tier_order.get(current, 0)

# 混合分层策略配置
tiering_policy:
  name: "hybrid-policy"
  rules:
    - name: "time-based-demotion"
      condition:
        age_days: 30
        source_tier: "hot"
        access_count_7d_less_than: 50     # 仅当访问频率确实下降时才降级
      action:
        target_tier: "warm"

    - name: "frequency-based-promotion"
      condition:
        source_tier: "cold"
        access_count_7d_greater_than: 100  # 冷层数据突然被频繁访问
      action:
        target_tier: "warm"
        cooldown_hours: 24                 # 避免频繁升降级

    - name: "force-demotion"
      condition:
        age_days: 180
        source_tier: "warm"               # 无论访问频率如何，180 天后强制降级
      action:
        target_tier: "cold"

    - name: "auto-archive"
      condition:
        age_days: 365
        source_tier: "cold"
      action:
        target_tier: "archive"

def should_migrate(obj: ObjectMetadata, target_tier: str) -> bool:
    """
    判断迁移是否划算。
    考虑迁移成本（API 调用 + 数据传输）与存储成本节省的对比。
    """
    # 各层每 GB 月成本（美元）
    tier_cost = {
        "hot": 0.023,
        "warm": 0.0125,
        "cold": 0.004,
        "archive": 0.00099,
    }

    # 迁移成本：PUT 请求费用 + 数据传输费用
    migration_cost_per_gb = 0.005  # 大约值

    size_gb = obj.size_bytes / (1024 ** 3)
    monthly_saving = (
        tier_cost[obj.current_tier] - tier_cost[target_tier]
    ) * size_gb
    migration_cost = migration_cost_per_gb * size_gb

    # 预计保留时间内的总节省必须大于迁移成本的 2 倍（安全系数）
    estimated_retention_months = 12
    total_saving = monthly_saving * estimated_retention_months
    return total_saving > migration_cost * 2

<!-- S3 生命周期配置示例 -->
<LifecycleConfiguration>
  <Rule>
    <ID>log-tiering</ID>
    <Filter>
      <Prefix>logs/</Prefix>
    </Filter>
    <Status>Enabled</Status>

    <!-- 30 天后转为低频访问 -->
    <Transition>
      <Days>30</Days>
      <StorageClass>STANDARD_IA</StorageClass>
    </Transition>

    <!-- 90 天后转为 Glacier Instant Retrieval -->
    <Transition>
      <Days>90</Days>
      <StorageClass>GLACIER_IR</StorageClass>
    </Transition>

    <!-- 180 天后转为 Glacier Flexible Retrieval -->
    <Transition>
      <Days>180</Days>
      <StorageClass>GLACIER</StorageClass>
    </Transition>

    <!-- 365 天后转为 Glacier Deep Archive -->
    <Transition>
      <Days>365</Days>
      <StorageClass>DEEP_ARCHIVE</StorageClass>
    </Transition>

    <!-- 2555 天（7 年）后删除 -->
    <Expiration>
      <Days>2555</Days>
    </Expiration>

    <!-- 删除未完成的分段上传（7 天后清理） -->
    <AbortIncompleteMultipartUpload>
      <DaysAfterInitiation>7</DaysAfterInitiation>
    </AbortIncompleteMultipartUpload>
  </Rule>

  <Rule>
    <ID>user-uploads-tiering</ID>
    <Filter>
      <And>
        <Prefix>uploads/</Prefix>
        <Tag>
          <Key>department</Key>
          <Value>marketing</Value>
        </Tag>
      </And>
    </Filter>
    <Status>Enabled</Status>

    <!-- 标记了 department=marketing 的上传文件，60 天后转冷存储 -->
    <Transition>
      <Days>60</Days>
      <StorageClass>STANDARD_IA</StorageClass>
    </Transition>

    <Expiration>
      <Days>730</Days>
    </Expiration>
  </Rule>
</LifecycleConfiguration>

# 应用生命周期配置
aws s3api put-bucket-lifecycle-configuration \
    --bucket my-data-bucket \
    --lifecycle-configuration file://lifecycle-config.json

# 查看当前生命周期配置
aws s3api get-bucket-lifecycle-configuration \
    --bucket my-data-bucket

# 生命周期配置的 JSON 格式
cat > lifecycle-config.json << 'EOF'
{
  "Rules": [
    {
      "ID": "log-tiering",
      "Status": "Enabled",
      "Filter": {
        "Prefix": "logs/"
      },
      "Transitions": [
        {"Days": 30, "StorageClass": "STANDARD_IA"},
        {"Days": 90, "StorageClass": "GLACIER_IR"},
        {"Days": 365, "StorageClass": "DEEP_ARCHIVE"}
      ],
      "Expiration": {"Days": 2555},
      "AbortIncompleteMultipartUpload": {
        "DaysAfterInitiation": 7
      }
    }
  ]
}
EOF

<!-- 阿里云 OSS 生命周期配置 -->
<LifecycleConfiguration>
  <Rule>
    <ID>log-tiering</ID>
    <Prefix>logs/</Prefix>
    <Status>Enabled</Status>

    <!-- 30 天后转为低频访问（IA） -->
    <Transition>
      <Days>30</Days>
      <StorageClass>IA</StorageClass>
    </Transition>

    <!-- 90 天后转为归档存储（Archive） -->
    <Transition>
      <Days>90</Days>
      <StorageClass>Archive</StorageClass>
    </Transition>

    <!-- 180 天后转为冷归档存储（ColdArchive） -->
    <Transition>
      <Days>180</Days>
      <StorageClass>ColdArchive</StorageClass>
    </Transition>

    <!-- 2555 天后删除 -->
    <Expiration>
      <Days>2555</Days>
    </Expiration>

    <!-- 清理未完成的分段上传 -->
    <AbortMultipartUpload>
      <Days>7</Days>
    </AbortMultipartUpload>
  </Rule>
</LifecycleConfiguration>

# 通过 ossutil 设置生命周期规则
ossutil lifecycle --method put oss://my-data-bucket lifecycle-config.xml

# 查看当前生命周期规则
ossutil lifecycle --method get oss://my-data-bucket

# 将对象上传为 Intelligent-Tiering 存储类
aws s3 cp large-dataset.tar.gz s3://my-bucket/datasets/ \
    --storage-class INTELLIGENT_TIERING

# 配置 Intelligent-Tiering 的归档访问层
aws s3api put-bucket-intelligent-tiering-configuration \
    --bucket my-bucket \
    --id "auto-archive-config" \
    --intelligent-tiering-configuration '{
        "Id": "auto-archive-config",
        "Status": "Enabled",
        "Tierings": [
            {
                "AccessTier": "ARCHIVE_ACCESS",
                "Days": 90
            },
            {
                "AccessTier": "DEEP_ARCHIVE_ACCESS",
                "Days": 180
            }
        ]
    }'

# 发起 Glacier 对象恢复请求

# 加急取回（Expedited）：1~5 分钟，费用最高
aws s3api restore-object \
    --bucket my-archive-bucket \
    --key "archives/2020-financial-report.tar.gz" \
    --restore-request '{
        "Days": 7,
        "GlacierJobParameters": {
            "Tier": "Expedited"
        }
    }'

# 标准取回（Standard）：3~5 小时
aws s3api restore-object \
    --bucket my-archive-bucket \
    --key "archives/2020-financial-report.tar.gz" \
    --restore-request '{
        "Days": 7,
        "GlacierJobParameters": {
            "Tier": "Standard"
        }
    }'

# 批量取回（Bulk）：5~12 小时，费用最低
aws s3api restore-object \
    --bucket my-archive-bucket \
    --key "archives/2020-financial-report.tar.gz" \
    --restore-request '{
        "Days": 7,
        "GlacierJobParameters": {
            "Tier": "Bulk"
        }
    }'

# 查看恢复状态
aws s3api head-object \
    --bucket my-archive-bucket \
    --key "archives/2020-financial-report.tar.gz" \
    --query "Restore"
# 输出示例：
# "ongoing-request=\"false\", expiry-date=\"Wed, 17 Jul 2024 00:00:00 GMT\""

# 加载磁带并挂载 LTFS
ltfs /mnt/tape -o devname=/dev/sg0

# 像普通文件系统一样读写
cp /data/archive/2020-reports.tar.gz /mnt/tape/
ls -la /mnt/tape/

# 卸载
umount /mnt/tape

#!/bin/bash
# archive_monthly.sh
# 将指定月份的数据打包归档到 S3 Glacier

YEAR="${1:?用法: $0 <年> <月>}"
MONTH="${2:?用法: $0 <年> <月>}"
DATA_DIR="/data/logs"
ARCHIVE_BUCKET="s3://archive-bucket/monthly"

ARCHIVE_NAME="archive-${YEAR}-${MONTH}.tar.gz"
INDEX_NAME="index-${YEAR}-${MONTH}.json"

echo "正在打包 ${YEAR}-${MONTH} 的数据..."

# 生成索引文件
find "${DATA_DIR}/${YEAR}/${MONTH}" -type f \
    -exec stat --printf='{"path":"%n","size":%s,"mtime":%Y}\n' {} \; | \
    jq -s '.' > "${INDEX_NAME}"

# 打包并压缩
tar czf "${ARCHIVE_NAME}" \
    -C "${DATA_DIR}" "${YEAR}/${MONTH}" \
    "${INDEX_NAME}"

# 计算校验和
sha256sum "${ARCHIVE_NAME}" > "${ARCHIVE_NAME}.sha256"

# 上传到 Glacier Deep Archive
aws s3 cp "${ARCHIVE_NAME}" \
    "${ARCHIVE_BUCKET}/${ARCHIVE_NAME}" \
    --storage-class DEEP_ARCHIVE \
    --metadata "sha256=$(cat ${ARCHIVE_NAME}.sha256 | awk '{print $1}')"

# 上传索引到标准存储（方便查询）
aws s3 cp "${INDEX_NAME}" \
    "${ARCHIVE_BUCKET}/indexes/${INDEX_NAME}" \
    --storage-class STANDARD

echo "归档完成：${ARCHIVE_NAME}"
echo "  大小：$(du -sh ${ARCHIVE_NAME} | awk '{print $1}')"
echo "  SHA256：$(cat ${ARCHIVE_NAME}.sha256 | awk '{print $1}')"

# 清理本地临时文件
rm -f "${ARCHIVE_NAME}" "${ARCHIVE_NAME}.sha256" "${INDEX_NAME}"

# 启用 S3 桶的对象锁定功能（必须在创建桶时启用）
aws s3api create-bucket \
    --bucket compliance-bucket \
    --object-lock-enabled-for-object-lock true

# 设置默认保留规则（合规模式，保留 7 年）
aws s3api put-object-lock-configuration \
    --bucket compliance-bucket \
    --object-lock-configuration '{
        "ObjectLockEnabled": "Enabled",
        "Rule": {
            "DefaultRetention": {
                "Mode": "COMPLIANCE",
                "Years": 7
            }
        }
    }'

# 合规模式（COMPLIANCE）：任何人（包括 root 账户）都不能在保留期内删除或修改对象
# 治理模式（GOVERNANCE）：拥有特定权限的用户可以覆盖保留期设置

# 为单个对象设置保留期
aws s3api put-object-retention \
    --bucket compliance-bucket \
    --key "audit-logs/2024-Q1.tar.gz" \
    --retention '{
        "Mode": "COMPLIANCE",
        "RetainUntilDate": "2031-04-01T00:00:00Z"
    }'

# 设置合法持有（Legal Hold）——诉讼期间冻结数据
aws s3api put-object-legal-hold \
    --bucket compliance-bucket \
    --key "financial/contract-2024.pdf" \
    --legal-hold '{"Status": "ON"}'

# GDPR 被遗忘权执行引擎（简化示意）
import logging
from typing import List
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class ErasureRequest:
    request_id: str
    user_id: str
    user_email: str
    requested_at: str
    deadline_days: int = 30      # GDPR 要求 30 天内完成


@dataclass
class DataLocation:
    storage_system: str          # "s3", "rds", "elasticsearch", "glacier"
    location: str                # 桶名/表名/索引名
    object_keys: List[str]       # 涉及的对象/记录


class GDPRErasureEngine:
    """GDPR 被遗忘权执行引擎。"""

    def __init__(self, data_catalog, storage_clients):
        self.catalog = data_catalog
        self.clients = storage_clients

    def execute_erasure(self, request: ErasureRequest) -> dict:
        """执行数据删除请求。"""
        logger.info(f"开始执行删除请求：{request.request_id}，"
                    f"用户：{request.user_id}")

        # 第一步：查找所有包含该用户数据的存储位置
        locations = self.catalog.find_user_data(request.user_id)
        logger.info(f"找到 {len(locations)} 个存储位置")

        results = {"success": [], "failed": [], "pending": []}

        for loc in locations:
            try:
                if loc.storage_system == "glacier":
                    # 归档数据无法立即删除，需要先恢复再删除
                    # 或者标记为待删除，在下次恢复时处理
                    self._schedule_glacier_deletion(loc, request)
                    results["pending"].append(loc.location)
                else:
                    self._delete_data(loc, request)
                    results["success"].append(loc.location)
            except Exception as e:
                logger.error(f"删除失败：{loc.location}，错误：{e}")
                results["failed"].append(loc.location)

        # 第二步：记录删除审计日志
        self._log_erasure_audit(request, results)

        return results

    def _delete_data(self, loc: DataLocation, req: ErasureRequest):
        """从指定存储位置删除用户数据。"""
        client = self.clients[loc.storage_system]
        for key in loc.object_keys:
            client.delete(location=loc.location, key=key)
            logger.info(f"已删除：{loc.storage_system}://"
                        f"{loc.location}/{key}")

    def _schedule_glacier_deletion(self, loc: DataLocation,
                                    req: ErasureRequest):
        """对归档数据，调度异步删除任务。"""
        logger.info(f"归档数据待删除：{loc.location}，"
                    f"将在下次恢复周期处理")

    def _log_erasure_audit(self, req: ErasureRequest, results: dict):
        """记录删除操作的审计日志，满足合规要求。"""
        logger.info(
            f"删除审计：request_id={req.request_id}, "
            f"user_id={req.user_id}, "
            f"success={len(results['success'])}, "
            f"failed={len(results['failed'])}, "
            f"pending={len(results['pending'])}"
        )

# HDD 安全擦除：使用 shred 命令覆写 3 次
shred -vfz -n 3 /dev/sdb

# SSD 安全擦除：使用 hdparm 发送 ATA Secure Erase 命令
# 警告：此操作不可逆，会擦除整块磁盘
hdparm --user-master u --security-set-pass password /dev/sdc
hdparm --user-master u --security-erase password /dev/sdc

# NVMe SSD 安全擦除：使用 nvme-cli
nvme format /dev/nvme0n1 --ses=1    # ses=1 表示用户数据擦除
nvme format /dev/nvme0n1 --ses=2    # ses=2 表示加密擦除（更安全）

# 对象存储：通过加密密钥销毁实现等效安全删除
# 如果对象使用了 SSE-KMS 加密，销毁 KMS 密钥即可使数据不可读
aws kms schedule-key-deletion \
    --key-id "arn:aws:kms:us-east-1:123456789:key/abc-123" \
    --pending-window-in-days 7

from dataclasses import dataclass, field
from typing import List


@dataclass
class TierConfig:
    name: str
    price_per_gb_month: float         # 每 GB 月存储费用（美元）
    put_price_per_1k: float           # 每千次 PUT 请求费用
    get_price_per_1k: float           # 每千次 GET 请求费用
    retrieval_per_gb: float = 0.0     # 每 GB 取回费用
    min_storage_days: int = 0         # 最短保留天数
    transition_price_per_1k: float = 0.0  # 每千次转换请求费用


@dataclass
class DataSegment:
    name: str
    size_tb: float
    monthly_get_requests: int
    monthly_put_requests: int
    monthly_retrieval_gb: float = 0.0


@dataclass
class CostReport:
    tier_name: str
    storage_cost: float
    request_cost: float
    retrieval_cost: float
    total_cost: float


def calculate_tier_cost(segment: DataSegment, tier: TierConfig) -> CostReport:
    """计算某个数据段在指定存储层的月成本。"""
    size_gb = segment.size_tb * 1024

    storage_cost = size_gb * tier.price_per_gb_month
    request_cost = (
        (segment.monthly_get_requests / 1000) * tier.get_price_per_1k +
        (segment.monthly_put_requests / 1000) * tier.put_price_per_1k
    )
    retrieval_cost = segment.monthly_retrieval_gb * tier.retrieval_per_gb

    total = storage_cost + request_cost + retrieval_cost
    return CostReport(
        tier_name=tier.name,
        storage_cost=round(storage_cost, 2),
        request_cost=round(request_cost, 2),
        retrieval_cost=round(retrieval_cost, 2),
        total_cost=round(total, 2),
    )


def compare_strategies(
    segments: List[DataSegment],
    tiers: List[TierConfig],
    assignment_before: dict,
    assignment_after: dict,
) -> dict:
    """
    对比分层前后的成本差异。
    assignment_before / assignment_after: {segment_name: tier_name}
    """
    tier_map = {t.name: t for t in tiers}

    cost_before = 0.0
    cost_after = 0.0
    details = []

    for seg in segments:
        tier_b = tier_map[assignment_before[seg.name]]
        tier_a = tier_map[assignment_after[seg.name]]

        report_b = calculate_tier_cost(seg, tier_b)
        report_a = calculate_tier_cost(seg, tier_a)

        cost_before += report_b.total_cost
        cost_after += report_a.total_cost

        details.append({
            "segment": seg.name,
            "before": report_b,
            "after": report_a,
            "monthly_saving": round(
                report_b.total_cost - report_a.total_cost, 2
            ),
        })

    return {
        "monthly_cost_before": round(cost_before, 2),
        "monthly_cost_after": round(cost_after, 2),
        "monthly_saving": round(cost_before - cost_after, 2),
        "annual_saving": round((cost_before - cost_after) * 12, 2),
        "saving_percentage": round(
            (cost_before - cost_after) / cost_before * 100, 1
        ),
        "details": details,
    }


# 使用示例
if __name__ == "__main__":
    tiers = [
        TierConfig("standard", 0.023, 0.005, 0.0004),
        TierConfig("standard_ia", 0.0125, 0.01, 0.001, retrieval_per_gb=0.01),
        TierConfig("glacier_ir", 0.004, 0.02, 0.01, retrieval_per_gb=0.03),
        TierConfig(
            "deep_archive", 0.00099, 0.05, 0.0004,
            retrieval_per_gb=0.02, min_storage_days=180,
        ),
    ]

    segments = [
        DataSegment("hot_0_30d", 50, 5_000_000, 500_000),
        DataSegment("warm_30_90d", 75, 100_000, 10_000),
        DataSegment("cold_90_365d", 125, 5_000, 1_000, 50),
        DataSegment("archive_365d_plus", 250, 500, 100, 10),
    ]

    before = {s.name: "standard" for s in segments}
    after = {
        "hot_0_30d": "standard",
        "warm_30_90d": "standard_ia",
        "cold_90_365d": "glacier_ir",
        "archive_365d_plus": "deep_archive",
    }

    result = compare_strategies(segments, tiers, before, after)
    print(f"分层前月成本：${result['monthly_cost_before']}")
    print(f"分层后月成本：${result['monthly_cost_after']}")
    print(f"月节省：${result['monthly_saving']}")
    print(f"年节省：${result['annual_saving']}")
    print(f"节省比例：{result['saving_percentage']}%")

# 策略引擎核心接口
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List


@dataclass
class ObjectInfo:
    key: str
    size_bytes: int
    created_at: datetime
    last_modified: datetime
    last_accessed: datetime
    current_tier: str
    access_stats: dict          # {"7d": 120, "30d": 450, "90d": 600}
    tags: dict                  # 用户自定义标签
    content_type: str
    is_locked: bool = False     # 是否被合规锁定


@dataclass
class TieringAction:
    object_key: str
    source_tier: str
    target_tier: str
    reason: str
    priority: int               # 优先级：1（最高）到 5（最低）
    scheduled_at: Optional[datetime] = None


class TieringPolicy(ABC):
    """分层策略的抽象基类。"""

    @abstractmethod
    def evaluate(self, obj: ObjectInfo) -> Optional[TieringAction]:
        """评估对象，返回分层动作或 None（不需要迁移）。"""
        pass


class CompositePolicy(TieringPolicy):
    """组合策略：按优先级依次评估多个子策略，取第一个匹配的结果。"""

    def __init__(self, policies: List[TieringPolicy]):
        self.policies = policies

    def evaluate(self, obj: ObjectInfo) -> Optional[TieringAction]:
        # 合规锁定的对象不允许降级
        if obj.is_locked:
            return None

        for policy in self.policies:
            action = policy.evaluate(obj)
            if action is not None:
                return action
        return None


class TimeBasedPolicy(TieringPolicy):
    """基于数据年龄的分层策略。"""

    def __init__(self, rules: List[dict]):
        self.rules = sorted(rules, key=lambda r: r["age_days"],
                            reverse=True)

    def evaluate(self, obj: ObjectInfo) -> Optional[TieringAction]:
        age = (datetime.utcnow() - obj.created_at).days

        for rule in self.rules:
            if age >= rule["age_days"]:
                target = rule["target_tier"]
                if target != obj.current_tier:
                    return TieringAction(
                        object_key=obj.key,
                        source_tier=obj.current_tier,
                        target_tier=target,
                        reason=f"数据年龄 {age} 天 >= 阈值 "
                               f"{rule['age_days']} 天",
                        priority=3,
                    )
                break
        return None


class FrequencyPolicy(TieringPolicy):
    """基于访问频率的分层策略。"""

    def __init__(self, thresholds: dict):
        self.thresholds = thresholds

    def evaluate(self, obj: ObjectInfo) -> Optional[TieringAction]:
        stats = obj.access_stats
        access_7d = stats.get("7d", 0)
        access_30d = stats.get("30d", 0)

        if access_7d >= self.thresholds.get("hot_7d", 100):
            target = "hot"
        elif access_30d >= self.thresholds.get("warm_30d", 10):
            target = "warm"
        elif access_30d < self.thresholds.get("cold_30d", 1):
            target = "cold"
        else:
            return None

        if target != obj.current_tier:
            return TieringAction(
                object_key=obj.key,
                source_tier=obj.current_tier,
                target_tier=target,
                reason=f"访问频率变化：7d={access_7d}, 30d={access_30d}",
                priority=2,
            )
        return None

import time
import logging
from typing import Iterator

logger = logging.getLogger(__name__)


class ScanScheduler:
    """
    分层扫描调度器。
    支持增量扫描和全量扫描两种模式。
    """

    def __init__(self, storage_adapter, policy_engine,
                 migration_queue, batch_size=1000):
        self.storage = storage_adapter
        self.policy = policy_engine
        self.queue = migration_queue
        self.batch_size = batch_size

    def run_incremental_scan(self, since: datetime):
        """
        增量扫描：仅扫描自上次扫描以来有变更的对象。
        适合高频运行（每小时或每天）。
        """
        logger.info(f"开始增量扫描，起始时间：{since}")
        scanned = 0
        actions = 0

        for batch in self._iter_modified_objects(since):
            for obj_info in batch:
                action = self.policy.evaluate(obj_info)
                if action is not None:
                    self.queue.enqueue(action)
                    actions += 1
                scanned += 1

        logger.info(f"增量扫描完成：扫描 {scanned} 个对象，"
                    f"生成 {actions} 个迁移任务")

    def run_full_scan(self, prefix: str = ""):
        """
        全量扫描：扫描指定前缀下的所有对象。
        适合低频运行（每周或每月）。
        """
        logger.info(f"开始全量扫描，前缀：{prefix or '/'}")
        scanned = 0
        actions = 0
        start_time = time.time()

        for batch in self._iter_all_objects(prefix):
            for obj_info in batch:
                action = self.policy.evaluate(obj_info)
                if action is not None:
                    self.queue.enqueue(action)
                    actions += 1
                scanned += 1

            if scanned % 100_000 == 0:
                elapsed = time.time() - start_time
                rate = scanned / elapsed
                logger.info(f"扫描进度：{scanned} 对象，"
                            f"速率 {rate:.0f} 对象/秒")

        elapsed = time.time() - start_time
        logger.info(
            f"全量扫描完成：扫描 {scanned} 个对象，"
            f"生成 {actions} 个迁移任务，"
            f"耗时 {elapsed:.1f} 秒"
        )

    def _iter_modified_objects(
        self, since: datetime
    ) -> Iterator[list]:
        """按批次迭代自指定时间以来有变更的对象。"""
        marker = ""
        while True:
            batch = self.storage.list_modified(
                since=since, marker=marker,
                max_keys=self.batch_size,
            )
            if not batch:
                break
            yield batch
            marker = batch[-1].key

    def _iter_all_objects(self, prefix: str) -> Iterator[list]:
        """按批次迭代指定前缀下的所有对象。"""
        marker = ""
        while True:
            batch = self.storage.list_objects(
                prefix=prefix, marker=marker,
                max_keys=self.batch_size,
            )
            if not batch:
                break
            yield batch
            marker = batch[-1].key

# Prometheus 监控指标定义
metrics:
  # 各层数据量
  - name: dlm_tier_size_bytes
    type: gauge
    labels: [tier, bucket]
    help: "各存储层的数据总量（字节）"

  # 各层对象数量
  - name: dlm_tier_object_count
    type: gauge
    labels: [tier, bucket]
    help: "各存储层的对象数量"

  # 迁移任务统计
  - name: dlm_migration_total
    type: counter
    labels: [source_tier, target_tier, status]
    help: "迁移任务累计数量"

  # 迁移数据量
  - name: dlm_migration_bytes_total
    type: counter
    labels: [source_tier, target_tier]
    help: "迁移数据累计字节数"

  # 扫描耗时
  - name: dlm_scan_duration_seconds
    type: histogram
    labels: [scan_type]
    help: "扫描任务耗时分布"

  # 成本估算
  - name: dlm_estimated_monthly_cost_dollars
    type: gauge
    labels: [tier, bucket]
    help: "各层预估月成本（美元）"

  # 合规保留状态
  - name: dlm_retention_locked_objects
    type: gauge
    labels: [bucket, retention_mode]
    help: "处于合规锁定状态的对象数量"