feat: add backup.retain_days config and local docker-compose for integration tests

- Add [backup] retain_days to config (default 1); --days becomes optional
- Add scripts/docker-compose.yml (azurite) for local integration testing
- Integration tests now use assets/local.toml and localhost endpoints
- Update .gitignore and CLAUDE.md accordingly

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.gitignore

@@ -1,6 +1,12 @@
 __pycache__/
 *.pyc
+.DS_Store
 .venv/
 *.egg-info/
 dist/
 build/
+backups/
+scan/
+dist-source/
+.worktrees/
+scripts/azurite/

CLAUDE.md

@@ -9,9 +9,70 @@ Order backup CLI tool — backs up order data by time range to CSV, then deletes
 ## Tech Stack
 
 - **Language**: Python
-- **Package management**: `uv` (install via `uv tool`)
-- **Storage**: Azure Table (Python SDK)
+- **Package management**: `uv`
+- **Storage**: Azure Table (`azure-data-tables` SDK)
+- **CLI framework**: Click
 
 ## Reference Implementation
 
 Order data model and format: `/Users/tech/workspace/n3-world/N3-Server/n3backend/n3order`
+
+## Commands
+
+```bash
+# Install dependencies for development
+uv sync --group dev
+
+# Run unit tests only (no Azure emulator required)
+uv run pytest tests/ -v -m "not integration"
+
+# Run a single test file
+uv run pytest tests/test_backup.py -v
+
+# Start Azure Storage emulator (required for integration tests)
+docker compose -f scripts/docker-compose.yml up -d
+
+# Run integration tests (requires emulator running via docker compose above)
+uv run pytest tests/test_integration.py -v
+
+# Seed test data into the emulator
+uv run python scripts/seed_test_data.py
+
+# Install as a CLI tool
+uv tool install .
+```
+
+## Architecture
+
+The tool implements a three-phase pipeline: **scan → backup → delete**.
+
+### Three-Phase Pipeline
+
+1. **`scan`** (`scan.py`): Full table scan of the `order` Azure Table, writing all entities to `scan/scan-<date>.csv`. Uses pagination with configurable page size and inter-page delay to limit production load.
+
+2. **`backup`** (`backup.py`): Reads the scan CSV offline (no Azure calls), filters to `state == 1` (`Recived`) orders with `OrderTime < cutoff`, and writes per-day CSVs to `backups/<YYYY-MM-DD>.csv`. The `--days` flag sets cutoff as `today_utc - N days`.
+
+3. **`delete`** (`delete.py`): Reads a backup CSV and deletes those records from Azure Table using batched transactions grouped by `PartitionKey`. Requires interactive confirmation before executing.
+
+### Data Flow
+
+- Azure Table `order` → scan CSV → backup CSVs → Azure Table deletion
+- Scan and backup are read-only with respect to Azure; only `delete` writes to production.
+- The `state` field is stored as an integer in Azure but legacy scan files may contain string values (`"Payed"` → 0, `"Recived"` → 1). `csv_utils.parse_row` handles both forms.
+
+### Key Modules
+
+- `cli.py`: Click command definitions; loads config and delegates to module functions
+- `config.py`: Reads `~/.config/order-bak/config.toml` (TOML via `tomllib`); only `[azure].connection_string` is required
+- `csv_utils.py`: Canonical column list (`CSV_COLUMNS`), `entity_to_row` / `parse_row` for serialization with type coercion
+- `scan.py`, `backup.py`, `delete.py`: One function each, thin and focused
+
+### Configuration
+
+Config file: `~/.config/order-bak/config.toml`. Template in `assets/config.toml`. For local development against the emulator, copy `assets/local.toml`.
+
+Override config path with `order-bak --config /path/to/config.toml <command>`.
+
+### Integration Tests
+
+Integration tests (`tests/test_integration.py`) require the Azure Storage emulator running locally via `scripts/docker-compose.yml` (azurite on `localhost:10002`). Start it with `docker compose -f scripts/docker-compose.yml up -d` before running. Tests are gated by the `integration` pytest marker. Override the endpoint via `ORDER_BAK_TEST_CONNECTION_STRING` env var if needed.

README.md

@@ -0,0 +1,85 @@
+# order-bak
+
+订单备份 CLI 工具。从 Azure Table 扫描订单数据，按时间范围备份到 CSV，并支持从生产表中删除已备份记录。
+
+## 安装
+
+```bash
+uv tool install .
+```
+
+## 配置
+
+创建 `~/.config/order-bak/config.toml`：
+
+```toml
+[azure]
+connection_string = "DefaultEndpointsProtocol=..."
+
+[scan]
+page_size = 500    # 每页行数
+delay_ms = 200     # 页间延迟 (ms)
+scan_dir = "./scan"
+
+[delete]
+batch_size = 100   # 每批删除行数
+delay_ms = 300     # 批间延迟 (ms)
+```
+
+仅 `[azure]` 为必填项，其余有默认值。
+
+## 使用
+
+### 1. 扫描全表
+
+```bash
+order-bak scan
+```
+
+全量扫描 `order` 表，结果保存到 `scan/scan-<date>.csv`。
+
+### 2. 按日期备份
+
+```bash
+order-bak backup --start 2025-01-01 --end 2025-06-30
+```
+
+从最新的 scan CSV 中筛选 `state == Recived` 且 `OrderTime` 在 `[start, end)` 范围内的订单，按日期生成独立 CSV 到 `backups/` 目录。
+
+可选指定 scan 文件：`--scan scan/scan-2025-01-01.csv`
+
+### 3. 删除已备份记录
+
+```bash
+order-bak delete --csv backups/2025-01-01.csv
+```
+
+读取 CSV 中的记录，按分区分批从 Azure Table 中删除。执行前需确认。
+
+## 目录结构
+
+```
+scan/
+└── scan-2025-06-01.csv    # 全表扫描结果
+
+backups/
+├── 2025-01-01.csv         # 按日期的备份
+├── 2025-01-02.csv
+└── ...
+```
+
+## 开发
+
+```bash
+# 安装依赖
+uv sync --group dev
+
+# 单元测试
+uv run pytest tests/ -v -m "not integration"
+
+# 集成测试（需要 Azure Storage Emulator）
+uv run pytest tests/test_integration.py -v
+
+# 初始化测试数据
+uv run python scripts/seed_test_data.py
+```

assets/config.toml

@@ -0,0 +1,17 @@
+# order-bak config template
+# Copy to ~/.config/order-bak/config.toml and fill in values
+
+[azure]
+connection_string = ""
+
+[scan]
+# page_size = 500
+# delay_ms = 200
+# scan_dir = "./scan"
+
+[backup]
+# retain_days = 1
+
+[delete]
+# batch_size = 100
+# delay_ms = 300

assets/local.toml

@@ -0,0 +1,15 @@
+
+[azure]
+connection_string = "AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;DefaultEndpointsProtocol=http;BlobEndpoint=http://localhost:10000/devstoreaccount1;QueueEndpoint=http://localhost:10001/devstoreaccount1;TableEndpoint=http://localhost:10002/devstoreaccount1;"
+
+[scan]
+page_size = 500
+delay_ms = 200
+scan_dir = "./scan"
+
+[backup]
+retain_days = 1
+
+[delete]
+batch_size = 100
+delay_ms = 300

assets/s101.toml

@@ -0,0 +1,15 @@
+
+[azure]
+connection_string = "AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;DefaultEndpointsProtocol=http;BlobEndpoint=http://192.168.9.101:10000/devstoreaccount1;QueueEndpoint=http://192.168.9.101:10001/devstoreaccount1;TableEndpoint=http://192.168.9.101:10002/devstoreaccount1;"
+
+[scan]
+page_size = 500
+delay_ms = 200
+scan_dir = "./scan"
+
+[backup]
+retain_days = 1
+
+[delete]
+batch_size = 100
+delay_ms = 300

assets/sandbox.toml

@@ -0,0 +1,15 @@
+
+[azure]
+connection_string = "DefaultEndpointsProtocol=https;AccountName=n3ordersmokestorage;AccountKey=aETnv18XBLUDVg7W1fu9ilnKchdLRy2sI2pJJyzqG6R/fqwcnIoSm7eM8ObyXFQAsvZrddZZTBN0+AStnOK/wg==;EndpointSuffix=core.windows.net;"
+
+[scan]
+page_size = 500
+delay_ms = 200
+scan_dir = "./scan"
+
+[backup]
+retain_days = 1
+
+[delete]
+batch_size = 100
+delay_ms = 300

docs/superpowers/plans/2026-04-21-order-bak.md

