"""Persist engine results to the database, and stream-insert transactions efficiently.""" from __future__ import annotations import json from typing import Any from sqlalchemy.orm import Session as OrmSession from ..core.pipeline import ProcessResult from ..core.settlements import RECEIVABLE_ACCOUNT_TYPES from ..db import models from ..db.database import ENGINE _TXN_COLS = ( "session_id", "source_file", "source_sheet", "source_row", "marketplace", "settlement_id", "order_id", "sku", "txn_type", "txn_type_en", "account_type", "posted_date", "total", "currency", "storage_flag", ) _INSERT_SQL = ( f"INSERT INTO transactions ({', '.join(_TXN_COLS)}) " f"VALUES ({', '.join('?' * len(_TXN_COLS))})" ) class TransactionSink: """Batched raw-DBAPI inserts of normalized records for one session.""" def __init__(self, session_id: int, batch: int = 20000): self.session_id = session_id self.batch = batch self._buf: list[tuple] = [] self._conn = ENGINE.raw_connection() # Bulk-import tuning: durability is not critical here (we can re-run a failed job). cur = self._conn.cursor() cur.execute("PRAGMA synchronous=OFF") cur.execute("PRAGMA temp_store=MEMORY") cur.execute("PRAGMA busy_timeout=30000") cur.close() self.count = 0 def add(self, rec: dict[str, Any]) -> None: d = rec.get("_date") self._buf.append(( self.session_id, rec.get("_source_file"), rec.get("_source_sheet"), rec.get("_source_row"), rec.get("_marketplace"), rec.get("settlement_id"), rec.get("order_id"), rec.get("sku"), rec.get("txn_type"), rec.get("_type_en") or rec.get("txn_type"), rec.get("account_type"), d.isoformat() if d else None, float(rec.get("total") or 0.0), rec.get("currency") or "USD", 1 if rec.get("_storage") else 0, )) self.count += 1 if len(self._buf) >= self.batch: self._flush_buf() def _flush_buf(self) -> None: if not self._buf: return cur = self._conn.cursor() cur.executemany(_INSERT_SQL, self._buf) cur.close() # Commit each batch so we don't hold a long write lock that starves the progress # updater (a separate connection). synchronous=OFF keeps these commits cheap. self._conn.commit() self._buf.clear() def finish(self) -> None: self._flush_buf() self._conn.commit() def apply_classification(self, result: ProcessResult) -> None: """Bulk-set settlement_status / receivable_flag per settlement after classification.""" agg = result.aggregation cur = self._conn.cursor() for (mkt, acct, sid), st in agg.settlements.items(): recv = 1 if (st.status == "receivable" and acct.lower() in RECEIVABLE_ACCOUNT_TYPES) else 0 cur.execute( "UPDATE transactions SET settlement_status=?, receivable_flag=? " "WHERE session_id=? AND settlement_id=? AND marketplace=? AND account_type=?", (st.status, recv, self.session_id, sid, mkt, acct), ) # transfer rows: never receivable (canonical type covers localized names) cur.execute( "UPDATE transactions SET receivable_flag=0 " "WHERE session_id=? AND txn_type_en='Transfer'", (self.session_id,), ) cur.close() self._conn.commit() def close(self) -> None: try: self._conn.close() except Exception: pass def clear_session_results(db: OrmSession, session_id: int) -> None: for model in (models.Settlement, models.Exception_, models.ReceivableResultRow, models.ReconciliationRow, models.Transaction, models.MarketPayout): db.query(model).filter(model.session_id == session_id).delete() db.commit() # Every table that references sessions.id. Foreign keys are enforced (PRAGMA foreign_keys=ON), # so all children must go before the parent row. _CHILD_MODELS = ( models.Transaction, models.Settlement, models.ReceivableResultRow, models.ReconciliationRow, models.MarketPayout, models.Exception_, models.OpeningBalance, models.FxRate, models.FxRateDaily, models.Reserve, models.JournalEntry, models.FinanceControl, models.ExportRecord, models.SessionFile, ) def purge_session(db: OrmSession, session_id: int) -> dict: """Delete a closing and everything belonging to it, including files on disk.""" import os import shutil removed_files = 0 # Uploaded source files + their session folder. for f in db.query(models.SessionFile).filter(models.SessionFile.session_id == session_id): if f.stored_path and os.path.exists(f.stored_path): try: os.remove(f.stored_path) removed_files += 1 except OSError: pass from ..config import UPLOAD_DIR folder = UPLOAD_DIR / f"session_{session_id}" if folder.exists(): shutil.rmtree(folder, ignore_errors=True) # Generated workbooks. for e in db.query(models.ExportRecord).filter(models.ExportRecord.session_id == session_id): if e.path and os.path.exists(e.path): try: os.remove(e.path) removed_files += 1 except OSError: pass counts: dict[str, int] = {} for model in _CHILD_MODELS: n = db.query(model).filter(model.session_id == session_id).delete( synchronize_session=False) if n: counts[model.__tablename__] = n s = db.get(models.Session, session_id) if s is not None: db.delete(s) db.commit() return {"deleted": session_id, "rows_deleted": counts, "files_removed": removed_files} def persist_aggregates(db: OrmSession, session_id: int, result: ProcessResult, files: list[models.SessionFile]) -> None: agg, cls, rec, recon = (result.aggregation, result.classification, result.receivable, result.reconciliation) # update file metas meta_by_name = {m.filename: m for m in result.file_metas} for f in files: m = meta_by_name.get(f.filename) if not m: continue f.worksheets = json.dumps(m.worksheets) f.data_sheet = m.data_sheet f.imported_rows = m.imported_rows f.min_date = m.min_date f.max_date = m.max_date f.currency = m.currency f.marketplace = m.marketplace f.status = "invalid" if m.missing_required else "parsed" if m.missing_required: f.message = f"missing required columns: {m.missing_required}" # settlements (+ attach boundary transfer info) boundary_tx = {k: t for k, t in (cls.boundary_transfer or {}).items() if t} for (mkt, acct, sid), st in agg.settlements.items(): row = models.Settlement( session_id=session_id, marketplace=mkt, account_type=acct, settlement_id=sid, order_total=st.order_total, transfer_total=st.transfer_total, row_count=st.row_count, first_date=st.first_date, last_date=st.last_date, status=st.status, ) t = boundary_tx.get((mkt, acct)) if t and t.settlement_id == sid: row.transfer_amount = t.amount row.transfer_date = t.txn_date row.transfer_received = t.received db.add(row) # receivable results (per account + TOTAL) for mkt, m in (rec.marketplaces.items() if rec else []): for acct, a in m.accounts.items(): db.add(models.ReceivableResultRow( session_id=session_id, marketplace=mkt, account_type=acct, additional_sales=a.additional_sales, reserve=a.reserve, receivable_local=a.receivable_local_unrounded, fx_rate=m.fx_rate, receivable_usd=a.receivable_local_unrounded * m.fx_rate, currency=m.currency, )) db.add(models.ReceivableResultRow( session_id=session_id, marketplace=mkt, account_type="TOTAL", additional_sales=m.additional_sales, reserve=m.reserve, receivable_local=m.receivable_local, fx_rate=m.fx_rate, receivable_usd=m.receivable_usd, currency=m.currency, )) # reconciliation + AR roll-forward payout components transfers = agg.transfers if agg else [] received_payouts = sum(t.amount for t in transfers if t.received) all_payouts = sum(t.amount for t in transfers) # Per-marketplace payouts, attributed to the OWNER of each transfer's settlement # (Amazon repeats account-level payouts across country files). owner = (cls.settlement_owner if cls else {}) or {} per_mkt: dict[str, list[float]] = {} for t in transfers: mkt = owner.get(t.settlement_id, t.marketplace) slot = per_mkt.setdefault(mkt, [0.0, 0.0]) slot[1] += t.amount if t.received: slot[0] += t.amount for mkt, (recv_amt, all_amt) in per_mkt.items(): db.add(models.MarketPayout(session_id=session_id, marketplace=mkt, received_payouts=recv_amt, all_payouts=all_amt)) if recon: db.add(models.ReconciliationRow( session_id=session_id, uploaded_total=recon.uploaded_total, receivable_orders=recon.receivable_orders, paid_orders=recon.paid_orders, transfers_total=recon.transfers_total, reserve_total=recon.reserve_total, manual_adjustments=recon.manual_adjustments, final_receivable_usd=recon.final_receivable_usd, identity_difference=recon.identity_difference, status=recon.status, notes=json.dumps(recon.notes), received_payouts=received_payouts, all_payouts=all_payouts, )) # exceptions (engine-detected + file-coverage validation) from ..core.validation import validate_coverage session = db.get(models.Session, session_id) coverage_exc = validate_coverage(result.file_metas, session.month_end_date if session else None) for exc in _exceptions_from(result) + coverage_exc: db.add(models.Exception_(session_id=session_id, **exc)) db.commit() def _exceptions_from(result: ProcessResult) -> list[dict]: from ..core.i18n import STORAGE_EXPECTED_REGIONS out: list[dict] = [] for m in result.file_metas: for col, txt in (m.unmapped_headers or {}).items(): out.append({"category": "unmapped_column", "severity": "warning", "detail": f"Header '{txt}' (col {col}) not mapped — classify it under " f"Upload & Mapping to include it in a category", "source": m.filename}) for fld in m.missing_required: out.append({"category": "missing_column", "severity": "error", "detail": f"Required field '{fld}' missing", "source": m.filename}) for col, s in (getattr(m, "unmapped_amount_sums", None) or {}).items(): if abs(s) > 0.005: out.append({"category": "unmapped_amounts", "severity": "error", "detail": (f"Column {col} is not mapped but contains amounts summing " f"to {s:,.2f} — map the header so nothing is excluded"), "source": m.filename}) if getattr(m, "helper_rows_skipped", 0): out.append({"category": "helper_header_row", "severity": "info", "detail": (f"{m.helper_rows_skipped} translation/helper header row(s) " f"below the real header were skipped"), "source": m.filename}) if m.imported_rows == 0: out.append({"category": "empty_file", "severity": "warning", "detail": "No transaction rows found (header only)", "source": m.filename}) if result.aggregation: if result.aggregation.duplicate_count: out.append({"category": "duplicate_rows", "severity": "warning", "detail": (f"{result.aggregation.duplicate_count} duplicate row(s) detected " f"(same settlement+type+order+sku+time+amount). Examples: " f"{'; '.join(result.aggregation.duplicate_samples[:5])}"), "source": "combined"}) if agg_undated := result.aggregation.undated_count: out.append({ "category": "undated_rows", "severity": "warning", "detail": (f"{agg_undated} row(s) have no usable transaction date, carrying " f"{result.aggregation.undated_total:,.2f}. These look like manually " f"appended total/footer rows rather than Amazon transactions — review " f"before relying on the daily ledger. Examples: " f"{'; '.join(result.aggregation.undated_samples[:3])}"), "source": "combined"}) for t in result.aggregation.transfers: if not t.received: out.append({"category": "in_transit_disbursement", "severity": "info", "detail": (f"Transfer {t.amount:,.2f} on {t.txn_date} " f"(settlement {t.settlement_id}) not received by month-end; " f"settlement stays receivable"), "source": t.source_file}) # Storage-fee detection (req #8) agg = result.aggregation for mkt, amt in sorted((agg.potential_storage_by_mkt or {}).items()): out.append({"category": "potential_storage_fee", "severity": "warning", "detail": (f"{mkt}: storage-like transactions totalling {amt:,.2f} were " f"recorded under non-storage categories — review and classify. " f"Examples: {'; '.join(agg.potential_storage_samples[:3])}"), "source": mkt}) for mkt in sorted(agg.marketplaces_seen): if (mkt in STORAGE_EXPECTED_REGIONS and abs(agg.storage_total_by_mkt.get(mkt, 0.0)) < 0.005 and abs(agg.potential_storage_by_mkt.get(mkt, 0.0)) < 0.005): out.append({"category": "missing_storage_fees", "severity": "warning", "detail": (f"{mkt}: no FBA storage fees found this month — Amazon " f"normally bills storage monthly; verify none are hiding " f"under another category"), "source": mkt}) if result.classification and result.classification.cross_market: cls = result.classification for key in sorted(cls.cross_market): mkt, _acct, sid = key st = result.aggregation.settlements.get(key) owner = cls.settlement_owner.get(sid, "?") out.append({"category": "cross_market_settlement", "severity": "warning", "detail": (f"{mkt}: {st.row_count} row(s) totalling {st.order_total:,.2f} " f"belong to settlement {sid} owned by {owner} — excluded from " f"the {mkt} receivable (review under Transaction Details)"), "source": mkt}) return out