import datetime import json import time import io # from msilib.schema import File import os from openai import OpenAI import environ import requests from bs4 import BeautifulSoup from django.core.files.images import ImageFile from django.core.files.base import ContentFile from data_scraper.models import OldData, OneNewHomesData, OneNewHomesDeveloperData, OneNewHomesDevelopmentsData, OneNewsHomesDeveloper from developer.models import ContactPerson, Developer, DeveloperTranslate from development.models import Attachment, BuildingType, BuildingTypeTranslate, Currency, Development, DevelopmentTranslate, Feature, FeatureTranslate, Flat, KeyFeature, PaymentOption, PaymentOptionTranslate from geo_location.models import City, Country, Location from main.models import Language from main.utils import log_func from django.contrib.gis.geos import Point from django.utils.text import slugify env = environ.Env() environ.Env.read_env() openai_key = env.str('openai_key') openai_pr_id = env.str('openai_pr_id') openai_org_id = env.str('openai_org_id') PARAPHRASE_PROMPT = """Paraphrase the following real estate description to make it more appealing and engaging for potential buyers, while maintaining all factual information. The description should be concise and clear: Original Description: "{text}" Paraphrased Description:""" def paraphrase_text(text, prompt=PARAPHRASE_PROMPT, model="gpt-4o-mini"): try: client = OpenAI( api_key=openai_key, # organization=openai_org_id, # project=openai_pr_id, ) prompt = prompt.format(text=text) chat = client.chat.completions.create( model = model, messages = [{"role": "system", "content": "You are an experienced real estate consultant with deep market knowledge."}, {"role": "user", "content": prompt}], temperature = 0.7 ) return chat.choices[0].message.content except Exception as e: log_func("paraphrase_text", e) return None TRANSLATE_PROMPT = """Translate the following real estate description from {from_language} to {to_language}, while maintaining all factual information. The description should be concise and clear: Original Description: "{text}" Translated Description:""" def translate_text(text, from_language="english", to_language="persian", prompt=TRANSLATE_PROMPT): try: client = OpenAI( api_key=openai_key, # organization=openai_org_id, # project=openai_pr_id, ) prompt = prompt.format(from_language=from_language, to_language=to_language, text=text) chat = client.chat.completions.create( model = "gpt-4o-mini", messages = [{"role": "system", "content": "You are an experienced real estate consultant with deep market knowledge."}, {"role": "user", "content": prompt}], temperature = 0.7 ) print(chat.choices[0].message.content) return chat.choices[0].message.content except Exception as e: print(e) log_func("translate_text", e) return None def translate_text_by_google(text, from_language="en", to_language="fa"): """Translates the given text from the specified source language to the target language using the Google Translator. Args: text (str): The text to be translated. from_language (str, optional): The source language code. Defaults to "en". to_language (str, optional): The target language code. Defaults to "fa". Returns: str: The translated text. """ from deep_translator import GoogleTranslator, ChatGptTranslator, DeeplTranslator # Initialize the translator # translator = ChatGptTranslator(source=from_language, target=to_language) # translator = DeeplTranslator(source=from_language, target=to_language) translator = GoogleTranslator(source=from_language, target=to_language) # Translate the text translated_text = translator.translate(text) return translated_text def get_old_developer(): base_url = "https://api.entralon.com/" url = f"{base_url}api/v1/developer?limit=1000" payload = "" headers = {} response = requests.request("GET", url, headers=headers, data=payload, timeout=30) if response.status_code == 200: language, _ = Language.objects.get_or_create(name="english", code="en") data = response.json() for item in data['data']: try: name = item.get('title') print(f"start developer:{name}") description = item.get('detailedDescription') # contact_persons = item.get('contactPersons', []) target_location = item.get('targetLocation') slug = slugify(name, allow_unicode=True) image = item.get('image') f = None try: image_file_request = requests.get(f"{base_url}{image}", timeout=30) if image_file_request.status_code == 200: f = ImageFile(io.BytesIO(image_file_request.content), name=f'{slug}.png') except Exception as e: print("image save error", e) website = item.get('website') developer, _ = Developer.objects.get_or_create(slug=slug) developer.name = name developer.image = f developer.target_location = target_location developer.web_address = website developer.save() dt, _ = DeveloperTranslate.objects.get_or_create(developer=developer,language=language) dt.name=name, dt.slug=slug, dt.description=description dt.save() except Exception as e: print("error for loop: ", e) def download_old_development(): base_url = "https://api.entralon.com/" url = f"{base_url}api/v1/developments?limit=1000" for i in range(1, 24): url = f"{base_url}api/v1/developments?page={i}&limit=50" payload = "" headers = {} response = requests.request("GET", url, headers=headers, data=payload, timeout=30) if response.status_code == 200: data = response.json() for item in data['data']: OldData.objects.create(data=item) def get_old_developments(): base_url = "https://api.entralon.com/" language_en, _ = Language.objects.get_or_create(name="english", code="en") language_fa, _ = Language.objects.get_or_create(name="persian", code="fa") language_ru, _ = Language.objects.get_or_create(name="russian", code="ru") language_ar, _ = Language.objects.get_or_create(name="arabic", code="ar") for data in OldData.objects.all(): item = data.data try: name = item.get('title') slug = slugify(name, allow_unicode=True) development, _ = Development.objects.get_or_create(slug=slug) development.title = name print(f"start development:{name}") translations = item.get('translations') en = translations.get('en') try: description_en = en.get('description') except: description_en = "" fa = translations.get('fa') try: description_fa = fa.get('description') except: description_fa = "" ru = translations.get('ru') try: description_ru = ru.get('description') except: description_ru = "" ar = translations.get('ar') try: description_ar = ar.get('description') except: description_ar = "" try: DevelopmentTranslate.objects.create(language=language_en, development=development, title=name, slug=slug, description=description_en) except: pass try: DevelopmentTranslate.objects.create(language=language_fa, development=development, title=name, slug=slug, description=description_fa) except: pass try: DevelopmentTranslate.objects.create(language=language_ar, development=development, title=name, slug=slug, description=description_ar) except: pass try: DevelopmentTranslate.objects.create(language=language_ru, development=development, title=name, slug=slug, description=description_ru) except: pass currency_ = item.get('currency') try: currency = Currency.objects.get(code=currency_.get('acronym')) except: currency = None development.default_currency = currency basePrice = item.get('basePrice') try: developer_name = item.get('developer').get('title') developer_slug = slugify(developer_name, allow_unicode=True) developer, _ = Developer.objects.get_or_create(slug=developer_slug) except: developer = None development.developer = developer address = item.get('address') development.address = address completionDate = item.get('completionDate',"") if completionDate.lower() == "ready to move": completed_date = None completed_at = None completed_status = 1 else: completed_date = completionDate[:4] try: completed_at = datetime.datetime.strptime(completed_date, "%Y") except: completed_at = None completed_status = 0 development.completed_date = completed_date development.completed_at = completed_at development.completed_status = completed_status keyFeatures = item.get('keyFeatures') for feature in keyFeatures: feature_name = feature.get('feature') feature_slug = slugify(feature_name, allow_unicode=True) feature_translations = feature.get('translations') try: feature_name_fa = feature_translations.get('fa').get('feature') feature_slug_fa = slugify(feature_name_fa, allow_unicode=True) except: feature_name_fa = "" feature_slug_fa = "" try: feature_name_ru = feature_translations.get('ru').get('feature') feature_slug_ru = slugify(feature_name_ru, allow_unicode=True) except: feature_name_ru = "" feature_slug_ru = "" try: feature_name_ar = feature_translations.get('ar').get('feature') feature_slug_ar = slugify(feature_name_ar, allow_unicode=True) except: feature_name_ar = "" feature_slug_ar = "" try: feature_, _ = Feature.objects.get_or_create(name=feature_name, slug=feature_slug) except: feature_ = None if feature_ is not None: try: FeatureTranslate.objects.create(feature=feature_, name=feature_name, slug=feature_slug, language=language_en, description=feature_name) except: pass try: FeatureTranslate.objects.create(feature=feature_, name=feature_name_fa, slug=feature_slug_fa, language=language_fa, description=feature_name_fa) except: pass try: FeatureTranslate.objects.create(feature=feature_, name=feature_name_ar, slug=feature_slug_ar, language=language_ar, description=feature_name_ar) except: pass try: FeatureTranslate.objects.create(feature=feature_, name=feature_name_ru, slug=feature_slug_ru, language=language_ru, description=feature_name_ru) except: pass KeyFeature.objects.get_or_create(development=development, feature=feature_) buildingTypes = item.get('buildingType') for building_type in buildingTypes: building_type_name = building_type.lower() building_type_slug = slugify(building_type_name, allow_unicode=True) try: building_type_, _ = BuildingType.objects.get_or_create(name=building_type_name, slug=building_type_slug) except: building_type_ = None if building_type_ is not None: try: BuildingTypeTranslate.objects.create(building_type=building_type_, name=building_type_name, slug=building_type_slug, language=language_en, description=building_type_name) except: pass try: development.building_type.add(building_type_) except: pass paymentOptions = item.get('paymentOptions') for payment_option in paymentOptions: payment_option_name = payment_option payment_option_slug = slugify(payment_option_name, allow_unicode=True) try: payment_option_, _ = PaymentOption.objects.get_or_create(name=payment_option_name, slug=payment_option_slug) except: payment_option_ = None if payment_option_ is not None: try: PaymentOptionTranslate.objects.create(payment_option=payment_option_, name=payment_option_name, slug=payment_option_slug, language=language_en, description=payment_option_name) except: pass try: development.payment_option.add(payment_option_) except: pass flats = item.get('flats') for flat in flats: bedRoom = flat.get('bedRoom', 0) currentPrice = flat.get('currentPrice', 0) isAvailable = int(flat.get('isAvailable', 0)) > 0 square = flat.get('square', 0) pricePerMeter = flat.get('pricePerMeter', 0) flat, _ = Flat.objects.get_or_create(development=development, bedrooms_num=bedRoom) flat.area = square flat.price_per_meter = pricePerMeter flat.base_price = currentPrice flat.is_active = isAvailable flat.save() attachments = item.get('attachment') for attachment in attachments: category = attachment.get('category', "") file = attachment.get('file') f = None try: image_file_request = requests.get(f"{base_url}{file}", timeout=10) file_name = file.split('/')[-1] if image_file_request.status_code == 200: f = ImageFile(io.BytesIO(image_file_request.content), name=f'{file_name}') except: pass file_category = 0 if category.lower() in ['exterior']: file_category = 1 elif category.lower() in ['interior']: file_category = 2 elif category.lower() in ['amenities']: file_category = 3 elif category.lower() in ['floor plans','floor plan','floorplan','floorplans']: file_category = 4 elif category.lower() in ['site plan','site plans','siteplan','siteplans']: file_category = 5 Attachment.objects.create( file=f, image=f, title=name, alt=slug, category=file_category, development=development, type=0 ) bedRoom = item.get('bedRoom') basePrice = item.get('basePrice') development.base_price = basePrice development.is_published = item.get('isPublished') development.reference_id = item.get('referenceId') try: location = item.get('location') if location is not None: coordinates=location.get('coordinates') address_=location.get('formattedAddress') try: city=location.get('city') city_, _= City.objects.get_or_create(name=city) except: city_ = None try: country=location.get('country') countryCode=location.get('countryCode') country_, _ = Country.objects.get_or_create(name=country, code=countryCode) except: country_ = None street_=location.get('street') street_number=location.get('streetNumber') place_id=location.get('place_id') zip_code=location.get('zipCode') location_, _ =Location.objects.get_or_create(coordinates=Point(coordinates[0], coordinates[1])) location_.address = address_ location_.street = street_ location_.country = country_ location_.city = city_ location_.street_number = street_number location_.zip_code = zip_code location_.place_id = place_id location_.save() development.location = location_ except Exception as e: print("location error",e) development.save() except Exception as e: print(e) def add_currency(): with open('currencies.json', encoding="utf-8") as f: d = json.load(f) for item in d: print(item) data = d[item] print(data) c, _ = Currency.objects.get_or_create(code = item) c.name = data.get("name") c.demonym = data.get("demonym") c.major_single = data.get("majorSingle") c.major_plural = data.get("majorPlural") c.iso_num = data.get("ISOnum", 0) c.symbol = data.get("symbol") c.symbol_native = data.get("symbolNative") c.minor_single = data.get("minorSingle") c.minor_plural = data.get("minorPlural") c.iso_digits = data.get("ISOdigits") c.decimal_digits = data.get("decimals") c.num_to_basic = data.get("numToBasic") c.save()