import ast, base64 import hmac import json import os, sys import requests from bs4 import BeautifulSoup from flask import Flask, request, jsonify from sqlalchemy import create_engine from sqlalchemy import text, inspect from sqlalchemy.exc import NoSuchTableError from urllib.parse import urlsplit app = Flask(__name__) ## Check requirements required_env_vars = ['DATABASE_URI', 'PRODUCT_API_TOKEN'] def check_env_variable(var_name): value = os.getenv(var_name) if value is None: print(f"Error: Environment variable '{var_name}' is missing.") sys.exit(1) # Exit the program with a non-zero status code return value for var in required_env_vars: check_env_variable(var) engine = create_engine(os.environ.get('DATABASE_URI')) ## Helper functions section def validate_token(provided_token): """Check if the provided Bearer token is valid""" if not provided_token or not provided_token.startswith("Bearer "): raise ValueError("Missing authorization Token.") provided_token = provided_token.split(" ")[1] expected_token = os.environ.get('PRODUCT_API_TOKEN') if not expected_token: raise ValueError("Server side authorization Token error.") if not hmac.compare_digest(provided_token, expected_token): raise ValueError("Authorization Token mismatch.") def validate_table_name(table_name): """Check if the table exists in the database.""" inspector = inspect(engine) tables = inspector.get_table_names() if table_name not in tables: raise ValueError(f"Table '{table_name}' does not exist in the database.") def validate_search_column(table_name, search_column): """Check if the column is a single character, capitalize it, and ensure it exists in the table.""" if not isinstance(search_column, str) or len(search_column) != 1 or not search_column.isalpha(): raise ValueError("Column value must be a single alphabetic character.") search_column = search_column.upper() # Check if the column exists in the specified table inspector = inspect(engine) columns = inspector.get_columns(table_name) column_names = [col['name'].upper() for col in columns] if search_column not in column_names: raise ValueError(f"Column '{search_column}' does not exist in table '{table_name}'.") return search_column def download_image(image_url, headers): response = requests.get(image_url, headers=headers) # Ensure the request was successful if response.status_code == 200: # Convert the image content to base64 image_base64 = base64.b64encode(response.content).decode('utf-8') return image_base64 return False ## Router section @app.route('/product/excel', methods=['POST']) def get_excel_data(): request_data = request.get_json() response = {} search_column = False search_string = False try: # Validate request token validate_token(request.headers.get('Authorization')) requested_article_number = request_data.pop('requested_article_number', None) requested_article_name = request_data.pop('requested_article_name', None) table_name = request_data.pop('table_name', None) if requested_article_number: search_column = request_data.get('article_number') search_string = requested_article_number elif requested_article_name: search_column = request_data.get('name') search_string = requested_article_name if not search_column or not search_string: raise ValueError("None of theres were provided: Article Number or Article Name") # Validate the table name validate_table_name(table_name) # Validate and capitalize the column, where we want to search, and checks if it exists in the table search_column = validate_search_column(table_name, search_column) with engine.connect() as db: query = text(f"SELECT * FROM {table_name} WHERE {search_column} = :search_string") result = db.execute(query, { "search_string": search_string, }) product = result.mappings().first() if product: for k, v in request_data.items(): if v and len(v) == 1 and v.isalpha(): response[k] = product[v] else: response[k] = False response['product_image'] = False return jsonify(response), 200 else: return jsonify({'message': 'Product not found in Excel table.'}), 404 except ValueError as e: return jsonify({"error": str(e)}), 400 except NoSuchTableError: return jsonify({"error": f"Table '{table_name}' does not exist."}), 400 except Exception as e: return jsonify({"error": f"An unexpected error occurred: '{e}'"}), 500 @app.route('/product/web', methods=['POST']) def get_website_data(): request_data = request.get_json() response = {} try: # Validate request token validate_token(request.headers.get('Authorization')) except ValueError as e: return jsonify({"error": str(e)}), 400 requested_article_number = request_data.pop('requested_article_number', None) requested_article_name = request_data.pop('requested_article_name', None) website_url = request_data.pop('website_url', None) if not website_url or not (requested_article_number or requested_article_name): return jsonify({"error": "URL, Article Number or Article Name is not provided."}), 400 if '{ArticleNumber}' in website_url: url = website_url.replace('{ArticleNumber}', requested_article_number) if '{Name}' in website_url: url = website_url.replace('{Name}', requested_article_name) headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' } website = requests.get(url, headers=headers) soup = BeautifulSoup(website.content, 'lxml') for k, v in request_data.items(): if v: v = json.loads(v) # We have to extract tag and remove it from the dictionary just to be sure, # that soap.find() gets it as first argument, otherwise it fails tag = v.pop('tag', None) next_sibling = v.pop('next_sibling', None) get_attribute = v.pop('get_attribute', None) splitter = v.pop('splitter', None) element = soup.find(tag, **v) if element: if next_sibling: result = element.find_next_sibling(next_sibling).get_text(strip=True) else: result = element.get_text(strip=True) if get_attribute: result = element.get(get_attribute) if splitter: # Convert str into a proper Python list type splitter = ast.literal_eval(splitter) splitting_char = splitter[0] requested_element = int(splitter[1]) result = result.split(splitting_char)[requested_element] if k == "product_image": if result.startswith("http"): response[k] = download_image(result, headers) continue if result.startswith("/"): url_parts = urlsplit(url, allow_fragments=False) download_url = f"{url_parts.scheme}://{url_parts.netloc}{result}" response[k] = download_image(download_url, headers) continue response[k] = result else: response[k] = False else: response[k] = False response['direct_url'] = url return jsonify(response), 200 # Enable this section for local development, but disable it on cPanel # if __name__ == "__main__": # app.run(debug=True)