@@ -1,984 +0,0 @@
-# Order Backup CLI — Implementation Plan
-
-> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
-
-**Goal:** Build a CLI tool that scans all orders from Azure Table, backs them up to per-date CSVs by time range, and deletes backed-up records from production.
-
-**Architecture:** Three-phase pipeline — `scan` (Azure → local CSV), `backup` (local CSV → filtered per-date CSVs), `delete` (CSV → Azure batch delete). Each phase is a separate CLI command. Config via TOML file.
-
-**Tech Stack:** Python 3.11+, `uv`, `azure-data-tables`, `click` (CLI), `tomllib` (stdlib)
-
----
-
-## File Structure
-
-```
-order-bak/
-├── pyproject.toml
-├── src/
-│   └── order_bak/
-│       ├── __init__.py
-│       ├── cli.py          # Click CLI entry point (3 commands)
-│       ├── config.py       # Load ~/.config/order-bak/config.toml
-│       ├── csv_utils.py    # CSV column defs, write/read helpers
-│       ├── scan.py         # Scan command logic
-│       ├── backup.py       # Backup command logic
-│       └── delete.py       # Delete command logic
-└── tests/
-    ├── __init__.py
-    ├── test_config.py
-    ├── test_csv_utils.py
-    ├── test_backup.py
-    └── test_delete.py
-```
-
----
-
-### Task 1: Project Scaffold
-
-**Files:**
-- Create: `pyproject.toml`
-- Create: `src/order_bak/__init__.py`
-- Create: `tests/__init__.py`
-
-- [ ] **Step 1: Create pyproject.toml**
-
-```toml
-[project]
-name = "order-bak"
-version = "0.1.0"
-requires-python = ">=3.11"
-dependencies = [
-    "azure-data-tables>=12.5",
-    "click>=8.1",
-]
-
-[project.scripts]
-order-bak = "order_bak.cli:main"
-
-[build-system]
-requires = ["hatchling"]
-build-backend = "hatchling.build"
-
-[tool.hatch.build.targets.wheel]
-packages = ["src/order_bak"]
-
-[dependency-groups]
-dev = [
-    "pytest>=8.0",
-]
-```
-
-- [ ] **Step 2: Create package init and test init**
-
-`src/order_bak/__init__.py` — empty file.
-
-`tests/__init__.py` — empty file.
-
-- [ ] **Step 3: Install dependencies**
-
-Run: `uv sync --group dev`
-Expected: dependencies installed successfully
-
-- [ ] **Step 4: Verify entry point works**
-
-`src/order_bak/cli.py`:
-```python
-import click
-
-
-@click.group()
-def main():
-    pass
-
-
-if __name__ == "__main__":
-    main()
-```
-
-Run: `uv run order-bak`
-Expected: no error, shows help prompt
-
-- [ ] **Step 5: Commit**
-
-```bash
-git add pyproject.toml src/ tests/
-git commit -m "feat: project scaffold with uv, click, azure-data-tables"
-```
-
----
-
-### Task 2: Config Module
-
-**Files:**
-- Create: `src/order_bak/config.py`
-- Create: `tests/test_config.py`
-
-- [ ] **Step 1: Write failing test**
-
-```python
-# tests/test_config.py
-import tomllib
-from pathlib import Path
-from unittest.mock import patch
-
-from order_bak.config import load_config
-
-
-def test_load_config_defaults(tmp_path):
-    config_path = tmp_path / "config.toml"
-    config_path.write_text(tomllib.dumps({
-        "azure": {"connection_string": "conn_str_val"},
-    }))
-    cfg = load_config(config_path)
-    assert cfg.azure_connection_string == "conn_str_val"
-    assert cfg.scan_page_size == 500
-    assert cfg.scan_delay_ms == 200
-    assert cfg.scan_dir == Path("./scan")
-    assert cfg.delete_batch_size == 100
-    assert cfg.delete_delay_ms == 300
-
-
-def test_load_config_custom_values(tmp_path):
-    config_path = tmp_path / "config.toml"
-    config_path.write_text(tomllib.dumps({
-        "azure": {"connection_string": "conn_str_val"},
-        "scan": {"page_size": 1000, "delay_ms": 500, "scan_dir": "/data/scan"},
-        "delete": {"batch_size": 50, "delay_ms": 100},
-    }))
-    cfg = load_config(config_path)
-    assert cfg.scan_page_size == 1000
-    assert cfg.scan_delay_ms == 500
-    assert cfg.scan_dir == Path("/data/scan")
-    assert cfg.delete_batch_size == 50
-    assert cfg.delete_delay_ms == 100
-
-
-def test_load_config_missing_azure_raises(tmp_path):
-    config_path = tmp_path / "config.toml"
-    config_path.write_text("[scan]\npage_size = 100\n")
-    try:
-        load_config(config_path)
-        assert False, "should have raised"
-    except SystemExit:
-        pass
-```
-
-- [ ] **Step 2: Run test to verify it fails**
-
-Run: `uv run pytest tests/test_config.py -v`
-Expected: FAIL — `ModuleNotFoundError: No module named 'order_bak.config'`
-
-- [ ] **Step 3: Implement config**
-
-```python
-# src/order_bak/config.py
-import sys
-import tomllib
-from dataclasses import dataclass
-from pathlib import Path
-
-
-@dataclass
-class Config:
-    azure_connection_string: str
-    scan_page_size: int = 500
-    scan_delay_ms: int = 200
-    scan_dir: Path = Path("./scan")
-    delete_batch_size: int = 100
-    delete_delay_ms: int = 300
-
-
-def load_config(config_path: Path) -> Config:
-    if not config_path.exists():
-        print(f"Config file not found: {config_path}")
-        sys.exit(1)
-
-    with open(config_path, "rb") as f:
-        raw = tomllib.load(f)
-
-    azure = raw.get("azure", {})
-    connection_string = azure.get("connection_string")
-    if not connection_string:
-        print("Missing azure.connection_string in config")
-        sys.exit(1)
-
-    scan = raw.get("scan", {})
-    delete = raw.get("delete", {})
-
-    return Config(
-        azure_connection_string=connection_string,
-        scan_page_size=scan.get("page_size", 500),
-        scan_delay_ms=scan.get("delay_ms", 200),
-        scan_dir=Path(scan.get("scan_dir", "./scan")),
-        delete_batch_size=delete.get("batch_size", 100),
-        delete_delay_ms=delete.get("delay_ms", 300),
-    )
-```
-
-- [ ] **Step 4: Run test to verify it passes**
-
-Run: `uv run pytest tests/test_config.py -v`
-Expected: 3 passed
-
-- [ ] **Step 5: Commit**
-
-```bash
-git add src/order_bak/config.py tests/test_config.py
-git commit -m "feat: config module with TOML loading"
-```
-
----
-
-### Task 3: CSV Utilities
-
-**Files:**
-- Create: `src/order_bak/csv_utils.py`
-- Create: `tests/test_csv_utils.py`
-
-- [ ] **Step 1: Write failing test**
-
-```python
-# tests/test_csv_utils.py
-import csv
-from datetime import datetime, timezone
-from pathlib import Path
-
-from order_bak.csv_utils import (
-    CSV_COLUMNS,
-    entity_to_row,
-    row_to_entity,
-    write_csv,
-    read_csv,
-)
-
-
-def test_csv_columns_defined():
-    assert "PartitionKey" in CSV_COLUMNS
-    assert "RowKey" in CSV_COLUMNS
-    assert "OrderTime" in CSV_COLUMNS
-    assert "state" in CSV_COLUMNS
-    assert len(CSV_COLUMNS) == 13
-
-
-def test_entity_to_row_converts_datetime():
-    entity = {
-        "PartitionKey": "user1",
-        "RowKey": "order1",
-        "OrderTime": datetime(2025, 6, 15, 10, 30, 0, tzinfo=timezone.utc),
-        "ReceiveTime": datetime(2025, 6, 15, 11, 0, 0, tzinfo=timezone.utc),
-        "revenue": 99.5,
-        "platformOrder": "plat123",
-        "prodCount": 2,
-        "pfId": "user1",
-        "prodId": "prod1",
-        "prodPrice": 49.75,
-        "state": 1,
-        "ver": 3,
-        "customData": '{"key":"val"}',
-    }
-    row = entity_to_row(entity)
-    assert row["OrderTime"] == "2025-06-15T10:30:00+00:00"
-    assert row["ReceiveTime"] == "2025-06-15T11:00:00+00:00"
-    assert row["revenue"] == "99.5"
-    assert row["state"] == "1"
-
-
-def test_entity_to_row_handles_none():
-    entity = {
-        "PartitionKey": "user1",
-        "RowKey": "order1",
-        "OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
-        "ReceiveTime": None,
-        "revenue": None,
-        "platformOrder": None,
-        "prodCount": None,
-        "pfId": None,
-        "prodId": None,
-        "prodPrice": None,
-        "state": 0,
-        "ver": None,
-        "customData": None,
-    }
-    row = entity_to_row(entity)
-    assert row["ReceiveTime"] == ""
-    assert row["revenue"] == ""
-
-
-def test_row_to_entity_roundtrip(tmp_path):
-    original = {
-        "PartitionKey": "user1",
-        "RowKey": "order1",
-        "OrderTime": "2025-06-15T10:30:00+00:00",
-        "ReceiveTime": "",
-        "revenue": "99.5",
-        "platformOrder": "plat123",
-        "prodCount": "2",
-        "pfId": "user1",
-        "prodId": "prod1",
-        "prodPrice": "49.75",
-        "state": "1",
-        "ver": "3",
-        "customData": '{"key":"val"}',
-    }
-    entity = row_to_entity(original)
-    assert entity["PartitionKey"] == "user1"
-    assert entity["state"] == 1
-    assert entity["revenue"] == 99.5
-    assert entity["prodCount"] == 2
-    assert entity["ReceiveTime"] is None
-
-
-def test_write_and_read_csv(tmp_path):
-    entities = [
-        {
-            "PartitionKey": "user1",
-            "RowKey": "order1",
-            "OrderTime": datetime(2025, 6, 15, 10, 30, tzinfo=timezone.utc),
-            "ReceiveTime": None,
-            "revenue": 99.5,
-            "platformOrder": "p1",
-            "prodCount": 2,
-            "pfId": "user1",
-            "prodId": "prod1",
-            "prodPrice": 49.75,
-            "state": 1,
-            "ver": 3,
-            "customData": None,
-        },
-        {
-            "PartitionKey": "user2",
-            "RowKey": "order2",
-            "OrderTime": datetime(2025, 6, 16, 12, 0, tzinfo=timezone.utc),
-            "ReceiveTime": datetime(2025, 6, 16, 13, 0, tzinfo=timezone.utc),
-            "revenue": 10.0,
-            "platformOrder": "p2",
-            "prodCount": 1,
-            "pfId": "user2",
-            "prodId": "prod2",
-            "prodPrice": 10.0,
-            "state": 1,
-            "ver": 3,
-            "customData": '{"note":"test"}',
-        },
-    ]
-    csv_path = tmp_path / "test.csv"
-    write_csv(csv_path, entities)
-
-    result = list(read_csv(csv_path))
-    assert len(result) == 2
-    assert result[0]["PartitionKey"] == "user1"
-    assert result[1]["PartitionKey"] == "user2"
-    assert result[1]["revenue"] == 10.0
-    assert result[0]["ReceiveTime"] is None
-    assert result[1]["ReceiveTime"] is not None
-```
-
-- [ ] **Step 2: Run test to verify it fails**
-
-Run: `uv run pytest tests/test_csv_utils.py -v`
-Expected: FAIL — `ModuleNotFoundError: No module named 'order_bak.csv_utils'`
-
-- [ ] **Step 3: Implement csv_utils**
-
-```python
-# src/order_bak/csv_utils.py
-import csv
-from datetime import datetime
-from pathlib import Path
-from typing import Any
-
-CSV_COLUMNS = [
-    "PartitionKey",
-    "RowKey",
-    "OrderTime",
-    "ReceiveTime",
-    "revenue",
-    "platformOrder",
-    "prodCount",
-    "pfId",
-    "prodId",
-    "prodPrice",
-    "state",
-    "ver",
-    "customData",
-]
-
-
-def _format_value(val: Any) -> str:
-    if val is None:
-        return ""
-    if isinstance(val, datetime):
-        return val.isoformat()
-    return str(val)
-
-
-def entity_to_row(entity: dict[str, Any]) -> dict[str, str]:
-    return {col: _format_value(entity.get(col)) for col in CSV_COLUMNS}
-
-
-def _parse_row(raw: dict[str, str]) -> dict[str, Any]:
-    result: dict[str, Any] = {}
-    for col in CSV_COLUMNS:
-        val = raw.get(col, "")
-        if val == "":
-            result[col] = None
-            continue
-        if col in ("OrderTime", "ReceiveTime"):
-            result[col] = datetime.fromisoformat(val)
-        elif col in ("revenue", "prodPrice"):
-            result[col] = float(val)
-        elif col in ("prodCount", "state", "ver"):
-            result[col] = int(val)
-        else:
-            result[col] = val
-    return result
-
-
-def row_to_entity(raw: dict[str, str]) -> dict[str, Any]:
-    return _parse_row(raw)
-
-
-def write_csv(path: Path, entities: list[dict[str, Any]]) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    with open(path, "w", newline="") as f:
-        writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
-        writer.writeheader()
-        for entity in entities:
-            writer.writerow(entity_to_row(entity))
-
-
-def read_csv(path: Path):
-    with open(path, newline="") as f:
-        reader = csv.DictReader(f)
-        for row in reader:
-            yield _parse_row(row)
-```
-
-- [ ] **Step 4: Run test to verify it passes**
-
-Run: `uv run pytest tests/test_csv_utils.py -v`
-Expected: 5 passed
-
-- [ ] **Step 5: Commit**
-
-```bash
-git add src/order_bak/csv_utils.py tests/test_csv_utils.py
-git commit -m "feat: CSV utilities for order entity serialization"
-```
-
----
-
-### Task 4: Scan Command
-
-**Files:**
-- Create: `src/order_bak/scan.py`
-- Modify: `src/order_bak/cli.py`
-
-- [ ] **Step 1: Implement scan logic**
-
-```python
-# src/order_bak/scan.py
-import time
-from datetime import datetime, timezone
-from pathlib import Path
-
-from azure.data.tables import TableClient
-
-from order_bak.csv_utils import CSV_COLUMNS, write_csv
-
-
-def scan_orders(
-    connection_string: str,
-    page_size: int,
-    delay_ms: int,
-    scan_dir: Path,
-) -> Path:
-    table = TableClient.from_connection_string(connection_string, table_name="order")
-    scan_dir.mkdir(parents=True, exist_ok=True)
-
-    now = datetime.now(timezone.utc)
-    csv_path = scan_dir / f"scan-{now.strftime('%Y-%m-%d')}.csv"
-
-    paged = table.list_entities(results_per_page=page_size)
-
-    total_scanned = 0
-    csv_path.parent.mkdir(parents=True, exist_ok=True)
-
-    with open(csv_path, "w", newline="") as f:
-        import csv
-        writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
-        writer.writeheader()
-
-        for page in paged.by_page():
-            for entity in page:
-                writer.writerow({col: _fmt(entity.get(col)) for col in CSV_COLUMNS})
-                total_scanned += 1
-
-            if delay_ms > 0:
-                time.sleep(delay_ms / 1000.0)
-
-    print(f"Scan complete: {total_scanned} orders saved to {csv_path}")
-    return csv_path
-
-
-def _fmt(val):
-    if val is None:
-        return ""
-    if isinstance(val, datetime):
-        return val.isoformat()
-    return str(val)
-```
-
-- [ ] **Step 2: Wire scan command into CLI**
-
-```python
-# src/order_bak/cli.py
-from pathlib import Path
-
-import click
-
-from order_bak.config import load_config
-
-
-@click.group()
-@click.option("--config", "-c", default="~/.config/order-bak/config.toml", type=click.Path())
-@click.pass_context
-def main(ctx, config):
-    ctx.ensure_object(dict)
-    ctx.obj["config_path"] = Path(config).expanduser()
-
-
-@main.command()
-@click.pass_context
-def scan(ctx):
-    cfg = load_config(ctx.obj["config_path"])
-    from order_bak.scan import scan_orders
-    scan_orders(
-        connection_string=cfg.azure_connection_string,
-        page_size=cfg.scan_page_size,
-        delay_ms=cfg.scan_delay_ms,
-        scan_dir=cfg.scan_dir,
-    )
-
-
-if __name__ == "__main__":
-    main()
-```
-
-- [ ] **Step 3: Verify CLI help works**
-
-Run: `uv run order-bak --help`
-Expected: shows `scan` command
-
-Run: `uv run order-bak scan --help`
-Expected: shows scan help
-
-- [ ] **Step 4: Commit**
-
-```bash
-git add src/order_bak/scan.py src/order_bak/cli.py
-git commit -m "feat: scan command — full table scan to CSV with throttling"
-```
-
----
-
-### Task 5: Backup Command
-
-**Files:**
-- Create: `src/order_bak/backup.py`
-- Create: `tests/test_backup.py`
-- Modify: `src/order_bak/cli.py`
-
-- [ ] **Step 1: Write failing test**
-
-```python
-# tests/test_backup.py
-from datetime import datetime, timezone
-from pathlib import Path
-
-from order_bak.csv_utils import write_csv
-from order_bak.backup import backup_orders
-
-
-def _make_entity(pk, rk, order_time, state, receive_time=None):
-    return {
-        "PartitionKey": pk,
-        "RowKey": rk,
-        "OrderTime": order_time,
-        "ReceiveTime": receive_time,
-        "revenue": 10.0,
-        "platformOrder": "plat1",
-        "prodCount": 1,
-        "pfId": pk,
-        "prodId": "prod1",
-        "prodPrice": 10.0,
-        "state": state,
-        "ver": 3,
-        "customData": None,
-    }
-
-
-def test_backup_filters_by_state_and_date(tmp_path):
-    scan_dir = tmp_path / "scan"
-    backup_dir = tmp_path / "backups"
-    scan_dir.mkdir()
-    backup_dir.mkdir()
-
-    entities = [
-        _make_entity("u1", "o1", datetime(2025, 1, 1, 10, tzinfo=timezone.utc), 1),
-        _make_entity("u2", "o2", datetime(2025, 1, 1, 12, tzinfo=timezone.utc), 0),  # Payed, skip
-        _make_entity("u3", "o3", datetime(2025, 1, 2, 8, tzinfo=timezone.utc), 1),
-        _make_entity("u4", "o4", datetime(2025, 1, 3, 8, tzinfo=timezone.utc), 1),  # outside range
-    ]
-    scan_csv = scan_dir / "scan-2025-01-03.csv"
-    write_csv(scan_csv, entities)
-
-    result = backup_orders(
-        scan_path=scan_csv,
-        backup_dir=backup_dir,
-        start=datetime(2025, 1, 1, tzinfo=timezone.utc),
-        end=datetime(2025, 1, 3, tzinfo=timezone.utc),
-    )
-
-    assert result == 2
-    jan1 = backup_dir / "2025-01-01.csv"
-    jan2 = backup_dir / "2025-01-02.csv"
-    assert jan1.exists()
-    assert jan2.exists()
-    assert not (backup_dir / "2025-01-03.csv").exists()
-
-    from order_bak.csv_utils import read_csv
-    jan1_rows = list(read_csv(jan1))
-    assert len(jan1_rows) == 1
-    assert jan1_rows[0]["RowKey"] == "o1"
-```
-
-- [ ] **Step 2: Run test to verify it fails**
-
-Run: `uv run pytest tests/test_backup.py -v`
-Expected: FAIL — `ModuleNotFoundError: No module named 'order_bak.backup'`
-
-- [ ] **Step 3: Implement backup**
-
-```python
-# src/order_bak/backup.py
-import csv
-from collections import defaultdict
-from datetime import datetime
-from pathlib import Path
-from typing import Iterator
-
-from order_bak.csv_utils import CSV_COLUMNS, _parse_row
-
-
-def backup_orders(
-    scan_path: Path,
-    backup_dir: Path,
-    start: datetime,
-    end: datetime,
-) -> int:
-    backup_dir.mkdir(parents=True, exist_ok=True)
-
-    by_date: dict[str, list[dict]] = defaultdict(list)
-
-    with open(scan_path, newline="") as f:
-        reader = csv.DictReader(f)
-        for raw_row in reader:
-            entity = _parse_row(raw_row)
-            if entity["state"] != 1:
-                continue
-            order_time = entity["OrderTime"]
-            if order_time is None:
-                continue
-            if order_time < start or order_time >= end:
-                continue
-            date_key = order_time.strftime("%Y-%m-%d")
-            by_date[date_key].append(raw_row)
-
-    total = 0
-    for date_key, rows in sorted(by_date.items()):
-        csv_path = backup_dir / f"{date_key}.csv"
-        with open(csv_path, "w", newline="") as f:
-            writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
-            writer.writeheader()
-            writer.writerows(rows)
-        total += len(rows)
-        print(f"  {date_key}: {len(rows)} orders")
-
-    print(f"Backup complete: {total} orders across {len(by_date)} days")
-    return total
-```
-
-- [ ] **Step 4: Run test to verify it passes**
-
-Run: `uv run pytest tests/test_backup.py -v`
-Expected: 1 passed
-
-- [ ] **Step 5: Wire backup command into CLI**
-
-Add to `src/order_bak/cli.py`:
-
-```python
-@main.command()
-@click.option("--start", required=True, type=click.DateTime(formats=["%Y-%m-%d"]), help="Start date (UTC, inclusive)")
-@click.option("--end", required=True, type=click.DateTime(formats=["%Y-%m-%d"]), help="End date (UTC, exclusive)")
-@click.option("--scan", "scan_path", default=None, type=click.Path(), help="Path to scan CSV (default: latest in scan_dir)")
-@click.pass_context
-def backup(ctx, start, end, scan_path):
-    cfg = load_config(ctx.obj["config_path"])
-    from order_bak.backup import backup_orders
-    import pathlib
-
-    if scan_path:
-        scan_file = pathlib.Path(scan_path)
-    else:
-        import glob
-        scan_files = sorted(cfg.scan_dir.glob("scan-*.csv"))
-        if not scan_files:
-            click.echo("No scan files found in scan_dir")
-            raise SystemExit(1)
-        scan_file = scan_files[-1]
-
-    start_utc = start.replace(tzinfo=__import__("datetime").timezone.utc)
-    end_utc = end.replace(tzinfo=__import__("datetime").timezone.utc)
-
-    backup_orders(
-        scan_path=scan_file,
-        backup_dir=pathlib.Path("./backups"),
-        start=start_utc,
-        end=end_utc,
-    )
-```
-
-- [ ] **Step 6: Verify CLI help**
-
-Run: `uv run order-bak backup --help`
-Expected: shows --start, --end, --scan options
-
-- [ ] **Step 7: Commit**
-
-```bash
-git add src/order_bak/backup.py tests/test_backup.py src/order_bak/cli.py
-git commit -m "feat: backup command — filter scan CSV by state and date range"
-```
-
----
-
-### Task 6: Delete Command
-
-**Files:**
-- Create: `src/order_bak/delete.py`
-- Create: `tests/test_delete.py`
-- Modify: `src/order_bak/cli.py`
-
-- [ ] **Step 1: Write failing test**
-
-```python
-# tests/test_delete.py
-from collections import defaultdict
-from unittest.mock import MagicMock, call
-
-from order_bak.delete import delete_orders
-
-
-def test_delete_groups_by_partition_and_batches(tmp_path):
-    # Create a CSV with entities across different partitions
-    from order_bak.csv_utils import write_csv
-    from datetime import datetime, timezone
-
-    entities = []
-    # 3 entities in partition "user1", 2 in "user2"
-    for i in range(3):
-        entities.append({
-            "PartitionKey": "user1",
-            "RowKey": f"order{i}",
-            "OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
-            "ReceiveTime": None, "revenue": 10.0, "platformOrder": "p",
-            "prodCount": 1, "pfId": "user1", "prodId": "prod",
-            "prodPrice": 10.0, "state": 1, "ver": 3, "customData": None,
-        })
-    for i in range(3, 5):
-        entities.append({
-            "PartitionKey": "user2",
-            "RowKey": f"order{i}",
-            "OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
-            "ReceiveTime": None, "revenue": 10.0, "platformOrder": "p",
-            "prodCount": 1, "pfId": "user2", "prodId": "prod",
-            "prodPrice": 10.0, "state": 1, "ver": 3, "customData": None,
-        })
-
-    csv_path = tmp_path / "2025-01-01.csv"
-    write_csv(csv_path, entities)
-
-    mock_table = MagicMock()
-
-    deleted_count = delete_orders(
-        table_client=mock_table,
-        csv_path=csv_path,
-        batch_size=2,
-        delay_ms=0,
-    )
-
-    assert deleted_count == 5
-    # user1 has 3 entities, batch_size=2 → 2 batches (2 + 1)
-    # user2 has 2 entities, batch_size=2 → 1 batch
-    assert mock_table.submit_transaction.call_count == 3
-
-
-def test_delete_respects_batch_size(tmp_path):
-    from order_bak.csv_utils import write_csv
-    from datetime import datetime, timezone
-
-    # 5 entities in same partition, batch_size=2
-    entities = [
-        {
-            "PartitionKey": "user1",
-            "RowKey": f"order{i}",
-            "OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
-            "ReceiveTime": None, "revenue": 10.0, "platformOrder": "p",
-            "prodCount": 1, "pfId": "user1", "prodId": "prod",
-            "prodPrice": 10.0, "state": 1, "ver": 3, "customData": None,
-        }
-        for i in range(5)
-    ]
-
-    csv_path = tmp_path / "2025-01-01.csv"
-    write_csv(csv_path, entities)
-
-    mock_table = MagicMock()
-
-    deleted_count = delete_orders(
-        table_client=mock_table,
-        csv_path=csv_path,
-        batch_size=2,
-        delay_ms=0,
-    )
-
-    assert deleted_count == 5
-    assert mock_table.submit_transaction.call_count == 3  # 2+2+1
-```
-
-- [ ] **Step 2: Run test to verify it fails**
-
-Run: `uv run pytest tests/test_delete.py -v`
-Expected: FAIL — `ModuleNotFoundError: No module named 'order_bak.delete'`
-
-- [ ] **Step 3: Implement delete**
-
-```python
-# src/order_bak/delete.py
-import csv
-import time
-from collections import defaultdict
-from itertools import islice
-from pathlib import Path
-
-
-def _chunked(iterable, size):
-    it = iter(iterable)
-    while chunk := list(islice(it, size)):
-        yield chunk
-
-
-def delete_orders(
-    table_client,
-    csv_path: Path,
-    batch_size: int,
-    delay_ms: int,
-) -> int:
-    by_partition: dict[str, list[dict]] = defaultdict(list)
-
-    with open(csv_path, newline="") as f:
-        reader = csv.DictReader(f)
-        for row in reader:
-            pk = row["PartitionKey"]
-            rk = row["RowKey"]
-            by_partition[pk].append({"PartitionKey": pk, "RowKey": rk})
-
-    total_deleted = 0
-    for pk, entities in by_partition.items():
-        for batch in _chunked(entities, batch_size):
-            operations = [("delete", e) for e in batch]
-            table_client.submit_transaction(operations)
-            total_deleted += len(batch)
-            if delay_ms > 0:
-                time.sleep(delay_ms / 1000.0)
-
-    print(f"Deleted {total_deleted} orders across {len(by_partition)} partitions")
-    return total_deleted
-```
-
-- [ ] **Step 4: Run test to verify it passes**
-
-Run: `uv run pytest tests/test_delete.py -v`
-Expected: 2 passed
-
-- [ ] **Step 5: Wire delete command into CLI**
-
-Add to `src/order_bak/cli.py`:
-
-```python
-@main.command()
-@click.option("--csv", "csv_path", required=True, type=click.Path(exists=True), help="Path to backup CSV to delete")
-@click.pass_context
-def delete(ctx, csv_path):
-    cfg = load_config(ctx.obj["config_path"])
-    from azure.data.tables import TableClient
-    from order_bak.delete import delete_orders
-    import pathlib
-
-    csv_file = pathlib.Path(csv_path)
-
-    # Count records first
-    import csv as csv_mod
-    with open(csv_file, newline="") as f:
-        count = sum(1 for _ in csv_mod.DictReader(f))
-
-    click.echo(f"Will delete {count} orders from {csv_file}")
-    if not click.confirm("Continue?"):
-        click.echo("Aborted")
-        return
-
-    table = TableClient.from_connection_string(cfg.azure_connection_string, table_name="order")
-    delete_orders(
-        table_client=table,
-        csv_path=csv_file,
-        batch_size=cfg.delete_batch_size,
-        delay_ms=cfg.delete_delay_ms,
-    )
-```
-
-- [ ] **Step 6: Verify CLI help**
-
-Run: `uv run order-bak delete --help`
-Expected: shows --csv option
-
-- [ ] **Step 7: Commit**
-
-```bash
-git add src/order_bak/delete.py tests/test_delete.py src/order_bak/cli.py
-git commit -m "feat: delete command — batch delete from CSV with confirmation"
-```
-
----
-
-### Task 7: Run All Tests
-
-- [ ] **Step 1: Run full test suite**
-
-Run: `uv run pytest tests/ -v`
-Expected: all tests pass
-
-- [ ] **Step 2: Run lint check**
-
-Run: `uv run python -m py_compile src/order_bak/cli.py`
-Expected: no errors
-
----
-
-## Self-Review Checklist
-
-- **Spec coverage:** scan (Task 4), backup (Task 5), delete (Task 6), config (Task 2), CSV format (Task 3) — all covered
-- **No placeholders:** all code blocks contain complete implementations
-- **Type consistency:** entity dict keys match across csv_utils, backup, delete — all use `CSV_COLUMNS`

