mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
141 lines
6.1 KiB
Python
141 lines
6.1 KiB
Python
import asyncio
|
|
from playwright.async_api import async_playwright
|
|
import os
|
|
import glob
|
|
import re
|
|
|
|
PASSWORD = "***REMOVED***"
|
|
IMPORT_DIR = "/home/allaun/.gemini/antigravity/scratch/imports"
|
|
SCREENSHOT_DIR = "/home/allaun/.gemini/antigravity/scratch/screenshots/multi_import"
|
|
os.makedirs(SCREENSHOT_DIR, exist_ok=True)
|
|
|
|
# List of account files to import and their original names
|
|
IMPORT_ACCOUNTS = [
|
|
{"file": "Chime-Chime_account.csv", "name": "Chime account"},
|
|
{"file": "Chime-Chime_Secured_Credit.csv", "name": "Chime Secured Credit"},
|
|
{"file": "Capital_One-Quicksilver.csv", "name": "Quicksilver"},
|
|
{"file": "Credit_One_Bank-Visa.csv", "name": "Visa"},
|
|
{"file": "Venmo_-_Personal-Personal_Profile.csv", "name": "Personal Profile"},
|
|
{"file": "Capital_One-credit_Account_8214.csv", "name": "credit Account 8214"},
|
|
{"file": "Capital_One-depository_Account_8109.csv", "name": "depository Account 8109"},
|
|
{"file": "Capital_One-depository_Account_9461.csv", "name": "depository Account 9461"},
|
|
{"file": "Capital_One-depository_Account_9555.csv", "name": "depository Account 9555"},
|
|
]
|
|
|
|
async def import_account(page, file_info):
|
|
csv_path = os.path.join(IMPORT_DIR, file_info["file"])
|
|
acc_name = file_info["name"]
|
|
|
|
if not os.path.exists(csv_path):
|
|
print(f"[-] Warning: CSV file {csv_path} does not exist. Skipping.")
|
|
return
|
|
|
|
print(f"\n[*] Processing account '{acc_name}' from {file_info['file']}...")
|
|
|
|
# 1. Check if account already exists in the sidebar
|
|
sidebar_text = await page.evaluate("document.body.innerText")
|
|
if acc_name in sidebar_text:
|
|
print(f"[+] '{acc_name}' already exists in sidebar. Clicking it...")
|
|
# To avoid clicking random text, let's find the navigation element or link matching the name
|
|
# In Actual, sidebar items are links or text
|
|
await page.click(f"text='{acc_name}'")
|
|
await page.wait_for_timeout(3000)
|
|
else:
|
|
print(f"[*] Creating '{acc_name}' account...")
|
|
# Click Add account button (sidebar or main page)
|
|
add_account_buttons = page.locator("text=Add account")
|
|
await add_account_buttons.first.click()
|
|
await page.wait_for_timeout(2000)
|
|
|
|
# Click Create local account
|
|
await page.click("button:has-text('Create local account')")
|
|
await page.wait_for_timeout(2000)
|
|
|
|
# Fill form (On-budget by default)
|
|
await page.fill("input[name='name']", acc_name)
|
|
|
|
# Click Create (using the last button to avoid clicking parent hidden button)
|
|
await page.locator("button:has-text('Create')").last.click()
|
|
await page.wait_for_timeout(5000)
|
|
|
|
# 2. Trigger Import
|
|
print(f"[*] Triggering import for '{acc_name}'...")
|
|
import_btn = page.locator("button:has-text('Import'), a:has-text('Import'), span:has-text('Import')")
|
|
count = await import_btn.count()
|
|
if count == 0:
|
|
print(f"[-] Error: Could not find Import button for '{acc_name}'")
|
|
await page.screenshot(path=f"{SCREENSHOT_DIR}/error_{acc_name.replace(' ', '_')}.png")
|
|
return
|
|
|
|
# Trigger file chooser
|
|
async with page.expect_file_chooser() as fc_info:
|
|
await import_btn.first.click()
|
|
file_chooser = await fc_info.value
|
|
print(f"[+] Selecting file {csv_path}...")
|
|
await file_chooser.set_files(csv_path)
|
|
await page.wait_for_timeout(5000)
|
|
|
|
# 3. Confirm mapping and import
|
|
print("[*] Confirming import mapping...")
|
|
# Locate the button that finishes the import
|
|
final_btn = page.locator("button:has-text('transactions'), button:has-text('Import')")
|
|
final_count = await final_btn.count()
|
|
print(f"[*] Found {final_count} final import buttons")
|
|
|
|
if final_count > 0:
|
|
await final_btn.last.click()
|
|
print("[*] Finalizing import...")
|
|
await page.wait_for_timeout(5000)
|
|
|
|
# Take a screenshot to document success
|
|
safe_name = acc_name.replace(' ', '_')
|
|
await page.screenshot(path=f"{SCREENSHOT_DIR}/{safe_name}_imported.png")
|
|
print(f"[+] Import of '{acc_name}' completed. Screenshot saved.")
|
|
else:
|
|
print(f"[-] Error: No final import button found for '{acc_name}'")
|
|
await page.screenshot(path=f"{SCREENSHOT_DIR}/error_confirm_{acc_name.replace(' ', '_')}.png")
|
|
|
|
async def main():
|
|
async with async_playwright() as p:
|
|
print("[*] Launching browser...")
|
|
browser = await p.chromium.launch(headless=True)
|
|
context = await browser.new_context(ignore_https_errors=True)
|
|
page = await context.new_page()
|
|
|
|
# Login
|
|
print("[*] Navigating to login page...")
|
|
await page.goto("https://budget.researchstack.info/login")
|
|
await page.wait_for_selector("input[type='password']")
|
|
await page.fill("input[type='password']", PASSWORD)
|
|
await page.click("button:has-text('Sign in')")
|
|
await page.wait_for_timeout(5000)
|
|
|
|
# Open Budget File
|
|
print("[*] Opening 'My Finances' budget...")
|
|
await page.click("text=My Finances")
|
|
await page.wait_for_timeout(8000)
|
|
|
|
# Loop through accounts and import them
|
|
for file_info in IMPORT_ACCOUNTS:
|
|
try:
|
|
await import_account(page, file_info)
|
|
except Exception as e:
|
|
print(f"[-] Exception occurred while processing '{file_info['name']}': {e}")
|
|
# Save screenshot of failure
|
|
safe_name = file_info["name"].replace(' ', '_')
|
|
await page.screenshot(path=f"{SCREENSHOT_DIR}/exception_{safe_name}.png")
|
|
|
|
# Final screenshot of budget sidebar
|
|
# Click budget menu to go to main dashboard or budget overview
|
|
try:
|
|
await page.click("text=Budget")
|
|
await page.wait_for_timeout(5000)
|
|
await page.screenshot(path=f"{SCREENSHOT_DIR}/final_sidebar_overview.png")
|
|
print("[+] Saved final budget overview screenshot.")
|
|
except Exception as e:
|
|
print(f"[-] Failed to capture final overview: {e}")
|
|
|
|
await browser.close()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|