python3 syntax_binding.py \
  --input-csv invoice.csv \
  --bindings-csv bindings.csv \
  --xml-out invoice.xml \
  --roundtrip-csv roundtrip.csv \
  --print-xml

import argparse
import csv
import io
import re
import sys
import xml.etree.ElementTree as ET
from decimal import Decimal
from pathlib import Path
def read_core_csv(text: str) -> list[dict[str, str]]:
    """CSV文字列を辞書行のリストとして読み込む。"""
    return list(csv.DictReader(io.StringIO(text)))


def read_core_csv_file(path: str | Path, encoding: str = "utf-8-sig") -> list[dict[str, str]]:
    """構造化CSVファイルを読み込む。Excel由来のBOMを考慮し，既定をutf-8-sigにする。"""
    return read_core_csv(Path(path).read_text(encoding=encoding))

[
    {
        "dInvoice": "1",
        "invoiceNumber": "INV-26-861",
        "issueDate": "2026-06-20",
    },
    {
        "dInvoice": "1",
        "dInvoiceParty": "1",
        "partyRole": "Seller",
        "partyName": "売手株式会社",
        "partyTaxID": "T1234567890123",
    },
]

def read_bindings_csv(text: str) -> list[Binding]:
    """semantic_path,xpath の2列を持つ構文結合表CSVを読み込む。"""
    bindings: list[Binding] = []

    reader = csv.DictReader(io.StringIO(text))
    if reader.fieldnames is None:
        raise ValueError("Binding CSV has no header")

    # UTF-8 BOM対策。
    fieldnames = [name.lstrip("\ufeff") for name in reader.fieldnames]
    semantic_col = "semantic_path"
    xpath_col = "xpath"

    if semantic_col not in fieldnames or xpath_col not in fieldnames:
        raise ValueError("Binding CSV must have semantic_path and xpath columns")

    for line_no, row in enumerate(reader, start=2):
        normalized = {k.lstrip("\ufeff"): (v or "").strip() for k, v in row.items()}
        semantic_path = normalized.get(semantic_col, "")
        xpath = normalized.get(xpath_col, "")

        if not semantic_path and not xpath:
            continue
        if not semantic_path or not xpath:
            raise ValueError(f"Incomplete binding at line {line_no}: {row!r}")
        if not xpath.startswith("/Invoice/") and xpath != "/Invoice":
            raise ValueError(f"XPath must start with /Invoice at line {line_no}: {xpath!r}")

        bindings.append((semantic_path, xpath))

    return bindings


def read_bindings_csv_file(path: str | Path, encoding: str = "utf-8-sig") -> list[Binding]:
    return read_bindings_csv(Path(path).read_text(encoding=encoding))

[
    ("$.Invoice.invoiceNumber", "/Invoice/cbc:ID"),
    ("$.Invoice.dInvoiceParty[?@.partyRole=Seller].partyName",
     "/Invoice/cac:AccountingSupplierParty/cac:Party/cac:PartyName/cbc:Name"),
]

$.Invoice.field
$.Invoice.dimension[?@.filterField=filterValue].field

$.Invoice.invoiceNumber

$.Invoice.dInvoiceParty[?@.partyRole=Seller].partyName

def rows_by_dimension(
    rows: list[dict[str, str]],
    dimension: str,
    d_invoice: str = "1",
) -> list[dict[str, str]]:
    return [r for r in rows if r.get("dInvoice") == d_invoice and r.get(dimension)]


def first_value(
    rows: list[dict[str, str]],
    column: str,
    d_invoice: str = "1",
) -> str:
    for row in rows:
        if row.get("dInvoice") == d_invoice and row.get(column):
            return row[column]
    return ""


def query_semantic_path(
    rows: list[dict[str, str]],
    path: str,
    d_invoice: str = "1",
) -> str:
    """
    このサンプルで扱う SemanticPath subset:
      $.Invoice.field
      $.Invoice.dimension[?@.filterField="filterValue"].field
      $.Invoice.dimension[?@.filterField=filterValue].field
    """

    m = re.fullmatch(r"\$\.Invoice\.([A-Za-z_][A-Za-z0-9_]*)", path)
    if m:
        return first_value(rows, m.group(1), d_invoice)

    m = re.fullmatch(
        r"\$\.Invoice\.([A-Za-z_][A-Za-z0-9_]*)"
        r"\[\?@\.([A-Za-z_][A-Za-z0-9_]*)=(?:\"([^\"]+)\"|([^\]]+))\]"
        r"\.([A-Za-z_][A-Za-z0-9_]*)",
        path,
    )
    if m:
        dimension, filter_field, quoted_value, bare_value, target_field = m.groups()
        filter_value = quoted_value if quoted_value is not None else bare_value

        for row in rows_by_dimension(rows, dimension, d_invoice):
            if row.get(filter_field) == filter_value:
                return row.get(target_field, "")

        return ""

    raise ValueError(f"Unsupported SemanticPath: {path}")