docs/superpowers/specs/2026-04-21-order-bak-design.md

@@ -1,103 +0,0 @@
-# Order Backup CLI — Design Spec
-
-## Overview
-
-CLI tool to back up completed (Recived) orders from production Azure Table by time range to CSV files, then optionally delete backed-up records. Operates on production storage, so all operations are throttled to minimize performance impact.
-
-## Scale
-
-- ~12,000 orders/day
-- ~6.5M total rows (1.5 years)
-- Backup targets: Recived orders only (state=1)
-
-## Configuration
-
-File: `~/.config/order-bak/config.toml`
-
-```toml
-[azure]
-connection_string = "DefaultEndpointsProtocol=..."
-
-[scan]
-page_size = 500
-delay_ms = 200
-scan_dir = "./scan"
-
-[delete]
-batch_size = 100
-delay_ms = 300
-```
-
-## Commands
-
-### `scan`
-
-```
-order-bak scan
-```
-
-Full table scan with continuation token pagination. No server-side filter — save all orders. Save results to `scan_dir/scan-<date>.csv`.
-
-- Single-threaded, sequential with configurable page size and delay
-- Default: 500 rows/page, 200ms inter-page delay
-- Estimated time: ~43 minutes for 6.5M rows
-
-### `backup`
-
-```
-order-bak backup --start 2025-01-01 --end 2025-06-30
-```
-
-Read from local scan CSV file (default: latest `scan-*.csv` in `scan_dir`, overridable with `--scan <path>`), filter by `state == Recived` and `OrderTime` in `[start, end)`, output one CSV per date under `backups/`.
-
-- Pure local operation, no Azure access
-- Print summary after completion: total matched, per-day distribution
-
-### `delete`
-
-```
-order-bak delete --csv backups/2025-01-01.csv
-```
-
-Read PartitionKey and RowKey from CSV. Group by PartitionKey, batch delete using Azure Table Transactions (max 100 per transaction, same partition). Print record count and require user confirmation before deleting.
-
-## Output Directory Structure
-
-```
-scan/
-└── scan-2026-04-21.csv        # scan output (all orders)
-
-backups/
-├── 2025-01-01.csv             # backup output (per-date CSV)
-├── 2025-01-02.csv
-└── ...
-```
-
-All CSV files share the same columns: `PartitionKey, RowKey, OrderTime, ReceiveTime, revenue, platformOrder, prodCount, pfId, prodId, prodPrice, state, ver, customData`
-
-## Azure Table Schema (reference)
-
-Table name: `order`
-
-| Field | Type | Notes |
-|---|---|---|
-| PartitionKey | string | User ID (pfId) |
-| RowKey | string | Order ID |
-| OrderTime | DateTime | Order creation time (UTC) |
-| ReceiveTime | DateTime | Confirmation time (UTC) |
-| revenue | double | Charged RMB amount |
-| platformOrder | string | Platform order ID |
-| prodCount | int | Product count |
-| pfId | string | Platform ID |
-| prodId | string | Product ID |
-| prodPrice | double | Product price |
-| state | byte | 0=Payed, 1=Recived |
-| ver | int | Schema version |
-| customData | string | JSON blob |
-
-## Tech Stack
-
-- Python, managed with `uv` (installable via `uv tool`)
-- Azure Data Tables SDK (`azure-data-tables`)
-- TOML config via `tomllib` (stdlib in Python 3.11+)
-- CSV output via `csv` stdlib module

