Jan Novak
69af4c1e3b
All checks were successful
Deploy to K8s / deploy (push) Successful in 24s
Add second Fio account (CZ0820100000002502035405 / 2502035405/2010). Both accounts are fetched on every sync run and combined before dedup, so the payments sheet accumulates transactions from either account. QR codes now default to the new account. Go: - config.go: hardcoded Accounts/LoadedAccount slice replaces scalar BankAccount + FioAPIToken; Config.BankAccount renamed QRAccount; per-account tokens via FIO_API_TOKEN_NEW / FIO_API_TOKEN_OLD - banksync.SyncToSheets: accepts []fio.Client, loops to combine txns - cmd/fuj/main.go: buildFioClients helper; both sync call sites updated - html_handler + build_adults/juniors: use Config.QRAccount - New TestSyncToSheets_MultiAccount covers cross-account dedup Python: - config.py: ACCOUNTS list + LOADED_ACCOUNTS (tokens from env) - fio_utils.py: fetch_transactions_for (per-account) + fetch_transactions_all (loops all accounts) - sync_fio_to_sheets.py: uses fetch_transactions_all Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
290 lines
10 KiB
Python
290 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Sync Fio bank transactions to a Google Sheet intermediary ledger."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import os
|
|
import pickle
|
|
from datetime import datetime, timedelta
|
|
|
|
from google.auth.transport.requests import Request
|
|
from google_auth_oauthlib.flow import InstalledAppFlow
|
|
from google.oauth2 import service_account
|
|
from googleapiclient.discovery import build
|
|
|
|
from fio_utils import fetch_transactions_all
|
|
|
|
from config import PAYMENTS_SHEET_ID as DEFAULT_SPREADSHEET_ID
|
|
SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]
|
|
TOKEN_FILE = "token.pickle"
|
|
COLUMN_LABELS = ["Date", "Amount", "manual fix", "Person", "Purpose", "Inferred Amount", "Sender", "VS", "Message", "Bank ID", "Sync ID"]
|
|
_SHEETS_SERVICE = None
|
|
|
|
def get_sheets_service(credentials_path: str):
|
|
"""Authenticate and return the Google Sheets API service."""
|
|
global _SHEETS_SERVICE
|
|
if _SHEETS_SERVICE is not None:
|
|
return _SHEETS_SERVICE
|
|
|
|
if not os.path.exists(credentials_path):
|
|
raise FileNotFoundError(f"Credentials file not found: {credentials_path}")
|
|
|
|
# Check if it's a service account
|
|
import json
|
|
with open(credentials_path, "r") as f:
|
|
creds_data = json.load(f)
|
|
|
|
if creds_data.get("type") == "service_account":
|
|
creds = service_account.Credentials.from_service_account_file(
|
|
credentials_path, scopes=SCOPES
|
|
)
|
|
else:
|
|
# Fallback to OAuth2 flow
|
|
creds = None
|
|
if os.path.exists(TOKEN_FILE):
|
|
with open(TOKEN_FILE, "rb") as token:
|
|
creds = pickle.load(token)
|
|
|
|
if not creds or not creds.valid:
|
|
if creds and creds.expired and creds.refresh_token:
|
|
creds.refresh(Request())
|
|
else:
|
|
flow = InstalledAppFlow.from_client_secrets_file(credentials_path, SCOPES)
|
|
creds = flow.run_local_server(port=0)
|
|
|
|
with open(TOKEN_FILE, "wb") as token:
|
|
pickle.dump(creds, token)
|
|
|
|
_SHEETS_SERVICE = build("sheets", "v4", credentials=creds)
|
|
return _SHEETS_SERVICE
|
|
|
|
|
|
def generate_sync_id(tx: dict) -> str:
|
|
"""Generate a unique SHA-256 hash for a transaction.
|
|
|
|
Hash components: date|amount|currency|sender|vs|message|bank_id
|
|
"""
|
|
components = [
|
|
str(tx.get("date", "")),
|
|
str(tx.get("amount", "")),
|
|
str(tx.get("currency", "CZK")),
|
|
str(tx.get("sender", "")),
|
|
str(tx.get("vs", "")),
|
|
str(tx.get("message", "")),
|
|
str(tx.get("bank_id", "")),
|
|
]
|
|
raw_str = "|".join(components).lower()
|
|
return hashlib.sha256(raw_str.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _trunc(s: str, n: int = 40) -> str:
|
|
s = str(s)
|
|
return s if len(s) <= n else s[: n - 1] + "…"
|
|
|
|
|
|
def _print_fio_table(transactions: list[dict], statuses: list[str]) -> None:
|
|
headers = ["DATE", "AMOUNT", "SENDER", "VS", "MESSAGE", "BANKID", "STATUS"]
|
|
rows = [
|
|
[
|
|
str(tx.get("date", "")),
|
|
f"{float(tx.get('amount', 0)):.2f}",
|
|
str(tx.get("sender", "")),
|
|
str(tx.get("vs", "")),
|
|
_trunc(str(tx.get("message", ""))),
|
|
str(tx.get("bank_id", "")),
|
|
status,
|
|
]
|
|
for tx, status in zip(transactions, statuses)
|
|
]
|
|
widths = [
|
|
max(len(headers[i]), max((len(r[i]) for r in rows), default=0))
|
|
for i in range(len(headers))
|
|
]
|
|
sep = " "
|
|
print(sep.join(h.ljust(w) for h, w in zip(headers, widths)))
|
|
for row in rows:
|
|
print(sep.join(cell.ljust(w) for cell, w in zip(row, widths)))
|
|
|
|
|
|
def sort_sheet_by_date(service, spreadsheet_id):
|
|
"""Sort the sheet by the Date column (Column B)."""
|
|
# Get the sheet ID (gid) of the first sheet
|
|
spreadsheet = service.spreadsheets().get(spreadsheetId=spreadsheet_id).execute()
|
|
sheet_id = spreadsheet['sheets'][0]['properties']['sheetId']
|
|
|
|
requests = [{
|
|
"sortRange": {
|
|
"range": {
|
|
"sheetId": sheet_id,
|
|
"startRowIndex": 1, # Skip header
|
|
"endRowIndex": 10000
|
|
},
|
|
"sortSpecs": [{
|
|
"dimensionIndex": 0, # Column A (Date)
|
|
"sortOrder": "ASCENDING"
|
|
}]
|
|
}
|
|
}]
|
|
|
|
service.spreadsheets().batchUpdate(
|
|
spreadsheetId=spreadsheet_id,
|
|
body={"requests": requests}
|
|
).execute()
|
|
print("Sheet sorted by date.")
|
|
|
|
|
|
def sync_to_sheets(
|
|
spreadsheet_id: str,
|
|
credentials_path: str,
|
|
days: int = None,
|
|
date_from_str: str = None,
|
|
date_to_str: str = None,
|
|
sort_by_date: bool = False,
|
|
dry_run: bool = False,
|
|
print_fio_table: bool = False,
|
|
):
|
|
print(f"Connecting to Google Sheets using {credentials_path}...")
|
|
service = get_sheets_service(credentials_path)
|
|
sheet = service.spreadsheets()
|
|
|
|
# 1. Read existing sync IDs from Column K
|
|
print(f"Reading existing sync IDs from sheet...")
|
|
try:
|
|
result = sheet.values().get(
|
|
spreadsheetId=spreadsheet_id,
|
|
range="A1:K" # Include header and all columns to check Sync ID
|
|
).execute()
|
|
values = result.get("values", [])
|
|
|
|
# Check and insert labels if missing
|
|
if not values or values[0] != COLUMN_LABELS:
|
|
if dry_run:
|
|
print("Dry run: would write header row")
|
|
else:
|
|
print("Inserting column labels...")
|
|
sheet.values().update(
|
|
spreadsheetId=spreadsheet_id,
|
|
range="A1",
|
|
valueInputOption="USER_ENTERED",
|
|
body={"values": [COLUMN_LABELS]}
|
|
).execute()
|
|
existing_ids = set()
|
|
else:
|
|
# Sync ID is the last column (index 10)
|
|
existing_ids = {row[10] for row in values[1:] if len(row) > 10}
|
|
except Exception as e:
|
|
print(f"Error reading sheet (maybe empty?): {e}")
|
|
existing_ids = set()
|
|
|
|
# 2. Fetch Fio transactions
|
|
if date_from_str and date_to_str:
|
|
df_str = date_from_str
|
|
dt_str = date_to_str
|
|
else:
|
|
now = datetime.now()
|
|
date_to = now
|
|
date_from = now - timedelta(days=days or 30)
|
|
df_str = date_from.strftime("%Y-%m-%d")
|
|
dt_str = date_to.strftime("%Y-%m-%d")
|
|
|
|
print(f"Fetching Fio transactions from {df_str} to {dt_str}...")
|
|
transactions = fetch_transactions_all(df_str, dt_str)
|
|
print(f"Found {len(transactions)} transactions.")
|
|
|
|
if dry_run:
|
|
print(f"Dry run: window {df_str} to {dt_str}, fetched {len(transactions)} transaction(s) from Fio")
|
|
|
|
# 3. Determine NEW/DUP for each transaction
|
|
new_rows = []
|
|
tx_statuses = []
|
|
for tx in transactions:
|
|
sync_id = generate_sync_id(tx)
|
|
if sync_id not in existing_ids:
|
|
# Schema: Date | Amount | Manual | Person | Purpose | Inferred Amount | Sender | VS | Message | Bank ID | Sync ID
|
|
new_rows.append([
|
|
tx.get("date", ""),
|
|
tx.get("amount", ""),
|
|
"", # Manual
|
|
"", # Person
|
|
"", # Purpose
|
|
"", # Inferred Amount
|
|
tx.get("sender", ""),
|
|
tx.get("vs", ""),
|
|
tx.get("message", ""),
|
|
tx.get("bank_id", ""),
|
|
sync_id,
|
|
])
|
|
tx_statuses.append("NEW")
|
|
else:
|
|
tx_statuses.append("DUP")
|
|
|
|
# 4. Print table (before early-return so all transactions are shown including DUPs)
|
|
if print_fio_table and transactions:
|
|
_print_fio_table(transactions, tx_statuses)
|
|
|
|
if not new_rows:
|
|
if dry_run:
|
|
print("Dry run: would sync 0 new transaction(s).")
|
|
else:
|
|
print("No new transactions to sync.")
|
|
return
|
|
|
|
# 5. Append to sheet or print dry-run would-write lines
|
|
if dry_run:
|
|
for tx, status in zip(transactions, tx_statuses):
|
|
if status == "NEW":
|
|
print(
|
|
f"Dry run: would append"
|
|
f" date={tx.get('date', '')}"
|
|
f" amount={tx.get('amount', '')}"
|
|
f" sender={tx.get('sender', '')}"
|
|
f" vs={tx.get('vs', '')}"
|
|
f" message={tx.get('message', '')}"
|
|
)
|
|
if sort_by_date:
|
|
print("Dry run: would sort by date")
|
|
print(f"Dry run: would sync {len(new_rows)} new transaction(s).")
|
|
else:
|
|
print(f"Appending {len(new_rows)} new transactions to the sheet...")
|
|
body = {"values": new_rows}
|
|
sheet.values().append(
|
|
spreadsheetId=spreadsheet_id,
|
|
range="A2", # Appends to the end of the sheet
|
|
valueInputOption="USER_ENTERED",
|
|
body=body
|
|
).execute()
|
|
print("Sync completed successfully.")
|
|
if sort_by_date:
|
|
sort_sheet_by_date(service, spreadsheet_id)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Sync Fio transactions to Google Sheets.")
|
|
parser.add_argument("--days", type=int, default=30, help="Days to look back (default: 30)")
|
|
parser.add_argument("--sheet-id", default=DEFAULT_SPREADSHEET_ID, help="Google Sheet ID")
|
|
parser.add_argument("--credentials", default="credentials.json", help="Path to Google API credentials JSON")
|
|
parser.add_argument("--from", dest="date_from", help="Start date YYYY-MM-DD")
|
|
parser.add_argument("--to", dest="date_to", help="End date YYYY-MM-DD")
|
|
parser.add_argument("--sort-by-date", action="store_true", help="Sort the sheet by date after sync")
|
|
parser.add_argument("--dry-run", action="store_true", help="Fetch and dedup without writing to the sheet")
|
|
parser.add_argument("--print-fio-table", action="store_true", help="Print aligned table of all fetched transactions with NEW/DUP status (use with --dry-run)")
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
sync_to_sheets(
|
|
spreadsheet_id=args.sheet_id,
|
|
credentials_path=args.credentials,
|
|
days=args.days,
|
|
date_from_str=args.date_from,
|
|
date_to_str=args.date_to,
|
|
sort_by_date=args.sort_by_date,
|
|
dry_run=args.dry_run,
|
|
print_fio_table=args.print_fio_table,
|
|
)
|
|
except Exception as e:
|
|
print(f"Sync failed: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|