UBL_NS = "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"
CBC_NS = "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2"
CAC_NS = "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"

NS = {
    "": UBL_NS,
    "cbc": CBC_NS,
    "cac": CAC_NS,
}

for prefix, uri in NS.items():
    ET.register_namespace(prefix, uri)


def qname(name: str) -> str:
    if ":" in name:
        prefix, local = name.split(":", 1)
        return f"{{{NS[prefix]}}}{local}"
    return f"{{{UBL_NS}}}{name}"

{名前空間URI}ローカル名

{urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2}ID

/Invoice/cac:AccountingSupplierParty/cac:Party/cac:PartyName/cbc:Name

def normalize_xpath(xpath: str) -> str:
    xpath = xpath.strip()
    # 結合表の正規表記では，途中空白を許容しない。
    if re.search(r"\s", xpath):
        raise ValueError(f"XPath contains whitespace: {xpath!r}")
    return xpath


def ensure_path(root: ET.Element, xpath: str) -> ET.Element:
    xpath = normalize_xpath(xpath)
    parts = [p for p in xpath.split("/") if p]

    if not parts or parts[0] != "Invoice":
        raise ValueError(f"XPath must start with /Invoice: {xpath}")

    elem = root
    for part in parts[1:]:
        tag = qname(part)
        child = elem.find(tag)
        if child is None:
            child = ET.SubElement(elem, tag)
        elem = child

    return elem


def set_text(root: ET.Element, xpath: str, value: str | None) -> None:
    if value not in (None, ""):
        ensure_path(root, xpath).text = str(value)


def get_text(root: ET.Element, xpath: str) -> str:
    xpath = normalize_xpath(xpath)
    parts = [p for p in xpath.split("/") if p]

    if not parts or parts[0] != "Invoice":
        raise ValueError(f"XPath must start with /Invoice: {xpath}")

    elem = root
    for part in parts[1:]:
        elem = elem.find(qname(part))
        if elem is None:
            return ""

    return elem.text or ""

elem = ensure_path(
    invoice,
    "/Invoice/cac:AccountingSupplierParty/cac:Party/cac:PartyName/cbc:Name"
)
elem.text = "売手株式会社"

<cac:AccountingSupplierParty>
  <cac:Party>
    <cac:PartyName>
      <cbc:Name>売手株式会社</cbc:Name>
    </cac:PartyName>
  </cac:Party>
</cac:AccountingSupplierParty>

1. SemanticPathでCSVから値を取得する。
2. XPathでXML上の設定先を決める。
3. 取得した値をXML要素に設定する。

def csv_to_ubl_xml(
    rows: list[dict[str, str]],
    bindings: list[Binding],
    d_invoice: str = "1",
) -> ET.Element:
    invoice = ET.Element(qname("Invoice"))

    # 単一項目及び条件付き単一項目は，構文結合表に従って設定する。
    for semantic_path, xpath in bindings:
        value = query_semantic_path(rows, semantic_path, d_invoice)
        set_text(invoice, xpath, value)

    currency = query_semantic_path(rows, "$.Invoice.currencyCode", d_invoice)

    # 繰返し構造は，行の集合をXMLの繰返し要素へ展開する。
    append_tax_total(invoice, rows, currency, d_invoice)
    append_invoice_lines(invoice, rows, currency, d_invoice)

    return invoice

def new_empty_row(fieldnames: list[str], d_invoice: str = "1") -> dict[str, str]:
    row = {name: "" for name in fieldnames}
    row["dInvoice"] = d_invoice
    return row


