### v1/management/commands/v1_transfer_mongodb_to_mongodb.py from django.core.management.base import BaseCommand from django.conf import settings import pymongo import time from retry import retry from datetime import datetime import os from math import radians, sin, cos, sqrt, atan2 # Import the translation dictionary from v1.management.commands.scripts.translation_dicts import * import logging logger = logging.getLogger(__name__) logging_file = os.path.join(settings.LOGGING_FILE_ROOT, 'v1_transfer_mongodb_to_mongodb.log') logging.basicConfig(filename=logging_file, level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') class Command(BaseCommand): help = 'Transfer data from MongoDB to MongoDB' def add_arguments(self, parser): ### limit argument parser.add_argument('--limit', type=int, nargs='?', default=None, help='Limit the number of records to transfer') ### step argument parser.add_argument('--step', type=int, nargs='?', default=None, help='Step to execute') ### batch_size argument parser.add_argument('--batch_size', type=int, nargs='?', default=None, help='batch_size to execute') def handle(self, *args, **options): #Start time start_time = time.time() ## Get the limit from the command line arguments limit = options['limit'] ## Get the step from the command line arguments step = options['step'] ## Get the batch_size from the command line arguments batch_size = options['batch_size'] # Coordinate to be used for distance calculation fixed_point = { "lat": -23.533773, "lng": -46.625290 } # pipeline_office = { # 'v1_beta_stage_offices': [ # # Step 1: Specify the source collection explicitly # { # '$unionWith': { # 'coll': "ilist_offices" # } # }, # # Step 2: Project core fields from `ilist_offices` # { # "$project": { # "office_id": "$ID", # "name": "$OfficeName", # "address": "$Address1", # "postal_code": "$PostalCode", # "website": "$URLToPrivatePage", # "logo": "$ImageURL", # "telephone": "$Phone", # "city_id": "$GeoData_CityID", # "lat": "$GeoCoordinates_Latitude", # "lng": "$GeoCoordinates_Longitude", # "total_properties": "$total_properties", # "creci": "$LicenseNumber", # "region_id": "$RegionID", # "base_complement": { # "slogan": "$Slogan", # "description": "$OfficeDescription", # "bullets": [ # "$BulletPoint1", # "$BulletPoint2", # "$BulletPoint3" # ], # "alternate_url_redirect_url": "$AlternateURL_RedirectURL", # "alternate_url_type": "$AlternateURL_UrlType", # "ultima_sincronia": { "$toDate": "$ModifiedDate" } # } # } # }, # # Step 3: Lookup city details from `ilist_cities` # { # "$lookup": { # "from": "ilist_cities", # "localField": "city_id", # "foreignField": "CityID", # "as": "city_info" # } # }, # # Step 4: Unwind the city information to flatten the array # { # "$unwind": { # "path": "$city_info", # "preserveNullAndEmptyArrays": True # } # }, # # Step 5: Lookup complements from `v1_mysql_offices_complements` # { # "$lookup": { # "from": "v1_mysql_offices_complements", # "localField": "office_id", # "foreignField": "referer_id", # "as": "mysql_complements" # } # }, # # Step 6: Merge base_complement with mysql_complements # { # "$addFields": { # "merged_complement": { # "$mergeObjects": [ # "$base_complement", # { # "$arrayElemAt": ["$mysql_complements", 0] # } # ] # } # } # }, # # Step 7: Add `distance` field using Haversine formula # { # "$addFields": { # "distance": { # "$let": { # "vars": { # "fixed_lat": -23.533773, # "fixed_lng": -46.625290, # "lat_radians": {"$multiply": [{"$divide": ["$lat", 180]}, 3.141592653589793]}, # "lng_radians": {"$multiply": [{"$divide": ["$lng", 180]}, 3.141592653589793]}, # "fixed_lat_radians": {"$multiply": [{"$divide": [-23.533773, 180]}, 3.141592653589793]}, # "fixed_lng_radians": {"$multiply": [{"$divide": [-46.625290, 180]}, 3.141592653589793]} # }, # "in": { # "$multiply": [ # 6371000, # Earth's radius in meters # { # "$acos": { # "$add": [ # { # "$multiply": [ # {"$cos": "$$lat_radians"}, # {"$cos": "$$fixed_lat_radians"}, # { # "$cos": { # "$subtract": [ # "$$lng_radians", # "$$fixed_lng_radians" # ] # } # } # ] # }, # { # "$multiply": [ # {"$sin": "$$lat_radians"}, # {"$sin": "$$fixed_lat_radians"} # ] # } # ] # } # } # ] # } # } # } # } # }, # # Step 8: Format the logo URL # { # "$addFields": { # "logo": { # "$concat": [ # "https://remax.azureedge.net/userimages/", # { "$toString": "$region_id" }, # "/", # "$logo" # ] # } # } # }, # # Step 9: Final projection to structure the document for `v1_beta_stage_offices` # { # "$project": { # "office_id": 1, # "name": 1, # "address": 1, # "postal_code": 1, # "website": 1, # "logo": 1, # "telephone": 1, # "lat": 1, # "lng": 1, # "distance": 1, # Include calculated distance # "creci": 1, # "complements_info": "$merged_complement", # "total_properties": 1, # # Map location fields from `city_info` # "city": "$city_info.CityName", # "state": "$city_info.ProvinceName", # "state_abbreviation": "$city_info.ProvinceAbbr", # "country": "$city_info.CountryName", # "country_abbreviation": "$city_info.CountryISOCode" # } # } # ] # } pipeline_office = { 'v1_beta_stage_offices': [ # Step 1: Specify the source collection explicitly { '$unionWith': { 'coll': "ilist_offices" } }, # Step 2: Project core fields from `ilist_offices` { "$project": { "office_id": "$ID", "name": "$OfficeName", "email": "$Email", "address": "$Address1", "complement": "$Address2", "postal_code": "$PostalCode", "website": "$URLToPrivatePage", "logo": "$ImageURL", "telephone": "$Phone", "zone_id": "$GeoData_LocalZoneID", "city_id": "$GeoData_CityID", "lat": { "$convert": { "input": "$GeoCoordinates_Latitude", "to": "double", "onError": None, "onNull": None } }, "lng": { "$convert": { "input": "$GeoCoordinates_Longitude", "to": "double", "onError": None, "onNull": None } }, "total_properties": "$total_properties", "creci": "$LicenseNumber", "region_id": "$RegionID", "complements_info": { "slogan": "$Slogan", "intro": "$Closer", "slug": { "$toLower": "$AlternateURL_URL" }, "description": "$OfficeDescription", "bullets": [ "$BulletPoint1", "$BulletPoint2", "$BulletPoint3" ], "alternate_url_redirect_url": "$AlternateURL_RedirectURL", "alternate_url_type": "$AlternateURL_UrlType", "ultima_sincronia": { "$toDate": "$ModifiedDate" } } } }, # Step 3: Lookup city details from `ilist_cities` { "$lookup": { "from": "ilist_cities", "localField": "city_id", "foreignField": "CityID", "as": "city_info" } }, # Step 4: Unwind the city information to flatten the array { "$unwind": { "path": "$city_info", "preserveNullAndEmptyArrays": True } }, # Step 3: Lookup city details from `ilist_cities` { "$lookup": { "from": "ilist_zones", "localField": "zone_id", "foreignField": "LocalZoneID", "as": "zone_info" } }, # Step 4: Unwind the city information to flatten the array { "$unwind": { "path": "$zone_info", "preserveNullAndEmptyArrays": True } }, # # Step 5: Lookup complements from `v1_mysql_offices_complements` # { # "$lookup": { # "from": "v1_mysql_offices_complements", # "localField": "office_id", # "foreignField": "referer_id", # "as": "mysql_complements" # } # }, # # Step 6: Merge base_complement with mysql_complements # { # "$addFields": { # "merged_complement": { # "$mergeObjects": [ # "$base_complement", # { # "$arrayElemAt": ["$mysql_complements", 0] # } # ] # } # } # }, # # Step 7: Add `distance` field using Haversine formula # { # "$addFields": { # "distance": { # "$let": { # "vars": { # "fixed_lat": -23.533773, # "fixed_lng": -46.625290, # "lat_radians": {"$multiply": [{"$divide": ["$lat", 180]}, 3.141592653589793]}, # "lng_radians": {"$multiply": [{"$divide": ["$lng", 180]}, 3.141592653589793]}, # "fixed_lat_radians": {"$multiply": [{"$divide": [-23.533773, 180]}, 3.141592653589793]}, # "fixed_lng_radians": {"$multiply": [{"$divide": [-46.625290, 180]}, 3.141592653589793]} # }, # "in": { # "$multiply": [ # 6371000, # Earth's radius in meters # { # "$acos": { # "$add": [ # { # "$multiply": [ # {"$cos": "$$lat_radians"}, # {"$cos": "$$fixed_lat_radians"}, # { # "$cos": { # "$subtract": [ # "$$lng_radians", # "$$fixed_lng_radians" # ] # } # } # ] # }, # { # "$multiply": [ # {"$sin": "$$lat_radians"}, # {"$sin": "$$fixed_lat_radians"} # ] # } # ] # } # } # ] # } # } # } # } # }, # Step 8: Format the logo URL { "$addFields": { "logo": { "$concat": [ "https://remax.azureedge.net/userimages/", { "$toString": "$region_id" }, "/", "$logo" ] }, "neighborhood": "$zone_info.LocalZoneName", "region_name": "$city_info.RegionName", "city": "$city_info.CityName", "state": "$city_info.ProvinceName", "state_abbreviation": "$city_info.ProvinceAbbr", "country": "$city_info.CountryName", "country_abbreviation": "$city_info.CountryISOCode", } }, # Step 9: Final projection to structure the document for `v1_beta_stage_offices` { "$project": { "office_id": 1, "name": 1, "email": 1, "address": 1, "complement": 1, "postal_code": 1, "website": 1, "logo": 1, "telephone": 1, "lat": 1, "lng": 1, #"distance": 1, # Include calculated distance "creci": 1, "complements_info": 1, "total_properties": 1, # Map location fields from `city_info` "neighborhood": 1, "region_name": 1, "city": 1, "state": 1, "state_abbreviation": 1, "country": 1, "country_abbreviation": 1, } } ] } # pipeline_agent = { # 'v1_beta_stage_agents': [ # # Pull data from ilist_associates # { # "$unionWith": { # "coll": "ilist_associates" # } # }, # # Project relevant fields # { # "$project": { # "agent_id": "$ID", # "office_id": "$OfficeID", # "name": "$AgentName", # "email": "$Email", # "creci": "$SalesLicenseNumber", # "telephone": "$Phone", # "profile_picture": "$ImageURL", # "total_properties": "$total_properties", # "website": "$URLToPrivatePage", # "region_id": "$RegionID", # "complements_info": [ # { # "slogan": "$Slogan", # "bullets": [ # "$BulletPoint1", # "$BulletPoint2", # "$BulletPoint3" # ], # "alternate_url": "$AlternateURL", # "alternate_url_redirect_url": "$AlternateURL_RedirectURL", # "alternate_url_type": "$AlternateURL_UrlType", # "description": "$Closer", # "ultima_sincronia": { "$toDate": "$ModifiedDate" } # } # ] # } # }, # # Add full URL to the profile picture # { # "$addFields": { # "profile_picture": { # "$concat": [ # "https://remax.azureedge.net/userimages/", # { "$toString": "$region_id" }, # "/", # "$profile_picture" # ] # } # } # }, # # Lookup `distance` from `v1_beta_stage_offices` # { # "$lookup": { # "from": "v1_offices", # "localField": "office_id", # "foreignField": "office_id", # "as": "office_details" # } # }, # # Unwind the office details array # { # "$unwind": { # "path": "$office_details", # "preserveNullAndEmptyArrays": True # Retain agents even if no matching office is found # } # }, # # Add the distance field from the office # { # "$addFields": { # "distance": "$office_details.distance" # } # }, # # Final projection of fields # { # "$project": { # "agent_id": 1, # "office_id": 1, # "name": 1, # "email": 1, # "creci": 1, # "website": 1, # "telephone": 1, # "profile_picture": 1, # "total_properties": 1, # "complements_info": 1, # "distance": 1 # Include the distance field # } # } # ] # } pipeline_languages = { 'v1_beta_stage_languages': [ { '$unionWith': { 'coll': 'ilist_lookups' } }, { '$match': { 'Name': 'Associate_LanguageSpoken' } }, { '$addFields': { 'language_id': { '$convert': { 'input': '$UID', 'to': 'int', 'onError': None } } } }, { '$match': { 'language_id': { '$ne': None } } }, { '$project': { '_id': 0, 'language_id': 1, 'language_name_en': '$Value', #'language_name_pt': '$Value' } } ] } pipeline_agent_languages = { "v1_beta_stage_agents_language_spoken": [ # Step 1: Start with ilist_associates collection { "$unionWith": { "coll": "ilist_associates" } }, # Step 2: Extract relevant fields { "$project": { "agent_id": "$ID", "languages_spoken_raw": "$LanguagesSpoken" } }, # Step 3: Split the LanguagesSpoken field into an array { "$addFields": { "language_ids": { "$cond": { "if": {"$ne": ["$languages_spoken_raw", None]}, "then": {"$split": ["$languages_spoken_raw", ", "]}, "else": [] } } } }, # Step 4: Convert language IDs to integers { "$addFields": { "language_ids": { "$map": { "input": "$language_ids", "as": "id", "in": { "$convert": {"input": "$$id", "to": "int", "onError": None, "onNull": None} } } } } }, # Step 5: Unwind the language IDs to match individually { "$unwind": { "path": "$language_ids", "preserveNullAndEmptyArrays": False } }, # Step 6: Lookup language details from v1_beta_stage_languages { "$lookup": { "from": "v1_beta_stage_languages", "localField": "language_ids", "foreignField": "language_id", "as": "language_details" } }, # Step 7: Unwind the language details { "$unwind": { "path": "$language_details", "preserveNullAndEmptyArrays": False } }, # Step 8: Group by agent_id and collect languages { "$group": { "_id": "$agent_id", "languages_spoken": { "$addToSet": "$language_details.language_name_en" } } }, # Step 9: Project the final structure { "$project": { "_id": 0, "agent_id": "$_id", "languages_spoken": 1 } } ] } pipeline_agent = { 'v1_beta_stage_agents': [ # Pull data from ilist_associates { "$unionWith": { "coll": "ilist_associates" } }, # Lookup `languages` from `v1_beta_stage_agents_language_spoken` { "$lookup": { "from": "v1_beta_stage_agents_language_spoken", "localField": "ID", "foreignField": "agent_id", "as": "languages_spoken" } }, # Unwind the languages_spoken array { "$unwind": { "path": "$languages_spoken", "preserveNullAndEmptyArrays": True # Retain agents even if no matching office is found } }, # Project relevant fields { "$project": { "agent_id": "$ID", "office_id": "$OfficeID", "name": "$AgentName", "email": "$Email", "creci": "$SalesLicenseNumber", "telephone": "$Phone", "profile_picture": "$ImageURL", "total_properties": "$total_properties", "website": "$URLToPrivatePage", "region_id": "$RegionID", "complements_info": [ { "slogan": "$Slogan", "slug": { "$toLower": "$AlternateURL_URL" }, "intro": "$Closer", "bullets": [ "$BulletPoint1", "$BulletPoint2", "$BulletPoint3" ], "languages": "$languages_spoken.languages_spoken", "alternate_url": "$AlternateURL", "alternate_url_redirect_url": "$AlternateURL_RedirectURL", "alternate_url_type": "$AlternateURL_UrlType", "description": "$Closer", "ultima_sincronia": { "$toDate": "$ModifiedDate" } } ] } }, # Lookup `distance` from `v1_beta_stage_offices` { "$lookup": { "from": "v1_beta_stage_offices", "localField": "office_id", "foreignField": "office_id", "as": "office_details" } }, # Unwind the office details array { "$unwind": { "path": "$office_details", "preserveNullAndEmptyArrays": True # Retain agents even if no matching office is found } }, # Add the distance field from the office { "$addFields": { "address": "$office_details.address", "street_number": "$office_details.street_number", "complement": "$office_details.complement", "postal_code": "$office_details.postal_code", "neighborhood": "$office_details.neighborhood", "city": "$office_details.city", "state": "$office_details.state", "state_abbreviation": "$office_details.state_abbreviation", "region_name": "$office_details.region_name", "country": "$office_details.country", "country_abbreviation": "$office_details.country_abbreviation", "lng": "$office_details.lng", "lat": "$office_details.lat", "profile_picture": { "$concat": [ "https://remax.azureedge.net/userimages/", { "$toString": "$region_id" }, "/", "$profile_picture" ] }, } }, # Final projection of fields { "$project": { "agent_id": 1, "office_id": 1, "name": 1, "email": 1, "telephone": 1, "creci": 1, "website": 1, "region_id": 1, "profile_picture": 1, "total_properties": 1, "address": 1, "street_number": 1, "complement": 1, "postal_code": 1, "neighborhood": 1, "city": 1, "state": 1, "state_abbreviation": 1, "region_name": 1, "country": 1, "country_abbreviation": 1, "lng": 1, "lat": 1, "complements_info": 1, #"distance": 1 # Include the distance field } } ] } # pipeline_agent = { # 'v1_beta_stage_agents': [ # # Pull data from ilist_associates # { # "$unionWith": { # "coll": "ilist_associates" # } # }, # # Project relevant fields # { # "$project": { # "agent_id": "$ID", # "office_id": "$OfficeID", # "name": "$AgentName", # "email": "$Email", # "creci": "$SalesLicenseNumber", # "telephone": "$Phone", # "profile_picture": "$ImageURL", # "total_properties": "$total_properties", # "website": "$URLToPrivatePage", # "region_id": "$RegionID", # "languages_spoken_ids": { # "$split": ["$LanguageSpoken", ", "] # Split IDs into an array # }, # "complements_info": [ # { # "slogan": "$Slogan", # "slug": { "$toLower": "$AlternateURL_URL" }, # "intro": "$Closer", # "bullets": [ # "$BulletPoint1", # "$BulletPoint2", # "$BulletPoint3" # ], # "alternate_url": "$AlternateURL", # "alternate_url_redirect_url": "$AlternateURL_RedirectURL", # "alternate_url_type": "$AlternateURL_UrlType", # "description": "$Closer", # "ultima_sincronia": { "$toDate": "$ModifiedDate" } # } # ] # } # }, # # Lookup `distance` from `v1_beta_stage_offices` # { # "$lookup": { # "from": "v1_beta_stage_offices", # "localField": "office_id", # "foreignField": "office_id", # "as": "office_details" # } # }, # # Unwind the office details array # { # "$unwind": { # "path": "$office_details", # "preserveNullAndEmptyArrays": True # } # }, # # Lookup languages from `v1_beta_stage_languages` # { # "$lookup": { # "from": "v1_beta_stage_languages", # "localField": "languages_spoken_ids", # "foreignField": "language_id", # "as": "languages_details" # } # }, # # Add processed fields, including concatenated language names # { # "$addFields": { # "languages_spoken": { # "$map": { # "input": "$languages_details", # "as": "language", # "in": "$$language.language_name_en" # } # }, # "address": "$office_details.address", # "street_number": "$office_details.street_number", # "complement": "$office_details.complement", # "postal_code": "$office_details.postal_code", # "neighborhood": "$office_details.neighborhood", # "city": "$office_details.city", # "state": "$office_details.state", # "state_abbreviation": "$office_details.state_abbreviation", # "region_name": "$office_details.region_name", # "country": "$office_details.country", # "country_abbreviation": "$office_details.country_abbreviation", # "lng": "$office_details.lng", # "lat": "$office_details.lat", # "profile_picture": { # "$concat": [ # "https://remax.azureedge.net/userimages/", # { "$toString": "$region_id" }, # "/", # "$profile_picture" # ] # } # } # }, # # Final projection of fields # { # "$project": { # "agent_id": 1, # "office_id": 1, # "name": 1, # "email": 1, # "telephone": 1, # "creci": 1, # "website": 1, # "region_id": 1, # "profile_picture": 1, # "total_properties": 1, # "address": 1, # "street_number": 1, # "complement": 1, # "postal_code": 1, # "neighborhood": 1, # "city": 1, # "state": 1, # "state_abbreviation": 1, # "region_name": 1, # "country": 1, # "country_abbreviation": 1, # "lng": 1, # "lat": 1, # "languages_spoken": 1, # Add translated languages # "complements_info": 1 # } # } # ] # } pipeline_neighborhoods = { 'v1_beta_stage_neighborhoods': [ # Step 1: Specify the source collection { '$unionWith': { 'coll': 'ilist_zones' } }, # Step 2: Add necessary fields and rename { '$addFields': { 'neighborhood_id': '$LocalZoneID', 'neighborhood_name': '$LocalZoneName', 'city_id': '$CityID', 'city_name': '$CityName', 'state_id': '$ProvinceID', 'country_id': '$CountryID', 'country_abbreviation': '$CountryISOCode', 'country_name': '$CountryName', 'state_abbreviation': '$ProvinceAbbr', 'state_name': '$ProvinceName' # 'bairro_id': '$LocalZoneID', # 'bairro_nome': '$LocalZoneName', # 'cidade_id': '$CityID', # 'cidade_nome': '$CityName', # 'estado_id': '$ProvinceID', # 'pais_id': '$CountryID', # 'pais_abr': '$CountryISOCode', # 'pais_nome': '$CountryName', # 'uf': '$ProvinceAbbr', # 'estado_nome': '$ProvinceName' } }, # Step 3: Project the final fields { '$project': { '_id': 0, # Exclude the original MongoDB `_id` field 'neighborhood_id': 1, 'neighborhood_name': 1, 'city_id': 1, 'city_name': 1, 'state_id': 1, 'country_id': 1, 'country_abbreviation': 1, 'country_name': 1, 'state_abbreviation': 1, 'state_name': 1, # 'bairro_id': 1, # 'bairro_nome': 1, # 'cidade_id': 1, # 'cidade_nome': 1, # 'estado_id': 1, # 'pais_id': 1, # 'pais_abr': 1, # 'pais_nome': 1, # 'uf': 1, # 'estado_nome': 1 } } ] } # pipeline_cidades = { # 'v1_beta_stage_cidades': [ # # Step 1: Specify the source collection # { # '$unionWith': { # 'coll': 'ilist_cities' # } # }, # # Step 2: Add necessary fields and rename # { # '$addFields': { # 'city_id': '$CityID', # 'city_name': '$CityName', # 'state_id': '$ProvinceID', # 'country_id': '$CountryID', # 'country_abbreviation': '$CountryISOCode', # 'country_name': '$CountryName', # 'state_abbreviation': '$ProvinceAbbr', # 'state_name': '$ProvinceName', # # 'cidade_id': '$CityID', # # 'cidade_nome': '$CityName', # # 'estado_id': '$ProvinceID', # # 'pais_id': '$CountryID', # # 'pais_abr': '$CountryISOCode', # # 'pais_nome': '$CountryName', # # 'uf': '$ProvinceAbbr', # # 'estado_nome': '$ProvinceName' # } # }, # # Step 3: Project the final fields # { # '$project': { # '_id': 0, # Exclude the original MongoDB `_id` field # 'city_id': 1, # 'city_name': 1, # 'state_id': 1, # 'country_id': 1, # 'country_abbreviation': 1, # 'country_name': 1, # 'state_abbreviation': 1, # 'state_name': 1, # # 'cidade_id': 1, # # 'cidade_nome': 1, # # 'estado_id': 1, # # 'pais_id': 1, # # 'pais_abr': 1, # # 'pais_nome': 1, # # 'uf': 1, # # 'estado_nome': 1 # } # } # ] # } # pipeline_estados = { # 'v1_beta_stage_estados': [ # # Step 1: Specify the source collection # { # '$unionWith': { # 'coll': 'ilist_provinces' # } # }, # # Step 2: Add necessary fields and rename # { # '$addFields': { # 'state_id': '$ProvinceID', # 'country_id': '$CountryID', # 'country_abbreviation': '$CountryISOCode', # 'country_name': '$CountryName', # 'state_abbreviation': '$ProvinceAbbr', # 'state_name': '$ProvinceName', # # 'estado_id': '$ProvinceID', # # 'pais_id': '$CountryID', # # 'pais_abr': '$CountryISOCode', # # 'pais_nome': '$CountryName', # # 'uf': '$ProvinceAbbr', # # 'estado_nome': '$ProvinceName' # } # }, # # Step 3: Project the final fields # { # '$project': { # '_id': 0, # Exclude the original MongoDB `_id` field # 'state_id': 1, # 'country_id': 1, # 'country_abbreviation': 1, # 'country_name': 1, # 'state_abbreviation': 1, # 'state_name': 1, # # 'estado_id': 1, # # 'pais_id': 1, # # 'pais_abr': 1, # # 'pais_nome': 1, # # 'uf': 1, # # 'estado_nome': 1 # } # } # ] # } pipeline_cities = { 'v1_beta_stage_cities': [ # Step 1: Specify the source collection { '$unionWith': { 'coll': 'ilist_cities' } }, # Step 2: Add necessary fields and rename { '$addFields': { 'city_id': '$CityID', 'city_name': '$CityName', 'state_id': '$ProvinceID', 'country_id': '$CountryID', 'country_abbreviation': '$CountryISOCode', 'country_name': '$CountryName', 'state_abbreviation': '$ProvinceAbbr', 'state_name': '$ProvinceName' } }, # Step 3: Lookup neighborhoods corresponding to the city { '$lookup': { 'from': 'ilist_zones', # Neighborhoods collection 'localField': 'city_id', # City ID in cities 'foreignField': 'CityID', # City ID in neighborhoods 'as': 'neighborhoods' # Output field for matched neighborhoods } }, # Step 4: Transform neighborhoods array into required format { '$addFields': { 'neighborhoods': { '$map': { 'input': '$neighborhoods', 'as': 'neighborhood', 'in': { 'neighborhood_id': '$$neighborhood.LocalZoneID', 'neighborhood_name': '$$neighborhood.LocalZoneName' } } } } }, # Step 5: Project the final fields { '$project': { '_id': 0, # Exclude the original MongoDB `_id` field 'city_id': 1, 'city_name': 1, 'state_id': 1, 'country_id': 1, 'country_abbreviation': 1, 'country_name': 1, 'state_abbreviation': 1, 'state_name': 1, 'neighborhoods': 1 } } ] } pipeline_states = { 'v1_beta_stage_states': [ # Step 1: Specify the source collection { '$unionWith': { 'coll': 'ilist_provinces' } }, # Step 2: Add necessary fields and rename { '$addFields': { 'state_id': '$ProvinceID', 'state_name': '$ProvinceName', 'state_abbreviation': '$ProvinceAbbr', 'country_id': '$CountryID', 'country_name': '$CountryName', 'country_abbreviation': '$CountryISOCode' } }, # Step 3: Lookup cities corresponding to the state { '$lookup': { 'from': 'ilist_cities', # Cities collection 'localField': 'state_id', # State ID in states 'foreignField': 'ProvinceID', # State ID in cities 'as': 'cities' # Output field for matched cities } }, # Step 4: Transform cities array into required format { '$addFields': { 'cities': { '$map': { 'input': '$cities', 'as': 'city', 'in': { 'city_id': '$$city.CityID', 'city_name': '$$city.CityName' } } } } }, # Step 5: Project the final fields { '$project': { '_id': 0, # Exclude the original MongoDB `_id` field 'state_id': 1, 'state_name': 1, 'state_abbreviation': 1, 'country_id': 1, 'country_name': 1, 'country_abbreviation': 1, 'cities': 1 } } ] } pipeline_estados_detalhes = { 'v1_beta_stage_estados_detalhes': [ # Step 1: Specify the source collection { '$unionWith': { 'coll': 'panel_estadodetalhe' } }, # Step 2: Add necessary fields and rename { '$addFields': { 'state_id': { '$convert': { 'input': { '$arrayElemAt': ['$estado_id', 0] }, 'to': 'int', 'onError': None } }, 'image': '$imagem', 'text': '$texto', } }, # Step 3: Project the final fields { '$project': { '_id': 0, # Exclude the original MongoDB `_id` field 'state_id': 1, 'image': 1, 'text': 1, } } ] } pipeline_cidades_detalhes = { 'v1_beta_stage_cidades_detalhes': [ # Step 1: Specify the source collection { '$unionWith': { 'coll': 'panel_cidadedetalhe' } }, # Step 2: Add necessary fields and rename { '$addFields': { 'city_id': { '$convert': { 'input': { '$arrayElemAt': ['$cidade_id', 0] }, 'to': 'int', 'onError': None } }, 'image': '$imagem', 'text': '$texto', } }, # Step 3: Project the final fields { '$project': { '_id': 0, # Exclude the original MongoDB `_id` field 'city_id': 1, 'image': 1, 'text': 1, } } ] } pipeline_bairros_detalhes = { 'v1_beta_stage_bairros_detalhes': [ # Step 1: Specify the source collection { '$unionWith': { 'coll': 'panel_bairrodetalhe' } }, # Step 2: Add necessary fields and rename { '$addFields': { 'neighborhood_id': { '$convert': { 'input': { '$arrayElemAt': ['$bairro_id', 0] }, 'to': 'int', 'onError': None } }, 'image': '$imagem', 'text': '$texto', } }, # Step 3: Project the final fields { '$project': { '_id': 0, # Exclude the original MongoDB `_id` field 'neighborhood_id': 1, 'image': 1, 'text': 1, } } ] } # Build lookup dictionary from types_translation type_translation_map = {item["type_id"]: item["type_name_pt"] for item in types_translation} pipeline_types = { 'v1_beta_stage_types': [ { '$unionWith': { 'coll': "ilist_lookups" } }, { "$match": { "Name": {"$in": ["Property_ResidentialType", "Property_CommercialType"]} } }, { "$addFields": { "type_id": { "$convert": { "input": "$UID", "to": "int", "onError": None } } } }, { "$match": { "type_id": {"$ne": None} } }, { "$addFields": { "type_name_pt": { "$switch": { "branches": [ {"case": {"$eq": ["$type_id", type_id]}, "then": type_name_pt} for type_id, type_name_pt in type_translation_map.items() ], "default": "$type_name_en" } } } }, { "$project": { "type_id": 1, "type_name_en": "$Value", "type_name_pt": 1 } } ] } # pipeline_types = { # 'v1_beta_stage_types': [ # { # '$unionWith': { # 'coll': "ilist_lookups" # } # }, # { # "$match": { # #Property_ResidentialType, 'Property_CommercialType # "Name": {"$in": ["Property_ResidentialType", "Property_CommercialType"]} # } # }, # { # "$addFields": { # "type_id": { # "$convert": { # "input": "$UID", # "to": "int", # "onError": None # } # } # } # }, # { # "$match": { # "type_id": {"$ne": None} # } # }, # { # "$project": { # "type_id": 1, # "type_name_en": "$Value", # "type_name_pt": "$Value" # } # } # ] # } pipeline_transaction_type = { 'v1_beta_stage_transaction_type': [ { '$unionWith': { 'coll': "ilist_lookups" } }, { "$match": { "Name": "Property_TransactionType" } }, { "$addFields": { "transaction_id": { "$convert": { "input": "$UID", "to": "int", "onError": None } } } }, { "$match": { "transaction_id": {"$ne": None} } }, { "$project": { "transaction_id": 1, "transaction_name_en": "$Value", "transaction_name_pt": "$Value" } } ] } pipeline_payment_period = { 'v1_beta_stage_payment_period': [ { '$unionWith': { 'coll': "ilist_lookups" } }, { "$match": { "Name": "Property_PaymentPeriod" } }, { "$addFields": { "payment_id": { "$convert": { "input": "$UID", "to": "int", "onError": None } } } }, { "$match": { "payment_id": {"$ne": None} } }, { "$project": { "payment_id": 1, "payment_name_en": "$Value", "payment_name_pt": "$Value" } } ] } # Build a lookup dictionary for translations listing_status_translation_map = { listing["listing_id"]: listing["listing_name_pt"] for listing in listings_status_translation } pipeline_listing_status = { 'v1_beta_stage_listing_status': [ # Step 1: Specify the source collection with `$unionWith` { '$unionWith': { 'coll': "ilist_lookups" } }, # Step 2: Match documents where Name is 'Property_ListingStatus' { "$match": { "Name": "Property_ListingStatus" } }, # Step 3: Convert UID to integer with fallback { "$addFields": { "listing_id": { "$convert": { "input": "$UID", "to": "int", "onError": None # Use None for invalid conversions } } } }, # Step 4: Filter out entries where `listing_id` is null { "$match": { "listing_id": {"$ne": None} } }, # Step 5: Add translations using a mapping stage { "$addFields": { "listing_name_pt": { "$switch": { "branches": [ {"case": {"$eq": ["$listing_id", listing_id]}, "then": listing_name_pt} for listing_id, listing_name_pt in listing_status_translation_map.items() ], "default": "$listing_name_en" # Fallback to English name if translation missing } } } }, # Step 6: Project the required fields { "$project": { "listing_id": 1, "listing_name_en": "$Value", "listing_name_pt": 1 } } ] } # pipeline_listing_status = { # 'v1_beta_stage_listing_status': [ # { # '$unionWith': { # 'coll': "ilist_lookups" # } # }, # { # "$match": { # "Name": "Property_ListingStatus" # } # }, # { # "$addFields": { # "listing_id": { # "$convert": { # "input": "$UID", # "to": "int", # "onError": None # } # } # } # }, # { # "$match": { # "listing_id": {"$ne": None} # } # }, # { # "$project": { # "listing_id": 1, # "listing_name_en": "$Value", # "listing_name_pt": "$Value" # } # } # ] # } # Build lookup dictionary from markets_status_translation market_translation_map = {item["market_id"]: item["market_name_pt"] for item in markets_status_translation} pipeline_market_status = { 'v1_beta_stage_market_status': [ { '$unionWith': { 'coll': "ilist_lookups" } }, { "$match": { "Name": "Property_MarketStatus" } }, { "$addFields": { "market_id": { "$convert": { "input": "$UID", "to": "int", "onError": None } } } }, { "$match": { "market_id": {"$ne": None} } }, { "$addFields": { "market_name_pt": { "$switch": { "branches": [ {"case": {"$eq": ["$market_id", market_id]}, "then": market_name_pt} for market_id, market_name_pt in market_translation_map.items() ], "default": "$market_name_en" } } } }, { "$project": { "market_id": 1, "market_name_en": "$Value", "market_name_pt": 1 } } ] } # pipeline_market_status = { # 'v1_beta_stage_market_status': [ # { # '$unionWith': { # 'coll': "ilist_lookups" # } # }, # { # "$match": { # "Name": "Property_MarketStatus" # } # }, # { # "$addFields": { # "market_id": { # "$convert": { # "input": "$UID", # "to": "int", # "onError": None # } # } # } # }, # { # "$match": { # "market_id": {"$ne": None} # } # }, # { # "$project": { # "market_id": 1, # "market_name_en": "$Value", # "market_name_pt": "$Value" # } # } # ] # } # Build lookup dictionary from floors_level_translation floor_translation_map = {item["floor_id"]: item["floor_name_pt"] for item in floors_level_translation} pipeline_floor_level = { 'v1_beta_stage_floor_level': [ { '$unionWith': { 'coll': "ilist_lookups" } }, { "$match": { "Name": "Property_FloorLevel" } }, { "$addFields": { "floor_id": { "$convert": { "input": "$UID", "to": "int", "onError": None } } } }, { "$match": { "floor_id": {"$ne": None} } }, { "$addFields": { "floor_name_pt": { "$switch": { "branches": [ {"case": {"$eq": ["$floor_id", floor_id]}, "then": floor_name_pt} for floor_id, floor_name_pt in floor_translation_map.items() ], "default": "$floor_name_en" } } } }, { "$project": { "floor_id": 1, "floor_name_en": "$Value", "floor_name_pt": 1 } } ] } # pipeline_floor_level = { # 'v1_beta_stage_floor_level': [ # { # '$unionWith': { # 'coll': "ilist_lookups" # } # }, # { # "$match": { # "Name": "Property_FloorLevel" # } # }, # { # "$addFields": { # "floor_id": { # "$convert": { # "input": "$UID", # "to": "int", # "onError": None # } # } # } # }, # { # "$match": { # "floor_id": {"$ne": None} # } # }, # { # "$project": { # "floor_id": 1, # "floor_name_en": "$Value", # "floor_name_pt": "$Value" # } # } # ] # } pipeline_property_description_types = { 'v1_beta_stage_property_description_types': [ { '$unionWith': { 'coll': "ilist_lookups" } }, { "$match": { "Name": "Property_DescriptionType" } }, { "$addFields": { "property_description_type_id": { "$convert": { "input": "$UID", "to": "int", "onError": None } } } }, { "$match": { "property_description_type_id": {"$ne": None} } }, { "$project": { "property_description_type_id": 1, "property_description_type": "$Value", } } ] } # pipeline_properties_description = { # 'v1_beta_stage_properties_description': [ # # Step 1: Load data from `ilist_property_descriptions` # { # '$unionWith': { # 'coll': "ilist_property_descriptions" # } # }, # # Step 2: Match documents with non-null and non-empty DescriptionText # { # "$match": { # "DescriptionText": {"$exists": True, "$ne": ""} # } # }, # # Step 3: Project core fields and transform property_id # { # "$project": { # "property_id": { # "$convert": { # "input": { "$replaceAll": { "input": "$PropertyID", "find": "-", "replacement": "" } }, # "to": "long", # "onError": None # } # }, # "description": "$DescriptionText", # "type": { # "$cond": { # "if": { "$eq": ["$DescriptionType", 1113] }, # "then": "title", # "else": "description" # } # } # } # }, # # Step 4: Final projection # { # "$project": { # "_id": 0, # "property_id": 1, # "description": 1, # "type": 1 # } # } # ] # } pipeline_properties_description = { 'v1_beta_stage_properties_description': [ # Step 1: Load data from `ilist_property_descriptions` { '$unionWith': { 'coll': "ilist_property_descriptions" } }, # Step 2: Match documents with non-null and non-empty DescriptionText { "$match": { "DescriptionText": {"$exists": True, "$ne": ""} } }, # Step 3: Project core fields and transform property_id { "$project": { "property_id": { "$convert": { "input": { "$replaceAll": { "input": "$PropertyID", "find": "-", "replacement": "" } }, "to": "long", "onError": None } }, "description": "$DescriptionText", "property_description_type_id": "$DescriptionType" } }, # Step 4: Lookup description type details from `v1_beta_stage_property_description_types` { "$lookup": { "from": "v1_beta_stage_property_description_types", "localField": "property_description_type_id", "foreignField": "property_description_type_id", "as": "description_type_details" } }, # Step 5: Unwind the description type details { "$unwind": { "path": "$description_type_details", "preserveNullAndEmptyArrays": True } }, # Step 6: Project the final structure { "$project": { "_id": 0, "property_description_id": 1, "property_description_type": "$description_type_details.property_description_type", "property_id": 1, "description": 1, } } ] } # Build a lookup dictionary for translations feature_translation_map = { feature["feature_id"]: feature["feature_name_pt"] for feature in features_translation } pipeline_features = { 'v1_beta_stage_features': [ # Step 1: Specify the source collection with `$unionWith` { '$unionWith': { 'coll': "ilist_lookups" } }, # Step 2: Match documents where Name is 'Property_Feature' { "$match": { "Name": "Property_Feature", "Value": {"$regex": "Features", "$options": "i"} } }, # Step 3: Convert UID to integer with fallback { "$addFields": { "feature_id": { "$convert": { "input": "$UID", "to": "int", "onError": None # Use None for invalid conversions } }, "feature_name": { "$trim": { "input": { "$arrayElemAt": [ {"$split": ["$Value", ". "]}, 1 ] } } } } }, # Step 4: Filter out entries where `feature_id` is null { "$match": { "feature_id": {"$ne": None} } }, # Step 5: Add translations using a mapping stage { "$addFields": { "feature_name_pt": { "$switch": { "branches": [ {"case": {"$eq": ["$feature_id", feature_id]}, "then": feature_name_pt} for feature_id, feature_name_pt in feature_translation_map.items() ], "default": "$feature_name" # Fallback to English name if translation missing } } } }, # Step 6: Project the required fields { "$project": { "feature_id": 1, "feature_name_en": "$feature_name", "feature_name_pt": 1 } } ] } # pipeline_features = { # 'v1_beta_stage_features': [ # # Step 1: Specify the source collection with `$unionWith` # { # '$unionWith': { # 'coll': "ilist_lookups" # } # }, # # Step 2: Match documents where Name is 'Property_Feature' # { # "$match": { # "Name": "Property_Feature", # "Value": {"$regex": "Features", "$options": "i"} # } # }, # # Step 3: Convert UID to integer with fallback # { # "$addFields": { # "feature_id": { # "$convert": { # "input": "$UID", # "to": "int", # "onError": None # Use None for invalid conversions # } # }, # "feature_name": { # "$arrayElemAt": [ # { "$split": ["$Value", ". "] }, # 1 # ] # } # } # }, # # Step 4: Filter out entries where `feature_id` is null # { # "$match": { # "feature_id": { "$ne": None } # } # }, # # Step 5: Project the required fields # { # "$project": { # "feature_id": 1, # "feature_name_en": "$feature_name", # "feature_name_pt": "$feature_name" # } # } # ], # } pipeline_properties_features = { 'v1_beta_stage_properties_features': [ # Step 1: Specify the source collection explicitly { "$unionWith": { "coll": "ilist_properties" } }, # Step 2: Match documents with non-empty Features field { "$match": { "Features": {"$exists": True, "$ne": ""} } }, # Step 3: Split the Features field into an array { "$addFields": { "features_array": {"$split": ["$Features", ", "]} } }, # Step 4: Unwind the array to work with individual features { "$unwind": "$features_array" }, # Step 5: Convert feature_id for joining with `v1_beta_stage_features` { "$addFields": { "feature_id": {"$toInt": "$features_array"} } }, # Step 6: Join with `v1_beta_stage_features` to get feature details { "$lookup": { "from": "v1_beta_stage_features", "localField": "feature_id", "foreignField": "feature_id", "as": "feature_details" } }, # Step 7: Unwind the joined data to flatten `feature_details` { "$unwind": { "path": "$feature_details", "preserveNullAndEmptyArrays": True # Ensure missing matches are not dropped } }, # Step 8: Project the required fields for final output { "$project": { "_id": 0, # Exclude MongoDB default `_id` field "property_id": { "$convert": { "input": {"$replaceAll": {"input": "$ID", "find": "-", "replacement": ""}}, "to": "long", "onError": None } }, "feature_id": "$feature_details.feature_id", "feature_name_en": "$feature_details.feature_name_en", "feature_name_pt": "$feature_details.feature_name_pt" } }, # Step 9: Group by `property_id` and `feature_id` to ensure unique entries { "$group": { "_id": {"property_id": "$property_id", "feature_id": "$feature_id"}, "feature_name_en": {"$first": "$feature_name_en"}, "feature_name_pt": {"$first": "$feature_name_pt"} } }, # Step 10: Reshape fields for final output { "$project": { "_id": 0, "property_id": "$_id.property_id", "feature_id": "$_id.feature_id", "feature_name_en": "$feature_name_en", "feature_name_pt": "$feature_name_pt" } } ] } # pipeline_types = { # 'v1_beta_stage_types': [ # # Specify the source collection # { # '$unionWith': { # 'coll': "ilist_properties" # } # }, # # Step 1: Match non-null PropertyType # { # "$match": { "PropertyType": { "$ne": None } } # }, # # Step 2: Group by unique PropertyType values # { # "$group": { # "_id": "$PropertyType" # } # }, # # Step 3: Add fields for type_id # { # "$addFields": { "type_id": { "$toString": "$_id" } } # }, # # Step 4: Lookup type details from `ilist_lookups` # { # "$lookup": { # "from": "ilist_lookups", # "localField": "type_id", # "foreignField": "UID", # "as": "type_details" # } # }, # # Step 5: Unwind `type_details` to flatten the array # { # "$unwind": "$type_details" # }, # # Step 6: Project final fields for `v1_beta_stage_types` # { # "$project": { # "_id": 0, # "type_id": "$type_id", # "type_name_en": "$type_details.Value", # "type_name_pt": "$type_details.Value" # } # } # ], # } pipeline_usages = { 'v1_beta_stage_usages': [ # Specify the source collection { '$unionWith': { 'coll': "ilist_properties" } }, # Step 1: Match documents with a non-null CommercialResidential field { "$match": { "CommercialResidential": { "$ne": None } } }, # Step 2: Group by CommercialResidential to get unique values { "$group": { "_id": "$CommercialResidential" } }, # Step 3: Add fields for usage_id and usage names based on the CommercialResidential value { "$addFields": { "usage_id": { "$toString": "$_id" }, "usage_name_en": { "$cond": { "if": { "$eq": ["$_id", 1] }, "then": "Commercial", "else": "Residential" } }, "usage_name_pt": { "$cond": { "if": { "$eq": ["$_id", 1] }, "then": "Comercial", "else": "Residencial" } } } }, # Step 4: Project the final structure for `v1_beta_stage_usages` { "$project": { "_id": 0, "usage_id": "$usage_id", "usage_name_en": "$usage_name_en", "usage_name_pt": "$usage_name_pt" } } ], } pipeline_location = { 'v1_beta_stage_properties_location': [ # Specify the source collection { '$unionWith': { 'coll': "ilist_properties" } }, # Step 1: Add necessary fields and trim text fields { "$addFields": { "property_id": { "$convert": { "input": { "$replaceAll": { "input": "$ID", "find": "-", "replacement": "" } }, "to": "long", "onError": None } }, "location_id": "$GeoData_LocalZoneID", "display_address": "$ShowAddressOnWeb", "address": { "$trim": { "input": "$StreetName" } }, "street_number": { "$trim": { "input": "$StreetNumber" } }, "postal_code": { "$trim": { "input": "$PostalCode" } }, "lat": { "$convert": { "input": "$GeoCoordinates_Latitude", "to": "double", "onError": None } }, "lng": { "$convert": { "input": "$GeoCoordinates_Longitude", "to": "double", "onError": None } } } }, # Step 2: Perform a lookup with `ilist_zones` { "$lookup": { "from": "ilist_cities", "localField": "GeoData_CityID", "foreignField": "CityID", "as": "city_data" } }, # Step 3: Unwind `zone_details` array { "$unwind": { "path": "$city_data", "preserveNullAndEmptyArrays": True } }, # Step 2: Perform a lookup with `ilist_zones` { "$lookup": { "from": "ilist_zones", "localField": "GeoData_LocalZoneID", "foreignField": "LocalZoneID", "as": "zone_details" } }, # Step 3: Unwind `zone_details` array { "$unwind": { "path": "$zone_details", "preserveNullAndEmptyArrays": True } }, { "$addFields": { "city_id": "$city_data.CityID", "state_id": "$city_data.ProvinceID", "neighborhood_id": "$zone_details.LocalZoneID" } }, # Step 2: Perform a lookup with `ilist_zones` { "$lookup": { "from": "v1_beta_stage_estados_detalhes", "localField": "state_id", "foreignField": "state_id", "as": "state_details" } }, # Step 3: Unwind `zone_details` array { "$unwind": { "path": "$state_details", "preserveNullAndEmptyArrays": True } }, # Step 2: Perform a lookup with `ilist_zones` { "$lookup": { "from": "v1_beta_stage_cidades_detalhes", "localField": "city_id", "foreignField": "city_id", "as": "city_details" } }, # Step 3: Unwind `zone_details` array { "$unwind": { "path": "$city_details", "preserveNullAndEmptyArrays": True } }, # Step 2: Perform a lookup with `ilist_zones` { "$lookup": { "from": "v1_beta_stage_bairros_detalhes", "localField": "neighborhood_id", "foreignField": "neighborhood_id", "as": "neighborhood_details" } }, # Step 3: Unwind `zone_details` array { "$unwind": { "path": "$neighborhood_details", "preserveNullAndEmptyArrays": True } }, # Step 4: Trim additional text fields from `zone_details` { "$addFields": { "country": { "$trim": { "input": "$city_data.CountryName" } }, "country_abbreviation": { "$trim": { "input": "$city_data.CountryISOCode" } }, "state": { "$trim": { "input": "$city_data.ProvinceName" } }, "state_abbreviation": { "$trim": { "input": "$city_data.ProvinceAbbr" } }, "city": { "$trim": { "input": "$city_data.CityName" } }, "region": { "$trim": { "input": "$city_data.RegionName" } }, "zone": { "$trim": { "input": "$GeoData_District" } }, "neighborhood": { "$trim": { "input": "$zone_details.LocalZoneName" } }, "complement": { "$trim": { "input": "$ApartmentNumber" } }, "state_details": { "$cond": { "if": { "$or": [{ "$ne": ["$state_details.text", None] }, { "$ne": ["$state_details.image", None] }] }, "then": [{ "text": "$state_details.text", "image": "$state_details.image" }], "else": None } }, "city_details": { "$cond": { "if": { "$or": [{ "$ne": ["$city_details.text", None] }, { "$ne": ["$city_details.image", None] }] }, "then": [{ "text": "$city_details.text", "image": "$city_details.image" }], "else": None } }, "neighborhood_details": { "$cond": { "if": { "$or": [{ "$ne": ["$neighborhood_details.text", None] }, { "$ne": ["$neighborhood_details.image", None] }] }, "then": [{ "text": "$neighborhood_details.text", "image": "$neighborhood_details.image" }], "else": None } } } }, # # Step 5: Add `distance` field using Haversine formula # { # "$addFields": { # "distance": { # "$let": { # "vars": { # "fixed_lat": fixed_point["lat"], # "fixed_lng": fixed_point["lng"], # "lat_radians": {"$multiply": [{"$divide": ["$lat", 180]}, 3.141592653589793]}, # "lng_radians": {"$multiply": [{"$divide": ["$lng", 180]}, 3.141592653589793]}, # "fixed_lat_radians": {"$multiply": [{"$divide": [fixed_point["lat"], 180]}, 3.141592653589793]}, # "fixed_lng_radians": {"$multiply": [{"$divide": [fixed_point["lng"], 180]}, 3.141592653589793]} # }, # "in": { # "$multiply": [ # 6371000, # Earth's radius in meters # { # "$acos": { # "$add": [ # { # "$multiply": [ # {"$cos": "$$lat_radians"}, # {"$cos": "$$fixed_lat_radians"}, # { # "$cos": { # "$subtract": [ # "$$lng_radians", # "$$fixed_lng_radians" # ] # } # } # ] # }, # { # "$multiply": [ # {"$sin": "$$lat_radians"}, # {"$sin": "$$fixed_lat_radians"} # ] # } # ] # } # } # ] # } # } # } # } # }, # Step 6: Project the final fields, including distance { "$project": { "location_id": 1, "property_id": 1, "display_address": 1, "country": 1, "country_abbreviation": 1, "state": 1, "state_abbreviation": 1, "city": 1, "region": 1, "zone": 1, "neighborhood": 1, "complement": 1, "address": 1, "street_number": 1, "postal_code": 1, "lat": 1, "lng": 1, "state_details": { "$cond": { "if": { "$and": [ { "$ne": ["$state_details", None] }, { "$ne": [{ "$arrayElemAt": ["$state_details", 0] }, {}] } ] }, "then": "$state_details", "else": None } }, "city_details": { "$cond": { "if": { "$and": [ { "$ne": ["$city_details", None] }, { "$ne": [{ "$arrayElemAt": ["$city_details", 0] }, {}] } ] }, "then": "$city_details", "else": None } }, "neighborhood_details": { "$cond": { "if": { "$and": [ { "$ne": ["$neighborhood_details", None] }, { "$ne": [{ "$arrayElemAt": ["$neighborhood_details", 0] }, {}] } ] }, "then": "$neighborhood_details", "else": None } } #"distance": 1 # Include calculated distance } } ] } # pipeline_location = { # 'v1_beta_stage_properties_location': [ # # Specify the source collection # { # '$unionWith': { # 'coll': "ilist_properties" # } # }, # # Step 1: Add necessary fields # { # "$addFields": { # "property_id": { # "$convert": { # "input": { "$replaceAll": { "input": "$ID", "find": "-", "replacement": "" } }, # "to": "long", # "onError": None # } # }, # "location_id": "$GeoData_LocalZoneID", # "display_address": "$ShowAddressOnWeb", # "address": "$StreetName", # "street_number": "$StreetNumber", # "postal_code": "$PostalCode", # "lat": "$GeoCoordinates_Latitude", # "lng": "$GeoCoordinates_Longitude" # } # }, # # Step 2: Perform a lookup with `ilist_cities` # { # "$lookup": { # "from": "ilist_zones", # "localField": "GeoData_LocalZoneID", # "foreignField": "LocalZoneID", # "as": "zone_details" # } # }, # # Step 3: Unwind `zone_details` array # { # "$unwind": { # "path": "$zone_details", # "preserveNullAndEmptyArrays": True # } # }, # # Step 4: Project the final fields # { # "$project": { # "location_id": 1, # "property_id": 1, # "display_address": 1, # "country": "$zone_details.CountryName", # "country_abbreviation": "$zone_details.CountryISOCode", # "state": "$zone_details.ProvinceName", # "state_abbreviation": "$zone_details.ProvinceAbbr", # "city": "$zone_details.CityName", # "zone": "$zone_details.LocalZoneName", # "neighborhood": "$GeoData_District", # "complement": "$ApartmentNumber", # "address": 1, # "street_number": 1, # "postal_code": 1, # "lat": 1, # "lng": 1 # } # }, # ], # } pipeline_properties_media = { 'v1_beta_stage_properties_media': [ # Specify the source collection { '$unionWith': { 'coll': "ilist_property_images" } }, # Lookup properties { "$lookup": { "from": "ilist_properties", "localField": "PropertyID", "foreignField": "ID", "as": "property_details" } }, { "$unwind": { "path": "$property_details", "preserveNullAndEmptyArrays": True } }, # Step 1: Add necessary fields for transformation, removing any "-" characters from PropertyID { "$addFields": { "region_id": "$property_details.RegionID", } }, # Step 1: Add necessary fields for transformation, removing any "-" characters from PropertyID { "$addFields": { "property_media_id": "$ID", "type": "image", "order": "$SequenceNumber", "main": { "$cond": { "if": { "$eq": ["$IsDefault", True] }, "then": 1, "else": 0 } }, "file_name": "$ImageURL", "property_id": { "$cond": { "if": { "$regexMatch": { "input": "$PropertyID", "regex": r"^[0-9\-]+$" } }, "then": { "$toLong": { "$replaceAll": { "input": "$PropertyID", "find": "-", "replacement": "" } } }, "else": None } }, "url": { "$cond": { "if": { "$and": [ { "$ne": ["$property_details.RegionID", None] }, { "$ne": ["$ImageURL", None] } ] }, "then": { "$concat": [ "https://remax.azureedge.net/userimages/", { "$toString": "$property_details.RegionID" }, "/", "$ImageURL" ] }, "else": None } } } }, # Step 2: Filter out any records with null `property_id` (invalid conversions) { "$match": { "property_id": { "$ne": None } } }, # Step 3: Order by `order` field in ascending order { "$sort": { "order": 1 } }, # Step 3: Project only the necessary fields { "$project": { "property_media_id": 1, "type": 1, "order": 1, "main": 1, "url": 1, "file_name": 1, "property_id": 1, "region_id": 1 } }, ] } pipeline_properties_nested = { 'v1_beta_stage_properties_nested': [ # Specify the source collection { '$unionWith': { 'coll': "ilist_properties" } }, # Step 1: Project core fields and convert IDs { "$project": { "property_id": { "$convert": { "input": { "$replaceAll": { "input": "$ID", "find": "-", "replacement": "" } }, "to": "long", "onError": None } }, "listing_id": "$ID", "transaction_type": "$TransactionType", "youtube_url": "$Youtube", "tour_virtual_url": "$VirtualTourURL", "list_price": "$CurrentListingPrice", "list_price_currency": "$CurrentListingCurrency", "property_administration_fee": "$MaintenanceFee", "property_administration_fee_currency": "$CurrentListingCurrency", "property_administration_fee_period": "$MaintenanceFeePeriod", "rental_price_period": "$RentalPriceGranularity", "yearly_tax": "$YearlyTax", "yearly_tax_currency": "$YearlyTaxCurrency", "unit_number": "$ApartmentNumber", "number_units_building" : "$NumberOfApartmentsInBuilding", "living_area": "$BuiltArea", "living_area_unit": "m2", "year_built": "$YearBuilt", "bedrooms": "$NumberOfBedrooms", "bathrooms": "$NumberOfBathrooms", "garage": "$ParkingSpaces", "garage_type": "Garage", "availability_date": "$AvailabilityDate", "commercial_residential": { "$toString": "$CommercialResidential" }, "total_area": "$TotalArea", "show_hide_flag": "$ShowHideFlag", "show_address_on_web": "$ShowAddressOnWeb", "is_public_available": "$IsPublicAvailable", "number_of_toilet_rooms": "$NumberOfToiletRooms", "total_num_of_rooms": "$TotalNumOfRooms", "hide_price_public": "$HidePricePublic", "market_id": "$MarketStatus", "floor_id": "$FloorLevel", "type_id": "$PropertyType", "office_id": "$OfficeID", "agent_id": "$AssociateID", "region_id": "$RegionID", "publish_date": { "$toDate": "$OrigListingDate" } } }, # Lookup Payment Period for Rental Payment Period { "$lookup": { "from": "v1_beta_stage_payment_period", "localField": "rental_price_period", "foreignField": "payment_id", "as": "rental_price_period" } }, { "$unwind": { "path": "$rental_price_period", "preserveNullAndEmptyArrays": True } }, # Lookup Payment Period for Property Administration Fee Period { "$lookup": { "from": "v1_beta_stage_payment_period", "localField": "property_administration_fee_period", "foreignField": "payment_id", "as": "property_administration_fee_period" } }, { "$unwind": { "path": "$property_administration_fee_period", "preserveNullAndEmptyArrays": True } }, # Lookup Transaction Type { "$lookup": { "from": "v1_beta_stage_transaction_type", "localField": "transaction_type", "foreignField": "transaction_id", "as": "transaction_type" } }, { "$unwind": { "path": "$transaction_type", "preserveNullAndEmptyArrays": True } }, # # Lookup Description # { # "$lookup": { # "from": "v1_beta_stage_properties_description", # "localField": "property_id", # "foreignField": "property_id", # "as": "description" # } # }, #{ "$unwind": { "path": "$description", "preserveNullAndEmptyArrays": True } }, # Lookup MarketStatus { "$lookup": { "from": "v1_beta_stage_market_status", "localField": "market_id", "foreignField": "market_id", "as": "market_status" } }, { "$unwind": { "path": "$market_status", "preserveNullAndEmptyArrays": True } }, { "$project": { "market_status._id": 0 } }, # Lookup ListingStatus { "$lookup": { "from": "v1_beta_stage_listing_status", "localField": "listing_id", "foreignField": "listing_id", "as": "listing_status" } }, { "$unwind": { "path": "$listing_status", "preserveNullAndEmptyArrays": True } }, { "$project": { "listing_status._id": 0 } }, # Lookup FloorLevel { "$lookup": { "from": "v1_beta_stage_floor_level", "localField": "floor_id", "foreignField": "floor_id", "as": "floor_level" } }, { "$unwind": { "path": "$floor_level", "preserveNullAndEmptyArrays": True } }, { "$project": { "floor_level._id": 0 } }, # Lookup Descriptions { "$lookup": { "from": "v1_beta_stage_properties_description", "localField": "property_id", "foreignField": "property_id", "as": "descriptions" } }, { "$project": { "descriptions._id": 0, "descriptions.property_id": 0 } }, # Lookup Features { "$lookup": { "from": "v1_beta_stage_properties_features", "localField": "property_id", "foreignField": "property_id", "as": "features" } }, { "$project": { "features._id": 0 } }, # Lookup Location { "$lookup": { "from": "v1_beta_stage_properties_location", "localField": "property_id", "foreignField": "property_id", "as": "location" } }, { "$unwind": { "path": "$location", "preserveNullAndEmptyArrays": True } }, { "$project": { "location._id": 0 } }, # Lookup Media { "$lookup": { "from": "v1_beta_stage_properties_media", "localField": "property_id", "foreignField": "property_id", "as": "medias" } }, { "$project": { "medias._id": 0 } }, # Lookup Agent { "$lookup": { "from": "v1_beta_stage_agents", "localField": "agent_id", "foreignField": "agent_id", "as": "agent" } }, { "$unwind": { "path": "$agent", "preserveNullAndEmptyArrays": True } }, { "$project": { "agent._id": 0 } }, # Lookup Office { "$lookup": { "from": "v1_beta_stage_offices", "localField": "office_id", "foreignField": "office_id", "as": "office" } }, { "$unwind": { "path": "$office", "preserveNullAndEmptyArrays": True } }, { "$project": { "office._id": 0 } }, # Lookup Type { "$lookup": { "from": "v1_beta_stage_types", "localField": "type_id", "foreignField": "type_id", "as": "type" } }, { "$unwind": { "path": "$type", "preserveNullAndEmptyArrays": True } }, { "$project": { "type._id": 0 } }, # Lookup Usage { "$lookup": { "from": "v1_beta_stage_usages", "localField": "commercial_residential", "foreignField": "usage_id", "as": "usage" } }, { "$unwind": { "path": "$usage", "preserveNullAndEmptyArrays": True } }, { "$project": { "usage._id": 0 } }, { "$addFields": { "sync_date": datetime.now().isoformat(), "transfer_data": datetime.now().isoformat(), "lat":{ "$convert": { "input": "$location.lat", "to": "double", "onError": None } }, "lng":{ "$convert": { "input": "$location.lng", "to": "double", "onError": None } }, # "title": { # "$cond": { # "if": { "$eq": ["$description.type", "ListingTitle"] }, # "then": "$description.description", # "else": None # } # }, "transaction_type": '$transaction_type.transaction_name_en', "detail_view_url": None, # "description": { # "$cond": { # "if": { "$eq": ["$description.type", "description"] }, # "then": "$description.description", # "else": None # } # }, "list_price": { "$cond": { "if": { "$eq": ["$transaction_type.transaction_id", 261] }, "then": { "$cond": { "if": { "$or": [{ "$isNumber": "$list_price" }, { "$type": "$list_price" == "decimal" }] }, "then": "$list_price", "else": None } }, "else": None } }, "list_price_currency": { "$cond": { "if": { "$eq": ["$transaction_type.transaction_id", 261] }, "then": "$list_price_currency", "else": None } }, "rental_price": { "$cond": { "if": { "$ne": ["$transaction_type.transaction_id", 261] }, "then": { "$cond": { "if": { "$or": [{ "$isNumber": "$list_price" }, { "$type": "$list_price" == "decimal" }] }, "then": "$list_price", "else": None } }, "else": None } }, "rental_price_currency": { "$cond": { "if": { "$ne": ["$transaction_type.transaction_id", 261] }, "then": "$list_price_currency", "else": None } }, "rental_price_period": "$rental_price_period.payment_name_en", "property_administration_fee": { "$cond": { "if": { "$or": [{ "$isNumber": "$property_administration_fee" }, { "$type": "$property_administration_fee" == "decimal" }] }, "then": "$property_administration_fee", "else": None } }, "property_administration_fee_period": "$property_administration_fee_period.payment_name_en", "bedrooms": { "$cond": { "if": {"$isNumber": "$bedrooms"}, "then": "$bedrooms", "else": None } }, "bathrooms": { "$cond": { "if": {"$isNumber": "$bathrooms"}, "then": "$bathrooms", "else": None } }, "garage": { "$cond": { "if": {"$isNumber": "$garage"}, "then": "$garage", "else": None } }, "unit_floor": None, "usage": [{ "usage_id": "$usage.usage_id", "usage_name_en": "$usage.usage_name_en", "usage_name_pt": "$usage.usage_name_pt" }], "type": [{ "type_id": "$type.type_id", "type_name_en": "$type.type_name_en", "type_name_pt": "$type.type_name_pt" }], "office": ["$office"], "agent": ["$agent"], "location": ["$location"], "market_status": [{ "market_id": "$market_status.market_id", "market_name_en": "$market_status.market_name_en", "market_name_pt": "$market_status.market_name_pt" }], "listing_status": [{ "listing_id": "$listing_status.listing_id", "listing_name_en": "$listing_status.listing_name_en", "listing_name_pt": "$listing_status.listing_name_pt" }], "floor_level": [{ "floor_id": "$floor_level.floor_id", "floor_name_en": "$floor_level.floor_name_en", "floor_name_pt": "$floor_level.floor_name_pt" }], "descriptions": "$descriptions", "features": "$features", "medias": "$medias" }, }, # Restructure the final document { "$project": { "_id": 0, "property_id": 1, "listing_id": 1, "title": 1, "transaction_type": 1, "detail_view_url": 1, "list_price": 1, "list_price_currency": 1, "rental_price":1, "rental_price_currency": 1, "rental_price_period": 1, "property_administration_fee": 1, "property_administration_fee_period": 1, "yearly_tax": 1, "yearly_tax_currency": 1, "living_area": 1, "living_area_unit": 1, "year_built": 1, "bedrooms": 1, "bathrooms": 1, "garage": 1, "garage_type": 1, "unit_floor": 1, "unit_number": 1, "publish_date": 1, "sync_date": 1, "transfer_data": 1, "usage": 1, "type": 1, "office": 1, "agent": 1, "location": 1, "number_units_building": 1, "availability_date": 1, "commercial_residential": 1, "total_area": 1, "show_hide_flag": 1, "show_address_on_web": 1, "is_public_available": 1, "number_of_toilet_rooms": 1, "total_num_of_rooms": 1, "hide_price_public": 1, "youtube_url": 1, "tour_virtual_url": 1, "region_id": 1, "lng": 1, "lat": 1, "market_status": 1, "listing_status": 1, "floor_level": 1, "descriptions": "$descriptions", "features": "$features", "medias": "$medias", } } ] } collection_indexes = { 'v1_beta_stage_offices': ['office_id'], 'v1_beta_stage_agents': ['agent_id', 'email'], 'v1_beta_stage_features': ['feature_id'], 'v1_beta_stage_transaction_type': ['trasaction_id'], 'v1_beta_stage_properties_features': ['feature_id', 'property_id'], 'v1_beta_stage_usages': ['usage_id'], 'v1_beta_stage_types': ['type_id'], 'v1_beta_stage_properties_media': ['property_id'], 'v1_beta_stage_properties_location': ['property_id'], 'v1_beta_stage_properties_nested': ['property_id'], 'v1_beta_stage_listing_status': ['listing_id'], 'v1_beta_stage_market_status': ['market_id'], 'v1_beta_stage_floor_level': ['floor_id'], 'v1_beta_stage_property_description_types': ['property_description_type_id'], 'v1_beta_stage_properties_description': ['property_id'], 'v1_beta_stage_payment_period': ['payment_id'], 'v1_beta_stage_states': ['state_id'], 'v1_beta_stage_cities': ['city_id'], 'v1_beta_stage_neighborhoods': ['neighborhood_id'], 'v1_beta_stage_languages': ['language_id'], 'v1_beta_stage_agents_language_spoken': ['agent_id'], 'v1_beta_stage_estados_detalhes': ['state_id'], 'v1_beta_stage_cidades_detalhes': ['city_id'], 'v1_beta_stage_bairros_detalhes': ['neighborhood_id'], } collection_names_to_be_changed = { 'v1_beta_stage_offices': 'v1_offices', 'v1_beta_stage_agents': 'v1_agents', 'v1_beta_stage_features': 'v1_features', 'v1_beta_stage_properties_features': 'v1_properties_features', 'v1_beta_stage_usages': 'v1_usages', 'v1_beta_stage_types': 'v1_types', 'v1_beta_stage_properties_media': 'v1_properties_media', 'v1_beta_stage_properties_location': 'v1_properties_location', 'v1_beta_stage_properties_nested': 'v1_properties_nested', 'v1_beta_stage_states': 'v1_states', 'v1_beta_stage_cities': 'v1_cities', 'v1_beta_stage_neighborhoods': 'v1_neighborhoods', } collection_names_to_be_deleted = [ 'v1_beta_stage_market_status', 'v1_beta_stage_listing_status', 'v1_beta_stage_floor_level', 'v1_beta_stage_transaction_type', 'v1_beta_stage_listing_status', 'v1_beta_stage_market_status', 'v1_beta_stage_floor_level', 'v1_beta_stage_property_description_types', 'v1_beta_stage_properties_description', 'v1_beta_stage_payment_period', 'v1_beta_stage_languages', 'v1_beta_stage_agents_language_spoken', 'v1_beta_stage_estados_detalhes', 'v1_beta_stage_cidades_detalhes', 'v1_beta_stage_bairros_detalhes', ] # Create features Collection step_number=1 if step is None or step == step_number: ### Create a collection from a pipeline (Stage) self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating collections from pipelines...')) try: if batch_size is None: batch_size_ = 20000 else: batch_size_ = batch_size # First pipeline self.create_collection_from_pipeline(pipeline_listing_status, collection_indexes, limit, batch_size_) # Second pipeline self.create_collection_from_pipeline(pipeline_market_status, collection_indexes, limit, batch_size_) # Third pipeline self.create_collection_from_pipeline(pipeline_floor_level, collection_indexes, limit, batch_size_) # Fourth pipeline self.create_collection_from_pipeline(pipeline_features, collection_indexes, limit, batch_size_) # Fifth pipeline self.create_collection_from_pipeline(pipeline_types, collection_indexes, limit, batch_size_) # Sixth pipeline self.create_collection_from_pipeline(pipeline_property_description_types, collection_indexes, limit, batch_size_) # Seventh pipeline self.create_collection_from_pipeline(pipeline_properties_description, collection_indexes, limit, batch_size_) # Eighth pipeline self.create_collection_from_pipeline(pipeline_transaction_type, collection_indexes, limit, batch_size_) # Ninth pipeline self.create_collection_from_pipeline(pipeline_payment_period, collection_indexes, limit, batch_size_) # Tenth pipeline self.create_collection_from_pipeline(pipeline_languages, collection_indexes, limit, batch_size_) # Eleventh pipeline self.create_collection_from_pipeline(pipeline_agent_languages, collection_indexes, limit, batch_size_) # Twelfth pipeline self.create_collection_from_pipeline(pipeline_estados_detalhes, collection_indexes, limit, batch_size_) # Thirteenth pipeline self.create_collection_from_pipeline(pipeline_cidades_detalhes, collection_indexes, limit, batch_size_) # Fourteenth pipeline self.create_collection_from_pipeline(pipeline_bairros_detalhes, collection_indexes, limit, batch_size_) except Exception as e: logger.error(f"Error in step {step_number} creating collection from pipeline: {str(e)}") # Create fregion collections #step_number=2 #if step is None or step == step_number: if step == 2: ### Create a collection from a pipeline (Stage) self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating collections from pipelines...')) try: if batch_size is None: batch_size_ = 5000 else: batch_size_ = batch_size # First pipeline self.create_collection_from_pipeline(pipeline_states, collection_indexes, limit, batch_size_) # Second pipeline self.create_collection_from_pipeline(pipeline_cities, collection_indexes, limit, batch_size_) # Third pipeline self.create_collection_from_pipeline(pipeline_neighborhoods, collection_indexes, limit, batch_size_) except Exception as e: logger.error(f"Error in step {step_number} creating collection from pipeline: {str(e)}") # Create office Collection step_number=3 if step is None or step == step_number: ### Create a collection from a pipeline (Stage) self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating collections from pipelines...')) try: if batch_size is None: batch_size_ = 5000 else: batch_size_ = batch_size self.create_collection_from_pipeline(pipeline_office, collection_indexes, limit, batch_size_) except Exception as e: logger.error(f"Error in step {step_number} creating collection from pipeline: {str(e)}") # Create agent Collection step_number=4 if step is None or step == step_number: ### Create a collection from a pipeline (Stage) self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating collections from pipelines...')) try: if batch_size is None: batch_size_ = 5000 else: batch_size_ = batch_size # Check if v1_beta_stage_features collection exists mongo_client, mongo_db = self.connect_to_mongodb() if "v1_beta_stage_offices" not in mongo_db.list_collection_names(): self.stdout.write(self.style.ERROR(f"Collection 'v1_beta_stage_offices' does not exist. Please create it before running step {step_number}.")) return if "v1_beta_stage_languages" not in mongo_db.list_collection_names(): self.stdout.write(self.style.ERROR(f"Collection 'v1_beta_stage_languages' does not exist. Please create it before running step {step_number}.")) return mongo_client.close() self.create_collection_from_pipeline(pipeline_agent, collection_indexes, limit, batch_size_) except Exception as e: logger.error(f"Error in step {step_number} creating collection from pipeline: {str(e)}") # Create properties_features Collection step_number=5 if step is None or step == step_number: self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating collections from pipelines...')) try: # Default batch size if not provided if batch_size is None: batch_size_ = 100000 else: batch_size_ = batch_size # Check if v1_beta_stage_features collection exists mongo_client, mongo_db = self.connect_to_mongodb() if "v1_beta_stage_features" not in mongo_db.list_collection_names(): self.stdout.write(self.style.ERROR(f"Collection 'v1_beta_stage_features' does not exist. Please create it before running step {step_number}.")) return mongo_client.close() # Create collection from pipeline self.create_collection_from_pipeline(pipeline_properties_features, collection_indexes, limit, batch_size_) except Exception as e: logger.error(f"Error in step {step_number} creating collection from pipeline: {str(e)}") self.stdout.write(self.style.ERROR(f"Error in step {step_number} creating collection from pipeline: {str(e)}")) # # Create types Collection # step_number=5 # if step is None or step == step_number: # ### Create a collection from a pipeline (Stage) # self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating collections from pipelines...')) # try: # if batch_size is None: # batch_size_ = 5000 # else: # batch_size_ = batch_size # self.create_collection_from_pipeline(pipeline_types, collection_indexes, limit, batch_size_) # except Exception as e: # logger.error(f"Error in step {step_number} creating collection from pipeline: {str(e)}") # Create usages Collection step_number=6 if step is None or step == step_number: ### Create a collection from a pipeline (Stage) self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating collections from pipelines...')) try: if batch_size is None: batch_size_ = 5000 else: batch_size_ = batch_size self.create_collection_from_pipeline(pipeline_usages, collection_indexes, limit, batch_size_) except Exception as e: logger.error(f"Error in step {step_number} creating collection from pipeline: {str(e)}") # Create location Collection step_number=7 if step is None or step == step_number: ### Create a collection from a pipeline (Stage) self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating collections from pipelines...')) try: if batch_size is None: batch_size_ = 20000 else: batch_size_ = batch_size self.create_collection_from_pipeline(pipeline_location, collection_indexes, limit, batch_size_) except Exception as e: logger.error(f"Error in step {step_number} creating collection from pipeline: {str(e)}") #Create properties_media collection step_number=8 if step is None or step == step_number: ### Create properties_media collection from a pipeline (Stage) self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating properties_media collection from pipeline...')) try: if batch_size is None: batch_size_ = 200000 else: batch_size_ = batch_size self.create_collection_from_pipeline(pipeline_properties_media, collection_indexes, limit, batch_size_) except Exception as e: logger.error(f"Error in step {step_number} creating properties_media collection from pipeline: {str(e)}") step_number=9 if step is None or step == step_number: ### Create properties_nested collection from a pipeline (Stage) self.stdout.write(self.style.WARNING(f'Step {step_number}: Creating properties_nested collection from pipeline...')) try: if batch_size is None: batch_size_ = 1000 else: batch_size_ = batch_size self.create_collection_from_pipeline(pipeline_properties_nested, collection_indexes, limit, batch_size_) except Exception as e: logger.error(f"Error in step {step_number} creating properties_nested collection from pipeline: {str(e)}") step_number=10 if step is None or step == step_number: ### Change collection names self.stdout.write(self.style.WARNING(f'Step {step_number}: Changing collection names...')) try: self.change_collection_names(collection_names_to_be_changed) except Exception as e: logger.error(f"Error in step {step_number} changing collection names: {str(e)}") #delete collections not used step_number=11 if step is None or step == step_number: ### Drop collections self.stdout.write(self.style.WARNING(f'Step {step_number}: Dropping collections...')) try: self.drop_collections(collection_names_to_be_deleted) except Exception as e: logger.error(f"Error in step {step_number} dropping collections: {str(e)}") #End time end_time = time.time() total_time = round((end_time - start_time),2) self.stdout.write(self.style.SUCCESS(f'Total Time: {total_time}')) logger.info(f"Data Transfer Completed Successfully at {datetime.now()}. Total Time: {total_time}") ### Transfer Methods ### def create_collection_from_pipeline(self, pipelines, indexes=None, limit=None, batch_size=None): try: start_pipeline_time = time.time() for mongo_collection_name, pipeline in pipelines.items(): try: mongo_client, mongo_db = self.connect_to_mongodb() mongo_db.drop_collection(mongo_collection_name) target_collection = mongo_db[mongo_collection_name] offset = 0 total_docs_transferred = 0 while True: # Determine the size of the next batch if limit is not None: remaining_docs = limit - total_docs_transferred current_batch_size = min(batch_size, remaining_docs) if batch_size is not None else remaining_docs if remaining_docs <= 0: break else: current_batch_size = batch_size # Add limit and skip stages to the pipeline batch_pipeline = pipeline.copy() if current_batch_size is not None: batch_pipeline.append({'$skip': offset}) batch_pipeline.append({'$limit': current_batch_size}) # Log the current state of the batch processing self.stdout.write(self.style.SUCCESS(f'Batch pipeline stages added. Offset: {offset}, Limit: {current_batch_size}')) # Execute the pipeline result = list(mongo_db['source_collection'].aggregate(batch_pipeline)) # Break if no more documents are retrieved if not result: self.stdout.write(self.style.WARNING('No more documents to process. Breaking loop.')) break # Remove '_id' field to avoid duplicate key errors result = [ {key: value for key, value in doc.items() if key != '_id'} for doc in result ] # Insert batch into the target collection try: target_collection.insert_many(result) except pymongo.errors.BulkWriteError as bwe: self.stdout.write(self.style.ERROR(f'Batch insert error: {bwe.details}')) logger.error(f"Batch insert error: {bwe.details}") break batch_docs_count = len(result) total_docs_transferred += batch_docs_count offset += current_batch_size self.stdout.write(self.style.SUCCESS(f'Batch processed. Documents retrieved: {batch_docs_count}')) self.stdout.write(self.style.SUCCESS(f'Batch inserted into {mongo_collection_name}. Documents: {batch_docs_count}')) self.stdout.write(self.style.SUCCESS(f'Total documents transferred so far: {total_docs_transferred}')) # Break if the last batch size is less than the batch size if batch_docs_count < current_batch_size: self.stdout.write(self.style.WARNING('Last batch processed. No more documents to retrieve.')) break # Create indexes from collection_indexes if mongo_collection_name in indexes: for index in indexes[mongo_collection_name]: target_collection.create_index([(index, pymongo.ASCENDING)]) self.stdout.write(self.style.SUCCESS(f'Index created: {index}')) mongo_client.close() self.stdout.write(self.style.SUCCESS(f'Collection {mongo_collection_name} successfully created!')) except Exception as e: logger.error(f"Error creating collection {mongo_collection_name}: {str(e)}") self.stdout.write(self.style.ERROR(f'An error occurred: {str(e)}')) except Exception as e: logger.error(f"Error creating collection from pipeline: {str(e)}") self.stdout.write(self.style.ERROR(f'An error occurred: {str(e)}')) finally: if mongo_client: mongo_client.close() total_pipeline_time = round((time.time() - start_pipeline_time), 2) self.stdout.write(self.style.SUCCESS(f'Total Pipeline Time: {total_pipeline_time}')) def change_collection_names(self, dict): try: start_change_time = time.time() mongo_client, mongo_db = self.connect_to_mongodb() for old_name, new_name in dict.items(): try: if old_name in mongo_db.list_collection_names(): mongo_db.drop_collection(new_name) mongo_db[old_name].rename(new_name) self.stdout.write(self.style.SUCCESS(f'Collection {old_name} successfully renamed to {new_name}!')) else: self.stdout.write(self.style.WARNING(f'Source collection {old_name} does not exist. Skipping...')) except pymongo.errors.OperationFailure as e: if e.code == 26: # NamespaceNotFound error code self.stdout.write(self.style.WARNING(f'Source collection {old_name} does not exist. Skipping...')) else: raise except Exception as e: logger.error(f"Error changing collection names: {str(e)}") self.stdout.write(self.style.ERROR(f'An error occurred: {str(e)}')) finally: mongo_client.close() total_change_time = round((time.time() - start_change_time), 2) self.stdout.write(self.style.SUCCESS(f'Total Change Time: {total_change_time}')) def drop_collections(self, list): try: start_drop_time = time.time() for collection_name in list: mongo_client, mongo_db = self.connect_to_mongodb() try: if collection_name in mongo_db.list_collection_names(): mongo_db.drop_collection(collection_name) self.stdout.write(self.style.SUCCESS(f'Collection {collection_name} successfully droped!')) else: self.stdout.write(self.style.WARNING(f'Collection {collection_name} does not exist. Skipping...')) except pymongo.errors.OperationFailure as e: if e.code == 26: # NamespaceNotFound error code self.stdout.write(self.style.WARNING(f'Collection {collection_name} does not exist. Skipping...')) else: raise finally: mongo_client.close() except Exception as e: logger.error(f"Error dropping collections: {str(e)}") self.stdout.write(self.style.ERROR(f'An error occurred: {str(e)}')) finally: total_drop_time = round((time.time() - start_drop_time), 2) self.stdout.write(self.style.SUCCESS(f'Total Drop Time: {total_drop_time}')) def drop_from_mysql_collections(self,dict1,dict2): try: start_drop_time = time.time() for table_to_collection in dict1: for mongo_data in table_to_collection.values(): mongo_client, mongo_db = self.connect_to_mongodb() mongo_collection_name = mongo_data[0] mongo_db.drop_collection(mongo_collection_name) mongo_client.close() self.stdout.write(self.style.SUCCESS(f'Collection {mongo_collection_name} successfully droped!')) for collection_from_mysql in dict2: for mongo_collection_name in collection_from_mysql.keys(): mongo_client, mongo_db = self.connect_to_mongodb() mongo_db.drop_collection(mongo_collection_name) mongo_client.close() self.stdout.write(self.style.SUCCESS(f'Collection {mongo_collection_name} successfully droped!')) except Exception as e: logger.error(f"Error dropping collections: {str(e)}") self.stdout.write(self.style.ERROR(f'An error occurred: {str(e)}')) finally: if mongo_client: mongo_client.close() total_drop_time = round((time.time() - start_drop_time),2) self.stdout.write(self.style.SUCCESS( f'Total Drop Time: {total_drop_time}' ) ) #@retry(pymongo.errors.ServerSelectionTimeoutError, delay=5, tries=3) def connect_to_mongodb(self): # Get MongoDB configuration from settings.py mongo_settings = settings.DATABASES['mongodb'] # Configure MongoDB connection using settings mongo_client = pymongo.MongoClient( host=mongo_settings['CLIENT']['host'], port=mongo_settings['CLIENT']['port'], username=mongo_settings['CLIENT']['username'], password=mongo_settings['CLIENT']['password'], authSource=mongo_settings['CLIENT']['authSource'], authMechanism=mongo_settings['CLIENT']['authMechanism'], ) return mongo_client, mongo_client[mongo_settings['NAME']] def haversine_distance(lat1, lon1, lat2, lon2): # Radius of Earth in meters R = 6371000 # Convert degrees to radians phi1, phi2 = radians(lat1), radians(lat2) delta_phi = radians(lat2 - lat1) delta_lambda = radians(lon2 - lon1) # Haversine formula a = sin(delta_phi / 2)**2 + cos(phi1) * cos(phi2) * sin(delta_lambda / 2)**2 c = 2 * atan2(sqrt(a), sqrt(1 - a)) return R * c ## /home/remax/virtualenv/api/3.8/bin/python3 /home/remax/api/PortalREMax/manage.py v1_transfer_mysql_to_mongodb 5000 >> /home/remax/logs/v1_transfer_mysql_to_mongodb.log 2>&1 # 11 0,6,12,19 * * * /home/remax/virtualenv/api/3.8/bin/python3 /home/remax/api/PortalREMax/manage.py v1_transfer_mongodb_to_mongodb > /dev/null 2>&1