erp/.claude/skills/dolibarr-tva-summary/scripts/tva-summary.sh

#!/usr/bin/env bash
# CA3-ready monthly TVA summary for Arcodange — collectée − déductible = net.
#
# Usage:
#   tva-summary.sh [--year YYYY] [--since YYYY-MM-DD] [--until YYYY-MM-DD]
#
# Composes:
#   - customer-side invoice lines (TVA collectée, CA3 A1 / A4 / E2)
#   - supplier-side invoice lines (TVA déductible, CA3 19 / 20 / 17+24)
# Outputs one table per month plus a cumulative net line.
#
# Net interpretation:
#   - net > 0 → TVA à reverser à l'État
#   - net < 0 → crédit de TVA, demande de remboursement / report
#   - net = 0 → équilibre

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
DOL_CURL="${SCRIPT_DIR}/../../dolibarr/scripts/dol-curl.sh"

SINCE=""; UNTIL=""; YEAR=""
while [[ $# -gt 0 ]]; do
  case "$1" in
    --since) SINCE="$2"; shift 2 ;;
    --until) UNTIL="$2"; shift 2 ;;
    --year)  YEAR="$2"; SINCE="$2-01-01"; UNTIL="$2-12-31"; shift 2 ;;
    -h|--help) sed -n '2,15p' "$0" | sed 's/^# \{0,1\}//'; exit 0 ;;
    *) echo "tva-summary.sh: unknown arg: $1" >&2; exit 2 ;;
  esac
done

WORK="$(mktemp -d -t tvasum.XXXXXX)"
trap 'rm -rf "${WORK}"' EXIT

# 1. Customer side (invoices)
"${DOL_CURL}" '/invoices?limit=500&sortfield=t.datef&sortorder=ASC' > "${WORK}/inv.json"
IDS=$(python3 -c "import json,sys; print(' '.join(str(r['id']) for r in json.load(open(sys.argv[1])) if r.get('id')))" "${WORK}/inv.json")
SOCIDS=$(python3 -c "import json,sys; print(' '.join(sorted({str(r.get('socid')) for r in json.load(open(sys.argv[1])) if r.get('socid')})))" "${WORK}/inv.json")
mkdir -p "${WORK}/inv" "${WORK}/tp"
for id in ${IDS};    do "${DOL_CURL}" "/invoices/${id}"     > "${WORK}/inv/${id}.json"; done
for s  in ${SOCIDS}; do "${DOL_CURL}" "/thirdparties/${s}"  > "${WORK}/tp/${s}.json"; done

# 2. Supplier side (supplier invoices)
"${DOL_CURL}" '/supplierinvoices?limit=500' > "${WORK}/sinv.json"
SIDS=$(python3 -c "import json,sys; print(' '.join(str(r['id']) for r in json.load(open(sys.argv[1])) if r.get('id')))" "${WORK}/sinv.json")
SSOCIDS=$(python3 -c "import json,sys; print(' '.join(sorted({str(r.get('socid')) for r in json.load(open(sys.argv[1])) if r.get('socid')})))" "${WORK}/sinv.json")
mkdir -p "${WORK}/sinv"
for id in ${SIDS};   do "${DOL_CURL}" "/supplierinvoices/${id}" > "${WORK}/sinv/${id}.json"; done
for s in ${SSOCIDS}; do
  [[ -f "${WORK}/tp/${s}.json" ]] || "${DOL_CURL}" "/thirdparties/${s}" > "${WORK}/tp/${s}.json"
done

python3 - "${WORK}" "${SINCE}" "${UNTIL}" "${YEAR}" <<'PY'
import json, sys, os, collections, datetime
work, since, until, year_arg = sys.argv[1:5]

EU = set("AT BE BG HR CY CZ DK EE FI DE GR HU IE IT LV LT LU MT NL PL PT RO SK SI ES SE".split())

# Thirdparty country map
tp_cnty = {}
for fn in os.listdir(os.path.join(work, "tp")):
    try: d = json.load(open(os.path.join(work, "tp", fn)))
    except json.JSONDecodeError: continue
    tp_cnty[fn[:-len(".json")]] = d.get("country_code") or ""

def in_window(ts):
    if not ts: return False
    d = datetime.date.fromtimestamp(int(ts))
    if since and d < datetime.date.fromisoformat(since): return False
    if until and d > datetime.date.fromisoformat(until): return False
    return True

# Customer-side buckets
def bucket_collected(country, tx):
    if tx and tx > 0:
        return f"A1+8 ({tx}% collectée)"
    if country == "FR":           return "A1 0% (FR atypique)"
    if country in EU:             return "A4 (autoliquidation intra-UE)"
    if country:                   return "E2 (export hors UE)"
    return "(country missing)"

# Supplier-side buckets
def bucket_deductible(country, tx):
    if tx and tx > 0:             return f"ligne 19/20 ({tx}% déductible)"
    if country == "FR":           return "FR exempt / HT seulement"
    if country in EU:             return "ligne 17+24 (autoliquidation intra-UE)"
    if country:                   return "ligne 7 (import hors UE)"
    return "(country missing)"

# Aggregate per month
months = collections.defaultdict(lambda: {
    "collected": {"ht": 0.0, "tva": 0.0, "by_bucket": collections.defaultdict(lambda: {"ht":0.0,"tva":0.0})},
    "deductible":{"ht": 0.0, "tva": 0.0, "by_bucket": collections.defaultdict(lambda: {"ht":0.0,"tva":0.0})},
})

# Customer side
for fn in sorted(os.listdir(os.path.join(work, "inv"))):
    try: inv = json.load(open(os.path.join(work, "inv", fn)))
    except json.JSONDecodeError: continue
    ts = int(inv.get("date") or 0)
    if not in_window(ts): continue
    ym  = datetime.date.fromtimestamp(ts).strftime("%Y-%m")
    sid = str(inv.get("socid") or "")
    cnty = tp_cnty.get(sid, "")
    for line in inv.get("lines") or []:
        tx  = round(float(line.get("tva_tx")  or 0), 4)
        ht  = float(line.get("total_ht")  or 0)
        tva = float(line.get("total_tva") or 0)
        b   = bucket_collected(cnty, tx)
        months[ym]["collected"]["ht"]  += ht
        months[ym]["collected"]["tva"] += tva
        months[ym]["collected"]["by_bucket"][b]["ht"]  += ht
        months[ym]["collected"]["by_bucket"][b]["tva"] += tva

# Supplier side
for fn in sorted(os.listdir(os.path.join(work, "sinv"))):
    try: inv = json.load(open(os.path.join(work, "sinv", fn)))
    except json.JSONDecodeError: continue
    ts = int(inv.get("date") or 0)
    if not in_window(ts): continue
    ym  = datetime.date.fromtimestamp(ts).strftime("%Y-%m")
    sid = str(inv.get("socid") or "")
    cnty = tp_cnty.get(sid, "")
    lines = inv.get("lines") or []
    if not lines:
        ht = float(inv.get("total_ht") or 0); tva = float(inv.get("total_tva") or 0)
        tx = round((tva / ht * 100), 4) if ht else 0.0
        lines = [{"tva_tx": tx, "total_ht": ht, "total_tva": tva}]
    for line in lines:
        tx  = round(float(line.get("tva_tx")  or 0), 4)
        ht  = float(line.get("total_ht")  or 0)
        tva = float(line.get("total_tva") or 0)
        b   = bucket_deductible(cnty, tx)
        months[ym]["deductible"]["ht"]  += ht
        months[ym]["deductible"]["tva"] += tva
        months[ym]["deductible"]["by_bucket"][b]["ht"]  += ht
        months[ym]["deductible"]["by_bucket"][b]["tva"] += tva

scope = f"window={since or '-inf'} → {until or '+inf'}"
if year_arg: scope = f"year {year_arg}"

print(f"# Arcodange TVA monthly summary — {scope}")
print(f"# (composition of dolibarr-tva-reconciliation + dolibarr-tva-deductible)")
print()

cum_collected = cum_deductible = 0.0
for ym in sorted(months):
    m = months[ym]
    col_tva = m["collected"]["tva"]
    ded_tva = m["deductible"]["tva"]
    net     = col_tva - ded_tva
    cum_collected  += col_tva
    cum_deductible += ded_tva

    print(f"=== {ym} ===")
    print(f"  Customer side (TVA collectée)   basis HT={m['collected']['ht']:>10.2f}   TVA={col_tva:>8.2f}")
    for b, s in sorted(m["collected"]["by_bucket"].items()):
        print(f"      {b:<42}  HT={s['ht']:>10.2f}  TVA={s['tva']:>8.2f}")
    print(f"  Supplier side (TVA déductible)  basis HT={m['deductible']['ht']:>10.2f}   TVA={ded_tva:>8.2f}")
    for b, s in sorted(m["deductible"]["by_bucket"].items()):
        print(f"      {b:<42}  HT={s['ht']:>10.2f}  TVA={s['tva']:>8.2f}")
    if abs(net) < 0.005:
        verdict = "(équilibre)"
    elif net > 0:
        verdict = f"→ TVA À REVERSER : {net:.2f} €"
    else:
        verdict = f"→ CRÉDIT DE TVA : {-net:.2f} €"
    print(f"  --- Net du mois : collectée − déductible = {col_tva:.2f} − {ded_tva:.2f} = {net:>8.2f}   {verdict}")
    print()

cum_net = cum_collected - cum_deductible
print(f"=== CUMUL {scope} ===")
print(f"  TVA collectée totale  : {cum_collected:>10.2f}")
print(f"  TVA déductible totale : {cum_deductible:>10.2f}")
print(f"  Net cumulé            : {cum_net:>10.2f}", end="")
if abs(cum_net) < 0.005:
    print("   (équilibre)")
elif cum_net > 0:
    print(f"   → TVA À REVERSER cumulée : {cum_net:.2f} €")
else:
    print(f"   → CRÉDIT DE TVA cumulé : {-cum_net:.2f} €")
PY