import uuid from django.conf import settings from django.db import models from abstractclass.models import NetworkMonitoringAbstractModel, NetworkMonitoringAbstractUpdateStatus from django.contrib.auth import get_user_model User = get_user_model() from django.db import transaction # Final ad model class Ads(NetworkMonitoringAbstractModel): pass class AdsManual(NetworkMonitoringAbstractModel): pass class DataManagement(models.Model): source = models.OneToOneField(Ads, on_delete=models.CASCADE, related_name='NetworkMonitoring_ad') isSendToMainServer = models.BooleanField(default=False) listing_date = models.DateField(null=True, blank=True) isMerged = models.BooleanField(default=False) DOWNLOAD_CHOICES = [ ("pc", "PC"), ("hously.space", "Hously Server"), ('server', 'Serwvr') ] DownloadManagment = models.CharField( max_length=50, choices=DOWNLOAD_CHOICES, blank=True, null=True ) worker_number = models.PositiveIntegerField(null=True, blank=True) ############## LEVEL 1 AGREGATOR ############## class SourceNetwork(models.Model): id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) title = models.CharField(max_length=255) type = models.CharField(default="portal") name = models.CharField(max_length=100) base_url = models.URLField() structure = models.CharField(max_length=255) skip_when_less = models.PositiveIntegerField(null=True) params = models.JSONField(default=dict) pagination = models.CharField(max_length=64, default="") selectors = models.JSONField(default=dict) is_ai = models.BooleanField(default=True) enabled = models.BooleanField(default=True) last_checked = models.DateTimeField(null=True, blank=True) last_status = models.CharField(max_length=64, blank=True, default="") created_at = models.DateTimeField(auto_now_add=True) @property def has_html_config(self): # Jeśli istnieje powiązany HtmlSourceNetwork, zwróci True return hasattr(self, "html_config") and self.html_config is not None def save_with_version(self, user=None, comment=""): with transaction.atomic(): # Pobierz ostatnią wersję last_version = NetworkSourceVersion.objects.filter(source=self).order_by("-version").first() new_version_number = (last_version.version + 1) if last_version else 1 # Stwórz wersję NetworkSourceVersion.objects.create( source=self, version=new_version_number, title=self.title, type=self.type, name=self.name, base_url=self.base_url, structure=self.structure, skip_when_less=self.skip_when_less, params=self.params, pagination=self.pagination, selectors=self.selectors, is_ai=self.is_ai, enabled=self.enabled, last_checked=self.last_checked, last_status=self.last_status, author=user, comment=comment, ) # Zapisz główny obiekt self.save() def __str__(self): return self.title ############## LEVEL 2 AGREGATOR ############## class SourceHtml(models.Model): source = models.OneToOneField(SourceNetwork, on_delete=models.CASCADE, related_name='html_config') type = models.CharField(default="portal") title = models.CharField(max_length=255) name = models.CharField(max_length=100) actions = models.JSONField(default=list, blank=True) selectors = models.JSONField(default=dict, blank=True) inactive = models.JSONField(default=dict, blank=True) raw_data_setup = models.JSONField(default=dict, blank=True) def __str__(self): return f"HTML config for {self.source.title}" class SourceManual(models.Model): source = models.OneToOneField(SourceNetwork, on_delete=models.CASCADE, related_name='manual_data_source_fetcher', blank=True, null=True) html_source = models.OneToOneField(SourceHtml, on_delete=models.CASCADE, related_name='manual_html_source', blank=True, null=True) title = models.CharField(max_length=255) name = models.CharField(max_length=100) enable = models.BooleanField(default=True) actions = models.JSONField(default=list, blank=True, null=True) rules = models.JSONField(default=dict, blank=True, null=True) # [ # { # "source": "name", # "match": ["adresowo", "adresowo.pl"], # "type": "adresowo" # }, # { # "selector": ".typ-ogloszenia", # "match": ["mieszkanie", "apartament"], # "type": "flat" # }, # { # "selector": ".typ-ogloszenia", # "match": ["dom", "bliźniak"], # "type": "house" # } # ] selectors = models.JSONField(default=dict, blank=True, null=True) # "selectors": { # "type1": { # "price": { "selector": ".price", "fieldType": "text" }, # "title": { "selector": ".title", "fieldType": "text" } # }, # "type2": { # "price": { "selector": ".price-dom", "fieldType": "text" }, # "garden": { "selector": ".ogród", "fieldType": "boolKey", "trueOptions": ["ogród"], "falseOptions": ["brak ogrodu"] } # } # }, # { # "parking": { # "selector": ".parking-info", # "fieldType": "boolKey", # "trueOptions": ["parking", "garaż", "parkingu", "parkinkiem"], # "falseOptions": ["brak parkingu", "bez garażu"], # "ifMissing": "false" # }, # "price": { # "selector": ".price", # "fieldType": "text", # "ifMissing": "default", # "defaultValue": "Cena niedostępna" # }, # "description": { # "selectors": [".desc1", ".desc2", ".desc3"], # "fieldType": "many", # "cleanOptions": ["\\n", "----"], # "ifMissing": "null" # } # } transferred = models.JSONField(default=dict, blank=True, null=True) # "transferred": { # "source_name": "name", # "meta_copy": "meta", # "listing_url": "url" # } inactive = models.JSONField(default=dict, blank=True, null=True) # [ # { # "type": "text_contains", # "text": "ogłoszenie wygasło" # }, # { # "type": "selector_missing", # "selector": ".contact-button" # }, # { # "type": "selector_contains", # "selector": "div.alert", # "text": "archiwalne" # }, # { # "type": "source_field_match", # "field": "note", # "match": "nieaktualne" # } # ] type = models.CharField(default="portal", blank=True, null=True) def __str__(self): return f"HTML config for {self.source.title}" class NetworkSourceVersion(models.Model): source = models.ForeignKey(SourceNetwork, on_delete=models.CASCADE, related_name="versions") type = models.CharField(default="portal") version = models.PositiveIntegerField() title = models.CharField(max_length=255) name = models.CharField(max_length=100) base_url = models.URLField() structure = models.CharField(max_length=255) skip_when_less = models.PositiveIntegerField(null=True) params = models.JSONField(default=dict) pagination = models.CharField(max_length=64, default="") selectors = models.JSONField(default=dict) html_selectors = models.JSONField(default=dict, blank=True) is_ai = models.BooleanField(default=True) enabled = models.BooleanField(default=True) last_checked = models.DateTimeField(null=True, blank=True) last_status = models.CharField(max_length=64, blank=True, default="") created_at = models.DateTimeField(auto_now_add=True) author = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True) comment = models.TextField(blank=True, default="") # np. "Zmiana selektorów po zmianie strony" class Meta: ordering = ["-created_at", "-version"] def __str__(self): return f"{self.source.title} v{self.version}" ############## AGREGATORS ERRORS ############## class NetworkSourceError(models.Model): source = models.ForeignKey(SourceNetwork, on_delete=models.CASCADE) error_message = models.TextField(blank=True, default="") error_type = models.CharField(max_length=255, blank=True, default="") created_at = models.DateTimeField(auto_now_add=True) ############## RAW OUTPUT ############## class NetworkMonitoredPage(models.Model): source = models.ForeignKey(SourceNetwork, on_delete=models.CASCADE) name = models.CharField(max_length=100) estate_type = models.CharField(max_length=255, blank=True, null=True) offer_type = models.CharField(max_length=20, null=True, blank=True) meta = models.JSONField(default=dict, null=True) url = models.CharField() html = models.TextField() sliced_html = models.TextField(blank=True, default="") raw_data = models.TextField(blank=True, null=True) image_links = models.JSONField(default=list, blank=True) parse_data = models.JSONField(default=list, blank=True) date_fetched = models.DateTimeField(auto_now_add=True) is_fetched = models.BooleanField(default=True) is_raw_data = models.BooleanField(default=False) is_complete = models.BooleanField(default=False) network_ad = models.OneToOneField(Ads, on_delete=models.SET_NULL, null=True, blank=True) # po przetworzeniu network_ad_manual = models.OneToOneField(AdsManual, on_delete=models.SET_NULL, null=True, blank=True) # po przetworzeniu inactive_date = models.DateTimeField(null=True, blank=True) meta = models.JSONField(default=dict, null=True) is_active = models.BooleanField(default=True) check_active = models.BooleanField(default=False) created_at = models.DateTimeField(auto_now_add=True) class Meta: unique_together = ('source', 'url') class NetworkPageError(models.Model): network_page = models.ForeignKey(NetworkMonitoredPage, on_delete=models.CASCADE) error_message = models.TextField(blank=True, default="") error_type = models.CharField(max_length=255, blank=True, default="") created_at = models.DateTimeField(auto_now_add=True) class ManualParserLog(models.Model): SKIP_TYPE_CHOICES = [ ('error', 'Error - Failed to Parse'), ('intentional', 'Intentional - Skipped by Rules'), ('inactive', 'Inactive - Detected as Expired'), ('validation', 'Validation - Field Length/Type Error'), ('no_data', 'No Data - Empty parsed_data'), ] network_page = models.ForeignKey(NetworkMonitoredPage, on_delete=models.CASCADE, related_name='manual_parser_logs') source = models.ForeignKey(SourceNetwork, on_delete=models.CASCADE) source_name = models.CharField(max_length=100) domain = models.CharField(max_length=255, blank=True, default="") url = models.CharField(max_length=2000) skip_type = models.CharField(max_length=20, choices=SKIP_TYPE_CHOICES, default='error') error_message = models.TextField(blank=True, default="") error_summary = models.CharField(max_length=500, blank=True, default="") parsed_successfully = models.BooleanField(default=False) saved_to_ads_manual = models.BooleanField(default=False) field_errors = models.JSONField(blank=True, null=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: indexes = [ models.Index(fields=['skip_type', 'created_at']), models.Index(fields=['source_name', 'skip_type']), models.Index(fields=['saved_to_ads_manual']), ] def __str__(self): return f"[{self.skip_type}] {self.source_name} - {self.url[:50]}" # need to be change class NetworkMonitoredObserve(models.Model): page = models.ForeignKey(NetworkMonitoredPage, on_delete=models.CASCADE) user = models.ForeignKey(User, on_delete=models.CASCADE) is_observe = models.BooleanField(default=False) last_scraped = models.DateTimeField(null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: unique_together = ('page', 'user') def __str__(self): return f"{self.user} obserwuje {self.page.url}" class RawMonitoredLink(models.Model): url = models.CharField() source = models.ForeignKey('SourceNetwork', on_delete=models.CASCADE, null=True) agregate_at = models.DateTimeField(auto_now_add=True) to_check = models.BooleanField(default=True) # <-- nowe pole! class Meta: unique_together = ('url', 'source') # class GeneralConfig(models.Moedel): # skip_words = models.JSONFIELD(dict=list, blank=True) # skip_images_key = models.JSONFIELD(dict=list, blank=True) # action_list = models.JSONFIELD(dict=list, blank=True) # setting_options = models.JSONFILED(dict=list, blank=True) # Ad status checker (is active or not) class NetworkStatus(models.Model): url = models.URLField(max_length=2000, unique=True) source = models.ForeignKey('SourceNetwork', on_delete=models.CASCADE) class Meta: unique_together = ('url', 'source')