cherbakov.py - master - NemcevaSI/git

cherbakov.py
535 строк24 KB
NemcevaSI
cherbakov
17 июн 2025, 19:23
17 июн 2025, 19:23b82015b
import requests
import re
import mysql.connector
import os
import urllib3
import time
import traceback
import json
from datetime import datetime, timezone
from typing import Optional, Dict, List, Tuple

# Отключение предупреждений о SSL
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Конфигурация базы данных (обновлена для вашей БД)
db_config = {
    'host': '127.0.0.1',
    'user': 'coni',
    'password': '778402',
    'database': 'wb1606'
}

API_URL = 'https://gigachat.devices.sberbank.ru/api/v1/chat/completions'
TOKEN_URL = 'https://ngw.devices.sberbank.ru:9443/api/v2/oauth'
CLIENT_ID = "your_client_id"  # Замените на реальный client_id если нужно
IMAGE_DIR = "images"  # Папка для сохранения изображений
REQUEST_DELAY = 1  # Задержка между запросами в секундах

# Авторизационные данные для получения токена
AUTH_HEADERS = {
    'Content-Type': 'application/x-www-form-urlencoded',
    'Accept': 'application/json',
    'RqUID': '3553f960-83d5-40db-aae8-47b6089ec700',
    'Authorization': 'Basic ZGU1ODk4OGEtNWZkZi00M2IwLTgxNDYtNGVmNzkxNWQ0ZjdlOjk4YzRjMjJjLTBjYzEtNDk5Mi04OWMxLTdjODg2OGM5MGQxMw=='
}

def create_tables(cursor, conn):
    """Создает необходимые таблицы, если они не существуют."""
    try:
        # Таблица для токенов
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS `tokens` (
                `id` int NOT NULL AUTO_INCREMENT,
                `token` text NOT NULL,
                `data` varchar(40) NOT NULL,
                PRIMARY KEY (`id`)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci
        """)
        
        # Таблица для статусов обработки товаров
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS `status_table` (
                `id` int NOT NULL AUTO_INCREMENT,
                `product_id` int NOT NULL,
                `status` tinyint NOT NULL DEFAULT 0 COMMENT '0 - не обработан, 1 - обработан',
                PRIMARY KEY (`id`),
                UNIQUE KEY `product_id` (`product_id`)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci
        """)
        
        conn.commit()
        print("Таблицы tokens и status_table созданы или уже существуют")
    except Exception as e:
        print(f"Ошибка при создании таблиц: {str(e)}")
        raise

def get_last_token_data(cursor):
    """Получает последний токен из базы данных."""
    cursor.execute("SELECT token, data FROM tokens ORDER BY id DESC LIMIT 1")
    return cursor.fetchone()

def insert_new_token(cursor, conn, token, expires_at_dt):
    """Сохраняет новый токен в базу данных."""
    try:
        expires_str = expires_at_dt.strftime('%y%m%d%H%M%S')
        sql = "INSERT INTO tokens (token, data) VALUES (%s, %s)"
        cursor.execute(sql, (token, expires_str))
        conn.commit()
        print(f"Вставлен новый токен с датой действия: {expires_str}")
    except Exception:
        print("Ошибка при вставке нового токена:")
        traceback.print_exc()
        raise

def fetch_new_token():
    """Получает новый токен от API."""
    payload = {'scope': 'GIGACHAT_API_CORP'}
    
    try:
        response = requests.post(TOKEN_URL, headers=AUTH_HEADERS, data=payload, verify=False)
        response.raise_for_status()
        data = response.json()

        access_token = data.get('access_token')
        expires_at = data.get('expires_at')

        if not access_token or not expires_at:
            raise ValueError("В ответе отсутствуют access_token или expires_at")

        if isinstance(expires_at, str):
            expires_at = int(expires_at)
        elif not isinstance(expires_at, int):
            raise ValueError("expires_at имеет неверный тип")

        expires_dt = datetime.fromtimestamp(expires_at / 1000)
        return access_token, expires_dt
    except Exception as e:
        print(f"Ошибка при получении токена: {str(e)}")
        raise

def get_valid_token(cursor):
    """Проверяет валидность токена в базе данных."""
    last_record = get_last_token_data(cursor)
    
    if last_record is None:
        print("Токенов в базе нет.")
        return None
    
    token, data_str = last_record
    try:
        token_expiry_dt = datetime.strptime(data_str, '%y%m%d%H%M%S').replace(tzinfo=timezone.utc)
        print(f"Последний токен истекает: {token_expiry_dt}")

        now_dt = datetime.now(timezone.utc)

        if token_expiry_dt > now_dt:
            print("Используем существующий токен.")
            return token
        else:
            print("Токен истёк.")
            return None
    except Exception:
        print("Неверный формат даты токена в базе.")
        return None

