generate_payment_report.py (3596B)
1 import argparse 2 import csv 3 from datetime import datetime, timedelta 4 import os 5 import sys 6 7 sys.path.append(os.path.join(os.path.dirname(__file__), "..")) 8 9 import config 10 from payments import database 11 from node import bitcoind, clightning, lnd, xpub 12 13 14 def valid_date(s): 15 try: 16 return datetime.strptime(s, "%Y-%m-%d") 17 except ValueError: 18 msg = "not a valid date: {0!r}".format(s) 19 raise argparse.ArgumentTypeError(msg) 20 21 22 def main(): 23 parser = argparse.ArgumentParser( 24 description="Generate CSV report about received payments.") 25 parser.add_argument("report_file") 26 parser.add_argument("--date-from", required=False, dest="date_from", 27 help="from date (YYYY-MM-DD)", type=valid_date) 28 parser.add_argument("--date-to", required=False, dest="date_to", 29 help="to date (YYYY-MM-DD)", type=valid_date) 30 31 try: 32 args = parser.parse_args() 33 except Exception as e: 34 print("Error: {}".format(e)) 35 return 36 37 nodes = {} 38 onchain = None 39 lightning = None 40 for method in config.payment_methods: 41 print("Connecting to {} node...".format(method["name"])) 42 if method["name"] == "bitcoind": 43 nodes["bitcoind"] = bitcoind.btcd(method) 44 onchain = "bitcoind" 45 elif method["name"] == "lnd": 46 nodes["lnd"] = lnd.lnd(method) 47 lightning = "lnd" 48 elif method["name"] == "clightning": 49 nodes["clightning"] = clightning.clightning(method) 50 lightning = "clightning" 51 elif method["name"] == "xpub": 52 nodes["xpub"] = xpub.xpub(method) 53 onchain = "xpub" 54 print("All nodes connected.") 55 56 where = "1" 57 if args.date_from: 58 where = where + " AND time >= {}".format( 59 datetime.timestamp(args.date_from)) 60 if args.date_to: 61 where = where + " AND time < {}".format( 62 datetime.timestamp(args.date_to + timedelta(days=1))) 63 invoices = database.load_invoices_from_db(where) 64 65 with open(args.report_file, "w", newline="") as csvfile: 66 reportwriter = csv.writer(csvfile) 67 reportwriter.writerow([ 68 "Date", "Invoice ID", "Base value", "Base currency", "BTC value", 69 "BTC paid", "Payment method", "Address" 70 ]) 71 num_rows = 0 72 for invoice in invoices: 73 if invoice["method"] == "onchain": 74 use_node_type = onchain 75 elif invoice["method"] == "lightning": 76 use_node_type = lightning 77 else: 78 use_node_type = invoice["method"] 79 if use_node_type == "lnd": 80 conf_paid, unconf_paid = nodes[use_node_type].check_payment( 81 invoice["rhash"]) 82 else: 83 conf_paid, unconf_paid = nodes[use_node_type].check_payment( 84 invoice["uuid"]) 85 86 if conf_paid > 0: 87 reportwriter.writerow([ 88 datetime.utcfromtimestamp( 89 int(invoice["time"])).strftime("%Y-%m-%d"), 90 invoice["uuid"], 91 invoice["base_value"], 92 invoice["base_currency"], 93 "%.8f" % float(invoice["btc_value"]), 94 "%.8f" % float(conf_paid), 95 invoice["method"], 96 invoice["address"] 97 ]) 98 num_rows = num_rows + 1 99 100 print("Report generated and saved to {} ({} rows).".format( 101 args.report_file, num_rows)) 102 103 104 if __name__ == "__main__": 105 main()