Open dinhopereira opened 5 months ago
response = wcapi.post("products", data)
response is a empty array
My server is a litepeed and the log below:
SERVER-IP - - [05/Apr/2024:11:32:34 +0000] "POST /wp-admin/admin-ajax.php?action=async_litespeed&nonce=b07b1c4d32&litespeed_type=imgoptm HTTP/1.1" 200 696 "-" "WordPress/6.5; https://www.mydomain.com"
MYIP - - [05/Apr/2024:11:32:33 +0000] "GET /wp-json/wc/v3/products?consumer_key=ck_***************************6feb6c7ecb9c5&consumer_secret=cs_3e0****************************f07e7de19 HTTP/1.1" 200 1097 "-" "WooCommerce-Python-REST-API/3.0.0"
SERVER-IP - - [05/Apr/2024:11:32:33 +0000] "POST /wp-cron.php?doing_wp_cron=1712316753.5901479721069335937500 HTTP/1.1" 200 424 "-" "WordPress/6.5; https://www.mydomain.com"
MYIP - - [05/Apr/2024:11:32:32 +0000] "POST /wp-json/wc/v3/products?consumer_key=ck_***************************6feb6c7ecb9c5&consumer_secret=cs_3e0****************************f07e7de19 HTTP/1.1" 301 1089 "-" "WooCommerce-Python-REST-API/3.0.0"
Hope this helps, remeber that your CSV file MUST have all the headers equal to whatever info you want to update, for example if you want to update a products name (or add) you need to have a 'name' header in your csv file and etc..
import pandas as pd import logging import threading import time import tkinter as tk from tkinter import filedialog, Button, Frame, ttk from tkinter.scrolledtext import ScrolledText from woocommerce import API
WC_URL = 'insert website here' WC_CONSUMER_KEY = 'insert key here' WC_CONSUMER_SECRET = 'insert secret here'
root = None
class TextHandler(logging.Handler): def init(self, widget): super().init() self.widget = widget
def emit(self, record):
self.widget.config(state='normal')
self.widget.insert('end', self.format(record) + '\n')
self.widget.config(state='disabled')
self.widget.see('end')
def configure_wc_api(): return API( url=WC_URL, consumer_key=WC_CONSUMER_KEY, consumer_secret=WC_CONSUMER_SECRET, version="wc/v3", timeout=300 # Set a higher timeout (in seconds) )
def make_api_call(wcapi, method, endpoint, data=None, retries=1, delay=0.5): for attempt in range(retries): try: if method.lower() == 'get': response = wcapi.get(endpoint) elif method.lower() == 'post': response = wcapi.post(endpoint, data) elif method.lower() == 'put': response = wcapi.put(endpoint, data) else: raise ValueError(f"Unsupported HTTP method: {method}")
if response.status_code == 200:
logging.info(f"Connection to database Successful... Continuing")
return response
elif response.status_code == 201:
logging.info(f"Product created successfully.")
return response
elif response.status_code == 404:
logging.warning(f"WARNING - Product not found")
return None
else:
logging.error(
f"ERROR - API call failed with status {response.status_code} for {endpoint}:\n{response.text}")
time.sleep(delay)
except Exception as e:
logging.error(f"ERROR - API call failed with exception for {endpoint}:\n{str(e)}")
time.sleep(delay)
return None
def process_csv(wcapi, filename, progress_var, total_rows_label): start_time = time.time() df = pd.read_csv(filename, encoding='utf-8') total_rows = len(df)
batch_size = 10
batch = []
final_batch = False # Flag to track if it's the final batch
for i, row in df.iterrows():
sku = row.get('sku')
id_column = row.get('id')
if not sku and not id_column:
continue # Skip rows without SKU and ID
data = create_product_data(row)
if id_column:
data['id'] = id_column
response = make_api_call(wcapi, 'get', f"products/{id_column}")
if response and response.status_code == 200:
batch.append((data, "update"))
process_product(data, i, total_rows, "update")
else:
logging.warning(
f"WARNING - Product not found for {data.get('name', '')} (SKU: {data.get('sku', '')}, ID: {data.get('id', '')}), creating...")
batch.append((data, "create"))
process_product(data, i, total_rows, "create")
if sku:
response = make_api_call(wcapi, 'get', f"products?sku={sku}")
if response and response.status_code == 200:
product_data = response.json()
if product_data:
product_id = product_data[0]['id']
data['id'] = product_id
batch.append((data, "update"))
process_product(data, i, total_rows, "update")
else:
logging.warning(
f"WARNING - Product not found for {data.get('name', '')} (SKU: {data.get('sku', '')}, ID: {data.get('id', '')}), creating...")
batch.append((data, "create"))
process_product(data, i, total_rows, "create")
if len(batch) >= batch_size:
process_batch(wcapi, batch)
batch = []
# Calculate and display estimated time remaining
elapsed_time = time.time() - start_time
if elapsed_time > 0:
completed_percentage = (i + 1) / total_rows * 100
estimated_total_time = elapsed_time / (completed_percentage / 100)
estimated_remaining_time = estimated_total_time - elapsed_time
total_rows_label.config(
text=f"{completed_percentage:.2f}% complete, estimated time remaining: {int(estimated_remaining_time)} seconds")
if batch:
process_batch(wcapi, batch)
logging.info("All updates done.")
def create_product_data(row):
data = {
"sku": row['sku'],
"name": row['name'],
"regular_price": row['regular_price'],
"description": row.get('description', ''),
"stock_quantity": int(row['stock_quantity']) if row['stock_quantity'] else None,
"manage_stock": True,
"images": [],
"attributes": [],
"categories": [],
}
image_url = row.get('image_url')
if not pd.isna(image_url):
data["images"].append({"src": image_url})
data = {k: v if v is not None and (not isinstance(v, float) or not pd.isna(v)) else None for k, v in data.items()}
dimensions_data = {
"weight": row.get('weight', ''),
"length": row.get('length', ''),
"width": row.get('width', ''),
"height": row.get('height', ''),
}
for key, value in dimensions_data.items():
if pd.isna(value):
dimensions_data[key] = ''
if any(dimensions_data.values()):
data["dimensions"] = dimensions_data
return data
def process_product(data, row_index, total_rows, action): product_name = data.get('name', 'No Product Name Found in CSV') if action == "update": logging.info(f"Processing Row {row_index + 1} of {total_rows}") logging.info(f"Product '{product_name}' (SKU: {data.get('sku', '')}, ID: {data.get('id', '')}) found") logging.info(f"Updating Product...\n") elif action == "create": logging.info( f"Product '{product_name}' (SKU: {data.get('sku', '')}, ID: {data.get('id', '')}) not found, creating...") logging.info(f"Creating Product...\n")
def process_batch(wcapi, batch_data): batch_action = batch_data[0][1] logging.info(f"Processing {batch_action} batch...") products = [data[0] for data in batch_data] for item in products: logging.info( f"Product Sent to be {batch_action.capitalize()}d in WooCommerce, ID:{item.get('id', '')} / SKU: {item.get('sku', '')} / Product: {item.get('name', '')}") response = make_api_call(wcapi, 'post', "products/batch", {batch_action: products}) if response and response.status_code != 200: logging.error( f"ERROR - API call failed with status {response.status_code} for products/batch:\n{response.text}") else: logging.info(f"Waiting for the batch to be processed...") time.sleep(0.5) # Add a sleep time for waiting logging.info('Processing Done! Moving on to next batch..\n')
def select_and_process_csv(progress_var, total_rows_label): file_path = filedialog.askopenfilename(filetypes=[("CSV files", "*.csv")])
if file_path:
wcapi = configure_wc_api()
thread = threading.Thread(target=process_csv, args=(wcapi, file_path, progress_var, total_rows_label))
thread.start()
def main(): global root root = tk.Tk() root.title("Change your title name") root.geometry("600x600")
frame = Frame(root)
frame.pack(fill='both', expand=True)
progress_var = tk.DoubleVar()
progress_bar = ttk.Progressbar(frame, orient="horizontal", length=300, variable=progress_var, mode='determinate')
progress_bar.pack(pady=10)
total_rows_label = tk.Label(frame, text="0% complete, estimated time remaining: calculating...")
total_rows_label.pack()
button = ttk.Button(frame, text="Select and Process CSV File",
command=lambda: select_and_process_csv(progress_var, total_rows_label))
button.pack(pady=10)
log_text = ScrolledText(frame, wrap='word', state='disabled', width=80, height=20, font=("Calibri", 12))
log_text.pack(fill='both', expand=True)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
handlers=[
logging.FileHandler('Product_Sync_Results.txt', mode='w'),
TextHandler(log_text)
]
)
root.mainloop()
if name == "main": main()
Since GitHub makes a mess with the code you can use this file and paste it on your IDE SynProducts.txt
I have been trying to use a simple code to post a product on WooCommerce from Python.
Im trying this code:
I have gotten the "200" status. Notthing was post on woocommerce.
After i tryed delete with the code:
print(wcapi.delete("products/108", params={"force": True}).json())
And deleted success. Where am I going wrong?
Tks