Skip to content

Instantly share code, notes, and snippets.

@svin24
Created January 25, 2026 20:41
Show Gist options
  • Select an option

  • Save svin24/9cfef0a5593e215c14f5b3454ca08b79 to your computer and use it in GitHub Desktop.

Select an option

Save svin24/9cfef0a5593e215c14f5b3454ca08b79 to your computer and use it in GitHub Desktop.
#!/usr/bin/env python3
import csv
from collections import defaultdict
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from pathlib import Path
import argparse
def parse_args():
p = argparse.ArgumentParser(
description="Aggregate total received per recipient from each donor."
)
p.add_argument("input_csv", nargs="?", default="Table2a_Data.csv")
p.add_argument("output_txt", nargs="?", default="donor_recipient_totals.txt")
p.add_argument(
"--currency",
choices=["millions", "usd"],
default="usd",
help="Output totals in full USD (default) or USD millions.",
)
p.add_argument(
"--include-donor-totals",
action="store_true",
help="Include aggregate donors like '..., Total' (can cause double counting).",
)
return p.parse_args()
def is_aggregate_donor(name: str) -> bool:
# Many datasets include aggregate rows labeled with "Total".
return "total" in name.lower()
def format_total(value: Decimal, currency: str) -> str:
# Round to whole numbers and add thousands separators.
if currency == "usd":
value = (value * Decimal("1000000")).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
return f"{value:,} USD"
value = value.quantize(Decimal("1"), rounding=ROUND_HALF_UP)
return f"{value:,} million USD"
def main():
args = parse_args()
input_path = Path(args.input_csv)
output_path = Path(args.output_txt)
totals = defaultdict(Decimal) # key: (donor, recipient)
recipient_totals = defaultdict(Decimal)
donor_totals = defaultdict(Decimal)
with input_path.open(newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
donor = row.get("Donor", "").strip()
recipient = row.get("Recipient", "").strip()
value_raw = row.get("Value", "").strip()
if not donor or not recipient or not value_raw:
continue
if (not args.include_donor_totals) and is_aggregate_donor(donor):
continue
try:
value = Decimal(value_raw)
except InvalidOperation:
continue
totals[(donor, recipient)] += value
recipient_totals[recipient] += value
donor_totals[donor] += value
with output_path.open("w", encoding="utf-8") as out:
for donor, recipient in sorted(totals.keys()):
total = totals[(donor, recipient)]
out.write(f"{donor} -> {recipient}: {format_total(total, args.currency)}\n")
out.write("\nTotals by recipient:\n")
for recipient in sorted(recipient_totals.keys()):
total = recipient_totals[recipient]
out.write(f"RECIEVED: {recipient} {format_total(total, args.currency)}\n")
out.write("\nTotals by donor:\n")
for donor in sorted(donor_totals.keys()):
total = donor_totals[donor]
out.write(f"DONATED: {donor} {format_total(total, args.currency)}\n")
if __name__ == "__main__":
main()
@Malix-Labs
Copy link

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment