"""Consolidation utilities for Orchard Sound Recordings.""" from datetime import datetime, timezone from typing import Any, Dict, List, NamedTuple, Optional, Set, Tuple from soundrecording_utils.constants.territories import RULE_TERRITORIES from soundrecording_utils.metadata.types import FingerprintRule, LabelType, OrchardSoundRecording, PolicyType class TerritoryDateRange(NamedTuple): """A date range with associated territory-policy combinations. - `start_date`: timezone-aware datetime as UTC - `end_date`: optional timezone-aware datetime as UTC or None - `territories`: list of territory codes (e.g. 'US', 'GB') """ start_date: datetime end_date: Optional[datetime] = None territories: List[str] = [] class ConsolidatedRights(NamedTuple): """Container for consolidated rights information. Contains a list of `TerritoryDateRange` instances; each range holds the territory-policy combinations that apply for that period. """ monetization_policies: List[TerritoryDateRange] sixtyseconds_policies: List[TerritoryDateRange] block_access_policies: List[TerritoryDateRange] def consolidate_rights( sound_recording: OrchardSoundRecording, service: str ) -> ConsolidatedRights: """Consolidate fingerprint rules across all tracks for a sound recording. This function processes all fingerprint rules (vendor, subaccount, track level, including carveouts) associated with the provided `OrchardSoundRecording` for the specified `service`. It produces a consolidated view of rights, combining overlapping and adjacent rules into date ranges with territory-policy combinations. Parameters sound_recording (OrchardSoundRecording): The sound recording object containing track and rule data. service (str): The service/vendor name for which to consolidate rights (e.g., 'tiktok', 'youtube'). Returns ConsolidatedRights: A named tuple containing: - monetization_policies: List of TerritoryDateRange objects for monetization periods/territories. - sixtyseconds_policies: List of TerritoryDateRange objects for sixty-second conditional periods/territories. - block_access_policies: List of TerritoryDateRange objects for block access periods/territories. Key business rules: - Policy precedence (highest to lowest): "block_access" > "sixtyseconds" > "monetize" - "block_access" policies override both "sixtyseconds" and "monetize" policies for the same territory and date range. - "sixtyseconds" policies override "monetize" policies for the same territory and date range. - "carveout" policies cancel monetization for the specified territory and date range. - Carveouts are converted into missing monetizations in the final result. - Rules are merged and deduplicated across all tracks. Steps: 1. Extract relevant rules for each track (including carveouts). 2. Consolidate rules into date ranges with territory-policy combinations per track. 3. Aggregate across all tracks to produce overall consolidated rights. """ # Step 1: for each track, extract the relevant rules track_rules = _extract_track_rules_by_service(sound_recording, service=service) # Step 2: For each track, consolidate the rules into date ranges with # territory-policy combinations (the results of this step no longer includes # carveouts - these are turned into missing monetizations) track_policies = _consolidate_intra_track_policies(track_rules) # Step 3: Aggregate across all tracks to get overall consolidated rights return _aggregate_inter_track_policies(track_policies) def _extract_track_rules_by_service( sound_recording: OrchardSoundRecording, service: str ) -> List[Dict[str, Any]]: """Extract rules for each track filtered by `service` and allowed policies. Returns a list where each item corresponds to a track and has the shape: { 'tuid': , 'track_level_rules': [ { 'territory': 'US', 'start_date': '2025-03-01', 'end_date': '2025-04-01' or None, 'policy': 'carveout'|'monetize'|'block_access' }, ... ], 'subaccount_level_rules': [ ... ], 'vendor_level_rules': [ ... ], } The function is permissive about input shapes: it accepts both dict-like rule objects and objects with attributes (NamedTuples). """ result: List[Dict[str, Any]] = [] tracks = sound_recording.track_connection.tracks for track in tracks: tuid = track.tuid track_level_rules = track.rules subaccount_level_rules: List[FingerprintRule] = [] vendor_level_rules: List[FingerprintRule] = [] if track.product.label.type.value == LabelType.Subaccount.value: vendor_level_rules = ( track.product.label.vendor.rules if track.product.label.vendor else [] ) subaccount_level_rules = track.product.label.rules else: vendor_level_rules = track.product.label.rules filtered_track_level_rules = [] filtered_subaccount_level_rules = [] filtered_vendor_level_rules = [] for rule in track_level_rules: if rule.service != service: continue filtered_track_level_rules.extend(_rule_to_entry(rule)) for rule in subaccount_level_rules: if rule.service != service: continue filtered_subaccount_level_rules.extend(_rule_to_entry(rule)) for rule in vendor_level_rules: if rule.service != service: continue filtered_vendor_level_rules.extend(_rule_to_entry(rule)) result.append( { 'tuid': tuid, 'track_level_rules': filtered_track_level_rules, 'subaccount_level_rules': filtered_subaccount_level_rules, 'vendor_level_rules': filtered_vendor_level_rules, } ) return result def _rule_to_entry(rule: FingerprintRule) -> List[Dict[str, Any]]: """Convert a rule object into a dict entry. This always returns a list of entries, as we may need to handle '*' territories -> In this case the function returns a list of all territories. """ if rule.territory != '*': return [ { 'territory': rule.territory, 'start_date': rule.start_date, 'end_date': rule.end_date, 'policy': rule.policy, } ] all_territories = [t for t in RULE_TERRITORIES if t != '*'] return [ { 'territory': territory, 'start_date': rule.start_date, 'end_date': rule.end_date, 'policy': rule.policy, } for territory in all_territories ] def _consolidate_intra_track_policies( track_rules: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: """Consolidate rules within a track into date ranges with territory-policy combos. Returns a list where each item corresponds to a track and has the shape: { 'tuid': , monetization_policies: [ { 'territory': 'US', 'start_date': datetime(2025, 1, 1, tzinfo=timezone.utc), 'end_date': datetime(2025, 2, 1, tzinfo=timezone.utc), 'policy': 'monetize' }, ... ], sixtyseconds_policies: [ { 'territory': 'US', 'start_date': datetime(2025, 1, 1, tzinfo=timezone.utc), 'end_date': datetime(2025, 2, 1, tzinfo=timezone.utc), 'policy': 'sixtyseconds' }, ... ], block_access_policies: [ { 'territory': 'US', 'start_date': datetime(2025, 3, 1, tzinfo=timezone.utc), 'end_date': datetime(2025, 4, 1, tzinfo=timezone.utc), 'policy': 'block_access' }, ... ], } All start_date/end_date values are timezone-aware datetime objects. """ consolidated_tracks = [] for track in track_rules: vendor_level_rules = track['vendor_level_rules'] subaccount_level_rules = track['subaccount_level_rules'] track_level_rules = track['track_level_rules'] territory_carveouts: Set[str] = set() relevant_territories: Set[str] = set() for rule_set in [vendor_level_rules, subaccount_level_rules, track_level_rules]: for rule in rule_set: relevant_territories.add(rule['territory']) if rule['policy'] == PolicyType.carveout.value: territory_carveouts.add(rule['territory']) sorted_relevant_territories = sorted(relevant_territories) # iterate over all territories # if a block_access exists at any level, that takes precedence # else if a sixtyseconds policy exists, that takes precedence over monetize # else if a monetize exists at any level, that is applied for that date range # unless there is a carveout at any level for that territory, in which case we # do not have any rights at all. # Note that there can be multiple date ranges for the same territory and policy monetization_policies: List[Dict[str, Any]] = [] sixtyseconds_policies: List[Dict[str, Any]] = [] block_access_policies: List[Dict[str, Any]] = [] for territory in sorted_relevant_territories: # If a carveout exists for this territory, we can claim neither monetize # nor block_access if territory in territory_carveouts: continue # Block on any level will result in block_access block_rules: List[Dict[str, Any]] = [] sixtyseconds_rules: List[Dict[str, Any]] = [] monetize_rules: List[Dict[str, Any]] = [] for rule_set in [ vendor_level_rules, subaccount_level_rules, track_level_rules, ]: block_rules.extend( [ r for r in rule_set if r['territory'] == territory and r['policy'] == PolicyType.block_access.value ] ) sixtyseconds_rules.extend( [ r for r in rule_set if r['territory'] == territory and r['policy'] == PolicyType.sixtyseconds.value ] ) if territory not in territory_carveouts: for rule_set in [ vendor_level_rules, subaccount_level_rules, track_level_rules, ]: monetize_rules.extend( [ r for r in rule_set if r['territory'] == territory and r['policy'] == PolicyType.monetize.value ] ) merged_monetize_rules = ( _merge_overlaps(monetize_rules) if monetize_rules else [] ) merged_sixtyseconds_rules = ( _merge_overlaps(sixtyseconds_rules) if sixtyseconds_rules else [] ) merged_block_rules = _merge_overlaps(block_rules) if block_rules else [] if merged_monetize_rules and merged_block_rules: merged_monetize_rules = _remove_overlapping_blocks( merged_monetize_rules, merged_block_rules, PolicyType.monetize.value, ) if merged_sixtyseconds_rules and merged_block_rules: merged_sixtyseconds_rules = _remove_overlapping_blocks( merged_sixtyseconds_rules, merged_block_rules, PolicyType.sixtyseconds.value, ) if merged_monetize_rules and merged_sixtyseconds_rules: merged_monetize_rules = _remove_overlapping_blocks( merged_monetize_rules, merged_sixtyseconds_rules, PolicyType.monetize.value, ) monetization_policies.extend(merged_monetize_rules) sixtyseconds_policies.extend(merged_sixtyseconds_rules) block_access_policies.extend(merged_block_rules) consolidated_tracks.append( { 'tuid': track['tuid'], 'monetization_policies': monetization_policies, 'sixtyseconds_policies': sixtyseconds_policies, 'block_access_policies': block_access_policies, } ) return consolidated_tracks def _merge_overlaps(date_ranges: List[Any]) -> List[Any]: """Merge overlapping date ranges. We assume all date ranges have the same territory and policy. We only need to merge based on start_date and end_date. """ if not date_ranges: return [] # Sort by start_date date_ranges = sorted(date_ranges, key=lambda dr: dr['start_date']) # iterate over each date range, check it against all other date ranges for overlaps merged_ranges: List[Any] = [] for current_range in date_ranges: current_start = current_range['start_date'] current_end = current_range['end_date'] if not merged_ranges: merged_ranges.append(current_range) continue last_merged = merged_ranges[-1] last_start = last_merged['start_date'] last_end = last_merged['end_date'] # Check for overlap if (last_end is None or current_start <= last_end) and ( current_end is None or last_start <= current_end ): # There is an overlap, merge the ranges new_start = min(last_start, current_start) if last_end is None or current_end is None: new_end = None else: new_end = max(last_end, current_end) merged_ranges[-1] = { 'territory': last_merged['territory'], 'start_date': new_start, 'end_date': new_end, 'policy': last_merged['policy'], } else: merged_ranges.append(current_range) return merged_ranges def _parse_date_string(date_str: str) -> datetime: """Parse an ISO date/datetime string into a timezone-aware datetime. Preserves the original UTC offset. If the source string has no offset (bare date or naive datetime), defaults to UTC midnight. """ dt = datetime.fromisoformat(date_str) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt def _remove_overlapping_blocks( merged_monetize_rules: List[Dict[str, Any]], merged_block_rules: List[Dict[str, Any]], target_policy: str = PolicyType.monetize.value, ) -> List[Dict[str, Any]]: """Remove dates from target rules where there's an equivalent block date. Ranges that share only a boundary timestamp (monetize_end == block_start) are treated as adjacent, not overlapping, so the target is not trimmed. """ # This can create gaps in target rules where block rules overlap new_monetization_ranges = merged_monetize_rules.copy() # Iterate over all target ranges for mon_range in new_monetization_ranges: monetize_start = mon_range['start_date'] monetize_end = mon_range['end_date'] # For each target range, find all collisions with block ranges # This can lead to: # 1. The block range completely covers the target range -> remove it # 2. The block range overlaps the start of the target range -> adjust start # 3. The block range overlaps the end of the target range -> adjust end # 4. The block range is in the middle -> split target range into two for block_range in merged_block_rules: block_start = block_range['start_date'] block_end = block_range['end_date'] # Check for overlap using strict inequality so that ranges sharing only a # boundary timestamp (e.g. monetize_end == block_start) are treated as # adjacent, not overlapping. if (block_end is None or monetize_start < block_end) and ( monetize_end is None or block_start < monetize_end ): # Case 4: The block is in the middle of a monetize range if block_end is not None and ( monetize_start < block_start and (monetize_end is None or monetize_end > block_end) ): # split the target range into two. Add the two new ranges to the # end of the loop, so they will be checked for further collisions new_monetization_ranges.remove(mon_range) new_monetization_ranges.append( { 'territory': mon_range['territory'], 'start_date': monetize_start, 'end_date': block_start, 'policy': target_policy, } ) new_monetization_ranges.append( { 'territory': mon_range['territory'], 'start_date': block_end, 'end_date': monetize_end, 'policy': target_policy, } ) # This target range has been removed, continue to next one break # Case 2: The block overlaps the start of the monetize range elif ( block_start <= monetize_start and block_end is not None and (monetize_end is None or monetize_end > block_end) ): mon_range['start_date'] = block_end # Case 3: The block overlaps the end of the monetize range # (this includes case where monetize_end is None, but only if # block_end is also None) elif monetize_start < block_start and ( (monetize_end is None and block_end is None) or ( (block_end is not None and monetize_end is not None) and block_end <= monetize_end ) ): mon_range['end_date'] = block_start # Case 1: The block completely covers the monetize range else: # Remove the target range new_monetization_ranges.remove(mon_range) # This target range has been removed, continue to next one break # sort the resulting new_monetization_ranges by start_date new_monetization_ranges = sorted( new_monetization_ranges, key=lambda dr: dr['start_date'] ) return new_monetization_ranges def _aggregate_inter_track_policies( track_policies: List[Dict[str, Any]], ) -> ConsolidatedRights: """Create the overall consolidated rights for the entire Sound Recording. Goes through all existing track policies, groups them by territory and policy, making sure that block_access overrides sixtyseconds, and sixtyseconds overrides monetize. """ monetization_policies: List[Dict[str, Any]] = [] sixtyseconds_policies: List[Dict[str, Any]] = [] block_access_policies: List[Dict[str, Any]] = [] # Iterate over all territories. For each territory, gather all date ranges from # all tracks. Merge overlapping date ranges for the same policy, and ensure # block_access overrides monetize. Collect all unique territories present in # the input track_policies relevant_territories: Set[str] = set() for track in track_policies: relevant_territories.update( r['territory'] for r in track['monetization_policies'] ) relevant_territories.update( r['territory'] for r in track['sixtyseconds_policies'] ) relevant_territories.update( r['territory'] for r in track['block_access_policies'] ) sorted_relevant_territories = sorted(relevant_territories) for territory in sorted_relevant_territories: territory_monetize_rules: List[Dict[str, Any]] = [] territory_sixtysecond_rules: List[Dict[str, Any]] = [] territory_block_rules: List[Dict[str, Any]] = [] for track in track_policies: territory_monetize_rules.extend( [ r for r in track['monetization_policies'] if r['territory'] == territory ] ) territory_sixtysecond_rules.extend( [ r for r in track['sixtyseconds_policies'] if r['territory'] == territory ] ) territory_block_rules.extend( [ r for r in track['block_access_policies'] if r['territory'] == territory ] ) merged_monetize_rules = ( _merge_overlaps(territory_monetize_rules) if territory_monetize_rules else [] ) merged_sixtysecond_rules = ( _merge_overlaps(territory_sixtysecond_rules) if territory_sixtysecond_rules else [] ) merged_block_rules = ( _merge_overlaps(territory_block_rules) if territory_block_rules else [] ) if merged_monetize_rules and merged_block_rules: merged_monetize_rules = _remove_overlapping_blocks( merged_monetize_rules, merged_block_rules, PolicyType.monetize.value, ) if merged_sixtysecond_rules and merged_block_rules: merged_sixtysecond_rules = _remove_overlapping_blocks( merged_sixtysecond_rules, merged_block_rules, PolicyType.sixtyseconds.value, ) if merged_monetize_rules and merged_sixtysecond_rules: merged_monetize_rules = _remove_overlapping_blocks( merged_monetize_rules, merged_sixtysecond_rules, PolicyType.monetize.value, ) monetization_policies.extend(merged_monetize_rules) sixtyseconds_policies.extend(merged_sixtysecond_rules) block_access_policies.extend(merged_block_rules) grouped_monetization_policies: List[TerritoryDateRange] = [] grouped_sixtyseconds_policies: List[TerritoryDateRange] = [] grouped_block_access_policies: List[TerritoryDateRange] = [] # Find territories with same date ranges and policies, group them together for policy_type, policies in [ (PolicyType.monetize.value, monetization_policies), (PolicyType.sixtyseconds.value, sixtyseconds_policies), (PolicyType.block_access.value, block_access_policies), ]: grouped: Dict[Tuple[datetime, Optional[datetime]], List[str]] = {} for policy in policies: key = (policy['start_date'], policy['end_date']) if key not in grouped: grouped[key] = [] grouped[key].append(policy['territory']) consolidated_list: List[TerritoryDateRange] = [] for (start_date, end_date), territories in grouped.items(): consolidated_list.append( TerritoryDateRange( start_date=start_date, end_date=end_date, territories=sorted(territories), ) ) # Sort consolidated list by start_date, then territories consolidated_list = sorted( consolidated_list, key=lambda x: (x.start_date, x.territories) ) if policy_type == PolicyType.monetize.value: grouped_monetization_policies = consolidated_list elif policy_type == PolicyType.sixtyseconds.value: grouped_sixtyseconds_policies = consolidated_list else: grouped_block_access_policies = consolidated_list return ConsolidatedRights( monetization_policies=grouped_monetization_policies, sixtyseconds_policies=grouped_sixtyseconds_policies, block_access_policies=grouped_block_access_policies, )