"""ProductTransferHistory model. Snapshot semantics: - Rows are bulk-inserted at job-queue time with source_artist_id populated from releases.artist_id and destination_artist_id NULL. - The transfer SFN later updates destination_artist_id per release after it resolves the destination artist on the destination vendor. The SFN uses destination_artist_id IS NULL/IS NOT NULL as an idempotency check. - Soft-delete columns (deleted_at, deleted_by_identity_id) cascade from the parent job: when a job is soft-deleted, its product_transfer_history rows are soft-deleted in the same transaction. """ from datetime import datetime, timezone from sqlalchemy import Column, DateTime, Integer, String, text from project_manager.connector import mysql class ProductTransferHistory(mysql.BaseModel): """product_transfer_history in art_relations.""" __tablename__ = 'product_transfer_history' product_transfer_history_id = Column(Integer, primary_key=True, autoincrement=True) job_id = Column(Integer, nullable=False) release_id = Column(Integer, nullable=False) source_artist_id = Column(Integer, nullable=True) destination_artist_id = Column(Integer, nullable=True) source_video_artist_id = Column(Integer, nullable=True) destination_video_artist_id = Column(Integer, nullable=True) deleted_at = Column(DateTime, nullable=True) deleted_by_identity_id = Column(String(36), nullable=True) def to_dict(self): """Return serializable dict.""" return { 'product_transfer_history_id': self.product_transfer_history_id, 'project_transfer_job_id': self.job_id, 'release_id': self.release_id, 'source_artist_id': self.source_artist_id, 'destination_artist_id': self.destination_artist_id, 'source_video_artist_id': self.source_video_artist_id, 'destination_video_artist_id': self.destination_video_artist_id, 'deleted_at': self.deleted_at.isoformat() if self.deleted_at else None, 'deleted_by_identity_id': self.deleted_by_identity_id, } @mysql.wrap_db_errors def get_products_for_job(job_id): """Return product rows for the given job, enriched with release metadata. Excludes soft-deleted snapshot rows (those cascade from a deleted job). LEFT JOIN against releases so that soft-deleted releases (present in releases with deletions flag) are included alongside active ones. release_name, upc, and display_upc are NULL when a release row cannot be found. """ with mysql.pm_session_scope() as session: rows = session.execute( text( 'SELECT' ' pth.product_transfer_history_id,' ' pth.job_id AS project_transfer_job_id,' ' pth.release_id,' ' pth.source_artist_id,' ' pth.destination_artist_id,' ' pth.source_video_artist_id,' ' pth.destination_video_artist_id,' ' pth.deleted_at,' ' pth.deleted_by_identity_id,' ' r.release_name,' ' r.upc,' ' r.display_upc' ' FROM product_transfer_history pth' ' LEFT JOIN releases r ON r.release_id = pth.release_id' ' WHERE pth.job_id = :job_id' ' AND pth.deleted_at IS NULL' ), {'job_id': job_id}, ) return [dict(row) for row in rows] @mysql.wrap_db_errors def snapshot_releases_for_job(job_id, releases): """Bulk-insert one product_transfer_history row per release. `releases` is a list of dicts with `release_id` and `source_artist_id`. `destination_artist_id` is left NULL for the SFN to populate. """ with mysql.pm_session_scope() as session: rows = [ ProductTransferHistory( job_id=job_id, release_id=r['release_id'], source_artist_id=r.get('source_artist_id'), source_video_artist_id=r.get('source_video_artist_id'), ) for r in releases ] session.add_all(rows) session.flush() return [row.to_dict() for row in rows] @mysql.wrap_db_errors def set_destination_artists(job_id, updates): """Bulk-update destination_artist_id for many (job_id, release_id) rows. `updates` is a list of {release_id, destination_artist_id} dicts. All updates run in a single transaction; if any release_id has no matching live row, returns (None, list_of_missing_release_ids) and the transaction rolls back. On success returns (list_of_row_dicts, []). Soft-deleted snapshot rows are not eligible. """ requested_ids = [u['release_id'] for u in updates] with mysql.pm_session_scope() as session: rows = (session.query(ProductTransferHistory) .filter(ProductTransferHistory.job_id == job_id, ProductTransferHistory.release_id.in_(requested_ids), ProductTransferHistory.deleted_at.is_(None)) .all()) rows_by_release = {r.release_id: r for r in rows} missing = [rid for rid in requested_ids if rid not in rows_by_release] if missing: session.rollback() return None, missing for u in updates: row = rows_by_release[u['release_id']] row.destination_artist_id = u['destination_artist_id'] if 'destination_video_artist_id' in u: row.destination_video_artist_id = u['destination_video_artist_id'] session.flush() return [rows_by_release[rid].to_dict() for rid in requested_ids], [] @mysql.wrap_db_errors def get_upcs_for_job(job_id): """Return distinct UPC strings for releases attached to the job. Returns the canonical bigint `releases.upc` cast to a string. Releases with `upc = 0` (no UPC assigned) are skipped. Soft-deleted snapshot rows are excluded. """ with mysql.pm_session_scope() as session: rows = session.execute( text( "SELECT DISTINCT CAST(r.upc AS CHAR) AS upc " "FROM product_transfer_history pth " "JOIN releases r ON r.release_id = pth.release_id " "WHERE pth.job_id = :job_id " " AND pth.deleted_at IS NULL " " AND r.upc != 0" ), {'job_id': job_id}, ) return [row[0] for row in rows] @mysql.wrap_db_errors def get_isrcs_for_job(job_id): """Return distinct ISRC strings for tracks under releases in the job. Joins product_transfer_history -> track on release_id (track.upc is nullable so the release_id path is safer than joining on UPC). Soft-deleted snapshot rows and NULL/empty ISRCs are excluded. """ with mysql.pm_session_scope() as session: rows = session.execute( text( "SELECT DISTINCT t.isrc " "FROM product_transfer_history pth " "JOIN track t ON t.release_id = pth.release_id " "WHERE pth.job_id = :job_id " " AND pth.deleted_at IS NULL " " AND t.isrc IS NOT NULL " " AND t.isrc != ''" ), {'job_id': job_id}, ) return [row[0] for row in rows] @mysql.wrap_db_errors def soft_delete_for_job(job_id, identity_id): """Soft-delete all live product_transfer_history rows for a job. Used when the parent job is soft-deleted. Returns the number of rows soft-deleted (0 if there were none). """ now = datetime.now(tz=timezone.utc).replace(tzinfo=None) with mysql.pm_session_scope() as session: return (session.query(ProductTransferHistory) .filter(ProductTransferHistory.job_id == job_id, ProductTransferHistory.deleted_at.is_(None)) .update({ ProductTransferHistory.deleted_at: now, ProductTransferHistory.deleted_by_identity_id: identity_id, }, synchronize_session=False))