You've already forked transaction-tracker
Aaron Guise
59c0bbde0c
Previously is_processed() returned True for any record file that existed, relying entirely on cleanup_expired() (called at __init__) to delete stale files. Because cleanup runs at container startup — before Akahu transactions are fetched — any record that expired exactly on that startup would be deleted and then immediately missed, letting the duplicate through. Fix: is_processed() reads the expires_at field from the JSON and returns False if the record has expired, regardless of whether cleanup has run. Also adds migrate_ttl.py script to retroactively extend expires_at on existing records that were written under a shorter TTL, and bumps version to 0.1.3.
66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Migrate existing transaction records to a new TTL.
|
|
|
|
Reads all .json record files in the given storage directory and rewrites
|
|
expires_at to processed_at + ttl_days. Records that have already expired
|
|
are left untouched (they will be cleaned up on next startup).
|
|
|
|
Usage:
|
|
python migrate_ttl.py <storage_dir> <ttl_days>
|
|
|
|
Example (run on the server):
|
|
python migrate_ttl.py /opt/data/transaction_records 14
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
def migrate(storage_dir: str, ttl_days: int) -> None:
|
|
if not os.path.isdir(storage_dir):
|
|
print(f"Directory not found: {storage_dir}")
|
|
sys.exit(1)
|
|
|
|
now = datetime.now()
|
|
updated = skipped = errors = 0
|
|
|
|
for filename in os.listdir(storage_dir):
|
|
if not filename.endswith(".json"):
|
|
continue
|
|
path = os.path.join(storage_dir, filename)
|
|
try:
|
|
with open(path) as f:
|
|
record = json.load(f)
|
|
|
|
processed_at = datetime.fromisoformat(record["processed_at"])
|
|
current_expires = datetime.fromisoformat(record["expires_at"])
|
|
|
|
if current_expires <= now:
|
|
skipped += 1
|
|
continue
|
|
|
|
new_expires = processed_at + timedelta(days=ttl_days)
|
|
if new_expires <= current_expires:
|
|
skipped += 1
|
|
continue
|
|
|
|
record["expires_at"] = new_expires.isoformat()
|
|
with open(path, "w") as f:
|
|
json.dump(record, f)
|
|
updated += 1
|
|
|
|
except (json.JSONDecodeError, KeyError, ValueError, OSError) as e:
|
|
print(f" ERROR {filename}: {e}")
|
|
errors += 1
|
|
|
|
print(f"Done: {updated} updated, {skipped} skipped (already expired or TTL already sufficient), {errors} errors")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 3:
|
|
print(__doc__)
|
|
sys.exit(1)
|
|
migrate(sys.argv[1], int(sys.argv[2]))
|