261 lines
7.8 KiB
Python
261 lines
7.8 KiB
Python
from fastapi import FastAPI, HTTPException, Body, File, UploadFile, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
import requests
|
|
from requests.auth import HTTPBasicAuth
|
|
import os
|
|
from typing import List, Optional, Dict, Any
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
import logging
|
|
|
|
# Configure logging to file
|
|
logging.basicConfig(
|
|
filename='debug.log',
|
|
level=logging.DEBUG,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI()
|
|
|
|
# Add CORS middleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # In production, replace with specific origin
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
@app.middleware("http")
|
|
async def log_requests(request: Request, call_next):
|
|
logger.info(f"Incoming request: {request.method} {request.url}")
|
|
try:
|
|
response = await call_next(request)
|
|
logger.info(f"Response status: {response.status_code}")
|
|
return response
|
|
except Exception as e:
|
|
logger.error(f"Request failed: {str(e)}")
|
|
raise e
|
|
|
|
class LoginRequest(BaseModel):
|
|
url: str
|
|
consumer_key: str
|
|
consumer_secret: str
|
|
|
|
class Product(BaseModel):
|
|
name: str
|
|
type: str = "simple"
|
|
regular_price: str = ""
|
|
description: str = ""
|
|
short_description: str = ""
|
|
categories: List[Dict[str, Any]] = []
|
|
images: List[Dict[str, Any]] = []
|
|
meta_data: List[Dict[str, Any]] = []
|
|
manage_stock: bool = False
|
|
stock_quantity: Optional[int] = None
|
|
stock_status: str = "instock"
|
|
|
|
@app.get("/")
|
|
def read_root():
|
|
return {"message": "WooCommerce Inventory API is running"}
|
|
|
|
@app.post("/login")
|
|
def login(request: LoginRequest):
|
|
logger.info(f"Login attempt for URL: {request.url}")
|
|
# Verify credentials by making a lightweight call to the API
|
|
try:
|
|
# Ensure URL has protocol
|
|
base_url = request.url.strip()
|
|
if not base_url.startswith('http'):
|
|
base_url = f'https://{base_url}'
|
|
|
|
url = f"{base_url.rstrip('/')}/wp-json/wc/v3/system_status"
|
|
logger.debug(f"Connecting to: {url}")
|
|
|
|
# Add User-Agent to avoid blocking by security plugins
|
|
headers = {
|
|
'User-Agent': 'WooCommerce-Inventory-App/1.0'
|
|
}
|
|
|
|
response = requests.get(
|
|
url,
|
|
auth=HTTPBasicAuth(request.consumer_key, request.consumer_secret),
|
|
headers=headers
|
|
)
|
|
logger.info(f"WooCommerce API Response: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
return {"status": "success", "message": "Login successful"}
|
|
else:
|
|
logger.warning(f"Login failed. Status: {response.status_code}, Body: {response.text}")
|
|
# Try to return a helpful error message
|
|
try:
|
|
error_detail = response.json().get('message', response.text)
|
|
except:
|
|
error_detail = response.text
|
|
raise HTTPException(status_code=401, detail=f"WooCommerce Error: {error_detail}")
|
|
|
|
except HTTPException as he:
|
|
raise he
|
|
except Exception as e:
|
|
logger.error(f"Login exception: {str(e)}")
|
|
raise HTTPException(status_code=400, detail=f"Connection failed: {str(e)}")
|
|
|
|
@app.get("/products")
|
|
def get_products(
|
|
url: str,
|
|
consumer_key: str,
|
|
consumer_secret: str,
|
|
page: int = 1,
|
|
per_page: int = 20,
|
|
search: Optional[str] = None
|
|
):
|
|
try:
|
|
api_url = f"{url.rstrip('/')}/wp-json/wc/v3/products"
|
|
params = {
|
|
"page": page,
|
|
"per_page": per_page
|
|
}
|
|
if search:
|
|
params["search"] = search
|
|
|
|
response = requests.get(
|
|
api_url,
|
|
auth=HTTPBasicAuth(consumer_key, consumer_secret),
|
|
params=params
|
|
)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise HTTPException(status_code=response.status_code, detail=response.text)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/products/{product_id}")
|
|
def get_product(
|
|
product_id: int,
|
|
url: str,
|
|
consumer_key: str,
|
|
consumer_secret: str
|
|
):
|
|
try:
|
|
api_url = f"{url.rstrip('/')}/wp-json/wc/v3/products/{product_id}"
|
|
response = requests.get(
|
|
api_url,
|
|
auth=HTTPBasicAuth(consumer_key, consumer_secret)
|
|
)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise HTTPException(status_code=response.status_code, detail=response.text)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.post("/products")
|
|
def create_product(
|
|
url: str,
|
|
consumer_key: str,
|
|
consumer_secret: str,
|
|
product: Dict[str, Any] = Body(...)
|
|
):
|
|
try:
|
|
api_url = f"{url.rstrip('/')}/wp-json/wc/v3/products"
|
|
response = requests.post(
|
|
api_url,
|
|
auth=HTTPBasicAuth(consumer_key, consumer_secret),
|
|
json=product
|
|
)
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise HTTPException(status_code=response.status_code, detail=response.text)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.put("/products/{product_id}")
|
|
def update_product(
|
|
product_id: int,
|
|
url: str,
|
|
consumer_key: str,
|
|
consumer_secret: str,
|
|
product: Dict[str, Any] = Body(...)
|
|
):
|
|
try:
|
|
api_url = f"{url.rstrip('/')}/wp-json/wc/v3/products/{product_id}"
|
|
response = requests.put(
|
|
api_url,
|
|
auth=HTTPBasicAuth(consumer_key, consumer_secret),
|
|
json=product
|
|
)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise HTTPException(status_code=response.status_code, detail=response.text)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.delete("/products/{product_id}")
|
|
def delete_product(
|
|
product_id: int,
|
|
url: str,
|
|
consumer_key: str,
|
|
consumer_secret: str,
|
|
force: bool = True
|
|
):
|
|
try:
|
|
api_url = f"{url.rstrip('/')}/wp-json/wc/v3/products/{product_id}"
|
|
params = {"force": str(force).lower()}
|
|
response = requests.delete(
|
|
api_url,
|
|
auth=HTTPBasicAuth(consumer_key, consumer_secret),
|
|
params=params
|
|
)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise HTTPException(status_code=response.status_code, detail=response.text)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.post("/upload")
|
|
async def upload_image(
|
|
url: str,
|
|
consumer_key: str,
|
|
consumer_secret: str,
|
|
file: UploadFile = File(...)
|
|
):
|
|
# Uploading images to WooCommerce via API is a bit different, typically POST to /wp-json/wp/v2/media
|
|
try:
|
|
api_url = f"{url.rstrip('/')}/wp-json/wp/v2/media"
|
|
|
|
# Read file content
|
|
content = await file.read()
|
|
|
|
headers = {
|
|
"Content-Disposition": f"attachment; filename={file.filename}",
|
|
"Content-Type": file.content_type
|
|
}
|
|
|
|
response = requests.post(
|
|
api_url,
|
|
auth=HTTPBasicAuth(consumer_key, consumer_secret),
|
|
data=content,
|
|
headers=headers
|
|
)
|
|
|
|
if response.status_code == 201:
|
|
return response.json()
|
|
else:
|
|
raise HTTPException(status_code=response.status_code, detail=response.text)
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|