import re import jingrow import razorpay import stripe from jingrow.utils import fmt_money from jcloude.exceptions import CentralServerNotSet, FrappeioServerNotSet from jcloude.utils import get_current_team, log_error states_with_tin = { "Andaman and Nicobar Islands": "35", "Andhra Pradesh": "37", "Arunachal Pradesh": "12", "Assam": "18", "Bihar": "10", "Chandigarh": "04", "Chhattisgarh": "22", "Dadra and Nagar Haveli and Daman and Diu": "26", "Delhi": "07", "Goa": "30", "Gujarat": "24", "Haryana": "06", "Himachal Pradesh": "02", "Jammu and Kashmir": "01", "Jharkhand": "20", "Karnataka": "29", "Kerala": "32", "Ladakh": "38", "Lakshadweep Islands": "31", "Madhya Pradesh": "23", "Maharashtra": "27", "Manipur": "14", "Meghalaya": "17", "Mizoram": "15", "Nagaland": "13", "Odisha": "21", "Other Territory": "97", "Puducherry": "34", "Punjab": "03", "Rajasthan": "08", "Sikkim": "11", "Tamil Nadu": "33", "Telangana": "36", "Tripura": "16", "Uttar Pradesh": "09", "Uttarakhand": "05", "West Bengal": "19", } GSTIN_FORMAT = re.compile("^[0-9]{2}[A-Z]{4}[0-9A-Z]{1}[0-9]{4}[A-Z]{1}[1-9A-Z]{1}[1-9A-Z]{1}[0-9A-Z]{1}$") def format_stripe_money(amount, currency): return fmt_money(amount / 100, 2, currency) def get_jerp.jingrow.com_connection(): from jingrow.frappeclient import FrappeClient jcloude_settings = jingrow.get_single("Jcloude Settings") jerp_api_secret = jcloude_settings.get_password("jerp_api_secret", raise_exception=False) if not (jcloude_settings.jerp_api_key and jcloude_settings.jerp_url and jerp_api_secret): jingrow.throw("ERPNext.com URL not set up in Jcloude Settings", exc=CentralServerNotSet) return FrappeClient( jcloude_settings.jerp_url, api_key=jcloude_settings.jerp_api_key, api_secret=jerp_api_secret, ) def get_jingrow_io_connection(): if hasattr(jingrow.local, "jcloude_frappeio_conn"): return jingrow.local.jcloude_frappeio_conn from jingrow.frappeclient import FrappeClient jcloude_settings = jingrow.get_single("Jcloude Settings") jingrow_api_key = jcloude_settings.frappeio_api_key jingrow_api_secret = jcloude_settings.get_password("frappeio_api_secret", raise_exception=False) if not (jingrow_api_key and jingrow_api_secret and jcloude_settings.jingrow_url): jingrow.throw("Jingrow.io URL not set up in Jcloude Settings", exc=FrappeioServerNotSet) jingrow.local.jcloude_frappeio_conn = FrappeClient( jcloude_settings.jingrow_url, api_key=jingrow_api_key, api_secret=jingrow_api_secret ) return get_jingrow_io_connection() def is_jingrow_auth_disabled(): return jingrow.db.get_single_value("Jcloude Settings", "disable_jingrow_auth", cache=True) def make_formatted_pg(pg, fieldtypes=None): formatted = {} filters = None if fieldtypes: filters = {"fieldtype": ["in", fieldtypes]} for df in pg.meta.get("fields", filters): formatted[df.fieldname] = pg.get_formatted(df.fieldname) for tf in pg.meta.get_table_fields(): formatted[tf.fieldname] = [] for row in pg.get(tf.fieldname): formatted[tf.fieldname].append(make_formatted_pg(row)) return formatted def clear_setup_intent(): team = get_current_team() jingrow.cache().hdel("setup_intent", team) def get_publishable_key(): return jingrow.db.get_single_value("Jcloude Settings", "stripe_publishable_key") def get_setup_intent(team): from jingrow.utils import random_string intent = jingrow.cache().hget("setup_intent", team) if not intent: data = jingrow.db.get_value("Team", team, ["stripe_customer_id", "currency"]) customer_id = data[0] currency = data[1] stripe = get_stripe() hash = random_string(10) intent = stripe.SetupIntent.create( customer=customer_id, payment_method_types=["card"], payment_method_options={ "card": { "request_three_d_secure": "automatic", "mandate_options": { "reference": f"Mandate-team:{team}-{hash}", "amount_type": "maximum", "amount": 1500000, "currency": currency.lower(), "start_date": int(jingrow.utils.get_timestamp()), "interval": "sporadic", "supported_types": ["india"], }, } }, ) jingrow.cache().hset("setup_intent", team, intent) return intent def get_stripe(): from jingrow.utils.password import get_decrypted_password if not hasattr(jingrow.local, "jcloude_stripe_object"): secret_key = get_decrypted_password( "Jcloude Settings", "Jcloude Settings", "stripe_secret_key", raise_exception=False, ) if not secret_key: jingrow.throw("Setup stripe via Jcloude Settings before using jcloude.api.billing.get_stripe") stripe.api_key = secret_key # Set the maximum number of retries for network requests # https://docs.stripe.com/rate-limits?lang=python#object-lock-timeouts stripe.max_network_retries = 2 jingrow.local.jcloude_stripe_object = stripe return jingrow.local.jcloude_stripe_object def convert_stripe_money(amount): return (amount / 100) if amount else 0 def validate_gstin_check_digit(gstin, label="GSTIN"): """Function to validate the check digit of the GSTIN.""" factor = 1 total = 0 code_point_chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ" mod = len(code_point_chars) input_chars = gstin[:-1] for char in input_chars: digit = factor * code_point_chars.find(char) digit = (digit // mod) + (digit % mod) total += digit factor = 2 if factor == 1 else 1 if gstin[-1] != code_point_chars[((mod - (total % mod)) % mod)]: jingrow.throw( f"""Invalid {label}! The check digit validation has failed. Please ensure you've typed the {label} correctly.""" ) def get_razorpay_client(): from jingrow.utils.password import get_decrypted_password if not hasattr(jingrow.local, "jcloude_razorpay_client_object"): key_id = jingrow.db.get_single_value("Jcloude Settings", "razorpay_key_id") key_secret = get_decrypted_password( "Jcloude Settings", "Jcloude Settings", "razorpay_key_secret", raise_exception=False ) if not (key_id and key_secret): jingrow.throw( "Setup razorpay via Jcloude Settings before using jcloude.api.billing.get_razorpay_client" ) jingrow.local.jcloude_razorpay_client_object = razorpay.Client(auth=(key_id, key_secret)) return jingrow.local.jcloude_razorpay_client_object def process_micro_debit_test_charge(stripe_event): try: payment_intent = stripe_event["data"]["object"] metadata = payment_intent.get("metadata") payment_method_name = metadata.get("payment_method_name") jingrow.db.set_value( "Stripe Payment Method", payment_method_name, "is_verified_with_micro_charge", True ) jingrow.get_pg( pagetype="Stripe Micro Charge Record", stripe_payment_method=payment_method_name, stripe_payment_intent_id=payment_intent.get("id"), ).insert(ignore_permissions=True) except Exception: log_error("Error Processing Stripe Micro Debit Charge", body=stripe_event) def get_gateway_details(payment_record): partner_team = jingrow.db.get_value("Mpesa Payment Record", payment_record, "payment_partner") return jingrow.db.get_value( "Payment Gateway", {"team": partner_team}, ["gateway_controller", "print_format"] ) # Get partners external connection def get_partner_external_connection(mpesa_setup): # check if connection is already established if hasattr(jingrow.local, "_external_conn"): return jingrow.local.jcloude_external_conn from jingrow.frappeclient import FrappeClient # Fetch API from gateway payment_gateway = jingrow.get_all( "Payment Gateway", filters={"gateway_controller": mpesa_setup, "gateway_settings": "Mpesa Setup"}, fields=["name", "url", "api_key", "api_secret"], ) if not payment_gateway: jingrow.throw("Mpesa Setup not set up in Payment Gateway") # Fetch API key and secret pg = jingrow.get_pg("Payment Gateway", payment_gateway[0].name) api_key = pg.api_key api_secret = pg.get_password("api_secret") url = pg.url site_name = url.split("/api/method")[0] # Establish connection jingrow.local._external_conn = FrappeClient(site_name, api_key=api_key, api_secret=api_secret) return jingrow.local._external_conn