pyproject.toml

@@ -21,3 +21,8 @@ packages = ["src/order_bak"]
 dev = [
     "pytest>=8.0",
 ]
+
+[tool.pytest.ini_options]
+markers = [
+    "integration: integration tests requiring Azure Storage emulator",
+]

scripts/docker-compose.yml

@@ -0,0 +1,9 @@
+services:
+  azure:
+    image: mcr.microsoft.com/azure-storage/azurite
+    ports:
+      - 10000:10000
+      - 10001:10001
+      - 10002:10002
+    volumes:
+      - ./azurite:/data

scripts/export_source.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+# Export project source code to dist-source/ directory.
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+PROJECT_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
+DIST_DIR="$PROJECT_DIR/dist-source"
+
+rm -rf "$DIST_DIR"
+mkdir -p "$DIST_DIR"
+
+# Copy project source and config files
+cp "$PROJECT_DIR/pyproject.toml" "$DIST_DIR/"
+cp "$PROJECT_DIR/uv.lock" "$DIST_DIR/"
+cp "$PROJECT_DIR/README.md" "$DIST_DIR/"
+cp "$PROJECT_DIR/CLAUDE.md" "$DIST_DIR/" 2>/dev/null || true
+
+# Copy directories
+for dir in src scripts docs; do
+    if [ -d "$PROJECT_DIR/$dir" ]; then
+        cp -r "$PROJECT_DIR/$dir" "$DIST_DIR/"
+    fi
+done
+
+# Remove __pycache__ from exported source
+find "$DIST_DIR" -type d -name "__pycache__" -exec rm -rf {} + 2>/dev/null || true
+find "$DIST_DIR" -name ".DS_Store" -delete 2>/dev/null || true
+
+echo "Exported to $DIST_DIR/"

