"""Bulk add/remove of UPCs/ISRCs from contract terms for product transfers (PORT-12).""" from abacus_common_logic.connectors.database import db from abacus_contract.constants.constants import CONTRACT_TERM_TYPES from abacus_contract.models.account_contract import AccountContract from abacus_contract.models.contract_term import ContractTerm from abacus_contract.models.contract_term_condition import ContractTermCondition _PRODUCT = CONTRACT_TERM_TYPES.PRODUCT _TRACK = CONTRACT_TERM_TYPES.TRACK def bulk_remove_from_contract_terms( account_id: int, upcs: list[str], isrcs: list[str], ) -> dict: """Remove transferred UPCs from product terms and ISRCs from track terms. Product terms store UPCs in their attachments list; track terms store ISRCs. Both are processed in a single DB transaction so a partial failure rolls back. This operation is idempotent: requesting removal of a value that is already absent from a term's attachments is a no-op for that term. If removing the specified values empties a term's attachments list entirely, the term is soft-deleted rather than left with an empty list (which is invalid per existing validation rules). Args: account_id: The originating_vendor_id from the product transfer job. Only terms whose contract belongs to this account are modified. upcs: UPC strings to remove from product-term attachments. isrcs: ISRC strings to remove from track-term attachments. Returns: dict with keys: updated_terms (list): one entry per modified term total_removed (int): count of terms that were actually modified Raises: ValueError: If both upcs and isrcs are empty. Exception: Re-raises any DB error after rolling back the transaction. """ if not upcs and not isrcs: raise ValueError('at least one of upcs or isrcs must be non-empty') upcs_set = set(upcs) isrcs_set = set(isrcs) # Fetch all active product/track terms for this account via AccountContract join. terms = ( ContractTerm.query.join( AccountContract, ContractTerm.contract_id == AccountContract.contract_id ) .filter(AccountContract.account_id == account_id) .filter(ContractTerm.term_type.in_((_PRODUCT, _TRACK))) .filter(ContractTerm.deleted_at.is_(None)) .filter(ContractTerm.attachments.is_not(None)) .all() ) updated_terms = [] try: for term in terms: current = term.attachments or [] removal_set = upcs_set if term.term_type == _PRODUCT else isrcs_set matching = [a for a in current if a in removal_set] if not matching: continue remaining = [a for a in current if a not in removal_set] soft_deleted = len(remaining) == 0 if soft_deleted: term._soft_delete() else: term.update_attributes(attachments=remaining) entry = { 'contract_id': term.contract_id, 'term_id': term.contract_term_id, 'term_type': term.term_type, 'removed_attachments': matching, 'remaining_attachments': remaining, 'soft_deleted': soft_deleted, } updated_terms.append(entry) if updated_terms: db.session.commit() except Exception: db.session.rollback() raise return { 'updated_terms': updated_terms, 'total_removed': len(updated_terms), } def bulk_add_to_contract_terms( account_id: int, contract_id: int, upcs: list[str], isrcs: list[str], conditions: list[dict], attachment_relations: dict = None, name: str = None, ) -> dict: """Add UPCs to a product term and ISRCs to a track term on a destination contract. If an active term of the required type already exists on the contract, the incoming values are merged into its attachments (duplicates are discarded). If no such term exists, a new ContractTerm is created and the provided conditions are attached as ContractTermCondition rows. Both product and track terms are processed in one DB transaction so a partial failure rolls back all changes. Args: account_id: The destination account ID (used only for authorization checks in the calling route; not used to filter terms here). contract_id: The destination contract that will receive the terms. upcs: UPC strings to add to the product term. isrcs: ISRC strings to add to the track term. conditions: List of condition dicts, each with keys: conditions (dict), term_rate (Decimal), priority (int). Applied only when a new term is created. attachment_relations: Optional label scoping (label_ids/upcs/contributors) stored on a newly created term. The contract detail page reads attachments_relations.label_ids to show the account on the term. Applied only when a new term is created; an existing term's attachments_relations is left untouched. name: Optional contract_term_name for a newly created term. Applied only when a new term is created; an existing term's name is left untouched. Returns: dict with keys: updated_terms (list): entries for terms whose attachments were merged created_terms (list): entries for newly created terms total_modified (int): count of terms created or updated Raises: ValueError: If both upcs and isrcs are empty. Exception: Re-raises any DB error after rolling back the transaction. """ if not upcs and not isrcs: raise ValueError('at least one of upcs or isrcs must be non-empty') updated_terms = [] created_terms = [] try: for term_type, values in ((_PRODUCT, upcs), (_TRACK, isrcs)): if not values: continue existing = _find_active_term(contract_id, term_type) if existing: merged = _merge_attachments(existing.attachments, values) existing.update_attributes(attachments=merged) updated_terms.append(_term_entry(existing, added=values)) else: new_term = ContractTerm.build( contract_id=contract_id, term_type=term_type, attachments=list(values), attachments_relations=attachment_relations, contract_term_name=name, ) db.session.flush() _create_conditions(new_term.contract_term_id, conditions) created_terms.append(_term_entry(new_term, added=values)) if updated_terms or created_terms: db.session.commit() except Exception: db.session.rollback() raise total = len(updated_terms) + len(created_terms) return { 'updated_terms': updated_terms, 'created_terms': created_terms, 'total_modified': total, } def create_transfer_contract_term( contract_id: int, term_type: str, attachments: list[str], conditions: list[dict], attachment_relations: dict = None, name: str = None, commit: bool = True, ) -> dict: """Create a single new contract_term for a product transfer staged term. Always creates a new term regardless of whether an active term of that type already exists on the contract. Each staged transfer term creates its own destination contract_term with its own conditions. Args: contract_id: The destination contract. term_type: 'product' or 'track'. attachments: UPCs (product) or ISRCs (track) to attach. conditions: List of condition dicts (conditions, term_rate, priority). attachment_relations: Optional label scoping stored on the new term. name: Optional contract_term_name for the new term. commit: When False, the term + conditions are flushed (so contract_term_id is populated) but NOT committed, leaving the transaction open for the caller to commit alongside its own changes (e.g. writing the staged term's destination_contract_term_id link in the same transaction). Returns: dict with keys: contract_term_id, contract_id, term_type, and conditions (a list of {contract_term_condition_id, priority} for each created row). Raises: Exception: Re-raises any DB error after rolling back the transaction. """ try: new_term = ContractTerm.build( contract_id=contract_id, term_type=term_type, attachments=list(attachments), attachments_relations=attachment_relations, contract_term_name=name, ) db.session.flush() created_conditions = _create_conditions(new_term.contract_term_id, conditions) # Flush so the new condition rows get their ids, which the caller links # back to the staged conditions by priority. db.session.flush() if commit: db.session.commit() except Exception: db.session.rollback() raise return { 'contract_term_id': new_term.contract_term_id, 'contract_id': new_term.contract_id, 'term_type': new_term.term_type, 'conditions': [ { 'contract_term_condition_id': c.contract_term_condition_id, 'priority': c.priority, } for c in created_conditions ], } # --------------------------------------------------------------------------- # Private helpers # --------------------------------------------------------------------------- def _find_active_term(contract_id: int, term_type: str): """Return the first active term of term_type for contract_id, or None.""" return ( ContractTerm.query.filter(ContractTerm.contract_id == contract_id) .filter(ContractTerm.term_type == term_type) .filter(ContractTerm.deleted_at.is_(None)) .first() ) def _merge_attachments(existing: list, incoming) -> list: """Return existing + incoming with duplicates removed, order preserved.""" existing_set = set(existing or []) result = list(existing or []) for value in incoming: if value not in existing_set: result.append(value) existing_set.add(value) return result def _create_conditions(contract_term_id: int, conditions: list[dict]) -> list: """Build ContractTermCondition rows for the given contract_term_id. Returns the built rows so callers can map them back to the staged conditions they came from (the transfer flow links each by priority). """ created = [] for cond in conditions: term_rate = float(cond.get('term_rate', 0)) commission = 100 - term_rate created.append( ContractTermCondition.build( contract_term_id=contract_term_id, conditions=cond.get('conditions', {}), term_rate=term_rate, commission=commission, priority=cond.get('priority'), ) ) return created def _term_entry(term, added) -> dict: """Build a summary entry for a created or updated term.""" return { 'contract_id': term.contract_id, 'term_id': term.contract_term_id, 'term_type': term.term_type, 'added_values': list(added), 'attachments': term.attachments or [], }