#### don't forget to add comments, specially for hard coded part import firebase_admin from firebase_admin import db, credentials, firestore from google.cloud import storage import time from selenium import webdriver from selenium.webdriver.chrome.service import Service from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from webdriver_manager.chrome import ChromeDriverManager from PIL import Image, ImageDraw, ImageFont import time import os from PIL import Image, ImageDraw, ImageFont import math from PIL import Image import math import requests from io import BytesIO import datetime from google.oauth2 import service_account from datetime import date import traceback import http.server import socketserver import threading import os import json # Port for the local server PORT = 8000 DIRECTORY = os.path.dirname(os.path.abspath(__file__)) httpd = None # Start a simple HTTP server to serve the files def start_server(): global httpd handler = http.server.SimpleHTTPRequestHandler handler.directory = DIRECTORY httpd = socketserver.TCPServer(("", PORT), handler) # with socketserver.TCPServer(("", PORT), handler) as httpd: print(f"Serving at port {PORT}") httpd.serve_forever() # Function to stop the server def stop_server(): #if httpd: print("Shutting down the server...") # httpd.shutdown() # Gracefully stop the server # httpd.server_close() # Close the server socket # print("Server stopped.") # Run the server in a separate thread try: server_thread = threading.Thread(target=start_server) server_thread.daemon = True server_thread.start() except: print(traceback.format_exc()) def create_image_collage(image_paths, output_path, images_per_row=3, spacing=10, logo_height=500): """ Create a collage from multiple images with a logo at the top. :param image_paths: List of paths to the watermarked images :param output_path: Path where the collage will be saved :param logo_url: URL of the logo image :param images_per_row: Number of images per row in the collage :param spacing: Spacing between images in pixels :param logo_height: Height of the logo in pixels """ logo_url = "https://farmonaut.com/Images/ir_header.png" # Open all images images = [Image.open(path) for path in image_paths] # Find the maximum dimensions max_width = max(img.width for img in images) max_height = max(img.height for img in images) # Calculate the number of rows needed num_images = len(images) num_rows = math.ceil(num_images / images_per_row) # Calculate the size of the collage collage_width = (max_width * images_per_row) + (spacing * (images_per_row - 1)) collage_height = (max_height * num_rows) + (spacing * (num_rows - 1)) # Download and resize the logo response = requests.get(logo_url) logo = Image.open(BytesIO(response.content)) logo_width = int((logo.width / logo.height) * logo_height) logo = logo.resize((logo_width, logo_height), Image.LANCZOS) # Create a new image for the collage, including space for the logo collage = Image.new('RGB', (collage_width, collage_height + logo_height + spacing), (255, 255, 255)) # Paste the logo at the top center logo_x = (collage_width - logo_width) // 2 collage.paste(logo, (logo_x, 0)) # Paste images into the collage for i, img in enumerate(images): row = i // images_per_row col = i % images_per_row x = col * (max_width + spacing) y = row * (max_height + spacing) + logo_height + spacing # Center the image if it's smaller than the maximum dimensions x_offset = (max_width - img.width) // 2 y_offset = (max_height - img.height) // 2 collage.paste(img, (x + x_offset, y + y_offset)) # Save the collage collage.save(output_path) # Example usage: # Assuming you have already created watermarked images # watermarked_images = ['watermarked1.jpg', 'watermarked2.jpg', 'watermarked3.jpg', ...] # logo_url = "https://farmonaut.com/wp-content/uploads/2020/01/Farmonaut_Logo_Black.png" # create_image_collage(watermarked_images, 'collage_with_logo.jpg', logo_url, images_per_row=3, spacing=10, logo_height=50) def add_diagonal_watermark(image_path, output_path, watermark_text, font_path=None): # Open the image font_size=300 opacity=30 with Image.open(image_path).convert('RGBA') as img: # Calculate the diagonal length of the image diagonal = int(math.sqrt(img.width**2 + img.height**2)) # Create a new transparent image for the watermark with the diagonal size watermark = Image.new('RGBA', (diagonal, diagonal), (0, 0, 0, 0)) # Try to use a font that supports special characters try: if font_path: font = ImageFont.truetype(font_path, font_size) else: # Try to use Arial, which usually supports ® font = ImageFont.truetype("arial.ttf", font_size) except IOError: # If Arial is not available, try a built-in font font = ImageFont.load_default().font_variant(size=font_size) # Create a draw object draw = ImageDraw.Draw(watermark) # Calculate the size of the watermark text left, top, right, bottom = font.getbbox(watermark_text) text_width = right - left text_height = bottom - top # Calculate the position to center the text on the diagonal image pos = ((diagonal - text_width) / 2, (diagonal - text_height) / 2) # Draw the text in green draw.text(pos, watermark_text, font=font, fill=(0, 255, 0, opacity)) # Rotate the watermark to align with the image diagonal angle = math.degrees(math.atan2(img.height, img.width)) rotated_watermark = watermark.rotate(angle, expand=False) # Calculate the position to paste the rotated watermark paste_pos = (-int((diagonal - img.width) / 2), int((diagonal - img.height) / 2)) # Paste the watermark onto the original image img.paste(rotated_watermark, paste_pos, rotated_watermark) # Save the watermarked image img.save(output_path) def capture_full_page_screenshot(url, uid, fieldid): # Set up Chrome options for mobile emulation mobile_emulation = { "deviceMetrics": {"width": 375, "height": 812, "pixelRatio": 3.0}, # iPhone X "userAgent": ( "Mozilla/5.0 (iPhone; CPU iPhone OS 13_3 like Mac OS X) " "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.4 Mobile/15E148 Safari/604.1" ) } chrome_options = Options() chrome_options.add_experimental_option("mobileEmulation", mobile_emulation) chrome_options.add_argument("--headless") # Run headless (without GUI) chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--disable-dev-shm-usage") chrome_options.add_argument("--window-size=1920x1080") # Initialize the WebDriver driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=chrome_options) # Open the URL driver.get(url) # Wait for the page to fully load time.sleep(25) # List of class names to scroll to class_names = ["cardHeader", "irrigation", "fertilizers"] # Initialize screenshot index and directory screenshot_index = 0 screenshot_dir = uid + "/" + fieldid os.makedirs(screenshot_dir, exist_ok=True) watermarked_images = [] for class_name in class_names: try: # Find the element by class name element = driver.find_element(By.CLASS_NAME, class_name) # Scroll to the element driver.execute_script("arguments[0].scrollIntoView();", element) # Wait for the element to be in view time.sleep(5) # Capture screenshot screenshot_path = os.path.join(screenshot_dir, f"{screenshot_index}.png") driver.save_screenshot(screenshot_path) print(f"Saved screenshot: {screenshot_path}") # Add watermark watermarked_path = os.path.join(screenshot_dir, f"watermarked_{screenshot_index}.png") watermarked_images.append((screenshot_dir + '/' + f"{screenshot_index}.png")) print(watermarked_images) add_diagonal_watermark(screenshot_path, screenshot_path, "FARMONAUT", font_path=None) #add_watermark(screenshot_path, watermarked_path, "Farmonaut®") print(f"Saved watermarked screenshot: {watermarked_path}") screenshot_index += 1 except Exception as e: print(f"Could not find or scroll to element with class name '{class_name}': {e}") # watermarked_images = ['watermarked1.jpg', 'watermarked2.jpg', 'watermarked3.jpg', ...] mediaUrl = uid + '/' + fieldid + '/collage.jpg' create_image_collage(watermarked_images, mediaUrl, images_per_row=3, spacing=10) # Close the WebDriver driver.quit() print(f"All screenshots saved in directory: {screenshot_dir}") return mediaUrl def get_firebase_storage_url(object_path): bucket_name = "farmbase-b2f7e.appspot.com" storage_client = storage.Client() # Get a reference to the storage service bucket = storage_client.get_bucket(bucket_name) cred = service_account.Credentials.from_service_account_file("servicekey.json") # Create a blob object representing the file blob = bucket.blob(object_path) # Generate a signed URL for the blob # You can adjust the expiration time as needed #url = blob.generate_signed_url(expiration=3600) # URL valid for 1 hour url = blob.generate_signed_url( version="v4", expiration=datetime.timedelta(minutes=10000), method="GET", credentials=cred, ) print(url) return url def upload_file_to_storage(source_file_name, destination_file_name, expire_minutes=1000): from google.cloud import storage from oauth2client.service_account import ServiceAccountCredentials import os import firebase_admin from firebase_admin import credentials from firebase_admin import db import datetime try: cred = credentials.Certificate('servicekey.json') firebase_admin.initialize_app(cred, {'databaseURL': 'https://farmbase-b2f7e-31c0c.firebaseio.com/'}) except: print('initialized') source_url = source_file_name storage_client = storage.Client.from_service_account_json('servicekey.json') bucket_name = 'farmbase-b2f7e.appspot.com' destination_blob_name = destination_file_name bucket = storage_client.get_bucket(bucket_name) blob = bucket.blob(destination_blob_name) storage_url = 'https://farmonaut.com/Images/data_not_found.jpg' try: blob.upload_from_filename(source_url) storage_url = blob.generate_signed_url(version="v4",expiration = datetime.timedelta(minutes=expire_minutes), method="GET") except: print('not there') return storage_url def generate_jeevnai_whatsapp_image(mediaUrl, uid, fieldid, lang, imagedate=None): try: is_admin = db.reference("SalesAssociatesPromo").child("AdminUIDs").child(uid).get() if uid == "mmC6X2DpreSxO6580MtUKhyaVgq1": is_admin = None if is_admin is not None: return mediaUrl else: #if imagedate is not None: db.reference("PaidMonitoredFields").child("PMF").child(uid).child(fieldid).child("LatestDay").set(imagedate) db.reference("PaidMonitoredFields").child("PMF").child(uid).child(fieldid).child("SensedDays").child(imagedate).set("yes") db_firestore = firestore.client() req_obj = { 'Processed':0, 'FieldID': fieldid } #upload_file_to_storage((uid + '/' + fieldid + '/hybrid.png'), ('PaidMonitoredFields/' + uid + '/' + fieldid + '/' + str(imagedate)+ '/hybrid'), expire_minutes=1000) # print(('hybrid uploaded...', imagedate)) #time.sleep(10000) db_firestore.collection("InteractiveReportRequests").document(uid).set(req_obj) print('generating jeevnai advisory...') # Call the function to start waiting while True: # Get the current state of the document doc = db_firestore.collection("InteractiveReportRequests").document(uid).get() # Check if 'Processed' field equals 1 if doc.to_dict().get('Processed') == 1: print("Document has been processed!") break else: print("Waiting for advisory to be ready...") time.sleep(10) # Wait for 5 seconds before checking again #time.sleep(120) # Example usage bucket_name = "farmbase-b2f7e.appspot.com" storage_client = storage.Client() # Get a reference to the storage service bucket = storage_client.get_bucket(bucket_name) cred = service_account.Credentials.from_service_account_file("servicekey.json") object_path = 'PaidMonitoredFields/' + uid + '/interactive_report.html' blob = bucket.blob(object_path) blob.download_to_filename((uid + '/' + fieldid + '/interactive_report.html')) url = f"http://localhost:{PORT}/{uid}/{fieldid}/interactive_report.html?FieldID={fieldid}&lang={lang}" #url = get_firebase_storage_url(object_path) # url = url + "?lang=" + lang mediaUrl = capture_full_page_screenshot(url, uid, fieldid) mediaUrl = upload_file_to_storage((uid + '/' + fieldid + '/collage.jpg'), ('PaidMonitoredFields/' + uid + '/interactive_report.jpg'), expire_minutes=1000) #upload_file_to_storage(uid, fieldid, "default", (uid + '/' + fieldid + '/collage.jpg'), ('PaidMonitoredFields/' + uid + '/interactive_report.jpg'), expire_minutes=15) # mediaUrl = get_firebase_storage_url(('PaidMonitoredFields/' + uid + '/interactive_report.jpg')) stop_server() return mediaUrl except: print(traceback.format_exc()) stop_server() return mediaUrl def send_whatsapp(mediaURL, paramsArray, lang, uid, fieldid, phoneNumber, whitelabel, imagedate): import requests import traceback try: if uid == "TCXcp5VIsfhHZrh0nm2VsgBtcGy2": phoneNumber = None # paramsArray[0] = userName # phoneNumber = "6366026267" phoneNumber = phoneNumber.split(",") phoneNumber = phoneNumber[0] phoneNumber = str(phoneNumber) userName = paramsArray[0] if userName is None: if lang == "en": userName = "Farmer" else: # for hi & else userName = "किसान" if lang == "hi": campaignName = "Main Campaign Hindi May 2023" elif lang == "kn": campaignName = "Main Campaign Kannada" elif lang == "mr": campaignName = "Main Campaign Marathi" elif lang == "gu": campaignName = "Main Campaign Gujarati" elif lang == "pa": campaignName = "Main Campaign Punjabi" elif lang == "ta": campaignName = "Main Campaign Tamil" elif lang == "te": campaignName = "Main Campaign Telugu" elif lang == "ml": campaignName = "Main Campaign Malayalam" else: campaignName = "Main Campaign English May 2023" # campaignName = 'Main Campaign 4' url = "https://backend.aisensy.com/campaign/t1/api" pg_campaign_names = { 'hi': 'Main Campaign Hindi 1', 'en': 'Main Campaign English 1', 'mr': 'Main Campaign Marathi 1' } pg_api_key = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjY2MzFkMGUwODRkMjA1MWZkMWQzZDNlMCIsIm5hbWUiOiJURUNIIFZJU0lUIElUIiwiYXBwTmFtZSI6IkFpU2Vuc3kiLCJjbGllbnRJZCI6IjYzODk3MDRlMDYyYTQ3NzIwZWMzYjc1ZiIsImFjdGl2ZVBsYW4iOiJOT05FIiwiaWF0IjoxNzE0NTQwNzY4fQ.FK-8s6AqOWBN3Ip4a4rEsMiWWa68JMBIsycNe_s8TvI' #pg_api_key = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjY1NzQ2ZmEwYzg0NjBjNTk5NjhhNzcwYSIsIm5hbWUiOiJUZWNoIFZpc2l0IElUIFB2dC4gTHRkLiIsImFwcE5hbWUiOiJBaVNlbnN5IiwiY2xpZW50SWQiOiI2Mzg5NzA0ZTA2MmE0NzcyMGVjM2I3NWYiLCJhY3RpdmVQbGFuIjoiTk9ORSIsImlhdCI6MTcwMjEyOTU2OH0.YxhKlTJWEi2L0-QKdhEtT4WWzo5WeFbW-Jmko7SI2AI' farmonaut_api_key = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjYzODk3MDRlMDYyYTQ3NzIwZWMzYjc2NCIsIm5hbWUiOiJGYXJtb25hdXQgVGVjaG5vbG9naWVzIFB2dC4gTHRkLiIsImFwcE5hbWUiOiJBaVNlbnN5IiwiY2xpZW50SWQiOiI2Mzg5NzA0ZTA2MmE0NzcyMGVjM2I3NWYiLCJhY3RpdmVQbGFuIjoiQkFTSUNfVFJJQUwiLCJpYXQiOjE2Njk5NTE1NjZ9.sm6bFj27KhdOPL7t_aAbLIJW7Il4pMSnLIQ7unVzenM' if whitelabel == None or "farmonaut" in whitelabel: #campaignName = 'Main Campaign English May 2023' api_key = farmonaut_api_key mediaURL = generate_jeevnai_whatsapp_image(mediaURL, uid, fieldid, lang, imagedate) paramsArray[4] = "https://farmonaut.com/app_redirect" elif whitelabel == "sat.precisiongrow.in": campaignName = pg_campaign_names.get('hi','Main Campaign English 1') api_key = pg_api_key myobj = { "apiKey": api_key, "campaignName": campaignName, "destination": phoneNumber, "userName": userName, "media": {"url": mediaURL, "filename": "Analysis.png"}, "templateParams": paramsArray, } print(myobj) x = requests.post(url, json=myobj) print(x.text) except: print("sendWhatsappErr " + traceback.format_exc())