def get_products_without_photo(cursor) -> List[Tuple]:
    """Получает список товаров без фото (адаптировано для вашей БД)."""
    try:
        query = """
            SELECT p.id, p.title, p.description
            FROM products p
            LEFT JOIN photos ph ON p.id = ph.product_id
            WHERE (ph.url IS NULL OR ph.url = '' OR ph.id IS NULL)
            AND p.description IS NOT NULL AND p.description != ''
            AND NOT EXISTS (
                SELECT 1 FROM status_table s 
                WHERE s.product_id = p.id AND s.status = 1
            )
            LIMIT 10  # Ограничиваем количество для тестирования
        """
        cursor.execute(query)
        return cursor.fetchall()
    except Exception as e:
        raise Exception(f"Ошибка при получении товаров без фото: {str(e)}")

def generate_image_from_description(description_text: str, filename: str, token: str) -> bool:
    """
    Генерирует и сохраняет изображение по описанию.
    Возвращает True при успехе, False при ошибке.
    """
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "Authorization": f"Bearer {token}",
        "X-Client-ID": CLIENT_ID,
        "X-Service-Name": "images"
    }
    
    payload = {
        "model": "GigaChat",
        "messages": [
            {"role": "system", "content": "Ты — профессиональный фотограф товаров для интернет-магазина."},
            {"role": "user", "content": f"Создай фотореалистичное изображение товара по описанию: {description_text}. Изображение должно быть на белом фоне, без водяных знаков, в высоком качестве."}
        ],
        "function_call": "auto"
    }

    try:
        # 1. Запрос на генерацию изображения
        response = requests.post(API_URL, headers=headers, json=payload, verify=False)
        response.raise_for_status()
        data = response.json()
        
        # Проверяем успешность генерации
        if data.get('choices', [{}])[0].get('finish_reason') != 'stop':
            raise Exception("API не завершило генерацию изображения")
        
        message_content = data['choices'][0]['message']['content']
        
        # Извлекаем file_id из ответа
        img_tags = re.findall(r'<img src="([a-f0-9\-]+)"', message_content)
        if not img_tags:
            raise Exception("Идентификатор изображения не найден в ответе API")
        
        file_id = img_tags[0]
        download_url = f'https://gigachat.devices.sberbank.ru/api/v1/files/{file_id}/content'
        
        # 2. Заголовки для скачивания
        headers_download = {
            'Accept': 'image/jpeg',
            'Authorization': f'Bearer {token}',
            'X-Client-ID': CLIENT_ID,
            'X-Service-Name': 'images'
        }
        
        # 3. Скачиваем изображение
        img_response = requests.get(download_url, headers=headers_download, verify=False, stream=True)
        img_response.raise_for_status()

        # Проверяем content-type
        if 'image/jpeg' not in img_response.headers.get('Content-Type', ''):
            raise Exception("Получен неожиданный Content-Type")

        # Создаем папку если не существует
        os.makedirs(os.path.dirname(filename), exist_ok=True)

        # Сохраняем изображение
        with open(filename, 'wb') as f:
            for chunk in img_response.iter_content(1024):
                f.write(chunk)

        # Проверяем что файл сохранен и не пустой
        if os.path.getsize(filename) == 0:
            os.remove(filename)
            raise Exception("Сохраненный файл изображения пуст")

        print(f"Изображение успешно сохранено: {filename}")
        return True
        
    except requests.HTTPError as http_err:
        print(f"HTTP ошибка ({http_err.response.status_code}): {http_err.response.text}")
        return False
    except Exception as e:
        print(f"Ошибка при генерации изображения: {str(e)}")
        if os.path.exists(filename):
            try:
                os.remove(filename)
            except:
                pass
        return False

def update_photo_url(cursor, conn, product_id: int, filename: str) -> bool:
    """Обновляет таблицу photos, добавляя путь к фото."""
    try:
        # Проверяем, есть ли уже запись для этого товара
        cursor.execute("SELECT id FROM photos WHERE product_id = %s", (product_id,))
        existing_record = cursor.fetchone()
        
        if existing_record:
            # Обновляем существующую запись
            update_query = """
                UPDATE photos 
                SET url = %s, status = 1
                WHERE product_id = %s
            """
            cursor.execute(update_query, (filename, product_id))
        else:
            # Вставляем новую запись
            insert_query = """
                INSERT INTO photos (url, product_id, status, isLoad) 
                VALUES (%s, %s, 1, 1)
            """
            cursor.execute(insert_query, (filename, product_id))
            
        # Обновляем статус обработки товара
        cursor.execute("""
            INSERT INTO status_table (product_id, status) 
            VALUES (%s, 1)
            ON DUPLICATE KEY UPDATE status = 1
        """, (product_id,))
        
        conn.commit()
        return True
    except Exception as e:
        print(f"Ошибка при обновлении URL фото: {str(e)}")
        conn.rollback()
        return False

