feat: add source export script and clean up .DS_Store tracking

Add scripts/export_source.sh to export project source to dist-source/,
add .DS_Store and dist-source/ to .gitignore, remove tracked .DS_Store.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

.gitignore

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

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,14 @@
+# 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"
+
+[delete]
+# batch_size = 100
+# delay_ms = 300

assets/s101.toml

@@ -0,0 +1,12 @@
+
+[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"
+
+[delete]
+batch_size = 100
+delay_ms = 300

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/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 sys
+from datetime import datetime, timezone
+
+from azure.data.tables import TableServiceClient
+
+DEFAULT_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;"
+)
+
+TABLE_NAME = "order"
+
+# Test users
+USERS = ["user001", "user002", "user003", "user004", "user005"]
+
+# 3 days of data: Jan 1-3, 2025
+DATES = [
+    datetime(2025, 1, 1, tzinfo=timezone.utc),
+    datetime(2025, 1, 2, tzinfo=timezone.utc),
+    datetime(2025, 1, 3, tzinfo=timezone.utc),
+]
+
+
+def seed(connection_string: str) -> None:
+    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 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")
+
+    # Seed new data
+    entities = []
+    order_idx = 0
+    for date in DATES:
+        for user in USERS:
+            # 2 Recived orders per user per day
+            for _ in range(2):
+                hour = 8 + (order_idx % 12)
+                entities.append({
+                    "PartitionKey": user,
+                    "RowKey": f"test-order-{order_idx:04d}",
+                    "OrderTime": date.replace(hour=hour, minute=order_idx % 60),
+                    "ReceiveTime": date.replace(hour=hour + 1, 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,  # Recived
+                    "ver": 3,
+                    "customData": None,
+                })
+                order_idx += 1
+
+            # 1 Payed order per user per day (should be skipped by backup)
+            entities.append({
+                "PartitionKey": user,
+                "RowKey": f"test-order-{order_idx:04d}",
+                "OrderTime": date.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, {len(DATES)} days")
+    print(f"  Recived (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__":
+    conn_str = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_CONNECTION_STRING
+    seed(conn_str)

src/order_bak/backup.py

@@ -0,0 +1,44 @@
+import csv
+from collections import defaultdict
+from datetime import datetime
+from pathlib import Path
+
+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

src/order_bak/cli.py

@@ -1,9 +1,94 @@
+from datetime import 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("--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
+
+    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]
+
+    start_utc = start.replace(tzinfo=timezone.utc)
+    end_utc = end.replace(tzinfo=timezone.utc)
+
+    if start_utc >= end_utc:
+        click.echo("Error: --start must be before --end")
+        raise SystemExit(1)
+
+    backup_orders(
+        scan_path=scan_file,
+        backup_dir=Path("./backups"),
+        start=start_utc,
+        end=end_utc,
+    )
+
+
+@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
+
+    csv_file = Path(csv_path)
+
+    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,
+    )
 
 
 if __name__ == "__main__":

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, "")
@@ -51,7 +51,7 @@ def _parse_row(raw: dict[str, str]) -> dict[str, Any]:
 
 
 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 +67,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,58 @@
+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"

tests/test_delete.py

@@ -0,0 +1,67 @@
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+
+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,
+    }
+
+
+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

tests/test_integration.py

@@ -0,0 +1,170 @@
+"""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 order_bak.backup import backup_orders
+from order_bak.csv_utils import read_csv
+from order_bak.delete import delete_orders
+from order_bak.scan import scan_orders
+
+CONNECTION_STRING = os.environ.get(
+    "ORDER_BAK_TEST_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;"
+    ),
+)
+
+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 (Jan 1-2 only) ---
+    backup_dir = tmp_path / "backups"
+    count = backup_orders(
+        scan_path=scan_path,
+        backup_dir=backup_dir,
+        start=datetime(2025, 1, 1, tzinfo=timezone.utc),
+        end=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