scripts/seed_test_data.py

@@ -0,0 +1,115 @@
+"""Seed test data into Azure Table emulator for integration testing."""
+
+import random
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from azure.data.tables import TableServiceClient
+
+from order_bak.config import load_config
+
+TABLE_NAME = "order"
+
+USER_COUNT = 1000
+ENTRIES_PER_USER_PER_DAY = 2
+PAYED_PER_USER_PER_DAY = 1
+DATE_START = datetime(2026, 1, 1, tzinfo=timezone.utc)
+DATE_END = datetime.now(timezone.utc)
+
+
+def seed(config_path: str) -> None:
+    cfg = load_config(Path(config_path))
+    service = TableServiceClient.from_connection_string(cfg.azure_connection_string)
+    service.create_table_if_not_exists(table_name=TABLE_NAME)
+    client = service.get_table_client(table_name=TABLE_NAME)
+
+    # Clear existing data
+    existing = list(client.list_entities(select=["PartitionKey", "RowKey"]))
+    by_partition: dict[str, list[dict]] = {}
+    for e in existing:
+        pk = e["PartitionKey"]
+        by_partition.setdefault(pk, []).append(
+            {"PartitionKey": pk, "RowKey": e["RowKey"]}
+        )
+    for pk, entities in by_partition.items():
+        for i in range(0, len(entities), 100):
+            batch = entities[i : i + 100]
+            ops = [("delete", e) for e in batch]
+            client.submit_transaction(ops)
+    if existing:
+        print(f"Cleared {len(existing)} existing records")
+
+    # Generate random users and dates
+    users = [f"user{i:03d}" for i in random.sample(range(1, USER_COUNT + 1), k=min(USER_COUNT, USER_COUNT))]
+    date_range_seconds = int((DATE_END - DATE_START).total_seconds())
+
+    # Seed new data
+    entities = []
+    order_idx = 0
+    for _ in range(len(users)):
+        user = random.choice(users)
+        # Random timestamp in [DATE_START, DATE_END)
+        offset = random.randint(0, max(date_range_seconds - 1, 0))
+        dt = DATE_START + timedelta(seconds=offset)
+
+        # Received orders (state=1)
+        for _ in range(ENTRIES_PER_USER_PER_DAY):
+            entities.append({
+                "PartitionKey": user,
+                "RowKey": f"test-order-{order_idx:04d}",
+                "OrderTime": dt.replace(hour=8 + (order_idx % 12), minute=order_idx % 60),
+                "ReceiveTime": dt.replace(hour=9 + (order_idx % 12), minute=order_idx % 60),
+                "revenue": 10.0 + order_idx,
+                "platformOrder": f"plat-{order_idx:04d}",
+                "prodCount": 1 + (order_idx % 5),
+                "pfId": user,
+                "prodId": f"prod-{order_idx % 3}",
+                "prodPrice": 10.0 + (order_idx % 10),
+                "state": 1,  # Received
+                "ver": 3,
+                "customData": None,
+            })
+            order_idx += 1
+
+        # Payed order (state=0, should be skipped by backup)
+        entities.append({
+            "PartitionKey": user,
+            "RowKey": f"test-order-{order_idx:04d}",
+            "OrderTime": dt.replace(hour=14, minute=30),
+            "ReceiveTime": None,
+            "revenue": 5.0,
+            "platformOrder": f"plat-payed-{order_idx:04d}",
+            "prodCount": 1,
+            "pfId": user,
+            "prodId": "prod-pay",
+            "prodPrice": 5.0,
+            "state": 0,  # Payed
+            "ver": 3,
+            "customData": None,
+        })
+        order_idx += 1
+
+    # Insert in batches grouped by PartitionKey
+    total = 0
+    by_pk: dict[str, list[dict]] = {}
+    for e in entities:
+        by_pk.setdefault(e["PartitionKey"], []).append(e)
+
+    for pk, pk_entities in by_pk.items():
+        for i in range(0, len(pk_entities), 100):
+            batch = pk_entities[i : i + 100]
+            ops = [("create", e) for e in batch]
+            client.submit_transaction(ops)
+            total += len(batch)
+
+    print(f"Seeded {total} orders across {len(by_pk)} partitions")
+    print(f"  Received (state=1): {len([e for e in entities if e['state'] == 1])}")
+    print(f"  Payed   (state=0): {len([e for e in entities if e['state'] == 0])}")
+
+
+if __name__ == "__main__":
+    config_path = sys.argv[1] if len(sys.argv) > 1 else str(
+        Path.home() / ".config" / "order-bak" / "config.toml"
+    )
+    seed(config_path)

