Scraper et importateur pour OpenCart : versions, solutions prêtes à l'emploi et exemple avec conversion de devises

pip install requests beautifulsoup4 pymysql lxml

import requests
import xml.etree.ElementTree as ET
from datetime import date

class CurrencyConverter:
    """Recupere le taux de la banque centrale et convertit le prix fournisseur dans la devise de la boutique."""

    CBR_URL = "https://www.cbr.ru/scripts/XML_daily.asp"

    def __init__(self, source_currency="USD", markup=1.20):
        # markup=1.20 — marge de la boutique : +20 % au-dessus du taux
        self.source_currency = source_currency
        self.markup = markup
        self._rate = None
        self._rate_date = None

    def _fetch_rate(self):
        """Recupere le taux de la devise voulue face au rouble depuis le XML de la banque centrale."""
        resp = requests.get(self.CBR_URL, timeout=15)
        resp.encoding = "windows-1251"
        tree = ET.fromstring(resp.text)

        for valute in tree.findall("Valute"):
            if valute.find("CharCode").text == self.source_currency:
                nominal = int(valute.find("Nominal").text)
                value = float(valute.find("Value").text.replace(",", "."))
                return value / nominal  # taux pour 1 unite de devise

        raise ValueError(f"Devise {self.source_currency} introuvable dans la reponse de la banque centrale")

    @property
    def rate(self):
        # on met le taux en cache pour la date du jour
        if self._rate is None or self._rate_date != date.today():
            self._rate = self._fetch_rate()
            self._rate_date = date.today()
        return self._rate

    def convert(self, amount):
        """Prix fournisseur -> prix boutique (en roubles) avec marge."""
        if amount is None:
            return None
        price = float(amount) * self.rate * self.markup
        return round(price, 2)

import time
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin

class SupplierParser:
    def __init__(self, base_url, converter, delay=1.0):
        self.base_url = base_url
        self.converter = converter
        self.delay = delay  # pause entre les requetes pour ne pas surcharger le site source
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (compatible; CatalogImporter/1.0)"
        })

    def _get_soup(self, url):
        resp = self.session.get(url, timeout=20)
        resp.raise_for_status()
        return BeautifulSoup(resp.text, "lxml")

    def parse_catalog(self, catalog_path, max_pages=5):
        """Collecte les liens produits sur les pages du catalogue paginé."""
        product_urls = []
        for page in range(1, max_pages + 1):
            url = urljoin(self.base_url, f"{catalog_path}?page={page}")
            soup = self._get_soup(url)
            cards = soup.select(".product-card a.product-link")
            if not cards:
                break  # plus de pages
            for a in cards:
                product_urls.append(urljoin(self.base_url, a["href"]))
            time.sleep(self.delay)
        return product_urls

    def parse_product(self, url):
        """Analyse une fiche produit."""
        soup = self._get_soup(url)

        name = soup.select_one("h1.product-title")
        price_raw = soup.select_one(".price .value")
        description = soup.select_one(".product-description")
        sku = soup.select_one(".sku")
        image = soup.select_one(".product-gallery img")

        # on nettoie le prix des espaces et symboles monetaires
        price_source = None
        if price_raw:
            digits = "".join(ch for ch in price_raw.text if ch.isdigit() or ch == ".")
            price_source = float(digits) if digits else None

        time.sleep(self.delay)

        return {
            "name": name.text.strip() if name else "Sans nom",
            "sku": sku.text.strip() if sku else "",
            "description": description.decode_contents().strip() if description else "",
            "image_url": urljoin(self.base_url, image["src"]) if image else "",
            "price_source": price_source,
            # c'est ici que se fait la conversion de devise :
            "price": self.converter.convert(price_source),
            "source_url": url,
        }

import pymysql
from datetime import datetime
from transliterate import slugify  # pip install transliterate

