missbleue/wcctl.py

print(f"📦 Script lancé : __name__ = {__name__}")
import argparse
from woocommerce import API as WoocommerceApi
from api import WoocommerceApiClient
from api_woocommerce import AuthentificationWpApi, MediaManager, CategoryManager, ProductManager, AttributeManager, VariationsManager, TabManager, WooCommerceManager
import os, sys

from logging.handlers import TimedRotatingFileHandler
import logging
logger = logging.getLogger(__name__)


class WcCtlBase:
    def __init__(self, args):
        self.args = args
        self._wcm = None

    @property
    def wcm(self):
        if self._wcm is None:
            wcapi = WoocommerceApiClient(
            url=self.args.wc_url,
            wc_consumer_key=self.args.wc_key,
            wc_consumer_secret=self.args.wc_secret,
            wp_application_user="admin_lcdm",
            wp_application_password= "Do4p tLYF 5uje PF2M 21Zo x3OR", #"eAB4 49W6 ZRBj zIZc fH62 tv5c", #"yTW8 Mc6J FUCN tPSq bnuJ 0Sdw", #
            wp_api=True,
            version="wc/v3",
            verify_ssl=True,
            timeout=30
            )   

            ath = AuthentificationWpApi()
            media_manager = MediaManager(ath, wcapi, filename_ods=self.args.ods_path)
            category_manager = CategoryManager(wcapi, ath, filename_ods=self.args.ods_path)
            product_manager = ProductManager(wcapi, ath, filename_ods=self.args.ods_path)
            attribute_manager = AttributeManager(wcapi, filename_ods=self.args.ods_path)
            tab_manager = TabManager(wcapi, filename_ods=self.args.ods_path)
            variation_manager = VariationsManager(wcapi, filename_ods=self.args.ods_path)
            self._wcm = WooCommerceManager(wcapi=wcapi, 
                                             media_manager=media_manager,
                                             category_manager=category_manager,
                                             product_manager=product_manager, 
                                             tab_manager=tab_manager, 
                                             attribute_manager=attribute_manager, 
                                             variation_manager=variation_manager,
                                             filename_ods=self.args.ods_path)
            
        return self._wcm
    
class DeleteWcCtl(WcCtlBase):
    def action(self):
        logger.info(f"DeleteWcCtl:action({self.args}) - begin")
        if self.args.all:
            return self.wcm.delete_all_informations()

        if self.args.attributes:
            return None

class ImportOdsCtl(WcCtlBase):
    def action(self):
        logger.info(f"ImportOdsCtl:action({self.args}) - begin")

        if self.args.media:
            self.import_medias_ods()
                    
        if self.args.category:
            medias = self.wmc.mm.get_all_as_slug_dict()
            self.wmc.cm.medias = medias
            regex = self.args.category_regex if self.args.category_regex else None
            self.wcm.cm.update_data_categories(regex)

        if self.args.attribute:
            regex = self.args.attribute_regex if self.args.attribute_regex else None
            self.wcm.am.create(regex)
            self.wcm.am.configure_term()
            
        if self.args.product:
            self.import_products_ods()
            
        if self.args.logo:
            self.wcm.mm.assign_image_logo()

    
    def import_medias_ods(self):
        if self.args.media:
            if self.args.media_regex:
                try:
                    self.wcm.mm.upload_media(self.args.media_regex)
                except ValueError:
                    logger.error("error name product_regex")   
            elif self.args.media_range:
                try:
                    parts = self.args.media_range.split(':')
                    start = int(parts[0]) -1 if parts[0] else 0
                    end = int(parts[1]) if len(parts) > 1 and parts[1] else None
                    logger.debug(f"start = {start}, end = {end or 'fin'}")
                    self.wcm.mm.upload_media_from_to(start, end)
                except ValueError:
                    print("❌ Mauvais format pour --media-range. Utilisez par exemple --media-range=1:40")
            else:
                start, end = 0, None
                print("ℹ️ --media activé, mais aucune plage spécifiée.")
                
    def import_products_ods(self):
        if self.args.product:
            if self.args.product_regex:
                try:
                    self.wcm.process_file(self.args.product_regex)
                except ValueError:
                    print("error name product_regex")    
            elif self.args.product_range:
                try:
                    parts = self.args.product_range.split(':')
                    start = int(parts[0]) -1 if parts[0] else 0
                    end = int(parts[1]) if len(parts) > 1 and parts[1] else None
                    print(f"start = {start}, end = {end or 'fin'}")
                    self.wcm.process_file_from_to(start, end)
                except ValueError:
                    print("❌ Mauvais format pour --product-range. Utilisez par exemple --product-range=1:40")
            else:
                start, end = 0, None
                print("ℹ️ --product activé, mais aucune plage spécifiée.")
            