src/order_bak/backup.py

@@ -0,0 +1,43 @@
+import csv
+from collections import defaultdict
+from datetime import datetime
+from pathlib import Path
+
+from order_bak.csv_utils import CSV_COLUMNS, entity_to_row, parse_row
+
+
+def backup_orders(
+    scan_path: Path,
+    backup_dir: Path,
+    cutoff: datetime,
+) -> int:
+    backup_dir.mkdir(parents=True, exist_ok=True)
+
+    by_date: dict[str, list[dict]] = defaultdict(list)
+
+    with open(scan_path, newline="") as f:
+        reader = csv.DictReader(f)
+        for raw_row in reader:
+            entity = parse_row(raw_row)
+            if entity["state"] != 1:
+                continue
+            order_time = entity["OrderTime"]
+            if order_time is None:
+                continue
+            if order_time >= cutoff:
+                continue
+            date_key = order_time.strftime("%Y-%m-%d")
+            by_date[date_key].append(entity)
+
+    total = 0
+    for date_key, rows in sorted(by_date.items()):
+        csv_path = backup_dir / f"{date_key}.csv"
+        with open(csv_path, "w", newline="") as f:
+            writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
+            writer.writeheader()
+            writer.writerows(entity_to_row(r) for r in rows)
+        total += len(rows)
+        print(f"  {date_key}: {len(rows)} orders")
+
+    print(f"Backup complete: {total} orders across {len(by_date)} days")
+    return total

src/order_bak/cli.py

@@ -1,9 +1,103 @@
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
 import click
 
+from order_bak.config import load_config
+
 
 @click.group()
-def main():
-    pass
+@click.option("--config", "-c", default="~/.config/order-bak/config.toml", type=click.Path())
+@click.pass_context
+def main(ctx, config):
+    ctx.ensure_object(dict)
+    ctx.obj["config_path"] = Path(config).expanduser()
+
+
+@main.command()
+@click.option("--count", default=0, type=int, help="Number of records to scan (0 = full table)")
+@click.pass_context
+def scan(ctx, count):
+    cfg = load_config(ctx.obj["config_path"])
+    from order_bak.scan import scan_orders
+
+    scan_orders(
+        connection_string=cfg.azure_connection_string,
+        page_size=cfg.scan_page_size,
+        delay_ms=cfg.scan_delay_ms,
+        scan_dir=cfg.scan_dir,
+        limit=count,
+    )
+
+
+@main.command()
+@click.option("--days", default=None, type=int, help="Backup orders older than N days (min 1); default from config backup.retain_days")
+@click.option("--scan", "scan_path", default=None, type=click.Path(), help="Path to scan CSV (default: latest in scan_dir)")
+@click.pass_context
+def backup(ctx, days, scan_path):
+    if days < 1:
+        click.echo("Error: --days must be at least 1")
+        raise SystemExit(1)
+
+    cfg = load_config(ctx.obj["config_path"])
+    from order_bak.backup import backup_orders
+
+    if days is None:
+        days = cfg.backup_retain_days
+
+    if scan_path:
+        scan_file = Path(scan_path)
+    else:
+        scan_files = sorted(cfg.scan_dir.glob("scan-*.csv"))
+        if not scan_files:
+            click.echo("No scan files found in scan_dir")
+            raise SystemExit(1)
+        scan_file = scan_files[-1]
+
+    today_utc = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
+    cutoff = today_utc - timedelta(days=days)
+
+    click.echo(f"Backing up orders before {cutoff.strftime('%Y-%m-%d')} UTC")
+    backup_orders(
+        scan_path=scan_file,
+        backup_dir=Path("./backups"),
+        cutoff=cutoff,
+    )
+
+
+@main.command()
+@click.pass_context
+def delete(ctx):
+    cfg = load_config(ctx.obj["config_path"])
+    from azure.data.tables import TableClient
+    from order_bak.delete import delete_orders
+
+    backup_dir = Path("./backups")
+    backup_files = sorted(backup_dir.glob("*.csv"))
+    if not backup_files:
+        click.echo("No backup files found in ./backups")
+        raise SystemExit(1)
+
+    import csv as csv_mod
+    total = 0
+    for f in backup_files:
+        with open(f, newline="") as fh:
+            total += sum(1 for _ in csv_mod.DictReader(fh))
+        click.echo(f"  {f}")
+
+    click.echo(f"Will delete {total} orders from {len(backup_files)} file(s)")
+    if not click.confirm("Continue?"):
+        click.echo("Aborted")
+        return
+
+    table = TableClient.from_connection_string(cfg.azure_connection_string, table_name="order")
+    for csv_file in backup_files:
+        delete_orders(
+            table_client=table,
+            csv_path=csv_file,
+            batch_size=cfg.delete_batch_size,
+            delay_ms=cfg.delete_delay_ms,
+        )
 
 
 if __name__ == "__main__":

src/order_bak/config.py

@@ -12,6 +12,7 @@ class Config:
     scan_dir: Path = Path("./scan")
     delete_batch_size: int = 100
     delete_delay_ms: int = 300
+    backup_retain_days: int = 1
 
 
 def load_config(config_path: Path) -> Config:
@@ -30,6 +31,7 @@ def load_config(config_path: Path) -> Config:
 
     scan = raw.get("scan", {})
     delete = raw.get("delete", {})
+    backup = raw.get("backup", {})
 
     return Config(
         azure_connection_string=connection_string,
@@ -38,4 +40,5 @@ def load_config(config_path: Path) -> Config:
         scan_dir=Path(scan.get("scan_dir", "./scan")),
         delete_batch_size=delete.get("batch_size", 100),
         delete_delay_ms=delete.get("delay_ms", 300),
+        backup_retain_days=backup.get("retain_days", 1),
     )

src/order_bak/csv_utils.py

@@ -32,7 +32,7 @@ def entity_to_row(entity: dict[str, Any]) -> dict[str, str]:
     return {col: _format_value(entity.get(col)) for col in CSV_COLUMNS}
 
 
-def _parse_row(raw: dict[str, str]) -> dict[str, Any]:
+def parse_row(raw: dict[str, str]) -> dict[str, Any]:
     result: dict[str, Any] = {}
     for col in CSV_COLUMNS:
         val = raw.get(col, "")
@@ -43,15 +43,21 @@ def _parse_row(raw: dict[str, str]) -> dict[str, Any]:
             result[col] = datetime.fromisoformat(val)
         elif col in ("revenue", "prodPrice"):
             result[col] = float(val)
-        elif col in ("prodCount", "state", "ver"):
+        elif col in ("prodCount", "ver"):
             result[col] = int(val)
+        elif col == "state":
+            try:
+                result[col] = int(val)
+            except ValueError:
+                _STATE_MAP = {"Payed": 0, "Recived": 1}
+                result[col] = _STATE_MAP.get(val, val)
         else:
             result[col] = val
     return result
 
 
 def row_to_entity(raw: dict[str, str]) -> dict[str, Any]:
-    return _parse_row(raw)
+    return parse_row(raw)
 
 
 def write_csv(path: Path, entities: list[dict[str, Any]]) -> None:
@@ -67,4 +73,4 @@ def read_csv(path: Path):
     with open(path, newline="") as f:
         reader = csv.DictReader(f)
         for row in reader:
-            yield _parse_row(row)
+            yield parse_row(row)

src/order_bak/delete.py

@@ -0,0 +1,39 @@
+import csv
+import time
+from collections import defaultdict
+from itertools import islice
+from pathlib import Path
+
+
+def _chunked(iterable, size):
+    it = iter(iterable)
+    while chunk := list(islice(it, size)):
+        yield chunk
+
+
+def delete_orders(
+    table_client,
+    csv_path: Path,
+    batch_size: int,
+    delay_ms: int,
+) -> int:
+    by_partition: dict[str, list[dict]] = defaultdict(list)
+
+    with open(csv_path, newline="") as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            pk = row["PartitionKey"]
+            rk = row["RowKey"]
+            by_partition[pk].append({"PartitionKey": pk, "RowKey": rk})
+
+    total_deleted = 0
+    for pk, entities in by_partition.items():
+        for batch in _chunked(entities, batch_size):
+            operations = [("delete", e) for e in batch]
+            table_client.submit_transaction(operations)
+            total_deleted += len(batch)
+            if delay_ms > 0:
+                time.sleep(delay_ms / 1000.0)
+
+    print(f"Deleted {total_deleted} orders across {len(by_partition)} partitions")
+    return total_deleted

src/order_bak/scan.py

