Flyers / app.py
SelimGilon's picture
colruyt
097ed17
Raw History Blame Contribute Delete
27.5 kB
import urllib
import pandas as pd
import numpy as np
from pdf2image import convert_from_path
import cv2
import pytesseract
import requests
import json
import re
from roboflow import Roboflow
import gradio as gr
from ultralytics import YOLO
import pypdfium2 as pdfium
search_google = True
GOOGLE_API_KEY = "AIzaSyCD5eIu3N2uD6Vf5GE0VLBpkPsJ3mEZgA0"
# GOOGLE_CSE_ID = "b536303d151b94476"
# GOOGLE_API_KEY = "AIzaSyBUvPKAZa2TddKYDJI9t0ac2A_JJqdgWHo"
# GOOGLE_CSE_ID = "f3446c60c52cd4dc2"
# GOOGLE_API_KEY = "AIzaSyCoBYZK4khX_W5eW-AqEXKOclUIer6CiC8"
# GOOGLE_CSE_ID = "c3ba01991c1a14c7d"
GOOGLE_CSE_ID = "b536303d151b94476"
mode = 'local'
# mode = 'rbf'
if mode == 'local':
model = YOLO('best-flyers-yolov8-1000img-v10.pt')
else:
rf = Roboflow(api_key="TN586SQsdwjUUvMN3H8e")
project = rf.workspace().project("smartflyers")
model = project.version(9).model
# def process(selected_shop):
# print(selected_shop)
# return selected_shop
hotspots_sku_links = []
shop = None
shops = {
'Colruyt': {
'url': 'https://www.colruyt.be/fr/producten?searchTerm=e',
'regex': r'',
'multi_items': False,
},
# Add more shops and their respective information here
}
regex_price = r'(\d+\.?\d*) ?([€$])|([€$]) ?(\d+\.?\d*)'
def pdf_to_image(pdf):
pdf = pdfium.PdfDocument(pdf)
n_pages = len(pdf) # get the number of pages in the document
print(f"Number of pages: {n_pages}")
assert n_pages > 0, "The PDF document is empty"
assert n_pages == 1, "The PDF document has more than one page"
page = pdf.get_page(0)
bitmap = page.render(
scale=5, # 72dpi resolution
rotation=0, # no additional rotation
# ... further rendering options
)
pil_image = bitmap.to_pil()
pil_image.save('test.png')
width_img, height_img = pil_image.size
print("width image", width_img, "height image", height_img)
width_pdf, height_pdf = page.get_size()
print("width pdf ", width_pdf, "height pdf ", height_pdf)
#assert int(width_pdf) == width_img, "The width of the image is not the same as the width of the page"
#assert int(height_pdf) == height_img, "The height of the image is not the same as the height of the page"
return pil_image, width_pdf, height_pdf, page
def app(pdf, selected_shop):
if pdf is None:
raise gr.Error("No file uploaded. Please upload PDF.")
# Check if the uploaded file is a PDF
if selected_shop is None:
raise gr.Error("Please select a shop.")
try:
image_annotated = section(pdf)
pass
except Exception as e:
return gr.Error(f"An error occurred: {str(e)}")
return image_annotated, hotspots_sku_links[selected_shop]
def build_request(query, start=1, num=10, date_restrict=None, **params):
request = {
'key': GOOGLE_API_KEY,
'q': query,
'cx': GOOGLE_CSE_ID,
# 'start': start,
# 'num': num,
# 'dateRestrict': date_restrict,
}
request.update(params)
return request
def make_request(request):
response = requests.get('https://www.googleapis.com/customsearch/v1', params=request)
# print(f"Status Code: {response.status_code}") # Print status code for debugging
if response.status_code != 200:
raise Exception("Google Custom Search request failed with status code {}".format(response.status_code))
search_results = response.json()
# print(f"Response JSON: {search_results}") # Print raw response for debugging
if 'items' in search_results and len(search_results['items']) > 0:
# find the first result with a 'linl' containing 'www.colruyt' and return it
first_result_url = None
for item in search_results['items']:
if 'link' in item and 'www.colruyt' in item['link']:
first_result_url = item['link']
break
#first_result_url = search_results['items'][0]['link']
return first_result_url
else:
return None
with gr.Blocks() as demo:
with gr.Row():
selected_shop = gr.Dropdown(shops.keys(), label="Shop")
# with gr.Row():
# input_type = gr.Radio(["PDF", "Image"], label="Input Type", value="PDF")
confidence_threshold = gr.Slider(minimum=0.1, maximum=1.0, step=0.1, label="Confidence Threshold", value=0.5)
with gr.Row():
pdf_input = gr.File(label="Input page (PDF)", file_types=["pdf"])
img_output = gr.AnnotatedImage(label="Annotated page (image)")
section_btn = gr.Button("Identify Hotspots")
# selected_section = gr.Textbox(label="Selected Hotspot")
selected_section = gr.Markdown(label="Selected Hotspot")
product_dataframe = gr.Dataframe(label="Products Found")
def get_yolo_predictions(image, confidence_threshold):
# Use Roboflow API to get predictions
predictions = model.predict(image, confidence=confidence_threshold, overlap=10).json()
# print("predictions", predictions)
return predictions
def get_yolo_predictions_local(image, confidence_threshold):
# Perform object detection on an image using the model
predictions = model(image, conf=confidence_threshold)
# print("predictions", predictions)
return predictions
def ocr(sub_image):
# Use OCR to get text from image
grey = cv2.cvtColor(sub_image, cv2.COLOR_BGR2GRAY)
text = pytesseract.image_to_string(grey, lang='eng')
return text
def section(pdf, selected_shop, confidence_threshold):
hostpots = []
# texts = []
sku = []
product_table_data = []
image, width_pdf, height_pdf, first_page = pdf_to_image(pdf)
image_name ="test.jpg"
image.save(image_name)
# # Load a text page helper
textpage = first_page.get_textpage()
text_all = textpage.get_text_range()
print("text_all", text_all)
#print("selected_shop:", selected_shop)
search_url = shops[selected_shop]['url']
ref_number_regex = shops[selected_shop]['regex']
# print("search_url:", search_url)
brands = shops[selected_shop]['brands'] if 'brands' in shops[selected_shop] else None
if brands:
brands = [brand.upper() for brand in brands]
if mode == 'rbf':
predictions = get_yolo_predictions(image)
else:
# local inference
results = get_yolo_predictions_local(image, confidence_threshold)
for result in results:
coordinates = result.boxes.xyxyn
coordinates = coordinates.to("cpu")
coordinates = coordinates.numpy()
print("FOUND {} objects".format(len(coordinates)))
# print("coordinates", coordinates)
for idx, coord in enumerate(coordinates):
print("==============")
# crop the image to the bounding box
# expansion_factor = 0.01
x = coord[0]
y = coord[1]
w = coord[2] - coord[0]
h = coord[3] - coord[1]
# sub_image = image[y:y + h, x:x + w]
print("x, y, w, h", x, y, w, h)
print('width_pdf', width_pdf)
print('height_pdf', height_pdf)
enlarge_factor = 0.0
pdf_x = x * width_pdf
pdf_y = y * height_pdf + enlarge_factor * height_pdf
pdf_w = w * width_pdf + enlarge_factor * width_pdf
pdf_h = h * height_pdf
print("PDF", pdf_x, pdf_y, pdf_w, pdf_h)
img_x = int(x * image.width)
img_y = int(y * image.height)
img_w = int(w * image.width)
img_h = int(h * image.height)
print("IMG", img_x, img_y, img_w, img_h)
text = textpage.get_text_bounded(left=pdf_x,
bottom=height_pdf - pdf_y - pdf_h,
right=pdf_x + pdf_w,
top=height_pdf - pdf_y)
print("TEXT", text)
if search_google:
# Use Google Search API to find items with the SKU
# keep only the first 2 lines of text to search
lines = text.split('\n')
# delete the \r in all lines
lines = [line.replace('\r', '') for line in lines]
# remove the blank spaces at the beginning and end of each line
lines = [line.strip() for line in lines]
text = '\n'.join(lines[:2])
# if the text is less than 3 words, add the next line
if len(text.split(' ')) < 2:
text = '\n'.join(lines[:3])
product_name = text
to_seach = str(selected_shop) + ' ' + text + str(selected_shop)
print("TO SEARCH: ", to_seach)
request = build_request(to_seach)
search_query = make_request(request)
print(f"First Result URL: {search_query}")
# search_query = "https://www.google.com/search?q="
sku.append(product_name)
# if search_query not in hotspots_sku_links:
hotspots_sku_links.append([{'sku': product_name, 'url': search_query, 'product_name': product_name}])
hostpots.append(((img_x, img_y, (img_x + img_w), (img_h+img_y)), str(product_name)))
print("----- ADDED ------")
print("hotspots_sku_links: ", hotspots_sku_links)
print("skus: ", sku)
for data in hotspots_sku_links:
for item in data:
product_table_data.append({
'Product Name': item['product_name'] if item['product_name'] else "Not Found",
'URL': item['url']
})
# Create a DataFrame
product_dataframe_data = pd.DataFrame(product_table_data)
# flattern the list hotspots_sku_links
# hotspots_sku_links = [item for sublist in hotspots_sku_links for item in sublist]
print("hotspots_sku_links: ", hotspots_sku_links)
print('length of hotspots_sku_links: ', len(hotspots_sku_links))
assert len(hotspots_sku_links) == len(sku)
return (image, hostpots), product_dataframe_data
section_btn.click(section, [pdf_input, selected_shop, confidence_threshold], [img_output, product_dataframe])
def select_section(evt: gr.SelectData):
hotspot_data = hotspots_sku_links[evt.index]
markdown_output = ""
for data in hotspot_data:
url = data['url']
product_name = data['product_name'] if data['product_name'] else "No product name found"
if product_name != "No product name found":
markdown_output += f"[{product_name}]({url})\n\n"
else:
markdown_output += f"[SKU: {data['sku']}]({url})\n\n"
return markdown_output
img_output.select(select_section, None, selected_section)
demo.queue(max_size=99).launch(debug=True)
# import urllib
# import pandas as pd
# import numpy as np
# from pdf2image import convert_from_path
# import cv2
# import pytesseract
# import requests
# import json
# import re
# from roboflow import Roboflow
# import gradio as gr
# from ultralytics import YOLO
#
#
#
#
# with gr.Blocks() as demo:
# search_google = False
# GOOGLE_API_KEY = "AIzaSyCD5eIu3N2uD6Vf5GE0VLBpkPsJ3mEZgA0"
# GOOGLE_CSE_ID = "b536303d151b94476"
# # GOOGLE_API_KEY = "AIzaSyBUvPKAZa2TddKYDJI9t0ac2A_JJqdgWHo"
# # GOOGLE_CSE_ID = "f3446c60c52cd4dc2"
# # GOOGLE_API_KEY = "AIzaSyCoBYZK4khX_W5eW-AqEXKOclUIer6CiC8"
# # GOOGLE_CSE_ID = "c3ba01991c1a14c7d"
# mode = 'local'
# # mode = 'rbf'
#
# if mode == 'local':
# model = YOLO('best-flyers-yolov8-1000img-v10.pt')
# else:
# rf = Roboflow(api_key="TN586SQsdwjUUvMN3H8e")
# project = rf.workspace().project("smartflyers")
# model = project.version(9).model
#
# # def process(selected_shop):
# # print(selected_shop)
# # return selected_shop
# hotspots_sku_links = []
# shop = None
# shops = {
# 'Latulippe': {
# 'url': 'https://latulippe.com/fr/recherche/?keywords=',
# 'regex': r'N° (M\d+)',
# 'multi_items': False,
# },
# 'Decathlon': {
# 'url': 'https://www.decathlon.fr/search?Ntt=',
# 'regex': r'Réf\. ?: ?([\d, ]+)',
# 'multi_items': True,
# 'brands': [
# "Allsix", "Aptonia", "Artengo", "Atorka", "B'Twin", "Caperlan", "Domyos", "Dreamscape",
# "FLX", "Forclaz", "Fouganza", "Geologic", "Geonaute", "Inesis", "Imviso", "Inovik",
# "Itiwit", "KOODZA", "KOROK", "Kalenji", "Kimjaly", "Kipsta", "Kuikma", "Nabaiji",
# "Newfeel", "Offload", "Olaian", "Opfeel", "Orao", "Oroks", "Outshock", "Oxelo",
# "Perfly", "Pongori", "Quechua", "Rockrider", "Sandever", "Simond", "Solognac",
# "Subea", "Tarmak", "Triban", "Tribord", "Urball", "Van Rysel", "Wed'Ze", "Decathlon"
# ]
# },
# 'Canadian Tire': {
# 'url': 'https://www.canadiantire.ca/en/search-results.html?q=',
# 'regex': r'(\d{2,3}-\d{4}-\d(?:/\d{4}-\d)?)',
# 'multi_items': True,
# },
# 'Walmart': {
# # 'url': 'https://www.walmart.ca/en/search?q=',
# 'url': '',
# 'regex': r'N°(\d+)',
# 'multi_items': False,
# },
# 'Colruyt': {
# 'url': 'https://www.colruyt.be/fr/producten?searchTerm=e',
# 'regex': r'',
# 'multi_items': False,
# },
# # Add more shops and their respective information here
# }
#
# regex_price = r'(\d+\.?\d*) ?([€$])|([€$]) ?(\d+\.?\d*)'
#
#
# # def app(img, selected_shop):
# # if img is None:
# # raise gr.Error("No file uploaded. Please upload an image.")
# #
# # # Check if the uploaded file is a PDF
# # if selected_shop is None:
# # raise gr.Error("Please select a shop.")
# #
# # try:
# # image_annotated = section(img)
# # pass
# # except Exception as e:
# # return gr.Error(f"An error occurred: {str(e)}")
# #
# # return image_annotated, hotspots_sku_links[selected_shop]
#
# def build_request(query, start=1, num=10, date_restrict=None, **params):
# request = {
# 'key': GOOGLE_API_KEY,
# 'q': query,
# 'cx': GOOGLE_CSE_ID,
# # 'start': start,
# # 'num': num,
# # 'dateRestrict': date_restrict,
# }
# request.update(params)
# return request
#
#
# def make_request(request):
# response = requests.get('https://www.googleapis.com/customsearch/v1', params=request)
# # print(f"Status Code: {response.status_code}") # Print status code for debugging
# if response.status_code != 200:
# raise Exception("Google Custom Search request failed with status code {}".format(response.status_code))
#
# search_results = response.json()
# # print(f"Response JSON: {search_results}") # Print raw response for debugging
# if 'items' in search_results and len(search_results['items']) > 0:
# first_result_url = search_results['items'][0]['link']
# return first_result_url
# else:
# return None
#
#
# with gr.Row():
# selected_shop = gr.Dropdown(shops.keys(), label="Shop")
# # with gr.Row():
# # input_type = gr.Radio(["PDF", "Image"], label="Input Type", value="PDF")
# confidence_threshold = gr.Slider(minimum=0.1, maximum=1.0, step=0.1, label="Confidence Threshold", value=0.5)
#
# with gr.Row():
# img_input = gr.Image(label="Input page (image)", sources=["upload", "webcam"])
# img_output = gr.AnnotatedImage(label="Annotated page (image)")
#
# section_btn = gr.Button("Identify Hotspots")
# # selected_section = gr.Textbox(label="Selected Hotspot")
# selected_section = gr.Markdown(label="Selected Hotspot")
# product_dataframe = gr.Dataframe(label="Products Found")
#
#
# def get_yolo_predictions(image, confidence_threshold):
# # Use Roboflow API to get predictions
# predictions = model.predict(image, confidence=confidence_threshold, overlap=10).json()
# # print("predictions", predictions)
# return predictions
#
# def get_yolo_predictions_local(image, confidence_threshold):
# # Perform object detection on an image using the model
# predictions = model(image, conf=confidence_threshold)
# # print("predictions", predictions)
# return predictions
#
# def ocr(sub_image):
# # Use OCR to get text from image
# grey = cv2.cvtColor(sub_image, cv2.COLOR_BGR2GRAY)
# text = pytesseract.image_to_string(grey, lang='eng')
# return text
#
# def section(img, selected_shop, confidence_threshold):
# hostpots = []
# sku = []
# product_table_data = []
#
# global hotspots_sku_links
# hotspots_sku_links = []
#
# #print("selected_shop:", selected_shop)
# search_url = shops[selected_shop]['url']
# ref_number_regex = shops[selected_shop]['regex']
# # print("search_url:", search_url)
# brands = shops[selected_shop]['brands'] if 'brands' in shops[selected_shop] else None
#
# if brands:
# brands = [brand.upper() for brand in brands]
#
# if mode == 'rbf':
# predictions = get_yolo_predictions(img)
# else:
# # local inference
# results = get_yolo_predictions_local(img, confidence_threshold)
# for result in results:
# coordinates = result.boxes.xyxy
# coordinates = coordinates.to("cpu")
# coordinates = coordinates.numpy()
# print("FOUND {} objects".format(len(coordinates)))
# for idx, coord in enumerate(coordinates):
# # crop the image to the bounding box
# x = int(coord[0])
# y = int(coord[1])
# w = int(coord[2] - coord[0])
# h = int(coord[3] - coord[1])
# sub_image = img[y:y + h, x:x + w]
# if shops[selected_shop]['url'] is None or shops[selected_shop]['url'] == "":
# print("No search url for this shop")
# # just append the hotspot to hostpots
# hostpots.append(((x, y, x + w, y + h), str(idx)))
# continue
#
# # Extract text using OCR
# text = ocr(sub_image)
# print("text: ", text)
#
# if selected_shop == "Latulippe":
# # if there is a 'i' between a "M" and a number, it is a sku. Replace the 'i' with a '1'
# text = re.sub(r'Mi(\d+)', r'M1\1', text)
#
# # for object_number, text in enumerate(texts, start=1):
# #print("object", object_number)
# print("text -----", text)
# ref_numbers = []
# if selected_shop == "Decathlon":
# full_match = re.search(ref_number_regex, text)
# if full_match:
# print("FULL MATCH: ", full_match)
# # Extracting all numbers from the captured sequence
# numbers_string = full_match.group(1)
# ref_numbers = re.findall(r'\d+', numbers_string)
# print("REF NUMBERS: ", ref_numbers)
# else:
# ref_numbers = re.findall(ref_number_regex, text)
# # print("REF NUMBERS: ", ref_numbers)
#
# if selected_shop == "Canadian Tire":
# temp_ref_numbers = ref_numbers
# ref_numbers = []
# split = False
# for ref_sku in temp_ref_numbers:
# if '/' in ref_sku:
# split = True
# sku_split = ref_sku.split('/')
# sku_last = sku_split[-1]
# length = len(sku_last)
# base_number = sku_split[0][:-length]
# ref_numbers.append(sku_split[0])
# # base number + sku for each sku in sku_split except the first one
# for i in range(1, len(sku_split)):
# ref_numbers.append(base_number + sku_split[i])
#
# if split == False:
# ref_numbers = temp_ref_numbers
#
# matches = re.findall(regex_price, text)
# prices = []
# for match in matches:
# if match[0]: # If the first group is not empty, the format is "price currency"
# prices.append(match[0] + match[1])
# else: # Otherwise, the format is "currency price"
# prices.append(match[2] + match[3])
# # price = prices[0] if len(prices) > 0 else ""
# #print("PRICES: ", prices)
#
# if brands:
# # print("BRANDS: yes")
# # Extracting the product name
# lines = text.split('\n')
# product_name = ""
# found_brand = False
# name_started = False
# for line in lines:
# if any(brand in line for brand in brands):
# found_brand = True
# #print("FOUND BRAND", line)
# continue
# if found_brand:
# #print('current line:', line)
# if line.strip() == "": # Skip blank lines
# continue
# if name_started and (line[0].isupper() or any(char.isdigit() for char in line)):
# break
# product_name += line.strip() + " "
# name_started = True
# product_name = product_name.strip()
#
# else:
# # Regex to match sequences of uppercase letters
# uppercase_words = re.findall(r'\b[A-Z]+\b', text)
#
# # Joining the extracted words with a space
# product_name = ' '.join(uppercase_words)
#
# #print("PRODUCT NAME: ", product_name)
#
# for ref in ref_numbers:
# print("ref", ref)
# if search_google:
# # Use Google Search API to find items with the SKU
# to_seach = str(selected_shop) + ' ' + str(ref)
# # print("to_search: ", to_seach)
# request = build_request(to_seach)
# search_query = make_request(request)
# # print(f"First Result URL: {search_query}")
# if search_query is None:
# search_query = search_url + urllib.parse.quote(ref)
# else:
# search_query = search_url + urllib.parse.quote(ref)
# if ref not in sku:
# sku.append(ref)
# # if search_query not in hotspots_sku_links:
# hotspots_sku_links.append([{'sku': ref, 'url': search_query, 'prices': prices, 'product_name': product_name}])
# hostpots.append(((int(coord[0]), int(coord[1]), int(coord[2]), int(coord[3])), str(ref)))
#
# for data in hotspots_sku_links:
# for item in data:
# product_table_data.append({
# 'Reference Number': item['sku'],
# 'Product Name': item['product_name'] if item['product_name'] else "Not Found",
# # 'Price': ", ".join(item['prices']) if item['prices'] else " ",
# 'URL': item['url']
# })
#
# # Create a DataFrame
# product_dataframe_data = pd.DataFrame(product_table_data)
#
# # flattern the list hotspots_sku_links
# # hotspots_sku_links = [item for sublist in hotspots_sku_links for item in sublist]
# print("hotspots_sku_links: ", hotspots_sku_links)
# print('length of hotspots_sku_links: ', len(hotspots_sku_links))
# print('shape of hotspots_sku_links: ', np.shape(hotspots_sku_links))
# print("skus: ", sku)
# print('length of skus: ', len(sku))
# assert len(hotspots_sku_links) == len(sku)
# return (img, hostpots), product_dataframe_data
#
#
# section_btn.click(section, [img_input, selected_shop, confidence_threshold], [img_output, product_dataframe])
#
#
#
# def select_section(evt: gr.SelectData):
# hotspot_data = hotspots_sku_links[evt.index]
# markdown_output = ""
# for data in hotspot_data:
# url = data['url']
# price_str = ", ".join(data['prices']) if data['prices'] else "No price found"
# product_name = data['product_name'] if data['product_name'] else "No product name found"
# if product_name != "No product name found":
# if price_str != "No price found":
# markdown_output += f"[{product_name}]({url}): {price_str}\n\n"
# else:
# markdown_output += f"[{product_name}]({url})\n\n"
# else:
# if price_str != "No price found":
# markdown_output += f"[SKU: {data['sku']}]({url}): {price_str}\n\n"
# else:
# markdown_output += f"[SKU: {data['sku']}]({url})\n\n"
# return markdown_output
#
# img_output.select(select_section, None, selected_section)
#
# # demo.launch(share=False)
# demo.queue(max_size=99).launch(debug=True)