286 lines
11 KiB
Python
286 lines
11 KiB
Python
"""
|
|
Amazon Custom Unified Transaction reports delivered as CSV (UTF-8, often with BOM).
|
|
|
|
Same public interface as TransactionReader / CalamineReader:
|
|
detect() / iter_records() / file_meta / column_mapping / close()
|
|
|
|
Amazon CSVs typically begin with a short preamble (scope, currency, definitions) before the
|
|
real header row. Amounts use a European decimal comma in localized EU reports ("13,49").
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import os
|
|
import re
|
|
from datetime import date
|
|
from typing import Iterator
|
|
|
|
from .column_map import FIELD_KIND, ColumnMapping, build_mapping
|
|
from .dates import parse_amazon_date_fast
|
|
from .xlsx_reader import FileMeta, ParseError
|
|
|
|
_DETECT_REQUIRED = {"settlement_id", "total", "date_time"}
|
|
_CURRENCY_RE = re.compile(
|
|
r"\b(USD|EUR|GBP|CAD|AUD|PLN|SEK|TRY|JPY)\b", re.IGNORECASE
|
|
)
|
|
|
|
|
|
# Amazon's localized reports are not all ASCII: Sweden writes negatives with a real
|
|
# MINUS SIGN (U+2212, "\u221278 690,40") and several locales group thousands with
|
|
# non-breaking / narrow spaces. float() rejects U+2212, and the fallback below would
|
|
# silently turn the cell into 0.0 \u2014 which dropped every negative amount (fees, taxes,
|
|
# transfers) of an entire Swedish month while the positives kept adding up.
|
|
_AMOUNT_CLEANUP = str.maketrans({
|
|
"\u2212": "-", "\u2010": "-", "\u2011": "-", "\u2013": "-", # minus / dash variants
|
|
"\u00a0": None, "\u202f": None, "\u2009": None, " ": None, # space variants
|
|
})
|
|
|
|
|
|
def parse_amount(raw) -> float:
|
|
"""Parse Amazon amount cells, including European '1.234,56' / '13,49' forms."""
|
|
if raw is None or raw == "":
|
|
return 0.0
|
|
if isinstance(raw, bool):
|
|
return 0.0
|
|
if isinstance(raw, (int, float)):
|
|
return float(raw)
|
|
s = str(raw).strip().translate(_AMOUNT_CLEANUP)
|
|
if not s:
|
|
return 0.0
|
|
# European: decimal comma, optional thousands dots / spaces.
|
|
if "," in s and "." in s:
|
|
if s.rfind(",") > s.rfind("."):
|
|
s = s.replace(".", "").replace(",", ".")
|
|
else:
|
|
s = s.replace(",", "")
|
|
elif "," in s:
|
|
# "13,49" or "1.234" — if one comma and digits after, treat as decimal.
|
|
left, _, right = s.partition(",")
|
|
if right.isdigit() and 1 <= len(right) <= 2:
|
|
s = f"{left.replace('.', '')}.{right}"
|
|
else:
|
|
s = s.replace(",", "")
|
|
try:
|
|
return float(s)
|
|
except ValueError:
|
|
return 0.0
|
|
|
|
|
|
def _index_to_col(idx: int) -> str:
|
|
idx += 1
|
|
s = ""
|
|
while idx:
|
|
idx, r = divmod(idx - 1, 26)
|
|
s = chr(65 + r) + s
|
|
return s
|
|
|
|
|
|
def _convert(field_name: str, raw):
|
|
kind = FIELD_KIND.get(field_name, "text")
|
|
if kind == "amount":
|
|
return parse_amount(raw)
|
|
if kind == "int":
|
|
if raw in (None, ""):
|
|
return None
|
|
try:
|
|
return int(parse_amount(raw))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if raw is None:
|
|
return None
|
|
s = str(raw).strip()
|
|
if kind == "id" and s.endswith(".0"):
|
|
s = s[:-2]
|
|
return s or None
|
|
|
|
|
|
class CsvReader:
|
|
def __init__(self, path: str, saved_overrides: dict[str, str] | None = None):
|
|
self.path = path
|
|
self.filename = os.path.basename(path)
|
|
self.saved_overrides = saved_overrides
|
|
self._rows: list[list[str]] | None = None
|
|
self.sheet_name = "CSV"
|
|
self.header_row = 0 # 1-based, matching Excel readers
|
|
self.column_mapping: ColumnMapping | None = None
|
|
self._field_to_idx: dict[str, int] = {}
|
|
self._sum_field_idx: dict[str, list[int]] = {}
|
|
self.file_meta = FileMeta(filename=self.filename)
|
|
|
|
def open(self) -> None:
|
|
if self._rows is not None:
|
|
return
|
|
try:
|
|
raw = open(self.path, "rb").read()
|
|
except OSError as e:
|
|
raise ParseError(f"'{self.filename}' could not be read: {e}") from e
|
|
if not raw:
|
|
raise ParseError(f"'{self.filename}' is empty.")
|
|
# Strip UTF-8 BOM; fall back through common Amazon encodings.
|
|
if raw.startswith(b"\xef\xbb\xbf"):
|
|
text = raw.decode("utf-8-sig")
|
|
else:
|
|
text = None
|
|
for enc in ("utf-8", "utf-16", "cp1252", "latin-1"):
|
|
try:
|
|
text = raw.decode(enc)
|
|
break
|
|
except UnicodeDecodeError:
|
|
continue
|
|
if text is None:
|
|
raise ParseError(f"'{self.filename}' is not a readable text/CSV file.")
|
|
# Sniff delimiter from the densest early line (comma vs semicolon EU exports).
|
|
sample = "\n".join(text.splitlines()[:40])
|
|
try:
|
|
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
|
delimiter = dialect.delimiter
|
|
except csv.Error:
|
|
delimiter = ";" if sample.count(";") > sample.count(",") else ","
|
|
self._rows = list(csv.reader(text.splitlines(), delimiter=delimiter))
|
|
self.file_meta.size_bytes = os.path.getsize(self.path)
|
|
self.file_meta.worksheets = [self.sheet_name]
|
|
# Currency hint from the preamble ("Tous les montants sont en EUR…").
|
|
for row in self._rows[:15]:
|
|
joined = " ".join(row)
|
|
m = _CURRENCY_RE.search(joined)
|
|
if m and ("montant" in joined.lower() or "amount" in joined.lower()
|
|
or "currency" in joined.lower() or "en " in joined.lower()):
|
|
self.file_meta.currency = m.group(1).upper()
|
|
break
|
|
|
|
def close(self) -> None:
|
|
self._rows = None
|
|
|
|
def __enter__(self):
|
|
self.open()
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
self.close()
|
|
|
|
def detect(self) -> ColumnMapping:
|
|
self.open()
|
|
assert self._rows is not None
|
|
best = None
|
|
best_score = -1
|
|
# Scan the first ~40 rows for the real Amazon header (skip preamble / definitions).
|
|
for r_idx, row in enumerate(self._rows[:40]):
|
|
cells = [(_index_to_col(i), str(v)) for i, v in enumerate(row) if str(v).strip()]
|
|
if len(cells) < 5:
|
|
continue
|
|
mapping = build_mapping(cells, r_idx + 1, self.saved_overrides)
|
|
if _DETECT_REQUIRED.issubset(set(mapping.field_to_col)):
|
|
score = len(mapping.field_to_col)
|
|
if score > best_score:
|
|
best, best_score = (r_idx, mapping), score
|
|
if not best:
|
|
raise ParseError(
|
|
f"'{self.filename}': could not find an Amazon transaction header row "
|
|
f"(need columns: date/time, settlement id, total)."
|
|
)
|
|
r_idx, mapping = best
|
|
self.header_row = r_idx + 1
|
|
self.column_mapping = mapping
|
|
self._field_to_idx = {
|
|
fld: _col_to_idx(col) for col, fld in mapping.col_to_field.items()
|
|
}
|
|
# Extra amount columns folded into an already-mapped field (ColumnMapping.sum_cols).
|
|
self._sum_field_idx = {
|
|
fld: [_col_to_idx(col) for col, _hdr in cols]
|
|
for fld, cols in mapping.sum_cols.items()
|
|
}
|
|
self.file_meta.data_sheet = self.sheet_name
|
|
self.file_meta.header_row = self.header_row
|
|
self.file_meta.unmapped_headers = mapping.unmapped
|
|
self.file_meta.missing_required = mapping.missing_required
|
|
self.file_meta.duplicate_fields = mapping.duplicate_fields
|
|
self.file_meta.summed_fields = mapping.sum_cols
|
|
self.file_meta.sheet_last_row = len(self._rows)
|
|
return mapping
|
|
|
|
def iter_records(self, only_fields: set[str] | None = None) -> Iterator[dict]:
|
|
if self.column_mapping is None:
|
|
self.detect()
|
|
assert self._rows is not None
|
|
idx_map = self._field_to_idx
|
|
sum_map = self._sum_field_idx
|
|
if only_fields is not None:
|
|
idx_map = {f: i for f, i in idx_map.items() if f in only_fields}
|
|
sum_map = {f: v for f, v in sum_map.items() if f in only_fields}
|
|
items = list(idx_map.items())
|
|
sum_items = list(sum_map.items())
|
|
mapped_idx = set(self._field_to_idx.values())
|
|
for _idxs in self._sum_field_idx.values():
|
|
mapped_idx.update(_idxs)
|
|
unmapped_sums = self.file_meta.unmapped_amount_sums
|
|
hdr = self.header_row
|
|
min_d: date | None = None
|
|
max_d: date | None = None
|
|
marketplace: str | None = None
|
|
count = 0
|
|
want_date = "date_time" in self._field_to_idx
|
|
|
|
for excel_row, row in enumerate(self._rows, start=1):
|
|
if excel_row <= hdr:
|
|
continue
|
|
rec = {
|
|
"_source_file": self.filename,
|
|
"_source_sheet": self.sheet_name,
|
|
"_source_row": excel_row,
|
|
}
|
|
has_value = False
|
|
for fld, i in items:
|
|
v = row[i] if i < len(row) else None
|
|
rec[fld] = _convert(fld, v)
|
|
if v not in (None, ""):
|
|
has_value = True
|
|
for fld, idxs in sum_items:
|
|
for i in idxs:
|
|
v = row[i] if i < len(row) else None
|
|
if v not in (None, ""):
|
|
rec[fld] = (rec.get(fld) or 0.0) + parse_amount(v)
|
|
has_value = True
|
|
if not has_value:
|
|
self.file_meta.blank_rows_skipped += 1
|
|
continue
|
|
if excel_row <= hdr + 2:
|
|
raw_sid = row[self._field_to_idx["settlement_id"]] \
|
|
if self._field_to_idx.get("settlement_id", 99999) < len(row) else None
|
|
sid = str(raw_sid or "").strip()
|
|
d_probe = parse_amazon_date_fast(rec.get("date_time")) \
|
|
if rec.get("date_time") else None
|
|
if d_probe is None and not sid.replace(".", "").isdigit():
|
|
self.file_meta.helper_rows_skipped += 1
|
|
continue
|
|
if len(row) > len(mapped_idx):
|
|
for i, v in enumerate(row):
|
|
if i not in mapped_idx and v not in (None, ""):
|
|
amt = parse_amount(v)
|
|
if amt:
|
|
col = _index_to_col(i)
|
|
unmapped_sums[col] = unmapped_sums.get(col, 0.0) + amt
|
|
if want_date and rec.get("date_time"):
|
|
d = parse_amazon_date_fast(rec["date_time"])
|
|
rec["_date"] = d
|
|
if d:
|
|
if min_d is None or d < min_d:
|
|
min_d = d
|
|
if max_d is None or d > max_d:
|
|
max_d = d
|
|
if marketplace is None and rec.get("marketplace"):
|
|
marketplace = rec["marketplace"]
|
|
count += 1
|
|
yield rec
|
|
|
|
self.file_meta.imported_rows = count
|
|
self.file_meta.min_date = min_d
|
|
self.file_meta.max_date = max_d
|
|
self.file_meta.marketplace = marketplace
|
|
|
|
|
|
def _col_to_idx(letters: str) -> int:
|
|
n = 0
|
|
for ch in letters:
|
|
n = n * 26 + (ord(ch) - 64)
|
|
return n - 1
|