@@ -0,0 +1,46 @@
+import csv
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+from azure.data.tables import TableClient
+
+from order_bak.csv_utils import CSV_COLUMNS, entity_to_row
+
+
+def scan_orders(
+    connection_string: str,
+    page_size: int,
+    delay_ms: int,
+    scan_dir: Path,
+    limit: int = 0,
+) -> Path:
+    table = TableClient.from_connection_string(connection_string, table_name="order")
+    scan_dir.mkdir(parents=True, exist_ok=True)
+
+    now = datetime.now(timezone.utc)
+    csv_path = scan_dir / f"scan-{now.strftime('%Y-%m-%d')}.csv"
+
+    paged = table.list_entities(results_per_page=page_size)
+
+    total_scanned = 0
+
+    with open(csv_path, "w", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
+        writer.writeheader()
+
+        for page in paged.by_page():
+            for entity in page:
+                writer.writerow(entity_to_row(entity))
+                total_scanned += 1
+                if limit > 0 and total_scanned >= limit:
+                    break
+
+            if limit > 0 and total_scanned >= limit:
+                break
+
+            if delay_ms > 0:
+                time.sleep(delay_ms / 1000.0)
+
+    print(f"Scan complete: {total_scanned} orders saved to {csv_path}")
+    return csv_path

tests/test_backup.py

@@ -0,0 +1,130 @@
+from datetime import datetime, timezone
+from order_bak.csv_utils import write_csv
+from order_bak.backup import backup_orders
+
+
+def _make_entity(pk, rk, order_time, state, receive_time=None):
+    return {
+        "PartitionKey": pk,
+        "RowKey": rk,
+        "OrderTime": order_time,
+        "ReceiveTime": receive_time,
+        "revenue": 10.0,
+        "platformOrder": "plat1",
+        "prodCount": 1,
+        "pfId": pk,
+        "prodId": "prod1",
+        "prodPrice": 10.0,
+        "state": state,
+        "ver": 3,
+        "customData": None,
+    }
+
+
+def test_backup_filters_by_state_and_date(tmp_path):
+    scan_dir = tmp_path / "scan"
+    backup_dir = tmp_path / "backups"
+    scan_dir.mkdir()
+    backup_dir.mkdir()
+
+    entities = [
+        _make_entity("u1", "o1", datetime(2025, 1, 1, 10, tzinfo=timezone.utc), 1),
+        _make_entity("u2", "o2", datetime(2025, 1, 1, 12, tzinfo=timezone.utc), 0),  # Payed, skip
+        _make_entity("u3", "o3", datetime(2025, 1, 2, 8, tzinfo=timezone.utc), 1),
+        _make_entity("u4", "o4", datetime(2025, 1, 3, 8, tzinfo=timezone.utc), 1),  # at cutoff, skip
+    ]
+    scan_csv = scan_dir / "scan-2025-01-03.csv"
+    write_csv(scan_csv, entities)
+
+    result = backup_orders(
+        scan_path=scan_csv,
+        backup_dir=backup_dir,
+        cutoff=datetime(2025, 1, 3, tzinfo=timezone.utc),
+    )
+
+    assert result == 2
+    jan1 = backup_dir / "2025-01-01.csv"
+    jan2 = backup_dir / "2025-01-02.csv"
+    assert jan1.exists()
+    assert jan2.exists()
+    assert not (backup_dir / "2025-01-03.csv").exists()
+
+    from order_bak.csv_utils import read_csv
+    jan1_rows = list(read_csv(jan1))
+    assert len(jan1_rows) == 1
+    assert jan1_rows[0]["RowKey"] == "o1"
+
+
+def test_backup_filters_by_cutoff(tmp_path):
+    scan_dir = tmp_path / "scan"
+    backup_dir = tmp_path / "backups"
+    scan_dir.mkdir()
+    backup_dir.mkdir()
+
+    entities = [
+        _make_entity("u1", "o1", datetime(2025, 1, 1, 10, tzinfo=timezone.utc), 1),
+        _make_entity("u2", "o2", datetime(2025, 1, 2, 8, tzinfo=timezone.utc), 1),
+        _make_entity("u3", "o3", datetime(2025, 1, 3, 8, tzinfo=timezone.utc), 1),  # at/after cutoff
+    ]
+    scan_csv = scan_dir / "scan-2025-01-03.csv"
+    write_csv(scan_csv, entities)
+
+    cutoff = datetime(2025, 1, 3, tzinfo=timezone.utc)
+    result = backup_orders(
+        scan_path=scan_csv,
+        backup_dir=backup_dir,
+        cutoff=cutoff,
+    )
+
+    assert result == 2
+    assert (backup_dir / "2025-01-01.csv").exists()
+    assert (backup_dir / "2025-01-02.csv").exists()
+    assert not (backup_dir / "2025-01-03.csv").exists()
+
+
+def test_backup_cutoff_at_midnight(tmp_path):
+    """Order exactly at cutoff midnight should be excluded."""
+    scan_dir = tmp_path / "scan"
+    backup_dir = tmp_path / "backups"
+    scan_dir.mkdir()
+    backup_dir.mkdir()
+
+    entities = [
+        _make_entity("u1", "o1", datetime(2025, 1, 3, 0, 0, 0, tzinfo=timezone.utc), 1),  # exactly at cutoff
+        _make_entity("u2", "o2", datetime(2025, 1, 2, 23, 59, 59, tzinfo=timezone.utc), 1),  # just before
+    ]
+    scan_csv = scan_dir / "scan.csv"
+    write_csv(scan_csv, entities)
+
+    result = backup_orders(
+        scan_path=scan_csv,
+        backup_dir=backup_dir,
+        cutoff=datetime(2025, 1, 3, tzinfo=timezone.utc),
+    )
+
+    assert result == 1
+    assert (backup_dir / "2025-01-02.csv").exists()
+    assert not (backup_dir / "2025-01-03.csv").exists()
+
+
+def test_backup_empty_result(tmp_path):
+    """No orders before cutoff returns 0 and no files."""
+    scan_dir = tmp_path / "scan"
+    backup_dir = tmp_path / "backups"
+    scan_dir.mkdir()
+    backup_dir.mkdir()
+
+    entities = [
+        _make_entity("u1", "o1", datetime(2025, 1, 5, 10, tzinfo=timezone.utc), 1),
+    ]
+    scan_csv = scan_dir / "scan.csv"
+    write_csv(scan_csv, entities)
+
+    result = backup_orders(
+        scan_path=scan_csv,
+        backup_dir=backup_dir,
+        cutoff=datetime(2025, 1, 1, tzinfo=timezone.utc),
+    )
+
+    assert result == 0
+    assert list(backup_dir.glob("*.csv")) == []

tests/test_delete.py

@@ -0,0 +1,145 @@
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+import pytest
+from click.testing import CliRunner
+
+from order_bak.cli import main
+from order_bak.csv_utils import write_csv
+from order_bak.delete import delete_orders
+
+
+def _make_entity(pk, rk):
+    return {
+        "PartitionKey": pk,
+        "RowKey": rk,
+        "OrderTime": datetime(2025, 1, 1, tzinfo=timezone.utc),
+        "ReceiveTime": None,
+        "revenue": 10.0,
+        "platformOrder": "p",
+        "prodCount": 1,
+        "pfId": pk,
+        "prodId": "prod",
+        "prodPrice": 10.0,
+        "state": 1,
+        "ver": 3,
+        "customData": None,
+    }
+
+
+# --- delete_orders unit tests ---
+
+def test_delete_groups_by_partition_and_batches(tmp_path):
+    entities = []
+    for i in range(3):
+        entities.append(_make_entity("user1", f"order{i}"))
+    for i in range(3, 5):
+        entities.append(_make_entity("user2", f"order{i}"))
+
+    csv_path = tmp_path / "2025-01-01.csv"
+    write_csv(csv_path, entities)
+
+    mock_table = MagicMock()
+
+    deleted_count = delete_orders(
+        table_client=mock_table,
+        csv_path=csv_path,
+        batch_size=2,
+        delay_ms=0,
+    )
+
+    assert deleted_count == 5
+    # user1: 3 entities, batch_size=2 -> 2 batches (2 + 1)
+    # user2: 2 entities, batch_size=2 -> 1 batch
+    assert mock_table.submit_transaction.call_count == 3
+
+
+def test_delete_respects_batch_size(tmp_path):
+    entities = [_make_entity("user1", f"order{i}") for i in range(5)]
+
+    csv_path = tmp_path / "2025-01-01.csv"
+    write_csv(csv_path, entities)
+
+    mock_table = MagicMock()
+
+    deleted_count = delete_orders(
+        table_client=mock_table,
+        csv_path=csv_path,
+        batch_size=2,
+        delay_ms=0,
+    )
+
+    assert deleted_count == 5
+    assert mock_table.submit_transaction.call_count == 3  # 2+2+1
+
+
+# --- delete CLI command tests ---
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+@pytest.fixture
+def config_file(tmp_path):
+    cfg = tmp_path / "config.toml"
+    cfg.write_text('[azure]\nconnection_string = "UseDevelopmentStorage=true"\n')
+    return cfg
+
+
+def test_delete_command_no_backups(runner, config_file, tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / "backups").mkdir()
+
+    result = runner.invoke(main, ["--config", str(config_file), "delete"])
+
+    assert result.exit_code == 1
+    assert "No backup files found" in result.output
+
+
+def test_delete_command_aborts(runner, config_file, tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    backup_dir = tmp_path / "backups"
+    backup_dir.mkdir()
+    write_csv(backup_dir / "2025-01-01.csv", [_make_entity("u1", "o1")])
+
+    mock_table = MagicMock()
+    with patch("azure.data.tables.TableClient.from_connection_string", return_value=mock_table):
+        result = runner.invoke(main, ["--config", str(config_file), "delete"], input="n\n")
+
+    assert "Aborted" in result.output
+    mock_table.submit_transaction.assert_not_called()
+
+
+def test_delete_command_single_file(runner, config_file, tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    backup_dir = tmp_path / "backups"
+    backup_dir.mkdir()
+    write_csv(backup_dir / "2025-01-01.csv", [_make_entity("u1", f"o{i}") for i in range(3)])
+
+    mock_table = MagicMock()
+    with patch("azure.data.tables.TableClient.from_connection_string", return_value=mock_table):
+        result = runner.invoke(main, ["--config", str(config_file), "delete"], input="y\n")
+
+    assert result.exit_code == 0
+    assert "3 orders" in result.output
+    assert "1 file" in result.output
+    assert mock_table.submit_transaction.called
+
+
+def test_delete_command_multiple_files(runner, config_file, tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    backup_dir = tmp_path / "backups"
+    backup_dir.mkdir()
+    write_csv(backup_dir / "2025-01-01.csv", [_make_entity("u1", f"o{i}") for i in range(2)])
+    write_csv(backup_dir / "2025-01-02.csv", [_make_entity("u2", f"o{i}") for i in range(3)])
+
+    mock_table = MagicMock()
+    with patch("azure.data.tables.TableClient.from_connection_string", return_value=mock_table):
+        result = runner.invoke(main, ["--config", str(config_file), "delete"], input="y\n")
+
+    assert result.exit_code == 0
+    assert "5 orders" in result.output
+    assert "2 file" in result.output
+    # u1 (2 entities, batch 100) → 1 call; u2 (3 entities, batch 100) → 1 call
+    assert mock_table.submit_transaction.call_count == 2

tests/test_integration.py

@@ -0,0 +1,232 @@
+"""Integration tests against Azure Storage emulator."""
+
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+from azure.data.tables import TableServiceClient
+from click.testing import CliRunner
+
+from order_bak.backup import backup_orders
+from order_bak.cli import main
+from order_bak.csv_utils import read_csv
+from order_bak.delete import delete_orders
+from order_bak.scan import scan_orders
+
+CONFIG_PATH = Path(__file__).parent.parent / "assets" / "local.toml"
+
+CONNECTION_STRING = os.environ.get(
+    "ORDER_BAK_TEST_CONNECTION_STRING",
+    (
+        "AccountName=devstoreaccount1;"
+        "AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/"
+        "K1SZFPTOtr/KBHBeksoGMGw==;"
+        "DefaultEndpointsProtocol=http;"
+        "BlobEndpoint=http://localhost:10000/devstoreaccount1;"
+        "QueueEndpoint=http://localhost:10001/devstoreaccount1;"
+        "TableEndpoint=http://localhost:10002/devstoreaccount1;"
+    ),
+)
+
+TABLE_NAME = "order"
+
+
+def _seed_orders(client, orders):
+    from collections import defaultdict
+
+    by_pk: dict[str, list[dict]] = defaultdict(list)
+    for o in orders:
+        by_pk[o["PartitionKey"]].append(o)
+
+    for pk, entities in by_pk.items():
+        for i in range(0, len(entities), 100):
+            batch = entities[i : i + 100]
+            ops = [("create", e) for e in batch]
+            client.submit_transaction(ops)
+
+
+def _clear_table(client):
+    existing = list(client.list_entities(select=["PartitionKey", "RowKey"]))
+    if not existing:
+        return
+    from collections import defaultdict
+
+    by_pk: dict[str, list[dict]] = defaultdict(list)
+    for e in existing:
+        by_pk[e["PartitionKey"]].append(
+            {"PartitionKey": e["PartitionKey"], "RowKey": e["RowKey"]}
+        )
+    for pk, entities in by_pk.items():
+        for i in range(0, len(entities), 100):
+            batch = entities[i : i + 100]
+            ops = [("delete", e) for e in batch]
+            client.submit_transaction(ops)
+
+
+@pytest.fixture
+def table_client():
+    service = TableServiceClient.from_connection_string(CONNECTION_STRING)
+    service.create_table_if_not_exists(table_name=TABLE_NAME)
+    client = service.get_table_client(table_name=TABLE_NAME)
+    _clear_table(client)
+    yield client
+    _clear_table(client)
+
+
+@pytest.mark.integration
+def test_full_pipeline(table_client, tmp_path):
+    # --- Seed ---
+    orders = []
+    for day in [1, 2, 3]:
+        for user in ["u1", "u2"]:
+            # 2 Recived per user per day
+            for j in range(2):
+                orders.append({
+                    "PartitionKey": user,
+                    "RowKey": f"o-d{day}-{user}-{j}",
+                    "OrderTime": datetime(2025, 1, day, 10 + j, 0, tzinfo=timezone.utc),
+                    "ReceiveTime": datetime(2025, 1, day, 11, 0, tzinfo=timezone.utc),
+                    "revenue": 10.0 + day,
+                    "platformOrder": f"plat-d{day}",
+                    "prodCount": 1,
+                    "pfId": user,
+                    "prodId": "prod1",
+                    "prodPrice": 10.0,
+                    "state": 1,
+                    "ver": 3,
+                    "customData": None,
+                })
+            # 1 Payed per user per day (should be excluded from backup)
+            orders.append({
+                "PartitionKey": user,
+                "RowKey": f"o-d{day}-{user}-payed",
+                "OrderTime": datetime(2025, 1, day, 14, 0, tzinfo=timezone.utc),
+                "ReceiveTime": None,
+                "revenue": 5.0,
+                "platformOrder": "plat-p",
+                "prodCount": 1,
+                "pfId": user,
+                "prodId": "prod-pay",
+                "prodPrice": 5.0,
+                "state": 0,
+                "ver": 3,
+                "customData": None,
+            })
+    _seed_orders(table_client, orders)
+
+    # --- Scan ---
+    scan_dir = tmp_path / "scan"
+    scan_path = scan_orders(
+        connection_string=CONNECTION_STRING,
+        page_size=10,
+        delay_ms=0,
+        scan_dir=scan_dir,
+    )
+    assert scan_path.exists()
+    scan_rows = list(read_csv(scan_path))
+    assert len(scan_rows) == len(orders)  # all orders, no filtering
+
+    # --- Backup (before Jan 3) ---
+    backup_dir = tmp_path / "backups"
+    count = backup_orders(
+        scan_path=scan_path,
+        backup_dir=backup_dir,
+        cutoff=datetime(2025, 1, 3, tzinfo=timezone.utc),
+    )
+    # 2 days × 2 users × 2 Recived = 8 orders
+    assert count == 8
+
+    jan1 = backup_dir / "2025-01-01.csv"
+    jan2 = backup_dir / "2025-01-02.csv"
+    assert jan1.exists()
+    assert jan2.exists()
+    assert not (backup_dir / "2025-01-03.csv").exists()
+
+    jan1_rows = list(read_csv(jan1))
+    assert len(jan1_rows) == 4  # 2 users × 2 Recived
+    for row in jan1_rows:
+        assert row["state"] == 1
+
+    # --- Delete (Jan 1 only) ---
+    deleted = delete_orders(
+        table_client=table_client,
+        csv_path=jan1,
+        batch_size=2,
+        delay_ms=0,
+    )
+    assert deleted == 4
+
+    # Verify Jan 1 orders are gone from table
+    remaining = list(table_client.list_entities())
+    remaining_jan1 = [
+        e for e in remaining
+        if e.get("OrderTime") and e["OrderTime"].day == 1 and e["state"] == 1
+    ]
+    assert len(remaining_jan1) == 0
+
+    # Jan 2 Recived orders should still exist
+    remaining_jan2_recived = [
+        e for e in remaining
+        if e.get("OrderTime") and e["OrderTime"].day == 2 and e["state"] == 1
+    ]
+    assert len(remaining_jan2_recived) == 4
+
+
+@pytest.mark.integration
+def test_delete_command_all_backups(table_client, tmp_path, monkeypatch):
+    """CLI delete command reads all backup CSVs and deletes them from Azure."""
+    # Seed: 3 days × 2 users × 1 Recived order = 6 orders to back up
+    orders = []
+    for day in [1, 2, 3]:
+        for user in ["u1", "u2"]:
+            orders.append({
+                "PartitionKey": user,
+                "RowKey": f"o-d{day}-{user}",
+                "OrderTime": datetime(2025, 1, day, 10, 0, tzinfo=timezone.utc),
+                "ReceiveTime": datetime(2025, 1, day, 11, 0, tzinfo=timezone.utc),
+                "revenue": 10.0,
+                "platformOrder": "plat",
+                "prodCount": 1,
+                "pfId": user,
+                "prodId": "prod1",
+                "prodPrice": 10.0,
+                "state": 1,
+                "ver": 3,
+                "customData": None,
+            })
+    _seed_orders(table_client, orders)
+
+    # Scan then backup (cutoff excludes nothing — all 6 are Recived before Jan 4)
+    scan_dir = tmp_path / "scan"
+    scan_path = scan_orders(
+        connection_string=CONNECTION_STRING,
+        page_size=10,
+        delay_ms=0,
+        scan_dir=scan_dir,
+    )
+    backup_dir = tmp_path / "backups"
+    backed_up = backup_orders(
+        scan_path=scan_path,
+        backup_dir=backup_dir,
+        cutoff=datetime(2025, 1, 4, tzinfo=timezone.utc),
+    )
+    assert backed_up == 6
+    assert len(list(backup_dir.glob("*.csv"))) == 3  # one file per day
+
+    # chdir so Path("./backups") in the CLI resolves to tmp_path/backups
+    monkeypatch.chdir(tmp_path)
+
+    result = CliRunner().invoke(
+        main,
+        ["--config", str(CONFIG_PATH), "delete"],
+        input="y\n",
+    )
+
+    assert result.exit_code == 0, result.output
+    assert "6 orders" in result.output
+    assert "3 file" in result.output
+
+    # All 6 Recived orders must be gone from the table
+    remaining = list(table_client.list_entities())
+    assert len(remaining) == 0