"""ProjectTransferJob model.""" from datetime import datetime, timezone from sqlalchemy import ( BigInteger, Column, Date, DateTime, Enum, Integer, String, Text, bindparam, text, ) from sqlalchemy.exc import IntegrityError from project_manager.connector import mysql JOB_STATUSES = ('QUEUED', 'PROCESSING', 'COMPLETED', 'FAILED', 'DELETED') class ProjectTransferJob(mysql.BaseModel): """project_transfer_job in art_relations.""" __tablename__ = 'project_transfer_job' job_id = Column(Integer, primary_key=True, autoincrement=True) project_id = Column(BigInteger, nullable=False) originating_vendor_id = Column(Integer, nullable=False) originating_subaccount_id = Column(Integer, nullable=True) originating_artist_id = Column(Integer, nullable=True) destination_vendor_id = Column(Integer, nullable=False) destination_subaccount_id = Column(Integer, nullable=True) destination_artist_id = Column(Integer, nullable=True) status = Column( Enum(*JOB_STATUSES, name='project_transfer_job_status', create_type=False), nullable=False, default='QUEUED', ) created_by_identity_id = Column(String(36), nullable=False) created_at = Column(DateTime, nullable=False) revenue_cutoff_date = Column(Date, nullable=True) transfer_completed_on = Column(DateTime, nullable=True) sfn_execution_arn = Column(String(2048), nullable=True) failure_reason = Column(Text, nullable=True) last_updated_at = Column(DateTime, nullable=True) last_updated_by_identity_id = Column(String(36), nullable=True) deleted_at = Column(DateTime, nullable=True) deleted_by_identity_id = Column(String(36), nullable=True) executed_by_identity_id = Column(String(36), nullable=True) def to_dict(self): """Return serializable dict.""" return { 'project_transfer_job_id': self.job_id, 'project_id': self.project_id, 'originating_vendor_id': self.originating_vendor_id, 'originating_subaccount_id': self.originating_subaccount_id, 'originating_artist_id': self.originating_artist_id, 'destination_vendor_id': self.destination_vendor_id, 'destination_subaccount_id': self.destination_subaccount_id, 'destination_artist_id': self.destination_artist_id, 'status': self.status, 'created_by_identity_id': self.created_by_identity_id, 'created_at': self.created_at.isoformat() if self.created_at else None, 'revenue_cutoff_date': ( self.revenue_cutoff_date.isoformat() if self.revenue_cutoff_date else None), 'transfer_completed_on': ( self.transfer_completed_on.isoformat() if self.transfer_completed_on else None), 'sfn_execution_arn': self.sfn_execution_arn, 'failure_reason': self.failure_reason, 'last_updated_at': self.last_updated_at.isoformat() if self.last_updated_at else None, 'last_updated_by_identity_id': self.last_updated_by_identity_id, 'deleted_at': self.deleted_at.isoformat() if self.deleted_at else None, 'deleted_by_identity_id': self.deleted_by_identity_id, 'executed_by_identity_id': self.executed_by_identity_id, } @mysql.wrap_db_errors def get_transfer_jobs(status=None, originating_vendor_id=None, destination_vendor_id=None, project_id=None, limit=100, offset=0): """Return (items, total) with optional filters. Excludes soft-deleted jobs.""" with mysql.pm_session_scope() as session: query = (session.query(ProjectTransferJob) .filter(ProjectTransferJob.deleted_at.is_(None))) if status is not None: query = query.filter(ProjectTransferJob.status == status) if originating_vendor_id is not None: query = query.filter( ProjectTransferJob.originating_vendor_id == originating_vendor_id) if destination_vendor_id is not None: query = query.filter( ProjectTransferJob.destination_vendor_id == destination_vendor_id) if project_id is not None: query = query.filter(ProjectTransferJob.project_id == project_id) total = query.count() items = (query.order_by(ProjectTransferJob.created_at.desc()) .limit(limit).offset(offset).all()) return [item.to_dict() for item in items], total @mysql.wrap_db_errors def get_transfer_job(job_id): """Return job dict or None. Returns None for soft-deleted jobs.""" with mysql.pm_session_scope() as session: job = (session.query(ProjectTransferJob) .filter(ProjectTransferJob.job_id == job_id, ProjectTransferJob.deleted_at.is_(None)) .one_or_none()) return job.to_dict() if job else None @mysql.wrap_db_errors def create_transfer_job(params, identity_id): """Insert a new job and return its dict.""" from oto import response as oto_response now = datetime.now(tz=timezone.utc).replace(tzinfo=None) try: with mysql.pm_session_scope() as session: job = ProjectTransferJob( project_id=params['project_id'], originating_vendor_id=params['originating_vendor_id'], originating_subaccount_id=params.get('originating_subaccount_id'), originating_artist_id=params.get('originating_artist_id'), destination_vendor_id=params['destination_vendor_id'], destination_subaccount_id=params.get('destination_subaccount_id'), revenue_cutoff_date=params.get('revenue_cutoff_date'), created_by_identity_id=identity_id, created_at=now, last_updated_by_identity_id=identity_id, status='QUEUED', ) session.add(job) session.flush() return job.to_dict() except IntegrityError as e: orig = getattr(e, 'orig', None) mysql_code = getattr(orig, 'args', [None])[0] if orig else None if mysql_code in (1216, 1452): return oto_response.create_error_response( code='bad_request', message='Invalid vendor or reference ID.', status=400) raise @mysql.wrap_db_errors def get_releases_for_project(project_id): """Return release_id + source artist for every release in the project. Used to seed product_transfer_history at job-queue time. Includes soft-deleted releases (no `deletions` filter) so the snapshot covers the full project even if rows were deleted between job creation and SFN run. """ with mysql.pm_session_scope() as session: rows = session.execute( text( 'SELECT r.release_id, r.artist_id AS source_artist_id, ' 'pv.primary_artist_id AS source_video_artist_id ' 'FROM releases r ' 'LEFT JOIN product_video pv ON pv.release_id = r.release_id ' 'WHERE r.project_id = :project_id' ), {'project_id': project_id}, ) return [dict(row) for row in rows] _RELEASE_CHUNK_SIZE = 500 def _chunks(lst, size): for i in range(0, len(lst), size): yield lst[i:i + size] @mysql.wrap_db_errors def soft_delete_transfer_job(job_id, identity_id): """Soft-delete a live job via a single UPDATE. Returns True if a live row was updated, None otherwise. """ now = datetime.now(tz=timezone.utc).replace(tzinfo=None) with mysql.pm_session_scope() as session: rowcount = (session.query(ProjectTransferJob) .filter(ProjectTransferJob.job_id == job_id, ProjectTransferJob.deleted_at.is_(None)) .update({ ProjectTransferJob.status: 'DELETED', ProjectTransferJob.deleted_at: now, ProjectTransferJob.deleted_by_identity_id: identity_id, ProjectTransferJob.last_updated_by_identity_id: identity_id, }, synchronize_session=False)) return True if rowcount else None @mysql.wrap_db_errors def update_transfer_job(job_id, fields): """Apply a partial update to a live job. `fields` maps ORM column attributes to values. Returns the updated job dict, or None if no live row matched. """ now = datetime.now(tz=timezone.utc).replace(tzinfo=None) fields[ProjectTransferJob.last_updated_at] = now with mysql.pm_session_scope() as session: rowcount = (session.query(ProjectTransferJob) .filter(ProjectTransferJob.job_id == job_id, ProjectTransferJob.deleted_at.is_(None)) .update(fields, synchronize_session=False)) if not rowcount: return None job = (session.query(ProjectTransferJob) .filter(ProjectTransferJob.job_id == job_id) .one()) return job.to_dict() @mysql.wrap_db_errors def set_executed_by_identity_id(job_id, identity_id): """Record who triggered ExecuteContentTransfer on the job row.""" with mysql.pm_session_scope() as session: session.query(ProjectTransferJob).filter( ProjectTransferJob.job_id == job_id ).update( {ProjectTransferJob.executed_by_identity_id: identity_id}, synchronize_session=False, ) @mysql.wrap_db_errors def execute_content_transfer(job_id, project_id, destination_vendor_id, destination_subaccount_id, destination_artist_id, products): """Execute all DB writes for Step 3 (ExecuteContentTransfer) in one transaction. Updates project, releases (one UPDATE per distinct destination_artist_id, chunked in batches of 500), and product_video (one UPDATE per distinct destination_video_artist_id, chunked in batches of 500). """ project_subaccount_id = ( destination_subaccount_id if destination_subaccount_id is not None else 0) with mysql.pm_session_scope() as session: session.execute( text( 'UPDATE project SET vendor_id = :vendor_id, ' 'subaccount_id = :subaccount_id, artist_id = :artist_id, ' 'updated_date_utc = NOW() ' 'WHERE project_id = :project_id' ), { 'vendor_id': destination_vendor_id, 'subaccount_id': project_subaccount_id, 'artist_id': destination_artist_id, 'project_id': project_id, }, ) by_artist = {} for p in products: by_artist.setdefault(p['destination_artist_id'], []).append(p['release_id']) releases_updated = 0 for artist_id, release_ids in by_artist.items(): for chunk in _chunks(release_ids, _RELEASE_CHUNK_SIZE): r = session.execute( text( 'UPDATE releases SET artist_id = :artist_id, ' 'subaccount_id = :subaccount_id ' 'WHERE release_id IN :release_ids' ).bindparams(bindparam('release_ids', expanding=True)), { 'artist_id': artist_id, 'subaccount_id': destination_subaccount_id, 'release_ids': chunk, }, ) releases_updated += r.rowcount by_video_artist = {} for p in products: if p.get('destination_video_artist_id') is not None: by_video_artist.setdefault( p['destination_video_artist_id'], [] ).append(p['release_id']) video_rows_updated = 0 for video_artist_id, release_ids in by_video_artist.items(): for chunk in _chunks(release_ids, _RELEASE_CHUNK_SIZE): r = session.execute( text( 'UPDATE product_video SET primary_artist_id = :artist_id ' 'WHERE release_id IN :release_ids' ).bindparams(bindparam('release_ids', expanding=True)), {'artist_id': video_artist_id, 'release_ids': chunk}, ) video_rows_updated += r.rowcount by_source_dest = {} for p in products: source = p.get('source_artist_id') dest = p['destination_artist_id'] if source is not None: by_source_dest.setdefault((source, dest), []).append(p['release_id']) release_artist_updated = 0 track_artist_updated = 0 track_writer_updated = 0 for (source_id, dest_id), release_ids in by_source_dest.items(): for chunk in _chunks(release_ids, _RELEASE_CHUNK_SIZE): params = {'dest_id': dest_id, 'source_id': source_id, 'release_ids': chunk} r = session.execute( text( 'UPDATE release_artist SET artist_info_id = :dest_id ' 'WHERE release_id IN :release_ids AND artist_info_id = :source_id' ).bindparams(bindparam('release_ids', expanding=True)), params, ) release_artist_updated += r.rowcount r = session.execute( text( 'UPDATE track_artist SET artist_info_id = :dest_id ' 'WHERE track_id IN ' '(SELECT id FROM track WHERE release_id IN :release_ids) ' 'AND artist_info_id = :source_id' ).bindparams(bindparam('release_ids', expanding=True)), params, ) track_artist_updated += r.rowcount r = session.execute( text( 'UPDATE track_writer SET artist_info_id = :dest_id ' 'WHERE unique_track_id IN ' '(SELECT id FROM track WHERE release_id IN :release_ids) ' 'AND artist_info_id = :source_id' ).bindparams(bindparam('release_ids', expanding=True)), params, ) track_writer_updated += r.rowcount return { 'project_updated': 1, 'releases_updated': releases_updated, 'video_rows_updated': video_rows_updated, 'release_artist_rows_updated': release_artist_updated, 'track_artist_rows_updated': track_artist_updated, 'track_writer_rows_updated': track_writer_updated, }