class OpenCartImporter:
    def __init__(self, db_config, prefix="oc_", store_id=0, language_id=1):
        self.conn = pymysql.connect(**db_config, charset="utf8mb4",
                                    cursorclass=pymysql.cursors.DictCursor)
        self.prefix = prefix
        self.store_id = store_id
        self.language_id = language_id

    def _seo_keyword(self, name):
        return slugify(name, "ru") or "product"

    def import_product(self, p, category_id=None):
        cur = self.conn.cursor()
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        px = self.prefix

        # 1) table principale du produit
        cur.execute(f"""
            INSERT INTO {px}product
                (model, sku, quantity, price, image, status,
                 date_added, date_modified, date_available)
            VALUES (%s, %s, %s, %s, %s, 1, %s, %s, %s)
        """, (p["sku"] or p["name"][:64], p["sku"], 100,
              p["price"], "", now, now, now))
        product_id = cur.lastrowid

        # 2) description (nom, texte, meta) — pour chaque langue
        cur.execute(f"""
            INSERT INTO {px}product_description
                (product_id, language_id, name, description,
                 meta_title, meta_description)
            VALUES (%s, %s, %s, %s, %s, %s)
        """, (product_id, self.language_id, p["name"],
              p["description"], p["name"], p["name"]))

        # 3) rattachement a la boutique
        cur.execute(f"""
            INSERT INTO {px}product_to_store (product_id, store_id)
            VALUES (%s, %s)
        """, (product_id, self.store_id))

        # 4) rattachement a la categorie (si definie)
        if category_id:
            cur.execute(f"""
                INSERT INTO {px}product_to_category (product_id, category_id)
                VALUES (%s, %s)
            """, (product_id, category_id))

        # 5) SEO URL
        keyword = f"{self._seo_keyword(p['name'])}-{product_id}"
        # — OpenCart 3.x : table url_alias ---
        cur.execute(f"""
            INSERT INTO {px}url_alias (query, keyword)
            VALUES (%s, %s)
        """, (f"product_id={product_id}", keyword))
        # — Pour OpenCart 4.x ce serait (table seo_url) :
        # cur.execute(f'''INSERT INTO {px}seo_url
        #   (store_id, language_id, key, value, keyword)
        #   VALUES (%s, %s, "product_id", %s, %s)''',
        #   (self.store_id, self.language_id, product_id, keyword))

        self.conn.commit()
        return product_id

    def close(self):
        self.conn.close()

def main():
    converter = CurrencyConverter(source_currency="USD", markup=1.25)
    parser = SupplierParser("https://supplier-example.com", converter, delay=1.0)

    importer = OpenCartImporter(
        db_config={
            "host": "localhost",
            "user": "db_user",
            "password": "db_password",
            "database": "opencart_db",
        },
        prefix="oc_",
        store_id=0,
        language_id=1,
    )

    product_urls = parser.parse_catalog("/catalog/category-1", max_pages=3)
    print(f"Produits trouves : {len(product_urls)}")

    imported = 0
    for url in product_urls:
        try:
            product = parser.parse_product(url)
            if product["price"] is None:
                print(f"Ignore (pas de prix) : {url}")
                continue
            pid = importer.import_product(product, category_id=20)
            imported += 1
            print(f"[{imported}] {product['name']} — "
                  f"{product['price_source']} {converter.source_currency} "
                  f"-> {product['price']} RUB  (ID {pid})")
        except Exception as e:
            print(f"Erreur sur {url} : {e}")

    importer.close()
    print(f"Termine. Importes : {imported}")

if __name__ == "__main__":
    main()

<?php
class CurrencyConverter
{
    private const CBR_URL = 'https://www.cbr.ru/scripts/XML_daily.asp';

    private ?float $rate = null;
    private ?string $rateDate = null;

    public function __construct(
        private string $sourceCurrency = 'USD',
        private float $markup = 1.20  // marge de la boutique : +20 % au-dessus du taux
    ) {}

    private function fetchRate(): float
    {
        $xmlRaw = file_get_contents(self::CBR_URL);
        // la reponse de la banque centrale est en windows-1251 — on la convertit en UTF-8
        $xmlRaw = mb_convert_encoding($xmlRaw, 'UTF-8', 'windows-1251');
        $xmlRaw = preg_replace('/encoding="windows-1251"/', 'encoding="UTF-8"', $xmlRaw);

        $xml = new SimpleXMLElement($xmlRaw);
        foreach ($xml->Valute as $valute) {
            if ((string)$valute->CharCode === $this->sourceCurrency) {
                $nominal = (int)$valute->Nominal;
                $value = (float)str_replace(',', '.', (string)$valute->Value);
                return $value / $nominal; // taux pour 1 unite de devise
            }
        }
        throw new RuntimeException("Devise {$this->sourceCurrency} introuvable dans la reponse de la banque centrale");
    }