def append_header_row(
    rows: list[dict[str, str]],
    invoice: ET.Element,
    fieldnames: list[str],
    d_invoice: str = "1",
) -> None:
    row = new_empty_row(fieldnames, d_invoice)
    row["invoiceNumber"] = get_text(invoice, "/Invoice/cbc:ID")
    row["issueDate"] = get_text(invoice, "/Invoice/cbc:IssueDate")
    row["invoiceTypeCode"] = get_text(invoice, "/Invoice/cbc:InvoiceTypeCode")
    row["currencyCode"] = get_text(invoice, "/Invoice/cbc:DocumentCurrencyCode")
    rows.append(row)


def append_party_rows(
    rows: list[dict[str, str]],
    invoice: ET.Element,
    fieldnames: list[str],
    d_invoice: str = "1",
) -> None:
    party_specs = [
        ("1", "Seller", "/Invoice/cac:AccountingSupplierParty/cac:Party"),
        ("2", "Buyer", "/Invoice/cac:AccountingCustomerParty/cac:Party"),
    ]

    for d_party, role, party_xpath in party_specs:
        parts = [p for p in party_xpath.split("/") if p]
        party = invoice

        for part in parts[1:]:
            party = party.find(qname(part))
            if party is None:
                break

        if party is None:
            continue

        row = new_empty_row(fieldnames, d_invoice)
        row["dInvoiceParty"] = d_party
        row["partyRole"] = role

        name = party.find(f"{qname('cac:PartyName')}/{qname('cbc:Name')}")
        tax_id = party.find(f"{qname('cac:PartyTaxScheme')}/{qname('cbc:CompanyID')}")

        row["partyName"] = name.text if name is not None and name.text else ""
        row["partyTaxID"] = tax_id.text if tax_id is not None and tax_id.text else ""

        rows.append(row)


def append_document_reference_rows(
    rows: list[dict[str, str]],
    invoice: ET.Element,
    fieldnames: list[str],
    d_invoice: str = "1",
) -> None:
    ref_specs = [
        ("Order", "/Invoice/cac:OrderReference/cbc:ID"),
        ("Delivery", "/Invoice/cac:DespatchDocumentReference/cbc:ID"),
    ]

    ref_no = 1
    for ref_type, ref_xpath in ref_specs:
        ref_id = get_text(invoice, ref_xpath)
        if ref_id:
            row = new_empty_row(fieldnames, d_invoice)
            row["dDocumentReference"] = str(ref_no)
            row["referenceType"] = ref_type
            row["referenceID"] = ref_id
            rows.append(row)
            ref_no += 1


def append_payment_row(
    rows: list[dict[str, str]],
    invoice: ET.Element,
    fieldnames: list[str],
    d_invoice: str = "1",
) -> None:
    due_date = get_text(invoice, "/Invoice/cbc:DueDate")
    payment_id = get_text(invoice, "/Invoice/cac:PaymentMeans/cbc:PaymentID")

    if due_date or payment_id:
        row = new_empty_row(fieldnames, d_invoice)
        row["dPayment"] = "1"
        row["paymentDueDate"] = due_date
        row["paymentReference"] = payment_id
        rows.append(row)


def append_tax_breakdown_rows(
    rows: list[dict[str, str]],
    invoice: ET.Element,
    fieldnames: list[str],
    d_invoice: str = "1",
) -> None:
    tax_total = invoice.find(qname("cac:TaxTotal"))

    if tax_total is None:
        return

    for i, subtotal in enumerate(tax_total.findall(qname("cac:TaxSubtotal")), start=1):
        row = new_empty_row(fieldnames, d_invoice)
        row["dTaxBreakdown"] = str(i)

        taxable = subtotal.find(qname("cbc:TaxableAmount"))
        tax_amount = subtotal.find(qname("cbc:TaxAmount"))
        category = subtotal.find(qname("cac:TaxCategory"))

        row["taxableAmount"] = taxable.text if taxable is not None and taxable.text else ""
        row["taxAmount"] = tax_amount.text if tax_amount is not None and tax_amount.text else ""

        if category is not None:
            cat_id = category.find(qname("cbc:ID"))
            percent = category.find(qname("cbc:Percent"))

            row["taxCategoryCode"] = cat_id.text if cat_id is not None and cat_id.text else ""
            row["taxRate"] = percent.text if percent is not None and percent.text else ""

        rows.append(row)


