""" Django models for broadcast playlist and advertising management system. This module contains models that manage the lifecycle of broadcast playlists, advertising windows, avails, ad spots placement, and verification data for a television/media broadcasting platform. Models hierarchy: Playlists -> Windows -> Avails -> AdspotsInAvail Author: Development Team Created: [Date] Last Modified: [Date] Version: 1.1 - Applied all TODO improvements """ from email.policy import default from django.db import models from django.core.validators import MinValueValidator, MaxValueValidator from django.core.exceptions import ValidationError from django.utils import timezone from django.utils.translation import gettext_lazy as _ from django.db.models import Q, Count, Sum from django.contrib.contenttypes.models import ContentType from datetime import time, datetime import logging # Import related models from other apps from apps.common.models import BaseModel from apps.campaigns.models import Adspots from apps.channels.models import Channel, ChannelZone, ChannelZoneRelation from apps.playlists.utils import ( convert_date_string_to_date, convert_duration_string_to_duration, convert_time_string_to_time, playlist_xml_upload_path ) # Configure logging for this module logger = logging.getLogger(__name__) # Constants for playlist status class PlaylistStatus: """Constants for playlist status values.""" DRAFT = 1 PUBLISHED = 0 CHOICES = [ (DRAFT, _('Draft')), (PUBLISHED, _('Published')), ] # Constants for verification status class VerificationStatus: """Constants for verification status values.""" PENDING = 'pending' COMPLETE = 'complete' FAILED = 'failed' PARTIAL = 'partial' CHOICES = [ (PENDING, _('Pending')), (COMPLETE, _('Complete')), (FAILED, _('Failed')), (PARTIAL, _('Partial')), ] class PlaylistManager(models.Manager): """Custom manager for Playlists model with useful query methods.""" def published(self): """Get only published playlists.""" return self.filter(is_draft=PlaylistStatus.PUBLISHED) def drafts(self): """Get only draft playlists.""" return self.filter(is_draft=PlaylistStatus.DRAFT) def for_channel(self, channel_id): """Get playlists for a specific channel.""" return self.filter(channel_id=channel_id) def for_date_range(self, start_date, end_date): """Get playlists within a specific date range.""" return self.filter( Q(start_date__gte=start_date, start_date__lte=end_date) | Q(end_date__gte=start_date, end_date__lte=end_date) | Q(start_date__lte=start_date, end_date__gte=end_date) ) def active_now(self): """Get playlists that are currently active.""" now = timezone.now() return self.filter( start_date__lte=now, end_date__gte=now, is_draft=PlaylistStatus.PUBLISHED ) class WindowsManager(models.Manager): """Custom manager for Windows model.""" def for_playlist(self, playlist_id): """Get windows for a specific playlist.""" return self.filter(playlist_id=playlist_id) def with_avails(self): """Get windows that have avails.""" return self.filter(avails__isnull=False).distinct() def by_start_time(self): """Order windows by start time.""" return self.order_by('window_start') class AvailsManager(models.Manager): """Custom manager for Avails model.""" def for_window(self, window_id): """Get avails for a specific window.""" return self.filter(window_id=window_id) def with_adspots(self): """Get avails that have ad spots placed.""" return self.filter(adspots_in_avail__isnull=False).distinct() def available(self): """Get avails that have no ad spots (available for placement).""" return self.filter(adspots_in_avail__isnull=True) def by_start_time(self): """Order avails by start time.""" return self.order_by('avail_start') class AdSpotsInAvailManager(models.Manager): """Custom manager for AdspotsInAvail model.""" def for_avail(self, avail_id): """Get ad spot placements for a specific avail.""" return self.filter(avail_id=avail_id) def for_adspot(self, adspot_id): """Get placements for a specific ad spot.""" return self.filter(adspot_id=adspot_id) def by_position(self): """Order placements by position in avail.""" return self.order_by('positioninavail') def for_traffic_id(self, traffic_id): """Get placements by traffic ID.""" return self.filter(trafficid=traffic_id) class VerifsManager(models.Manager): """Custom manager for Verifs model.""" def for_network(self, network_name): """Get verifications for a specific network.""" return self.filter(networkname=network_name) def for_zone(self, zone_name): """Get verifications for a specific zone.""" return self.filter(zonename=zone_name) def for_date(self, broadcast_date): """Get verifications for a specific broadcast date.""" return self.filter(broadcast_date=broadcast_date) def completed(self): """Get only completed verifications.""" return self.filter(vercomplete__in=['complete', 'yes', '1', 'true']) def pending(self): """Get pending verifications.""" return self.exclude(vercomplete__in=['complete', 'yes', '1', 'true']) def for_traffic_id(self, traffic_id): """Get verifications for a specific traffic ID.""" return self.filter(trafficId=traffic_id) def by_air_time(self): """Order verifications by air time.""" return self.order_by('air_time') class Playlists(BaseModel): """ Model representing broadcast playlists for television channels. A playlist contains the scheduling information for a specific broadcast period, including windows where advertisements can be placed. Each playlist is associated with a channel and potentially a specific zone within that channel. Attributes: channel (ForeignKey): Reference to the broadcasting channel id_zone_channel (ForeignKey): Reference to specific zone within the channel version (str): Version identifier for playlist tracking broadcast_date (DateField): Date when the playlist is scheduled to broadcast start_date (datetime): Start datetime for the playlist broadcast end_date (datetime): End datetime for the playlist broadcast filepath (str): File system path to the playlist file is_draft (bool): Flag indicating if playlist is in draft state draft_version (int): Version number for draft iterations """ # Foreign key relationship to the broadcasting channel # Using DO_NOTHING to prevent cascade deletion of playlists when channel is deleted channel = models.ForeignKey( Channel, on_delete=models.DO_NOTHING, db_column='channel', blank=True, null=True, help_text="The channel this playlist belongs to", related_name='playlists' ) # Foreign key relationship to specific channel zone (optional) # Allows for zone-specific playlist management within a channel zone_channel = models.ForeignKey( ChannelZone, on_delete=models.DO_NOTHING, db_column='zone_channel', blank=True, null=True, help_text="Specific zone within the channel (optional)", related_name='playlists' ) # Version string for playlist version control and tracking version = models.IntegerField( default=1, help_text="Version identifier for playlist versioning", validators=[MinValueValidator(0)] ) broadcast_date = models.DateField( blank=True, null=True, help_text="Scheduled broadcast date", db_index=True ) # Precise start datetime for playlist broadcast window start_date = models.DateTimeField( blank=True, null=True, help_text="Start datetime for playlist broadcast" ) # Precise end datetime for playlist broadcast window end_date = models.DateTimeField( blank=True, null=True, help_text="End datetime for playlist broadcast" ) # File system path where the playlist file is stored xml_file = models.FileField( upload_to=playlist_xml_upload_path, blank=True, null=True, help_text="Generated XML schedule file for this playlist", ) is_draft = models.BooleanField( default=True, help_text="Draft status flag (True=draft, False=published)", db_index=True ) # Version number for draft iterations draft_version = models.IntegerField( help_text="Version number for draft management", validators=[MinValueValidator(0)] ) # Add custom manager objects = PlaylistManager() class Meta: db_table = 'Playlists' verbose_name = _('Playlist') verbose_name_plural = _('Playlists') ordering = ['-created_at', '-id'] indexes = [ models.Index(fields=['channel', 'broadcast_date'], name='playlist_channel_date_idx'), models.Index(fields=['start_date', 'end_date'], name='playlist_datetime_range_idx'), models.Index(fields=['is_draft', 'created_at'], name='playlist_draft_created_idx'), models.Index(fields=['version'], name='playlist_version_idx'), models.Index(fields=['zone_channel'], name='playlist_zone_idx'), ] constraints = [ # Ensure version is unique per channel and broadcast date models.UniqueConstraint( fields=['channel', 'broadcast_date', 'version'], name='unique_playlist_version_per_channel_date' ), # Ensure start_date is before end_date when both are set models.CheckConstraint( check=Q(start_date__lt=models.F('end_date')) | Q(start_date__isnull=True) | Q(end_date__isnull=True), name='playlist_valid_date_range' ), ] def __str__(self): """String representation of the playlist.""" channel_name = self.channel.channel_name if self.channel else "No Channel" status = "Draft" if self.is_draft else "Published" return f"{channel_name} - {self.version} ({status}) - {self.broadcast_date}" def clean(self): """Custom validation for the model.""" super().clean() # Validate that end_date is after start_date if self.start_date and self.end_date and self.start_date >= self.end_date: raise ValidationError({ 'end_date': _("End date must be after start date.") }) # Validate version format if needed if self.version and len(self.version) > 255: raise ValidationError({ 'version': _("Version identifier cannot exceed 255 characters.") }) # Log validation logger.debug(f"Validating playlist {self.id}: {self.version}") def save(self, *args, **kwargs): """Custom save method with additional logic.""" # Auto-increment draft version if this is a new draft if not self.pk and self.is_draft: max_draft = Playlists.objects.filter( channel=self.channel, broadcast_date=self.broadcast_date ).aggregate( max_draft=models.Max('draft_version') )['max_draft'] self.draft_version = (max_draft or 0) + 1 # Log save operation logger.info(f"Saving playlist {self.version} for channel {self.channel}") super().save(*args, **kwargs) @property def is_published(self): """Check if the playlist is published (not in draft).""" return not self.is_draft @property def is_active(self): """Check if the playlist is currently active.""" if not self.start_date or not self.end_date: return False now = timezone.now() return self.start_date <= now <= self.end_date and self.is_published @property def duration_minutes(self): """Calculate playlist duration in minutes.""" if self.start_date and self.end_date: delta = self.end_date - self.start_date return round(delta.total_seconds() / 60, 2) return None @property def duration_hours(self): """Calculate playlist duration in hours.""" minutes = self.duration_minutes return round(minutes / 60, 2) if minutes else None @property def status_display(self): """Get human-readable status.""" return "Draft" if self.is_draft else "Published" def get_windows_count(self): """Get count of windows in this playlist.""" return self.windows.count() def get_total_avails_count(self): """Get total count of avails across all windows.""" return self.windows.aggregate( total_avails=Count('avails') )['total_avails'] or 0 def publish(self, user=None): """Publish a draft playlist.""" if self.is_draft: self.is_draft = False self.save(update_fields=['is_draft']) logger.info(f"Playlist {self.id} published by user {user}") return True return False def create_copy(self, new_version=None): """Create a copy of this playlist with a new version.""" new_playlist = Playlists( channel=self.channel, zone_channel=self.zone_channel, version=new_version or self.version+1, broadcast_date=self.broadcast_date, start_date=self.start_date, end_date=self.end_date, filepath=self.filepath, is_draft=True, draft_version=1 ) new_playlist.save() # Copy windows and their relationships for window in self.windows.all(): window.create_copy(new_playlist) logger.info(f"Created copy of playlist {self.id} as {new_playlist.id}") return new_playlist class Windows(BaseModel): """ Model representing advertising windows within a playlist. Windows are time segments within a playlist where advertisements can be scheduled. Each window has a defined start time, end time, and duration within the broader playlist timeframe. Attributes: id_window (AutoField): Primary key for the window id_playlist (ForeignKey): Reference to the parent playlist window_start (TimeField): Start time of the advertising window window_end (TimeField): End time of the advertising window window_duration (DurationField): Duration of the advertising window """ playlist = models.ForeignKey( Playlists, on_delete=models.CASCADE, db_column='playlist', blank=True, null=True, help_text=_("The playlist this window belongs to"), related_name='windows', verbose_name=_("Playlist") ) window_start = models.TimeField( blank=True, null=True, help_text=_("Start time of the advertising window"), verbose_name=_("Window Start"), db_index=True ) window_end = models.TimeField( blank=True, null=True, help_text=_("End time of the advertising window"), verbose_name=_("Window End") ) window_duration = models.DurationField( blank=True, null=True, help_text=_("Duration of the advertising window"), verbose_name=_("Window Duration") ) # Add custom manager objects = WindowsManager() class Meta: db_table = 'Windows' verbose_name = _('Advertising Window') verbose_name_plural = _('Advertising Windows') # ordering = ['window_start', 'id_window'] ordering = ['window_start',] indexes = [ models.Index(fields=['playlist', 'window_start'], name='window_playlist_start_idx'), models.Index(fields=['window_start'], name='window_start_idx'), ] constraints = [ # Ensure window_start comes before window_end when both are set models.CheckConstraint( check=Q(window_start__lt=models.F('window_end')) | Q(window_start__isnull=True) | Q(window_end__isnull=True), name='window_valid_time_range' ), ] def __str__(self): """String representation of the advertising window.""" playlist_info = f"Playlist {self.playlist.version}" if self.playlist else "No Playlist" return f"Window {self.id} ({playlist_info}): {self.window_start} - {self.window_end}" def clean(self): """Custom validation for the model.""" super().clean() # Validate that window_end is after window_start if self.window_start and self.window_end and self.window_start >= self.window_end: raise ValidationError({ 'window_end': _("Window end time must be after start time.") }) # Auto-calculate duration if start and end times are provided if self.window_start and self.window_end and not self.window_duration: # Convert times to datetime for calculation start_dt = datetime.combine(datetime.today(), self.window_start) end_dt = datetime.combine(datetime.today(), self.window_end) # Handle case where end time is next day if end_dt <= start_dt: end_dt = datetime.combine(datetime.today() + timezone.timedelta(days=1), self.window_end) self.window_duration = end_dt - start_dt def get_avails_count(self): """Get count of avails in this window.""" return self.avails.count() def get_total_adspots_count(self): """Get total count of ad spots across all avails in this window.""" return self.avails.aggregate( total_spots=Count('adspots_in_avail') )['total_spots'] or 0 def create_copy(self, new_playlist): """Create a copy of this window for a new playlist.""" new_window = Windows( playlist=new_playlist, window_start=self.window_start, window_end=self.window_end, window_duration=self.window_duration ) new_window.save() # Copy avails for avail in self.avails.all(): avail.create_copy(new_window) return new_window class Avails(BaseModel): """ Model representing available advertising slots within windows. Avails are specific time slots within advertising windows where individual advertisements can be placed. They represent the granular inventory units for ad placement scheduling. Attributes: id_window (ForeignKey): Reference to the parent advertising window avail_start (TimeField): Start time of the available slot availinwindow (str): Position or identifier of avail within the window datetime (datetime): Timestamp associated with the avail """ # Foreign key relationship to parent advertising window window = models.ForeignKey( Windows, on_delete=models.CASCADE, db_column='window', blank=True, null=True, help_text=_("The advertising window this avail belongs to"), related_name='avails', verbose_name=_("Window") ) avail_start = models.TimeField( blank=True, null=True, help_text=_("Start time of the available advertising slot"), verbose_name=_("Avail Start"), db_index=True ) availinwindow = models.IntegerField( db_column='availInWindow', # Preserve original database column name blank=True, null=True, help_text=_("Position or identifier of avail within the window"), verbose_name=_("Avail in Window") ) # Add custom manager objects = AvailsManager() class Meta: db_table = 'Avails' verbose_name = _('Available Slot') verbose_name_plural = _('Available Slots') ordering = ['avail_start', 'availinwindow'] indexes = [ models.Index(fields=['window', 'avail_start'], name='avail_window_start_idx'), models.Index(fields=['availinwindow'], name='avail_position_idx'), ] def __str__(self): """String representation of the available slot.""" window_info = f"Window {self.window.id}" if self.window else "No Window" return f"Avail {self.id} ({window_info}) at {self.avail_start}" def get_adspots_count(self): """Get count of ad spots in this avail.""" return self.adspots_in_avail.count() def is_available(self): """Check if this avail has no ad spots placed.""" return self.adspots_in_avail.count() == 0 def get_next_position(self): """Get the next available position for ad spot placement.""" max_position = self.adspots_in_avail.aggregate( max_pos=models.Max('positioninavail') )['max_pos'] return (max_position or 0) + 1 def create_copy(self, new_window): """Create a copy of this avail for a new window.""" new_avail = Avails( window=new_window, avail_start=self.avail_start, availinwindow=self.availinwindow ) new_avail.save() # Copy ad spot placements for adspot_placement in self.adspots_in_avail.all(): adspot_placement.create_copy(new_avail) return new_avail class AdspotsInAvail(BaseModel): """ Model representing the placement of specific ad spots within available slots. This is the junction model that connects advertising spots to specific avails, managing the actual placement and positioning of advertisements within the broadcast schedule. Attributes: id_avail (ForeignKey): Reference to the available slot id_adspot (ForeignKey): Reference to the advertisement spot positioninavail (int): Position of the ad spot within the avail trafficid (int): Traffic system identifier for the placement """ # Foreign key relationship to the available advertising slot avail = models.ForeignKey( Avails, on_delete=models.CASCADE, db_column='avail', blank=True, null=True, help_text=_("The available slot where this ad spot is placed"), related_name='adspots_in_avail', verbose_name=_("Avail") ) # Foreign key relationship to the advertisement spot from campaigns app adspot = models.ForeignKey( Adspots, models.DO_NOTHING, db_column='adspot', blank=True, null=True, help_text=_("The advertisement spot being placed"), related_name='placements_in_avail', verbose_name=_("Ad Spot") ) # Position/sequence number of the ad spot within the available slot # Allows for multiple ads within a single avail with specific ordering positioninavail = models.IntegerField( db_column='positionInAvail', # Preserve original database column name blank=True, null=True, help_text=_("Position of the ad spot within the avail (1-based)"), verbose_name=_("Position in Avail"), validators=[MinValueValidator(1)], db_index=True ) # Traffic system identifier for integration with external traffic systems trafficid = models.IntegerField( db_column='trafficId', # Preserve original database column name blank=True, null=True, help_text=_("Traffic system identifier for the placement"), verbose_name=_("Traffic ID"), db_index=True ) # Add custom manager objects = AdSpotsInAvailManager() class Meta: db_table = 'Adspots_in_avail' verbose_name = _('Ad Spot Placement') verbose_name_plural = _('Ad Spot Placements') ordering = ['positioninavail', 'id'] indexes = [ models.Index(fields=['avail', 'positioninavail'], name='adspot_avail_position_idx'), models.Index(fields=['adspot'], name='adspot_spot_idx'), models.Index(fields=['trafficid'], name='adspot_traffic_idx'), ] constraints = [ # Ensure unique positioning within each avail models.UniqueConstraint( fields=['avail', 'positioninavail'], name='unique_adspot_position_per_avail', condition=Q(positioninavail__isnull=False) ), ] def __str__(self): """String representation of the ad spot placement.""" adspot_info = f"AdSpot {self.adspot_id}" if self.adspot_id else "No AdSpot" avail_info = f"Avail {self.avail_id}" if self.avail_id else "No Avail" position_info = f"Position {self.positioninavail}" if self.positioninavail else "No Position" return f"{adspot_info} in {avail_info} at {position_info}" def clean(self): """Custom validation for the model.""" super().clean() # Validate position is within reasonable bounds if self.positioninavail and self.positioninavail > 100: raise ValidationError({ 'positioninavail': _("Position cannot exceed 100.") }) # Check for position conflicts within the same avail if self.avail and self.positioninavail: conflicting = AdspotsInAvail.objects.filter( avail=self.avail, positioninavail=self.positioninavail ).exclude(pk=self.pk) if conflicting.exists(): raise ValidationError({ 'positioninavail': _("This position is already occupied in this avail.") }) def save(self, *args, **kwargs): """Custom save method with additional logic.""" # Auto-assign position if not provided if not self.positioninavail and self.avail: self.positioninavail = self.avail.get_next_position() # Log placement creation logger.info(f"Saving ad spot placement: {self.adspot_id} in avail {self.avail_id}") super().save(*args, **kwargs) def create_copy(self, new_avail): """Create a copy of this placement for a new avail.""" new_placement = AdspotsInAvail( avail=new_avail, adspot=self.adspot, positioninavail=self.positioninavail, trafficid=self.trafficid ) new_placement.save() return new_placement class Verifs(BaseModel): """ Model for storing advertisement verification and broadcast confirmation data. This model tracks the actual broadcast verification of advertisements, recording when ads were aired, their duration, status, and other verification metadata for billing and compliance purposes. Attributes: networkname (str): Name of the broadcasting network zonename (str): Name of the broadcasting zone broadcast_date (DateField): Date when the ad was broadcast trafficId (int): Traffic system identifier spotId (str): Unique identifier for the advertisement spot air_time (TimeField): Time when the ad was aired air_length (DurationField): Duration of the aired advertisement airStatuscode (str): Status code indicating broadcast result revision (int): Revision number for the verification record vercomplete (str): Verification completion status """ # Name of the broadcasting network networkname = models.CharField( max_length=255, help_text=_("Name of the broadcasting network"), verbose_name=_("Network Name"), db_index=True ) # Name of the broadcasting zone within the network zonename = models.CharField( max_length=255, help_text=_("Name of the broadcasting zone"), verbose_name=_("Zone Name"), db_index=True ) # TODO APPLIED: Changed from CharField to DateField for better data validation broadcast_date = models.DateField( help_text=_("Date when the advertisement was broadcast"), verbose_name=_("Broadcast Date"), db_index=True ) # Traffic system identifier linking to traffic management system trafficId = models.IntegerField( help_text=_("Traffic system identifier for the advertisement"), verbose_name=_("Traffic ID"), db_index=True ) # Unique identifier for the advertisement spot spotId = models.CharField( max_length=255, help_text=_("Unique identifier for the advertisement spot"), verbose_name=_("Spot ID"), db_index=True ) # TODO APPLIED: Changed from CharField to TimeField for better data validation air_time = models.TimeField( help_text=_("Time when the advertisement was aired"), verbose_name=_("Air Time"), db_index=True ) # TODO APPLIED: Changed from CharField to DurationField for better data handling air_length = models.DurationField( help_text=_("Duration of the aired advertisement"), verbose_name=_("Air Length") ) # Status code indicating the result of the broadcast attempt # (e.g., success, failure, partial, etc.) airStatuscode = models.CharField( max_length=255, help_text=_("Status code indicating broadcast result"), verbose_name=_("Air Status Code"), db_index=True ) # Revision number for version control of verification records revision = models.IntegerField( default=1, help_text=_("Revision number for the verification record"), verbose_name=_("Revision"), validators=[MinValueValidator(1)], db_index=True ) # Verification completion status flag vercomplete = models.CharField( db_column='verComplete', # Preserve original database column name max_length=50, blank=True, null=True, help_text=_("Verification completion status"), verbose_name=_("Verification Complete"), db_index=True ) # Add custom manager objects = VerifsManager() class Meta: db_table = "Verifs" verbose_name = _('Advertisement Verification') verbose_name_plural = _('Advertisement Verifications') ordering = ['-broadcast_date', '-air_time', '-revision'] indexes = [ models.Index(fields=['networkname', 'zonename'], name='verif_network_zone_idx'), models.Index(fields=['broadcast_date', 'air_time'], name='verif_broadcast_air_idx'), models.Index(fields=['trafficId', 'revision'], name='verif_traffic_revision_idx'), models.Index(fields=['spotId', 'broadcast_date'], name='verif_spot_date_idx'), models.Index(fields=['airStatuscode'], name='verif_status_idx'), models.Index(fields=['vercomplete'], name='verif_complete_idx'), ] constraints = [ # Ensure unique combination of traffic ID, spot ID, and revision models.UniqueConstraint( fields=['trafficId', 'spotId', 'revision'], name='unique_verif_per_traffic_spot_revision' ), ] def __str__(self): """String representation of the verification record.""" status = "Complete" if self.is_verification_complete else "Pending" return f"Verification {self.spotId} - {self.networkname}/{self.zonename} on {self.broadcast_date} ({status})" def clean(self): """Custom validation for the model.""" super().clean() # Validate required fields combination if self.trafficId and not self.spotId: raise ValidationError({ 'spotId': _("Spot ID is required when Traffic ID is provided.") }) # Validate status code format if needed if self.airStatuscode and len(self.airStatuscode) > 50: raise ValidationError({ 'airStatuscode': _("Status code cannot exceed 50 characters.") }) def save(self, *args, **kwargs): """Custom save method with additional logic.""" # Log verification save logger.info(f"Saving verification for spot {self.spotId} on {self.broadcast_date}") super().save(*args, **kwargs) @property def is_verification_complete(self): """Check if verification is marked as complete.""" if not self.vercomplete: return False return self.vercomplete.lower() in ['complete', 'yes', '1', 'true'] @property def status_display(self): """Get human-readable verification status.""" if self.is_verification_complete: return _("Complete") elif self.vercomplete: return self.vercomplete.title() else: return _("Pending") @property def is_successful_broadcast(self): """Check if the broadcast was successful based on status code.""" success_codes = ['success', 'ok', '200', 'aired', 'broadcast'] return self.airStatuscode and self.airStatuscode.lower() in success_codes def mark_complete(self, user=None): """Mark verification as complete.""" self.vercomplete = 'complete' self.save(update_fields=['vercomplete']) logger.info(f"Verification {self.id} marked complete by user {user}") def create_revision(self): """Create a new revision of this verification.""" new_revision = Verifs( networkname=self.networkname, zonename=self.zonename, broadcast_date=self.broadcast_date, trafficId=self.trafficId, spotId=self.spotId, air_time=self.air_time, air_length=self.air_length, airStatuscode=self.airStatuscode, revision=self.revision + 1, vercomplete=None # New revision starts as incomplete ) new_revision.save() logger.info(f"Created revision {new_revision.revision} for verification {self.spotId}") return new_revision @classmethod def get_latest_revision(cls, traffic_id, spot_id): """Get the latest revision for a specific traffic ID and spot ID.""" return cls.objects.filter( trafficId=traffic_id, spotId=spot_id ).order_by('-revision').first() def get_broadcast_summary(self): """Get a summary of broadcast information.""" return { 'network': self.networkname, 'zone': self.zonename, 'date': self.broadcast_date, 'time': self.air_time, 'duration': self.air_length, 'status': self.airStatuscode, 'complete': self.is_verification_complete, 'successful': self.is_successful_broadcast } # # Enhanced version of the Placement model with full implementation # class PlacementManager(models.Manager): # """Custom manager for Placement model.""" # def for_campaign(self, campaign_id): # """Get placements for a specific campaign.""" # return self.filter(id_campaign=campaign_id) # def for_channel(self, channel_id): # """Get placements for a specific channel.""" # return self.filter(channel_id=channel_id) # def for_time_slot(self, time_id): # """Get placements for a specific time slot.""" # return self.filter(id_time=time_id) # def active(self): # """Get active placements.""" # return self.filter(is_active=True) # class Placement(models.Model): # """ # Model for managing advertisement placements across campaigns and time slots. # This model handles the broader placement strategy, linking campaigns # to specific time slots and channels for strategic ad placement planning. # It serves as a high-level scheduling and campaign management tool. # Attributes: # id_placement (AutoField): Primary key for the placement # id_time (int): Time slot identifier # id_campaign (int): Campaign identifier # channel (ForeignKey): Reference to the broadcasting channel # placement_date (DateField): Date when the placement is scheduled # priority (int): Priority level for placement (1-10, higher is more important) # is_active (bool): Whether the placement is currently active # created_at (DateTimeField): When the placement was created # updated_at (DateTimeField): When the placement was last updated # notes (TextField): Additional notes about the placement # """ # # Primary key with auto-increment # id_placement = models.AutoField( # primary_key=True, # help_text=_("Unique identifier for the placement"), # verbose_name=_("Placement ID") # ) # # Time slot identifier (would link to a TimeSlot model) # id_time = models.IntegerField( # help_text=_("Time slot identifier for the placement"), # verbose_name=_("Time Slot ID"), # db_index=True # ) # # Campaign identifier (would link to a Campaign model) # id_campaign = models.IntegerField( # help_text=_("Campaign identifier for the placement"), # verbose_name=_("Campaign ID"), # db_index=True # ) # # Foreign key relationship to the broadcasting channel # channel = models.ForeignKey( # Channels, # models.DO_NOTHING, # db_column='channel', # blank=True, # null=True, # help_text=_("The channel where the placement will occur"), # related_name='placements', # verbose_name=_("Channel") # ) # # Date when the placement is scheduled # placement_date = models.DateField( # blank=True, # null=True, # help_text=_("Date when the placement is scheduled"), # verbose_name=_("Placement Date"), # db_index=True # ) # # Priority level for placement conflicts # priority = models.IntegerField( # default=5, # help_text=_("Priority level for placement (1-10, higher is more important)"), # verbose_name=_("Priority"), # validators=[MinValueValidator(1), MaxValueValidator(10)] # ) # # Active status flag # is_active = models.BooleanField( # default=True, # help_text=_("Whether the placement is currently active"), # verbose_name=_("Is Active"), # db_index=True # ) # # Timestamp fields for tracking # created_at = models.DateTimeField( # auto_now_add=True, # help_text=_("When the placement was created"), # verbose_name=_("Created At") # ) # updated_at = models.DateTimeField( # auto_now=True, # help_text=_("When the placement was last updated"), # verbose_name=_("Updated At") # ) # # Additional notes # notes = models.TextField( # blank=True, # null=True, # help_text=_("Additional notes about the placement"), # verbose_name=_("Notes") # ) # # Add custom manager # objects = PlacementManager() # class Meta: # db_table = 'Placement' # verbose_name = _('Advertisement Placement') # verbose_name_plural = _('Advertisement Placements') # ordering = ['-placement_date', '-priority', '-created_at'] # indexes = [ # models.Index(fields=['id_campaign', 'placement_date'], name='placement_campaign_date_idx'), # models.Index(fields=['channel', 'id_time'], name='placement_channel_time_idx'), # models.Index(fields=['is_active', 'priority'], name='placement_active_priority_idx'), # models.Index(fields=['placement_date'], name='placement_date_idx'), # ] # constraints = [ # # Ensure unique placement per campaign, channel, time, and date # models.UniqueConstraint( # fields=['id_campaign', 'channel', 'id_time', 'placement_date'], # name='unique_placement_per_campaign_channel_time_date', # condition=Q(is_active=True) # ), # ] # def __str__(self): # """String representation of the placement.""" # channel_name = self.channel.name if self.channel else "No Channel" # status = "Active" if self.is_active else "Inactive" # return f"Placement {self.id_placement} - Campaign {self.id_campaign} on {channel_name} ({status})" # def clean(self): # """Custom validation for the model.""" # super().clean() # # Validate placement date is not in the past for new placements # if not self.pk and self.placement_date and self.placement_date < timezone.now().date(): # raise ValidationError({ # 'placement_date': _("Placement date cannot be in the past.") # }) # def save(self, *args, **kwargs): # """Custom save method with additional logic.""" # # Log placement save # logger.info(f"Saving placement {self.id_placement} for campaign {self.id_campaign}") # super().save(*args, **kwargs) # def deactivate(self, user=None): # """Deactivate this placement.""" # self.is_active = False # self.save(update_fields=['is_active']) # logger.info(f"Placement {self.id_placement} deactivated by user {user}") # def activate(self, user=None): # """Activate this placement.""" # self.is_active = True # self.save(update_fields=['is_active']) # logger.info(f"Placement {self.id_placement} activated by user {user}") # @property # def status_display(self): # """Get human-readable status.""" # return _("Active") if self.is_active else _("Inactive") # def get_conflicts(self): # """Get other placements that might conflict with this one.""" # return Placement.objects.filter( # channel=self.channel, # id_time=self.id_time, # placement_date=self.placement_date, # is_active=True # ).exclude(pk=self.pk) # def has_conflicts(self): # """Check if this placement has conflicts.""" # return self.get_conflicts().exists() # Utility functions for the models def get_playlist_statistics(): """Get overall statistics for playlists.""" return { 'total_playlists': Playlists.objects.count(), 'published_playlists': Playlists.objects.published().count(), 'draft_playlists': Playlists.objects.drafts().count(), 'active_playlists': Playlists.objects.active_now().count(), } def get_verification_statistics(): """Get overall statistics for verifications.""" return { 'total_verifications': Verifs.objects.count(), 'completed_verifications': Verifs.objects.completed().count(), 'pending_verifications': Verifs.objects.pending().count(), 'completion_rate': ( Verifs.objects.completed().count() / max(Verifs.objects.count(), 1) * 100 ), } def cleanup_old_drafts(days=30): """Clean up old draft playlists.""" cutoff_date = timezone.now() - timezone.timedelta(days=days) old_drafts = Playlists.objects.filter( is_draft=True, creation_at__lt=cutoff_date ) count = old_drafts.count() old_drafts.delete() logger.info(f"Cleaned up {count} old draft playlists older than {days} days") return count # Migration helper class class DataTypeMigrationHelper: """ Helper class for migrating data when changing field types. Use this in Django data migrations to convert existing string data to proper field types. """ @staticmethod def migrate_playlist_dates(): """Migrate playlist broadcastdate from CharField to DateField.""" from django.db import transaction with transaction.atomic(): playlists = Playlists.objects.filter( broadcast_date__isnull=True, broadcastdate__isnull=False ) for playlist in playlists: converted_date = convert_date_string_to_date(playlist.broadcastdate) if converted_date: playlist.broadcast_date = converted_date playlist.save(update_fields=['broadcast_date']) logger.info("Completed playlist date migration") @staticmethod def migrate_window_times(): """Migrate window times from CharField to TimeField.""" from django.db import transaction with transaction.atomic(): windows = Windows.objects.filter( models.Q(window_start__isnull=True) | models.Q(window_end__isnull=True) ) for window in windows: if hasattr(window, 'window_start_old') and window.window_start_old: converted_start = convert_time_string_to_time(window.window_start_old) if converted_start: window.window_start = converted_start if hasattr(window, 'window_end_old') and window.window_end_old: converted_end = convert_time_string_to_time(window.window_end_old) if converted_end: window.window_end = converted_end if hasattr(window, 'window_duration_old') and window.window_duration_old: converted_duration = convert_duration_string_to_duration(window.window_duration_old) if converted_duration: window.window_duration = converted_duration window.save() logger.info("Completed window times migration") @staticmethod def migrate_verification_data(): """Migrate verification data from string fields to proper types.""" from django.db import transaction with transaction.atomic(): verifs = Verifs.objects.filter( models.Q(broadcast_date__isnull=True) | models.Q(air_time__isnull=True) | models.Q(air_length__isnull=True) ) for verif in verifs: if hasattr(verif, 'broadcastDate') and verif.broadcastDate: converted_date = convert_date_string_to_date(verif.broadcastDate) if converted_date: verif.broadcast_date = converted_date if hasattr(verif, 'airTime') and verif.airTime: converted_time = convert_time_string_to_time(verif.airTime) if converted_time: verif.air_time = converted_time if hasattr(verif, 'airLength') and verif.airLength: converted_duration = convert_duration_string_to_duration(verif.airLength) if converted_duration: verif.air_length = converted_duration verif.save() logger.info("Completed verification data migration")