    public function getRate(): float
    {
        $today = date('Y-m-d');
        if ($this->rate === null || $this->rateDate !== $today) {
            $this->rate = $this->fetchRate();
            $this->rateDate = $today;
        }
        return $this->rate;
    }

    public function convert(?float $amount): ?float
    {
        if ($amount === null) {
            return null;
        }
        return round($amount * $this->getRate() * $this->markup, 2);
    }
}

<?php
class SupplierParser
{
    public function __construct(
        private string $baseUrl,
        private CurrencyConverter $converter,
        private float $delay = 1.0  // pause entre les requetes, en secondes
    ) {}

    private function getHtml(string $url): string
    {
        $ch = curl_init($url);
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_FOLLOWLOCATION => true,
            CURLOPT_TIMEOUT        => 20,
            CURLOPT_USERAGENT      => 'Mozilla/5.0 (compatible; CatalogImporter/1.0)',
        ]);
        $html = curl_exec($ch);
        if ($html === false) {
            throw new RuntimeException('cURL: ' . curl_error($ch));
        }
        curl_close($ch);
        return $html;
    }

    private function makeXPath(string $html): DOMXPath
    {
        $doc = new DOMDocument();
        libxml_use_internal_errors(true); // on ignore les erreurs du balisage « sale »
        $doc->loadHTML('<?xml encoding="UTF-8">' . $html);
        libxml_clear_errors();
        return new DOMXPath($doc);
    }

    private function absolute(string $href): string
    {
        return str_starts_with($href, 'http') ? $href : rtrim($this->baseUrl, '/') . '/' . ltrim($href, '/');
    }

    /** Collecte les liens produits sur les pages du catalogue paginé. */
    public function parseCatalog(string $catalogPath, int $maxPages = 5): array
    {
        $urls = [];
        for ($page = 1; $page <= $maxPages; $page++) {
            $html = $this->getHtml($this->absolute("{$catalogPath}?page={$page}"));
            $xpath = $this->makeXPath($html);
            $links = $xpath->query("//div[contains(@class,'product-card')]//a[contains(@class,'product-link')]");

            if ($links->length === 0) {
                break; // plus de pages
            }
            foreach ($links as $a) {
                $urls[] = $this->absolute($a->getAttribute('href'));
            }
            usleep((int)($this->delay * 1_000_000));
        }
        return $urls;
    }

    /** Analyse une fiche produit. */
    public function parseProduct(string $url): array
    {
        $xpath = $this->makeXPath($this->getHtml($url));

        $text = fn(string $q) => trim($xpath->query($q)->item(0)?->textContent ?? '');

        $name        = $text("//h1[contains(@class,'product-title')]") ?: 'Sans nom';
        $sku         = $text("//*[contains(@class,'sku')]");
        $description = $text("//*[contains(@class,'product-description')]");
        $priceRaw    = $text("//*[contains(@class,'price')]//*[contains(@class,'value')]");

        $imgNode = $xpath->query("//*[contains(@class,'product-gallery')]//img")->item(0);
        $imageUrl = $imgNode ? $this->absolute($imgNode->getAttribute('src')) : '';

        // on nettoie le prix des espaces et symboles monetaires
        $priceSource = null;
        if ($priceRaw !== '') {
            $digits = preg_replace('/[^0-9.]/', '', str_replace(',', '.', $priceRaw));
            $priceSource = $digits !== '' ? (float)$digits : null;
        }

        usleep((int)($this->delay * 1_000_000));

        return [
            'name'         => $name,
            'sku'          => $sku,
            'description'  => $description,
            'image_url'    => $imageUrl,
            'price_source' => $priceSource,
            // conversion de devise :
            'price'        => $this->converter->convert($priceSource),
            'source_url'   => $url,
        ];
    }
}

<?php
class OpenCartImporter
{
    private PDO $pdo;

    public function __construct(
        array $dbConfig,
        private string $prefix = 'oc_',
        private int $storeId = 0,
        private int $languageId = 1
    ) {
        $dsn = "mysql:host={$dbConfig['host']};dbname={$dbConfig['database']};charset=utf8mb4";
        $this->pdo = new PDO($dsn, $dbConfig['user'], $dbConfig['password'], [
            PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
        ]);
    }

