import os import mysql.connector from datetime import datetime from dotenv import load_dotenv import requests import time import json import traceback from google.oauth2 import service_account from googleapiclient.discovery import build from fastapi import APIRouter, Depends, Request router = APIRouter() # ---------- Load Environment ---------- load_dotenv() DB_HOST = os.getenv("DB_HOST", "localhost") DB_PORT = int(os.getenv("DB_PORT", 3306)) DB_USER = os.getenv("DB_USER", "root") DB_PASSWORD = os.getenv("DB_PASSWORD", "") DB_NAME = os.getenv("DB_NAME", "") DB_SECRET = os.getenv("SUBSCRIPTION_SECRET", "") IOS_PLANS = ["com.tattoo.apps.yearly", "com.tattoo.apps.montly"] ANDROID_PLANS = ["com.vpn.check", "com.vpn.check.basic", "com.vpn.check.premium"] appstore_receipt_validator_url_prod= os.getenv("APPSTORE_RECEIPT_VALIDATOR_URL_PROD") appstore_receipt_validator_url_sandbox= os.getenv("APPSTORE_RECEIPT_VALIDATOR_URL_SANDBOX") sandbox_secret=os.getenv("SANDBOX_SECRET") itunes_secret=os.getenv("ITUNES_SECRET") google_json_key_path = os.getenv("JSON_KEY_PATH_ANDROID_SUB", "") android_pkg_name = os.getenv("ANDROID_PKG_NAME", "") # ---------- Android Validator ---------- class AndroidReceiptValidator: def __init__(self): try: credentials = service_account.Credentials.from_service_account_file( google_json_key_path, scopes=["https://www.googleapis.com/auth/androidpublisher"], ) self.android_publisher = build( "androidpublisher", "v3", credentials=credentials ) except Exception as e: print("❌ Google Auth Init Failed:", str(e)) self.android_publisher = None def validate_subscription(self, product_id: str, purchase_token: str): """ Validate Android subscription purchase """ if not self.android_publisher: return {"isValid": False, "response": {"error": "Android Publisher not initialized"}} try: response = ( self.android_publisher.purchases() .subscriptions() .get( packageName=android_pkg_name, subscriptionId=product_id, token=purchase_token, ) .execute() ) return {"isValid": True, "response": response} except Exception as e: return { "isValid": False, "response": {"error": {"message": str(e), "trace": traceback.format_exc()}}, } def from_ms(ms: int) -> datetime: """Convert milliseconds since epoch to datetime (UTC).""" return datetime.utcfromtimestamp(ms / 1000.0) # ---------- DB Utility ---------- def get_db_conn(): return mysql.connector.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME ) # ---------- Expiry Updater ---------- def update_expired_subscriptions(ids): conn, cursor = None, None try: conn = get_db_conn() cursor = conn.cursor() update_q = """ UPDATE subscription_receipts SET is_expired = 1, modifiedDate = %s WHERE id = %s """ cursor.execute(update_q, (datetime.utcnow(), ids)) conn.commit() print(f"❌ Marked subscription {ids} as expired.") except Exception as e: print("DB Expire Update Error:", str(e)) finally: if cursor: cursor.close() if conn: conn.close() # ---------- Get Active Subs ---------- def get_active_subscriptions(after_date=None): conn, cursor = None, None try: conn = get_db_conn() cursor = conn.cursor(dictionary=True) if after_date is None: after_date = datetime.now() print('after_date',after_date) query = """ SELECT id, user_id, user_device_id, receipt, expiry_date, is_android FROM subscription_receipts WHERE expiry_date < %s AND is_expired = 0 """ # query = """ # SELECT id, user_id, user_device_id, receipt, expiry_date, is_android # FROM subscription_receipts WHEREis_expired = 0 # """ cursor.execute(query, (after_date,)) results = cursor.fetchall() return results except Exception as e: print("DB Active Subs Error:", str(e)) return [] finally: if cursor: cursor.close() if conn: conn.close() # ---------- Apple Receipt Validation ---------- def validate_receipt(receipt, secret, env="prod", exclude_old_transactions=True): payload = { "password": itunes_secret, "exclude-old-transactions": exclude_old_transactions, "receipt-data": receipt } url = appstore_receipt_validator_url_prod if env != 'prod': url = appstore_receipt_validator_url_sandbox if env == 'sandbox': payload['password'] = sandbox_secret else: payload['password'] = itunes_secret print("url",url) print("payload",payload) try: resp = requests.post(url, json=payload, timeout=10) decoded_response = resp.json() print('decoded_response',decoded_response) status = decoded_response.get("status") print('status',status) if status == 0: return {"isValid": True, "response": decoded_response.get("latest_receipt_info", [])} else: return {"isValid": False, "response": status} except Exception as e: return {"isValid": False, "response": str(e)} def get_latest_receipt(receipt, secret): response = validate_receipt(receipt, secret, env="prod") if response["isValid"]: return response["response"] elif response["response"] in [21002, 21004, 21007]: print(f"Receipt validation failed with code {response['response']}, trying sandbox.") response = validate_receipt(receipt, secret, env="sandbox") if response["isValid"]: return response["response"] else: print(f"Receipt validation failed for sandbox, trying testflight.") response = validate_receipt(receipt, secret, env="testflight") if response["isValid"]: return response["response"] else: print(f"Receipt validation failed with response: {response['response']}") return False elif response["response"] in [501, 21010]: print(f"Receipt validation failed with response: {response['response']}") return False else: print(f"Receipt validation failed with response: {response['response']}") return False # ---------- Android Receipt Wrapper ---------- def get_latest_receipt_android(receipt: str, productId: str): validator = AndroidReceiptValidator() receiptData = validator.validate_subscription(productId, receipt) if not receiptData.get("isValid"): return [] response = receiptData["response"] purchaseDateMS = response.get("startTimeMillis") expireDateMS = response.get("expiryTimeMillis") autoRenewStatus = response.get("autoRenewing") isInTrialPeriod = "true" if response.get("paymentState") == 2 else "false" cancellationDateMS = response.get("userCancellationTimeMillis", 0) order_id = response.get("orderId", "").split("..")[0] result = { "product_id": productId, "purchase_date_ms": purchaseDateMS, "original_purchase_date_ms": purchaseDateMS, "expires_date_ms": expireDateMS, "transaction_id": response.get("orderId"), "original_transaction_id": order_id, "auto_renew_status": autoRenewStatus, "is_trial_period": isInTrialPeriod, "cancellation_date_ms": cancellationDateMS, "receipt": receipt, "is_android": True, } # ✅ Always return as list return [result] # ---------- Receipt Check ---------- def check_receipt(receipt, user_id, user_device_id, is_android, secret, product_id=None): if is_android == 1: user_subscriptions = get_latest_receipt_android(receipt, product_id or "") else: user_subscriptions = get_latest_receipt(receipt, secret) if not user_subscriptions or not isinstance(user_subscriptions, list): return {"code": 400, "message": "Invalid receipt", "data": []} # Deduplicate by original_transaction_id original_transaction_ids = list({r["original_transaction_id"] for r in user_subscriptions}) latest_receipts = [] for original_txn_id in original_transaction_ids: transactions = [row for row in user_subscriptions if row["original_transaction_id"] == original_txn_id] for row in transactions: row["expires_date_ms"] = float(row.get("expires_date_ms", 0)) if transactions: latest_expiration = max([t["expires_date_ms"] for t in transactions]) latest_txn = next(t for t in transactions if t["expires_date_ms"] == latest_expiration) latest_receipts.append(latest_txn) for latest_receipt in latest_receipts: product_id = latest_receipt["product_id"] if product_id in IOS_PLANS: is_android = 0 elif product_id in ANDROID_PLANS: is_android = 1 else: print("❌ Unknown product_id:", product_id) return False print("Latest Receipt:", latest_receipt.get("expires_date_ms", 0)) if latest_receipt.get("expires_date_ms", 0) > time.time() * 1000: try: conn = get_db_conn() cursor = conn.cursor() update_query = """ UPDATE subscription_receipts SET receipt = %s, transaction_id = %s, original_transaction_id = %s, subscription_date = %s, is_trial_period = %s, subscription_type = %s, expiry_date = %s, is_expired = %s, is_android = %s, modifiedDate = %s WHERE user_id = %s AND user_device_id = %s """ cursor.execute( update_query, ( receipt, latest_receipt["transaction_id"], latest_receipt["original_transaction_id"], from_ms(float(latest_receipt["purchase_date_ms"])), 1 if latest_receipt.get("is_trial_period") == "true" else 0, latest_receipt["product_id"], from_ms(float(latest_receipt["expires_date_ms"])), 0, # is_expired is_android, datetime.utcnow(), user_id, user_device_id ) ) conn.commit() print("✅ DB Updated for transaction:", latest_receipt["transaction_id"]) except Exception as e: print("DB Update Error:", str(e)) finally: if cursor: cursor.close() if conn: conn.close() return {"code": 200, "message": "Subscription active", "data": latest_receipt} else: print("❌ Subscription expired:", latest_receipt["transaction_id"]) return False return {"code": 200, "message": "Subscription receipt processed", "data": []} @router.get("/update_receipts_cron") async def update_receipts(request: Request): try: active_subs = get_active_subscriptions() print("Active Subs:", json.dumps(active_subs, indent=2, default=str)) for sub in active_subs: if sub["receipt"]: check_data = check_receipt( sub["receipt"], sub["user_id"], sub["user_device_id"], sub["is_android"], secret=DB_SECRET, product_id=sub.get("subscription_type") ) print("Check Result:", check_data) if check_data == False: update_expired_subscriptions(sub['id']) return {"code": 200, "message": "Receipt update process completed."} except Exception as e: print("Error in /update_receipts:", str(e), traceback.format_exc()) return {"code": 500, "message": "Internal server error."}