def append_invoice_line_rows(
    rows: list[dict[str, str]],
    invoice: ET.Element,
    fieldnames: list[str],
    d_invoice: str = "1",
) -> None:
    for i, line in enumerate(invoice.findall(qname("cac:InvoiceLine")), start=1):
        row = new_empty_row(fieldnames, d_invoice)
        row["dInvoiceLine"] = str(i)

        line_id = line.find(qname("cbc:ID"))
        quantity = line.find(qname("cbc:InvoicedQuantity"))
        line_amount = line.find(qname("cbc:LineExtensionAmount"))
        item = line.find(qname("cac:Item"))
        price = line.find(qname("cac:Price"))

        row["lineID"] = line_id.text if line_id is not None and line_id.text else ""

        if quantity is not None:
            row["invoicedQuantity"] = quantity.text or ""
            row["unitCode"] = quantity.attrib.get("unitCode", "")

        row["lineExtensionAmount"] = line_amount.text if line_amount is not None and line_amount.text else ""

        if item is not None:
            name = item.find(qname("cbc:Name"))
            row["itemName"] = name.text if name is not None and name.text else ""

            tax_category = item.find(qname("cac:ClassifiedTaxCategory"))
            if tax_category is not None:
                cat_id = tax_category.find(qname("cbc:ID"))
                percent = tax_category.find(qname("cbc:Percent"))

                row["lineTaxCategoryCode"] = cat_id.text if cat_id is not None and cat_id.text else ""
                row["lineTaxRate"] = percent.text if percent is not None and percent.text else ""

        if price is not None:
            price_amount = price.find(qname("cbc:PriceAmount"))
            row["unitPriceAmount"] = price_amount.text if price_amount is not None and price_amount.text else ""

        rows.append(row)


def ubl_xml_to_csv_rows(
    invoice: ET.Element,
    fieldnames: list[str],
    d_invoice: str = "1",
) -> list[dict[str, str]]:
    rows: list[dict[str, str]] = []

    append_header_row(rows, invoice, fieldnames, d_invoice)
    append_party_rows(rows, invoice, fieldnames, d_invoice)
    append_document_reference_rows(rows, invoice, fieldnames, d_invoice)
    append_payment_row(rows, invoice, fieldnames, d_invoice)
    append_tax_breakdown_rows(rows, invoice, fieldnames, d_invoice)
    append_invoice_line_rows(rows, invoice, fieldnames, d_invoice)

    return rows

def rows_to_csv(rows: list[dict[str, str]], fieldnames: list[str]) -> str:
    out = io.StringIO()
    writer = csv.DictWriter(out, fieldnames=fieldnames, lineterminator="\n")
    writer.writeheader()
    writer.writerows(rows)
    return out.getvalue()


def pretty_xml(elem: ET.Element) -> str:
    ET.indent(elem, space="  ")
    return ET.tostring(elem, encoding="unicode")


def write_xml(elem: ET.Element, path: str | Path) -> None:
    tree = ET.ElementTree(elem)
    ET.indent(tree, space="  ")
    tree.write(path, encoding="utf-8", xml_declaration=True)