    private function seoKeyword(string $name): string
    {
        // slugification simplifiee pour l'URL propre
        $slug = mb_strtolower(trim($name));
        $slug = preg_replace('/[^a-z0-9\s-]/u', '', $slug);
        $slug = preg_replace('/\s+/u', '-', $slug);
        return $slug !== '' ? $slug : 'product';
    }

    public function importProduct(array $p, ?int $categoryId = null): int
    {
        $px = $this->prefix;
        $now = date('Y-m-d H:i:s');

        // 1) table principale du produit
        $stmt = $this->pdo->prepare("
            INSERT INTO {$px}product
                (model, sku, quantity, price, image, status,
                 date_added, date_modified, date_available)
            VALUES (:model, :sku, 100, :price, '', 1, :added, :modified, :available)
        ");
        $stmt->execute([
            ':model'     => $p['sku'] ?: mb_substr($p['name'], 0, 64),
            ':sku'       => $p['sku'],
            ':price'     => $p['price'],
            ':added'     => $now,
            ':modified'  => $now,
            ':available' => $now,
        ]);
        $productId = (int)$this->pdo->lastInsertId();

        // 2) description (nom, texte, meta)
        $this->pdo->prepare("
            INSERT INTO {$px}product_description
                (product_id, language_id, name, description, meta_title, meta_description)
            VALUES (?, ?, ?, ?, ?, ?)
        ")->execute([
            $productId, $this->languageId, $p['name'],
            $p['description'], $p['name'], $p['name'],
        ]);

        // 3) rattachement a la boutique
        $this->pdo->prepare("
            INSERT INTO {$px}product_to_store (product_id, store_id) VALUES (?, ?)
        ")->execute([$productId, $this->storeId]);

        // 4) rattachement a la categorie
        if ($categoryId !== null) {
            $this->pdo->prepare("
                INSERT INTO {$px}product_to_category (product_id, category_id) VALUES (?, ?)
            ")->execute([$productId, $categoryId]);
        }

        // 5) SEO URL
        $keyword = $this->seoKeyword($p['name']) . '-' . $productId;
        // — OpenCart 3.x : table url_alias ---
        $this->pdo->prepare("
            INSERT INTO {$px}url_alias (query, keyword) VALUES (?, ?)
        ")->execute(["product_id={$productId}", $keyword]);
        // — Pour OpenCart 4.x ce serait (table seo_url) :
        // INSERT INTO {$px}seo_url (store_id, language_id, `key`, `value`, keyword)
        //   VALUES (:store, :lang, 'product_id', :pid, :keyword)

        return $productId;
    }
}

<?php
require 'CurrencyConverter.php';
require 'SupplierParser.php';
require 'OpenCartImporter.php';

$converter = new CurrencyConverter('USD', 1.25);
$parser    = new SupplierParser('https://supplier-example.com', $converter, 1.0);

// les acces peuvent etre externalises dans un config ou tires du config.php d'OpenCart
$importer = new OpenCartImporter([
    'host'     => 'localhost',
    'user'     => 'db_user',
    'password' => 'db_password',
    'database' => 'opencart_db',
], prefix: 'oc_', storeId: 0, languageId: 1);

$urls = $parser->parseCatalog('/catalog/category-1', maxPages: 3);
echo 'Produits trouves : ' . count($urls) . PHP_EOL;

$imported = 0;
foreach ($urls as $url) {
    try {
        $product = $parser->parseProduct($url);
        if ($product['price'] === null) {
            echo "Ignore (pas de prix) : {$url}" . PHP_EOL;
            continue;
        }
        $pid = $importer->importProduct($product, categoryId: 20);
        $imported++;
        printf("[%d] %s — %s USD -> %s RUB  (ID %d)%s",
            $imported, $product['name'], $product['price_source'], $product['price'], $pid, PHP_EOL);
    } catch (Throwable $e) {
        echo "Erreur sur {$url} : {$e->getMessage()}" . PHP_EOL;
    }
}

echo "Termine. Importes : {$imported}" . PHP_EOL;

# chaque jour a 4h00, on synchronise le catalogue avec le fournisseur
0 4 * * * /usr/bin/php /var/www/opencart/parser/run.php >> /var/www/opencart/parser/parser.log 2>&1