def generate_new_description(token: str, current_description: str, product_id: int, cursor, conn) -> bool:
    """Генерирует новое описание товара с помощью GigaChat API и сохраняет в базу."""
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': f'Bearer {token}'
    }
    
    system_message = {
        "role": "system",
        "content": "Ты — профессиональный копирайтер для маркетплейсов. Перепиши данное описание товара, сделав его более привлекательным для покупателей. Сохрани все ключевые характеристики, но изложи их более продающим языком. Описание должно быть четким, структурированным и убедительным."
    }
    
    user_message = {
        "role": "user",
        "content": current_description
    }
    
    data = {
        "model": "GigaChat",
        "messages": [system_message, user_message],
        "stream": False,
        "update_interval": 0
    }

    try:
        response = requests.post(API_URL, headers=headers, json=data, verify=False)
        response.raise_for_status()
        response_json = response.json()
        new_description = response_json['choices'][0]['message']['content']
        
        # Обновляем описание в базе данных
        update_query = "UPDATE products SET description = %s WHERE id = %s"
        cursor.execute(update_query, (new_description, product_id))
        
        # Обновляем статус обработки товара
        cursor.execute("""
            INSERT INTO status_table (product_id, status) 
            VALUES (%s, 1)
            ON DUPLICATE KEY UPDATE status = 1
        """, (product_id,))
        
        conn.commit()
        
        print(f"\nНовое описание для товара ID {product_id} успешно сгенерировано и сохранено.")
        return True
        
    except Exception as e:
        print(f"\nОшибка при генерации нового описания: {str(e)}")
        conn.rollback()
        return False

def evaluate_product_card(token, message, product_id, cursor, conn):
    """Оценивает карточку товара с помощью GigaChat API и генерирует новое описание если оценка < 40."""
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': f'Bearer {token}'
    }
    
    system_message = {
        "role": "system",
        "content": "Ты — профессиональный маркетолог с опытом создания карточек товаров для маркетплейсов. Твоя задача — оценить карточку товара для маркетплейсов по 100-бальной шкале. Ответ представь в формате JSON, где элемент 'procent' - это оценка, если оценка меньше 40, то создается объект 'text' - это рекомендации по исправлению"
    }
    
    data = {
        "model": "GigaChat",
        "messages": [system_message, message],
        "stream": False,
        "update_interval": 0
    }

    try:
        response = requests.post(API_URL, headers=headers, json=data, verify=False)
        response.raise_for_status()
        response_json = response.json()
        answer = response_json['choices'][0]['message']['content']
        print("\nОтвет от сервера:")
        print(answer)

        result = json.loads(answer)
        procent = result.get('procent')
        text = result.get('text')

        print(f"\nОценка товара: {procent}/100")
        if procent is not None and float(procent) < 40 and text:
            print(f"Рекомендации по улучшению: {text}")
            
            # Получаем текущее описание из сообщения
            current_description = message['content'].split("Описание: ")[1].split("\n")[0]
            
            # Генерируем новое описание
            generate_new_description(token, current_description, product_id, cursor, conn)
            
    except requests.RequestException as e:
        print("Ошибка при запросе:", e)
    except (KeyError, IndexError) as e:
        print("Некорректный формат ответа:", e)
    except json.JSONDecodeError:
        print("Ответ не является валидным JSON.")