def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Convert structured invoice CSV to a JP PINT UBL subset and back."
    )
    parser.add_argument("--input-csv", default="invoice.csv", help="structured invoice CSV")
    parser.add_argument("--bindings-csv", default="bindings.csv", help="semantic_path,xpath binding CSV")
    parser.add_argument("--xml-out", default="invoice.xml", help="output XML file")
    parser.add_argument("--roundtrip-csv", default="roundtrip.csv", help="round-trip CSV output")
    parser.add_argument("--d-invoice", default="1", help="target dInvoice value")
    parser.add_argument("--print-xml", action="store_true", help="print generated XML to stdout")
    parser.add_argument("--print-csv", action="store_true", help="print round-trip CSV to stdout")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(sys.argv[1:] if argv is None else argv)

    rows = read_core_csv_file(args.input_csv)
    if not rows:
        raise ValueError(f"No rows in input CSV: {args.input_csv}")

    bindings = read_bindings_csv_file(args.bindings_csv)
    fieldnames = list(rows[0].keys())

    invoice_xml = csv_to_ubl_xml(rows, bindings, args.d_invoice)
    write_xml(invoice_xml, args.xml_out)

    roundtrip_rows = ubl_xml_to_csv_rows(invoice_xml, fieldnames, args.d_invoice)
    Path(args.roundtrip_csv).write_text(
        rows_to_csv(roundtrip_rows, fieldnames),
        encoding="utf-8-sig",
    )

    if args.print_xml:
        print("=== CSV -> JP PINT UBL subset ===")
        print(pretty_xml(invoice_xml))

    if args.print_csv:
        print("=== JP PINT UBL subset -> structured CSV ===")
        print(rows_to_csv(roundtrip_rows, fieldnames))

    print(f"XML written: {args.xml_out}")
    print(f"CSV written: {args.roundtrip_csv}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python3 syntax_binding.py \
  --input-csv invoice.csv \
  --bindings-csv bindings.csv \
  --xml-out invoice.xml \
  --roundtrip-csv roundtrip.csv \
  --print-xml

<?xml version='1.0' encoding='utf-8'?>
<Invoice xmlns="urn:oasis:names:specification:ubl:schema:xsd:Invoice-2" xmlns:cac="urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2" xmlns:cbc="urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2">
  <cbc:ID>INV-26-861</cbc:ID>
  <cbc:IssueDate>2026-06-20</cbc:IssueDate>
  <cbc:InvoiceTypeCode>380</cbc:InvoiceTypeCode>
  <cbc:DocumentCurrencyCode>JPY</cbc:DocumentCurrencyCode>
  <cac:AccountingSupplierParty>
    <cac:Party>
      <cac:PartyName>
        <cbc:Name>売手株式会社</cbc:Name>
      </cac:PartyName>
      <cac:PartyTaxScheme>
        <cbc:CompanyID>T1234567890123</cbc:CompanyID>
      </cac:PartyTaxScheme>
    </cac:Party>
  </cac:AccountingSupplierParty>
  <cac:AccountingCustomerParty>
    <cac:Party>
      <cac:PartyName>
        <cbc:Name>買手株式会社</cbc:Name>
      </cac:PartyName>
    </cac:Party>
  </cac:AccountingCustomerParty>
  <cac:OrderReference>
    <cbc:ID>PO-2026-0001</cbc:ID>
  </cac:OrderReference>
  <cac:DespatchDocumentReference>
    <cbc:ID>DN-2026-0001</cbc:ID>
  </cac:DespatchDocumentReference>
  <cbc:DueDate>2026-07-31</cbc:DueDate>
  <cac:PaymentMeans>
    <cbc:PaymentID>INV-2026-0001</cbc:PaymentID>
  </cac:PaymentMeans>
  <cac:TaxTotal>
    <cbc:TaxAmount currencyID="JPY">1472</cbc:TaxAmount>

dInvoice,dInvoiceParty,dDocumentReference,dPayment,dTaxBreakdown,dInvoiceLine,invoiceNumber,issueDate,invoiceTypeCode,currencyCode,totalAmount,partyRole,partyName,partyTaxID,referenceType,referenceID,paymentDueDate,paymentMeansCode,paymentReference,taxCategoryCode,taxRate,taxableAmount,taxAmount,lineID,itemName,lineTaxCategoryCode,lineTaxRate,invoicedQuantity,unitCode,unitPriceAmount,lineExtensionAmount
1,,,,,,INV-26-861,2026-06-20,380,JPY,,,,,,,,,,,,,,,,,,,,,
1,1,,,,,,,,,,Seller,売手株式会社,T1234567890123,,,,,,,,,,,,,,,,,
1,2,,,,,,,,,,Buyer,買手株式会社,,,,,,,,,,,,,,,,,,
1,,1,,,,,,,,,,,,Order,PO-2026-0001,,,,,,,,,,,,,,,
1,,2,,,,,,,,,,,,Delivery,DN-2026-0001,,,,,,,,,,,,,,,
1,,,1,,,,,,,,,,,,,2026-07-31,,INV-2026-0001,,,,,,,,,,,,
1,,,,1,,,,,,,,,,,,,,,S,10,12320,1232,,,,,,,,
1,,,,2,,,,,,,,,,,,,,,AA,8,3010,240,,,,,,,,
1,,,,,1,,,,,,,,,,,,,,,,,,INV-26-861-1,商品A,S,10,10,EA,1000,10000
1,,,,,2,,,,,,,,,,,,,,,,,,INV-26-861-2,商品B,S,10,4,EA,580,2320
1,,,,,3,,,,,,,,,,,,,,,,,,INV-26-861-3,食品C,AA,8,7,EA,430,3010

SemanticPathで値を取得する。
XPathでXML要素を作る。
値を設定する。
逆方向では，XPathで値を取得し，構造化CSV行へ戻す。