tech
ef449729d7
Switch backup to write converted entities instead of raw table rows via entity_to_row, and map string state values like "Payed"/"Recived" to their integer equivalents during CSV parsing. Add sandbox.toml config. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
77 lines
1.9 KiB
Python
77 lines
1.9 KiB
Python
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", "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)
|
|
|
|
|
|
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)
|