def process_descriptions_and_photos(cursor, conn):
    """Обрабатывает описания товаров для оценки (адаптировано для вашей БД)."""
    try:
        # Проверяем, есть ли товары с описанием без фото и статусом 0
        query = """
        SELECT p.id, p.title, p.description, GROUP_CONCAT(ph.url SEPARATOR '; ') as photos
        FROM products p
        LEFT JOIN photos ph ON p.id = ph.product_id
        WHERE p.description IS NOT NULL AND p.description != ''
        AND NOT EXISTS (
            SELECT 1 FROM status_table s 
            WHERE s.product_id = p.id AND s.status = 1
        )
        GROUP BY p.id, p.title, p.description
        HAVING photos IS NULL OR photos = ''
        LIMIT 10  # Ограничиваем количество для тестирования
        """
        cursor.execute(query)
        results = cursor.fetchall()
        
        messages = []
        
        for row in results:
            product_id, title, description, photos = row
            if description:
                content = f"Название: {title}\nОписание: {description}"
                if photos:
                    content += f"\nФотографии: {photos}"
                
                message = {
                    "role": "user",
                    "content": content
                }
                messages.append((message, product_id))  # Возвращаем кортеж (message, product_id)
                
                # Обновляем статус обработки товара
                cursor.execute("""
                    INSERT INTO status_table (product_id, status) 
                    VALUES (%s, 1)
                    ON DUPLICATE KEY UPDATE status = 1
                """, (product_id,))
                conn.commit()
                
        print(f"Обработано {len(messages)} товаров")
        return messages
    
    except Exception as e:
        print(f"Ошибка при обработке описаний и фотографий: {e}")
        traceback.print_exc()
        return []

def main():
    """Основная функция для управления процессом."""
    while True:  # Бесконечный цикл
        try:
            print("\n=== Начало новой итерации обработки ===")
            conn = mysql.connector.connect(**db_config)
            cursor = conn.cursor()

            # Создаем необходимые таблицы, если они не существуют
            create_tables(cursor, conn)

            # Проверяем есть ли валидный токен
            token = get_valid_token(cursor)

            if token is None:
                try:
                    access_token, expires_dt = fetch_new_token()
                    print(f"\nПолучен новый токен, действует до {expires_dt}")
                    insert_new_token(cursor, conn, access_token, expires_dt)
                    token = access_token
                except Exception:
                    print("\nОшибка при получении или сохранении нового токена:")
                    traceback.print_exc()
                    time.sleep(60)  # Пауза перед следующей попыткой
                    continue

            if not token:
                print("Не удалось получить валидный токен")
                time.sleep(60)  # Пауза перед следующей попыткой
                continue

            print("\n=== Начало генерации изображений ===")
            # Режим генерации изображений
            try:
                products = get_products_without_photo(cursor)
                print(f"Найдено {len(products)} товаров без изображений")

                for i, product in enumerate(products, 1):
                    product_id, title, description = product  # Распаковываем кортеж
                    print(f"\nОбработка товара {i}/{len(products)}: ID {product_id} - {title}")
                    
                    try:
                        filename = f"{IMAGE_DIR}/product_{product_id}.jpg"
                        
                        if generate_image_from_description(description, filename, token):
                            if update_photo_url(cursor, conn, product_id, filename):
                                print(f"Успешно обновлено фото для товара ID {product_id}")
                            else:
                                print(f"Не удалось обновить URL фото")
                        else:
                            print(f"Не удалось сгенерировать изображение")
                        
                        if i < len(products):
                            time.sleep(REQUEST_DELAY)
                            
                    except Exception as e:
                        print(f"Ошибка при обработке товара: {str(e)}")
                        traceback.print_exc()
                        continue
            
            except Exception as e:
                print(f"Ошибка в режиме генерации изображений: {str(e)}")
                traceback.print_exc()

            print("\n=== Начало оценки карточек товаров ===")
            # Режим оценки карточек товаров
            try:
                messages = process_descriptions_and_photos(cursor, conn)
                
                if not messages:
                    print("Нет товаров для обработки (статус 0 не найден)")
                else:
                    print(f"\nНачинаем обработку {len(messages)} товаров...")
                    for i, (message, product_id) in enumerate(messages, 1):
                        print(f"\nОбработка товара {i} из {len(messages)}")
                        evaluate_product_card(token, message, product_id, cursor, conn)
                        if i < len(messages):
                            time.sleep(REQUEST_DELAY)
            
            except Exception as e:
                print(f"Ошибка в режиме оценки карточек: {str(e)}")
                traceback.print_exc()

            # Закрываем соединение с БД
            if 'conn' in locals() and conn.is_connected():
                cursor.close()
                conn.close()

            # Пауза перед следующей итерацией (например,  минут)
            print("\n=== Итерация завершена. Ожидание следующего цикла... ===")
            time.sleep(60)  #  минут паузы

        except KeyboardInterrupt:
            print("\nЗавершение работы по запросу пользователя...")
            break
        except Exception as e:
            print(f"\nКритическая ошибка в основном цикле: {str(e)}")
            traceback.print_exc()
            time.sleep(5)  # Пауза перед следующей попыткой
            continue

    print("\nРабота завершена")

if __name__ == "__main__":
    main()