"""Generate vend_contact report: Auth0 users × Neo4j LabelProfiles × MySQL vend_contact.""" import argparse import json import logging import sys from datetime import datetime from pathlib import Path from typing import Any from sqlalchemy import text from sqlalchemy.engine import Engine from src.mysql.connection import create_mysql_engine from src.neo4j.cypher_runner import CypherRunner from src.readers import read_auth0_users logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) def format_file_size(size_bytes: int) -> str: """Format bytes as human-readable string (KB or MB).""" if size_bytes < 1024 * 1024: return f'{size_bytes / 1024:.1f} KB' return f'{size_bytes / (1024 * 1024):.1f} MB' def list_job_files(directory: str = 'data/input') -> list[tuple[Path, datetime, int]]: """Find job_*.json files with modification date and size.""" input_dir = Path(directory) if not input_dir.exists(): return [] files: list[tuple[Path, datetime, int]] = [] for f in input_dir.glob('job_*.json'): stat = f.stat() files.append((f, datetime.fromtimestamp(stat.st_mtime), stat.st_size)) return sorted(files, key=lambda x: x[1], reverse=True) def prompt_file_selection(files: list[tuple[Path, datetime, int]]) -> Path: """Display numbered menu of files and return selected path.""" print('\nAvailable input files in data/input/:') for i, (path, mtime, size) in enumerate(files, 1): date_str = mtime.strftime('%Y-%m-%d %H:%M:%S') print(f' {i}) {path.name} ({date_str}, {format_file_size(size)})') while True: choice = input(f'\nSelect file (1-{len(files)}): ').strip() if choice.isdigit() and 1 <= int(choice) <= len(files): return files[int(choice) - 1][0] print(f'Invalid choice. Enter a number between 1 and {len(files)}.') VEND_CONTACT_QUERY = """ SELECT id, auth0_user_id, auth0_primary, active FROM vend_contact WHERE id IN :ids """ def query_label_profiles(runner: CypherRunner, identity_ids: list[str]) -> dict[str, list[int]]: """Query Neo4j for LabelProfile profileIds per identity.""" cypher_file = Path(__file__).parent.parent / 'neo4j' / 'cypher' / 'get_label_profiles.cypher' query = cypher_file.read_text() results = runner.run_query(query, {'identityIds': identity_ids}) profiles_by_identity: dict[str, list[int]] = {} for record in results: identity_id = record.get('identityId') label_profiles = record.get('labelProfiles', []) if identity_id: profile_ids = [int(p['profileId']) for p in label_profiles if p.get('profileId') is not None] profiles_by_identity[identity_id] = profile_ids return profiles_by_identity def query_vend_contacts(engine: Engine, profile_ids: list[int]) -> dict[int, dict[str, Any]]: """Query MySQL vend_contact records by id.""" if not profile_ids: return {} with engine.connect() as conn: rows = conn.execute(text(VEND_CONTACT_QUERY), {'ids': tuple(profile_ids)}) return {row['id']: dict(row) for row in rows.mappings()} def _compute_mismatch(vend_contact_id: int | None, vend_contacts: list[dict[str, Any]]) -> bool: """Determine if Auth0's vend_contact_id mismatches the vend_contact records. Mismatch is True when: - Auth0's vend_contact_id is not found in the vend_contacts list, OR - The vend_contact with auth0_primary='Y' has a different id than Auth0's vend_contact_id """ if vend_contact_id is None: return False vc_ids = {vc['id'] for vc in vend_contacts} if vend_contact_id not in vc_ids: return True for vc in vend_contacts: if vc.get('auth0_primary') == 'Y' and vc['id'] != vend_contact_id: return True return False def generate_report( input_file: str, output_file: str, batch_size: int = 1000, max_batches: int | None = None, ) -> None: """Generate vend_contact report combining Auth0, Neo4j, and MySQL data.""" logger.info(f'Reading Auth0 data from {input_file}') auth0_users = list(read_auth0_users(input_file)) total_users = len(auth0_users) logger.info(f'Found {total_users} Auth0 users') results: list[dict[str, Any]] = [] mismatch_count = 0 engine = create_mysql_engine() with CypherRunner() as runner: batches_processed = 0 for i in range(0, total_users, batch_size): if max_batches and batches_processed >= max_batches: logger.info(f'Reached max batch limit of {max_batches}, stopping') break batch = auth0_users[i : i + batch_size] batch_num = i // batch_size + 1 total_batches = (total_users + batch_size - 1) // batch_size logger.info(f'Processing batch {batch_num}/{total_batches} ({len(batch)} users)') # Step 1: Query Neo4j for LabelProfile profileIds identity_ids = [user.identity_id for user in batch if user.identity_id] profiles_by_identity = query_label_profiles(runner, identity_ids) if identity_ids else {} # Step 2: Collect all unique profile IDs and query MySQL all_profile_ids: set[int] = set() for pid_list in profiles_by_identity.values(): all_profile_ids.update(pid_list) vend_contacts_by_id = query_vend_contacts(engine, list(all_profile_ids)) if all_profile_ids else {} logger.info(f' Neo4j: {len(profiles_by_identity)} identities with profiles, MySQL: {len(vend_contacts_by_id)} vend_contacts') # Step 3: Build report items for user in batch: label_profile_ids = profiles_by_identity.get(user.identity_id, []) vend_contacts = [vend_contacts_by_id[pid] for pid in label_profile_ids if pid in vend_contacts_by_id] mismatch = _compute_mismatch(user.vend_contact_id, vend_contacts) if mismatch: mismatch_count += 1 results.append( { 'auth0_id': user.id, 'vend_contact_id': user.vend_contact_id, 'identity_id': user.identity_id, 'mismatch': mismatch, 'label_profile_ids': label_profile_ids, 'vend_contacts': vend_contacts, } ) batches_processed += 1 engine.dispose() # Write results logger.info(f'Writing {len(results)} records to {output_file}') Path(output_file).parent.mkdir(parents=True, exist_ok=True) with open(output_file, 'w', encoding='utf-8') as f: json.dump(results, f, indent=2, default=str) logger.info('=' * 60) logger.info('Report complete!') logger.info(f'Total users: {len(results)}') logger.info(f'Mismatches: {mismatch_count}') logger.info(f'Output: {output_file}') logger.info('=' * 60) def main() -> None: """Main entry point.""" parser = argparse.ArgumentParser(description='Generate vend_contact report (Auth0 + Neo4j LabelProfiles + MySQL)') parser.add_argument('--input', help='Path to Auth0 export JSON Lines file (interactive prompt if omitted)') parser.add_argument('--output', help='Path to output JSON file (default: data/input/vend_contact_report_TIMESTAMP.json)') parser.add_argument('--batch-size', type=int, default=1000, help='Users per batch (default: 1000)') parser.add_argument('--max-batches', type=int, help='Max batches to process (for testing)') args = parser.parse_args() if not args.input: job_files = list_job_files() if not job_files: logger.error('No job_*.json files found in data/input/') sys.exit(1) args.input = str(prompt_file_selection(job_files)) if not args.output: timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') args.output = f'data/input/vend_contact_report_{timestamp}.json' generate_report(args.input, args.output, args.batch_size, args.max_batches) if __name__ == '__main__': main()