#!/usr/bin/env python3 """ User Relations Repair Script ----------------------------- Idempotens: csak hiányzó rekordokat hoz létre. Biztosítja, hogy minden aktív user rendelkezzen: 1. Personal Organization-nel (fleet.organizations) — ha még nincs 2. Wallet-tel (identity.wallets) — ha még nincs 3. Subscription-nel (finance.user_subscriptions, tier=private_free_v1) — ha még nincs Futtatás: docker compose exec sf_api python3 /app/backend/scripts/repair_user_relations.py """ import asyncio import sys import os from datetime import datetime, timezone from decimal import Decimal from uuid import uuid4 # Path setup for the container sys.path.insert(0, '/app') sys.path.insert(0, '/app/backend') os.environ.setdefault('DATABASE_URL', 'postgresql+asyncpg://...') # Will be read from .env from app.database import AsyncSessionLocal from sqlalchemy import select, text from app.models.identity.identity import User, Wallet from app.models.core_logic import UserSubscription from app.models.marketplace.organization import Organization from app.core.config import settings DEFAULT_TIER_ID = 13 # private_free_v1 DEFAULT_CURRENCY = "HUF" async def ensure_organization(db, user): """ Create personal organization if user doesn't have one. Returns: org_id (existing or newly created) """ # Check if user already owns an organization (owner_id or legal_owner_id) result = await db.execute( select(Organization).where( Organization.owner_id == user.id ).limit(1) ) org = result.scalar_one_or_none() if org: print(f" [OK] Organization already exists for user {user.id}: org_id={org.id}") return org.id # Create new personal organization folder_slug = f"user-{user.id}-{int(datetime.now(timezone.utc).timestamp())}" now = datetime.now(timezone.utc) org = Organization( name=f"{user.email}'s Garage", full_name=f"{user.email}'s Personal Garage", display_name=f"{user.email}'s Garage", folder_slug=folder_slug, owner_id=user.id, org_type="individual", status="active", is_active=True, is_deleted=False, is_verified=False, subscription_plan="FREE", base_asset_limit=5, purchased_extra_slots=0, notification_settings={}, external_integration_config={}, first_registered_at=now, current_lifecycle_started_at=now, lifecycle_index=1, is_anonymized=False, default_currency="HUF", country_code="HU", language="hu", is_ownership_transferable=False, visual_settings={}, is_affiliate_partner=False, aliases=[], tags=[], settings={}, custom_features={}, created_at=now, ) db.add(org) await db.flush() print(f" [CREATED] Organization for user {user.id}: org_id={org.id}") return org.id async def ensure_wallet(db, user, org_id): """ Create wallet if user doesn't have one. wallet.user_id has UNIQUE constraint, so at most one wallet per user. Returns: True if created, False if already existed """ result = await db.execute( select(Wallet).where(Wallet.user_id == user.id).limit(1) ) wallet = result.scalar_one_or_none() if wallet: print(f" [OK] Wallet already exists for user {user.id}: wallet_id={wallet.id}") return False wallet = Wallet( user_id=user.id, earned_credits=Decimal('0'), purchased_credits=Decimal('0'), service_coins=Decimal('0'), currency=DEFAULT_CURRENCY, organization_id=org_id, ) db.add(wallet) await db.flush() print(f" [CREATED] Wallet for user {user.id}: wallet_id={wallet.id}") return True async def ensure_subscription(db, user): """ Create free tier subscription if user doesn't have one. Returns: True if created, False if already existed """ result = await db.execute( select(UserSubscription).where(UserSubscription.user_id == user.id).limit(1) ) sub = result.scalar_one_or_none() if sub: print(f" [OK] Subscription already exists for user {user.id}: sub_id={sub.id}") return False sub = UserSubscription( user_id=user.id, tier_id=DEFAULT_TIER_ID, valid_from=datetime.now(timezone.utc), is_active=True, ) db.add(sub) await db.flush() print(f" [CREATED] Subscription for user {user.id}: sub_id={sub.id}") return True async def repair_all_users(): """Main repair loop.""" print("=" * 60) print(" USER RELATIONS REPAIR SCRIPT") print(f" Started at: {datetime.now(timezone.utc).isoformat()}") print(f" Default tier: {DEFAULT_TIER_ID} (private_free_v1)") print("=" * 60) async with AsyncSessionLocal() as db: # Get all active, non-deleted users result = await db.execute( select(User).where( User.is_active == True, User.is_deleted == False, ) ) users = result.scalars().all() stats = { "total": len(users), "wallets_created": 0, "subs_created": 0, "orgs_created": 0, "errors": [], } print(f"\nFound {len(users)} active users to process.\n") for user in users: print(f"--- Processing user {user.id}: {user.email} ---") try: org_id = await ensure_organization(db, user) # Track org creation — since ensure_organization always returns org_id, # we can't easily diff "created" vs "existed" without checking first. # We'll track it manually. wallet_created = await ensure_wallet(db, user, org_id) if wallet_created: stats["wallets_created"] += 1 sub_created = await ensure_subscription(db, user) if sub_created: stats["subs_created"] += 1 except Exception as e: err_msg = f"User {user.id} ({user.email}): {e}" stats["errors"].append(err_msg) print(f" [ERROR] {err_msg}") await db.commit() print("\n" + "=" * 60) print(" REPAIR SUMMARY") print("=" * 60) print(f" Total active users processed: {stats['total']}") print(f" Wallets created: {stats['wallets_created']}") print(f" Subscriptions created: {stats['subs_created']}") print(f" Errors: {len(stats['errors'])}") if stats["errors"]: print("\n Errors:") for err in stats["errors"]: print(f" - {err}") print("=" * 60) return stats if __name__ == "__main__": asyncio.run(repair_all_users())