def main():

    # initialize logging
    log_directory = "logs"

    # 🔧 Configuration du handler avec rotation quotidienne
    log_file = os.path.join(log_directory, "woocommerce.log")
    handler = TimedRotatingFileHandler(
        filename=log_file,
        when="midnight",         # ⏰ Rotation tous les jours à minuit
        interval=1,              # 📅 Chaque 1 jour
        backupCount=7,           # ♻️ Garde les 7 derniers fichiers de log
        encoding='utf-8'         # 🧾 Pour supporter tous les caractères
    )

    # 📋 Format du log
    formatter = logging.Formatter(
        fmt="%(asctime)s - %(levelname)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )
    handler.setFormatter(formatter)

    # 🔌 Récupère le logger
    logger.setLevel(logging.DEBUG)   # 👁 Niveau minimum à capturer
    logger.addHandler(handler)
            
    parser = argparse.ArgumentParser(prog='wcctl', description='WooCommerce CLI controller')

    # 🌐 Options globales
    parser.add_argument('--wc-url', required=True, help='WooCommerce site URL')
    parser.add_argument('--wc-key', required=True, help='WooCommerce API consumer key')
    parser.add_argument('--wc-secret', required=True, help='WooCommerce API consumer secret')

    #parser.add_argument('--media', action='store_true', help='Process media items')
    #parser.add_argument('--media-range', type=str, help='Range of media rows to process (e.g., 10:30)')

    # 🧱 Sous-commandes
    subparsers = parser.add_subparsers(dest='command', required=True)

    delete_parser = subparsers.add_parser('delete', help='delete entities')
    delete_parser.add_argument('--attributes', action="store_true", required=False, default=False, help='delete all attributes')
    delete_parser.add_argument('--all', action='store_true', required=False, default=False, help='Delete media, categories, products, attributes, tabs')

    # 📥 Commande : import-ods
    import_parser = subparsers.add_parser('import-ods', help='Import ODS file data')
    import_parser.add_argument('--ods-path', required=True, help='Path to the ODS file')
    
    # media
    import_parser.add_argument('--media', action='store_true', help='Process media items')
    import_parser.add_argument('--media-range', type=str, help='Range of media rows to process (e.g., 10:30)')
    import_parser.add_argument('--media-regex', type=str, help='Regex to filter and import media by name')
    import_parser.add_argument('--logo', action='store_true', help='Process logo')

    # category
    import_parser.add_argument('--category', action='store_true', help='import all categories')
    import_parser.add_argument('--category-regex', type=str, help='Regex to filter and import categories by name')

    # attribute
    import_parser.add_argument('--attribute', action='store_true', help='import all attributes and terms')
    import_parser.add_argument('--attribute-regex', type=str, help='Regex to filter and import attribute by name')

    # tab
    #import_parser.add_argument('--tab', action='store_true', help='import all tabs')
    #import_parser.add_argument('--tab-regex', type=str, help='Regex to filter and import tab by name')
 
    # product
    import_parser.add_argument('--product', action='store_true', help='import all products')
    import_parser.add_argument('--product-regex', type=str, help='Regex to filter and import product by name')
    import_parser.add_argument('--product-range', type=str, help='Range of product rows to process (e.g., 10:30)')    

    # 📥 Commande : import-ods
    checkssl_parser = subparsers.add_parser('checkssl', help='Check ssl connectivity')
    check_parser = subparsers.add_parser('check', help='Check connectivity')



    # Analyse des arguments
    args = parser.parse_args()  

    if args.command == "delete":
        sys.exit(DeleteWcCtl(args).action())

    if args.command == "import-ods":
        sys.exit(ImportOdsCtl(args).action())

    if args.command == "check":
        #wcctl-api-rw
        #class WoocommerceApiClient(API) def __init__(self, url, wc_consumer_key, wc_consumer_secret, wp_application_user, wp_application_password, verify_ssl=False, **kwargs): 
        wcapi = WoocommerceApiClient(
            url=args.wc_url,
            wc_consumer_key=args.wc_key,
            wc_consumer_secret=args.wc_secret,
            wp_application_user="admin_lcdm",
            wp_application_password= "Do4p tLYF 5uje PF2M 21Zo x3OR",#"eAB4 49W6 ZRBj zIZc fH62 tv5c", #"yTW8 Mc6J FUCN tPSq bnuJ 0Sdw", #
            wp_api=True,
            version="wc/v3",
            verify_ssl=True,
            timeout=30
        )
        print("recupération d'un media", end="")
        response = wcapi.get("media", params={"per_page": 100, "page": 1})
        print(response)
        print("OK")
        sys.exit(0)

    if args.command == "checkssl":
        for i in range(10):
            response = wcapi.get("products", params={"per_page": 1, "page": 1})
            print(i)
        print(f'connection OK')
        sys.exit(0)

if __name__ == "__main__":
    main()

# wcctl --wc-url=https://lescreationsdemissbleue.local --wc-key=<consumer_key> --wc-secret=<consumer_secret> import-ods --ods-path=fichier.ods     
# ods_file = donnees_site_internet_missbleue_corrige.ods


#python wcctl.py --wc-url="https://lescreationsdemissbleue.local" --wc-key="ck_604e9b7b5d290cce72346efade6b31cb9a1ff28e" --wc-secret="cs_563974c7e59532c1ae1d0f8bbf61f0500d6bc768" import-ods --ods-path="donnees_site_internet_missbleue_version_finale.ods"

#python wcctl.py --wc-url="https://les-creations-de-missbleue.local" --wc-key="ck_604e9b7b5d290cce72346efade6b31cb9a1ff28e" --wc-secret="cs_563974c7e59532c1ae1d0f8bbf61f0500d6bc768" import-ods --ods-path="donnees_site_internet_missbleue_version_finale.ods"

# python wcctl.py --wc-url="https://les-creations-de-missbleue.local" --wc-key="ck_604e9b7b5d290cce72346efade6b31cb9a1ff28e" --wc-secret="cs_563974c7e59532c1ae1d0f8bbf61f0500d6bc768" import-ods --ods-path="donnees_site_internet_missbleue_version_finale.ods" --product --product-range="111:112" 
# 
# # python wcctl.py --wc-url="https://les-creations-de-missbleue.com" --wc-key="ck_4125db027f75040ce9c7ca1bbc95b7031d1a6263" --wc-secret="cs_50d0a1de003fa8f8d0deb5f427b7769071dd4bfd" import-ods --ods-path="donnees_site_internet_missbleue_version_finale.ods"