sap_test_data.py

"""VynFi → SAP-compatible test data.

Backs the blog post "How to Generate SAP-Compatible Test Data with VynFi".

Generates journal entries that map cleanly onto SAP's BKPF/BSEG tables, plus
master data (vendor, customer, G/L accounts) compatible with LFA1/LFB1, KNA1/KNB1,
and SKA1. Useful for:
- S/4HANA migration dress rehearsals
- FICO integration testing
- UAT dataset seeding
- Performance testing at production volumes
"""

import os

import pandas as pd

import vynfi

client = vynfi.VynFi(api_key=os.environ["VYNFI_API_KEY"], timeout=180.0)

# ── Generate SAP-shaped data ─────────────────────────────────────────────────
#
# VynFi's manufacturing + US GAAP preset produces SAP-compatible doc types
# (SA, KR, DR, AB), standard posting key semantics, and 4-digit company codes.

config = {
    "sector": "manufacturing",
    "country": "US",
    "accountingFramework": "us_gaap",
    "rows": 500,
    "companies": 3,
    "periods": 3,
    "periodLength": "monthly",
    "processModels": ["p2p", "o2c", "manufacturing", "h2r"],
    "exportFormat": "json",
    "fraudPacks": [],
    "fraudRate": 0.0,
}

print("Generating manufacturing data with SAP-compatible doc types...")
job = client.jobs.generate_config(config=config)
done = client.jobs.wait(job.id, timeout=300)
print(f"  Job: {done.id}, status: {done.status}")

if done.status != "completed":
    print(f"  Error: {done.error_detail}")
    raise SystemExit(1)

archive = client.jobs.download_archive(done.id)
print(f"  Archive: {archive}")

# ── Map journal entries to BKPF (header) + BSEG (line) ──────────────────────
#
# SAP Journal entry tables:
#   BKPF = document header   (BUKRS, BELNR, GJAHR, BUDAT, BLDAT, BLART, XBLNR)
#   BSEG = line item         (BUKRS, BELNR, BUZEI, BSCHL, HKONT, WRBTR, SHKZG)

entries = archive.json("journal_entries.json")

bkpf_rows, bseg_rows = [], []
for entry in entries:
    h = entry.get("header", entry)
    lines = entry.get("lines", [entry])

    bkpf_rows.append(
        {
            "BUKRS": h.get("company_code", ""),          # company code
            "BELNR": h.get("document_id", "")[:10],      # doc number (truncated)
            "GJAHR": h.get("fiscal_year", 2024),         # fiscal year
            "BLDAT": h.get("document_date", ""),         # document date
            "BUDAT": h.get("posting_date", ""),          # posting date
            "BLART": h.get("document_type", "SA")[:2],   # doc type
            "XBLNR": h.get("reference", ""),             # reference
            "BKTXT": h.get("header_text", ""),           # header text
            "MONAT": h.get("fiscal_period", 1),          # posting period
            "WAERS": h.get("currency", "USD"),           # currency
            "USNAM": h.get("created_by", ""),            # user
        }
    )

    for i, line in enumerate(lines, start=1):
        debit = line.get("debit_amount", 0) or 0
        credit = line.get("credit_amount", 0) or 0
        is_debit = float(str(debit)) > 0 if debit else False
        amount = float(str(debit if is_debit else credit))
        bseg_rows.append(
            {
                "BUKRS": h.get("company_code", ""),
                "BELNR": h.get("document_id", "")[:10],
                "GJAHR": h.get("fiscal_year", 2024),
                "BUZEI": f"{i:03d}",                     # line number
                "BSCHL": "40" if is_debit else "50",     # posting key (simple)
                "HKONT": line.get("gl_account", ""),     # G/L account
                "WRBTR": amount,                         # amount in doc currency
                "SHKZG": "S" if is_debit else "H",       # debit/credit indicator
                "KOSTL": line.get("cost_center", ""),    # cost center
                "PRCTR": line.get("profit_center", ""),  # profit center
                "SGTXT": line.get("line_text", ""),      # line item text
            }
        )

bkpf = pd.DataFrame(bkpf_rows)
bseg = pd.DataFrame(bseg_rows)
print(f"\n  BKPF: {len(bkpf):,} document headers")
print(f"  BSEG: {len(bseg):,} line items")

# ── Map vendor master → LFA1 (general) + LFB1 (company-code) ─────────────────

try:
    vendors = archive.json("master_data/vendors.json")
    lfa1_rows, lfb1_rows = [], []
    for v in vendors:
        lifnr = v.get("entity_id", v.get("vendor_id", ""))[:10]
        lfa1_rows.append(
            {
                "LIFNR": lifnr,
                "NAME1": v.get("name", ""),
                "ORT01": v.get("city", ""),
                "LAND1": v.get("country", "US"),
                "STCD1": v.get("tax_id", ""),           # tax number 1
                "STCD2": "",
            }
        )
        for cc in config.get("companies_list", ["1000", "2000", "3000"])[: config["companies"]]:
            lfb1_rows.append(
                {
                    "LIFNR": lifnr,
                    "BUKRS": cc,
                    "AKONT": v.get("recon_account", "200000"),  # reconciliation account
                    "ZTERM": v.get("payment_terms", "NT30"),    # payment terms
                }
            )
    lfa1 = pd.DataFrame(lfa1_rows)
    lfb1 = pd.DataFrame(lfb1_rows)
    print(f"  LFA1: {len(lfa1):,} vendors")
    print(f"  LFB1: {len(lfb1):,} vendor-company combos")
except KeyError:
    print("  (no vendor master in archive)")

# ── Export CSVs ──────────────────────────────────────────────────────────────

from pathlib import Path

out = Path("sap_export")
out.mkdir(exist_ok=True)
bkpf.to_csv(out / "bkpf.csv", index=False)
bseg.to_csv(out / "bseg.csv", index=False)
if "lfa1" in locals():
    lfa1.to_csv(out / "lfa1.csv", index=False)
    lfb1.to_csv(out / "lfb1.csv", index=False)

print(f"\nExported to {out}/")
for p in sorted(out.iterdir()):
    print(f"  {p.name}: {p.stat().st_size:,} bytes")

# ── Validate totals match SAP rules ──────────────────────────────────────────

print("\nSAP validation:")
bseg["WRBTR"] = pd.to_numeric(bseg["WRBTR"], errors="coerce").fillna(0)
debit_sum = bseg.loc[bseg["SHKZG"] == "S", "WRBTR"].sum()
credit_sum = bseg.loc[bseg["SHKZG"] == "H", "WRBTR"].sum()
print(f"  Debits  (S): ${debit_sum:>15,.2f}")
print(f"  Credits (H): ${credit_sum:>15,.2f}")
print(f"  Balanced: {abs(debit_sum - credit_sum) < 0.01}")

archive.close()