feat: CSV utilities for order entity serialization

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

src/order_bak/csv_utils.py

@@ -0,0 +1,70 @@
+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)