Problem with ASIN Check Based on MAX_DAYS

[peppees]

Which installation method have you chosen?

Python Virtual Environment (venv)

What version of the software are you using?

3.9.6

What operating system are you using (Name and Version)?

Windows 11

What browser are you using (Name and Version)?

No response

Describe your issue

Hi Piero,

Sorry to bother you, but I wanted to ask you something. I made some modifications to my database file to add new functions, and everything is working perfectly. However, I’m experiencing some minor issues with verifying ASINs that have already been sent based on the MAX_DAYS parameter.

The check works correctly at times, logging messages like "ASIN already sent on...", but other times it seems to skip this validation entirely, sending products that were already sent on previous days (even the day before).

I’ve reviewed the file multiple times, but I can't pinpoint the issue. Could it be related to database connections or table handling?

If you could help me out, I’d really appreciate it.

Describe the steps to reproduce the issue

import os
import logging
import sqlite3
from typing import Optional
from datetime import datetime, timedelta

from utils.product import Product
from utils.log_manager import setup_logger
from utils import functions_toolbox
from configs import settings  # Importiamo la configurazione

setup_logger()
logger = logging.getLogger(__name__)

MAX_KEYWORD = settings.MAX_KEYWORD  # Numero massimo di parole chiave da non ripetere

last_keywords = []

asin_keyword_map = {}
e
_removal_in_progress = False

def manage_keyword_in_database():
    """Rimuove la parola chiave più vecchia dal database se si supera il limite massimo.

    Questa funzione utilizza un lock interno (_removal_in_progress) per evitare che il processo
    venga eseguito ripetutamente prima di completarsi.
    """
    global _removal_in_progress
    if _removal_in_progress:
        logging.debug("Rimozione della parola chiave già in corso, salto l'esecuzione.")
        return

    _removal_in_progress = True
    try:
        now = datetime.now()
        db_path = f"./database/{now.strftime('%Y')}/"
        db_name = f"{db_path}/{now.strftime('%m')}.db"
        table_name = f"day_{now.strftime('%d')}"

        os.makedirs(db_path, exist_ok=True)
        conn = sqlite3.connect(db_name)
        cursor = conn.cursor()

        try:
            cursor.execute("BEGIN IMMEDIATE;")
            cursor.execute(f"SELECT COUNT(*) FROM {table_name};")
            keyword_count = cursor.fetchone()[0]

            if keyword_count > MAX_KEYWORD:
                cursor.execute(f"""
                    SELECT ID FROM {table_name}
                    ORDER BY "DATE ADDED" ASC
                    LIMIT 1;
                """)
                oldest_id = cursor.fetchone()

                if oldest_id is None:
                    logging.warning("Nessuna parola chiave da rimuovere, la tabella potrebbe essere vuota.")
                else:
                    cursor.execute(f"DELETE FROM {table_name} WHERE ID = ?;", (oldest_id[0],))
                    conn.commit()
                    logging.info(f"Parola chiave più vecchia rimossa con ID: {oldest_id[0]}.")
            else:
                logging.debug("Il numero di parole chiave non supera il limite, nessuna rimozione effettuata.")

        except sqlite3.Error as e:
            logging.error(f"Errore SQLite durante la gestione delle parole chiave: {e}")
        finally:
            conn.close()
    finally:
        _removal_in_progress = False

def save_keyword_history(product_keyword: str):
    """Salva la parola chiave nel database e nella memoria locale, rimuovendo la più vecchia se necessario."""
    manage_keyword_in_database()

    if len(last_keywords) >= MAX_KEYWORD:
        removed_keyword = last_keywords.pop(0)
        logging.info(f"Rimossa parola chiave vecchia dalla cronologia locale: {removed_keyword}")

    last_keywords.append(product_keyword)
    logging.info(f"Parola chiave '{product_keyword}' aggiunta alla cronologia locale.")

def get_keyword_from_log(asin: str, log_file_path: str, max_lines: int = 25) -> Optional[str]:
    """Recupera la parola chiave dal log associata all'ASIN specificato.

    Args:
        asin (str): L'ASIN del prodotto.
        log_file_path (str): Il percorso del file di log da cui leggere.
        max_lines (int, opzionale): Numero massimo di righe da leggere indietro. Defaults to 25.

    Returns:
        str: La parola chiave associata all'ASIN, se trovata.
    """
    # Verifica se il file di log esiste
    if not os.path.exists(log_file_path):
        logging.error(f"Il file di log {log_file_path} non esiste.")
        return None

    # Apre il file di log e legge le ultime righe
    with open(log_file_path, 'r') as file:
        lines = file.readlines()

    # Cerca l'ASIN nel log e analizza le righe precedenti
    for i in range(len(lines) - 1, -1, -1):  # Legge all'indietro
        line = lines[i]
        if asin in line:
            # Cerca nella riga precedente per trovare la categoria e la parola chiave
            for j in range(i - 1, max(i - max_lines, -1), -1):
                prev_line = lines[j]
                if 'Keyword: ' in prev_line:
                    # Estrai la parola chiave dalla riga
                    keyword = prev_line.split('Keyword: ')[1].strip()
                    logging.debug(f"Recuperata parola chiave '{keyword}' per l'ASIN {asin} dal log.")
                    return keyword
    logging.warning(f"Parola chiave non trovata per l'ASIN {asin} nel file di log.")
    return None

def get_product_keyword_by_asin(asin: str) -> Optional[str]:
    """Recupera la parola chiave del prodotto dato il suo ASIN.

    Args:
        asin (str): L'ASIN del prodotto.

    Returns:
        str: La parola chiave del prodotto, se disponibile.
    """
    # Se esiste già una keyword associata a questo ASIN nella mappa, la restituisco
    if asin in asin_keyword_map:
        return asin_keyword_map[asin]

    # Altrimenti, prova a recuperarla dal log
    log_file_path = 'log/2025/01.log'  # Percorso del file di log
    keyword = get_keyword_from_log(asin, log_file_path)
    if keyword:
        return keyword
    else:
        return "Keyword_Fittizia"

def add_to_database(product: Product, name: Optional[str] = '') -> str:
    """Aggiunge un prodotto al database.

    Args:
        product (Product): Il prodotto da aggiungere al database.
        name (str, opzionale): Il nome del database. Defaults to ''.

    Returns:
        str: L'ASIN del prodotto aggiunto.
    """
    now = datetime.now()

    if not name:
        name = f"{now.strftime('%m')}"
        db_path = f"./database/{now.strftime('%Y')}/"
        db_name = f"{db_path}/{name}.db"
        table_name = f"day_{now.strftime('%d')}"
    else:
        db_path = f"./database/"
        db_name = f"{db_path}/{name}.db"
        table_name = "waiting_list"

    os.makedirs(db_path, exist_ok=True)

    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()

    # Crea la tabella se non esiste
    conn.execute(
        f'''CREATE TABLE IF NOT EXISTS {table_name} (
            ID INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
            ASIN TEXT,
            "DATE ADDED" DATE,
            "KEYWORD" TEXT
        );'''
    )
    conn.commit()

    # Verifica se la colonna "KEYWORD" esiste; se non esiste, aggiungila
    cursor.execute(f"PRAGMA table_info({table_name});")
    columns = [column[1] for column in cursor.fetchall()]

    if "KEYWORD" not in columns:
        cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN KEYWORD TEXT")
        conn.commit()
        logging.info(f"Aggiunta la colonna 'KEYWORD' alla tabella {table_name}.")

    try:
        if product.date_added is None:
            current_date = datetime.now().strftime('%Y-%m-%d')
        else:
            current_date = product.date_added

        # Inserisci i dati nel database
        cursor.execute(f"INSERT INTO {table_name} "
                       "(ASIN, 'DATE ADDED', 'KEYWORD') "
                       "VALUES (?, ?, ?)",
                       (product.asin, current_date, get_product_keyword_by_asin(product.asin)))

        conn.commit()

    except sqlite3.IntegrityError:
        logging.error(f"Errore durante l'inserimento del prodotto: {product.asin} nel database {name}.")
        return ''

    conn.close()

    # Recupera la parola chiave (già nota) e salvala nella cronologia locale
    product_keyword = get_product_keyword_by_asin(product.asin)
    if product_keyword:
        save_keyword_history(product_keyword)

    return product.asin

def correctly_added(asin_list: list[str]) -> None:
    """Logga l'aggiunta corretta dei prodotti con gli ASIN al database.

    Args:
        asin_list (list[str]): Una lista di ASIN dei prodotti che sono stati aggiunti al database.
    """
    logging.debug(f"I prodotti con gli ASIN: {asin_list} sono stati aggiunti correttamente al database.")

def is_valid_for_resend(product: Product, max_days: int) -> bool:
    """Check if this product has already been sent in the latest messages.

    If not enough products have already been shipped on the same day,
        check the number of products remaining in the previous days.

    Args:
        product (Product): Product object with all its characteristics.
        max_days (int): Number of days to check for product validity.

    Returns:
        Return True if this product has not been sent. False otherwise.
    """
    for day_index in range(0, max_days):
        now = datetime.now()
        new_date = now - timedelta(days=day_index)

        try:
            db_path = f"./database/{new_date.strftime('%Y')}/"
            db_name = f"{db_path}/{new_date.strftime('%m')}.db"
            os.makedirs(db_path, exist_ok=True)

            conn = sqlite3.connect(db_name)
            cursor = conn.cursor()

        except sqlite3.Error as e:
            logging.error("SQLite error:", e)

        try:
            cursor.execute(f'''
                            SELECT *
                            FROM day_{new_date.strftime("%d")}
                            WHERE ASIN = ?''', (product.asin,))

        except sqlite3.OperationalError as e:
            error_message = str(e)
            err_msg = f"no such table: day_{new_date.strftime('%d')}"

            if error_message != err_msg:
                logging.warning(f"SQLite error: {error_message} - Can't "
                                f"connect to the table "
                                f"day_{new_date.strftime('%d')}.")
            continue

        try:
            # Fetch all ASINs from the database for the specific day
            rows = cursor.fetchall()

            if len(rows) < 1:
                conn.close()
                continue

            else:
                conn.close()
                logging.debug(f"Asin: {product.asin} already sent on "
                              f"{new_date.strftime('%d-%m-%Y')}.")
                return False

        except Exception as e:
            logging.error(f"Error during the ferch of the price "
                            f"for the asin:{product.asin} - {e}")
    return True

def get_last_sent_keywords(max_days: int, max_keywords: int = MAX_KEYWORD) -> list:
    """Recupera le parole chiave degli ultimi 'max_keywords' prodotti inviati.

    Args:
        max_days (int): Numero massimo di giorni da cui considerare i prodotti inviati.
        max_keywords (int): Numero massimo di parole chiave da cui recuperare le parole chiave.

    Returns:
        list: Lista di parole chiave degli ultimi prodotti inviati.
    """
    last_keywords_list = []

    now = datetime.now()
    new_date = now

    try:
        db_path = f"./database/{new_date.strftime('%Y')}/"
        db_name = f"{db_path}/{new_date.strftime('%m')}.db"
        os.makedirs(db_path, exist_ok=True)

        conn = sqlite3.connect(db_name)
        cursor = conn.cursor()

        # Verifica se la tabella per il giorno corrente esiste
        table_name = f"day_{new_date.strftime('%d')}"
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?;", (table_name,))
        if not cursor.fetchone():
            logging.info(f"Tabella {table_name} non esistente per il giorno corrente.")
            conn.close()
            return []

        try:
            cursor.execute(f'''
                            SELECT ASIN, "KEYWORD"
                            FROM {table_name}
                            ORDER BY "DATE ADDED" DESC
                            LIMIT ?
                            ''', (max_keywords,))
            rows = cursor.fetchall()

            for row in rows:
                asin = row[0]
                keyword = row[1]
                if keyword and keyword not in last_keywords_list:
                    last_keywords_list.append(keyword)

                if len(last_keywords_list) >= max_keywords:
                    break

        except sqlite3.OperationalError as e:
            logging.warning(f"Errore SQLite: {e} - Impossibile recuperare i dati per il giorno {new_date.strftime('%d')}")
            logging.info("Nessun dato trovato per il giorno corrente. Aggiungerò la nuova parola chiave.")
            conn.close()
            return []

    except sqlite3.Error as e:
        logging.error("Errore SQLite:", e)
        return []

    logging.debug(f"Ultime parole chiave recuperate: {last_keywords_list}")
    conn.close()
    return last_keywords_list[:max_keywords]

def check_products_in_list(products_list, max_days):
    """Verifica i prodotti in una lista e restituisce solo quelli validi.

    Args:
        products_list (list): Lista di oggetti Product.
        max_days (int): Numero massimo di giorni da cui considerare la validità del prodotto.

    Returns:
        list: Lista dei prodotti validi.
    """
    valid_products = []

    for product in products_list:
        if is_valid_for_resend(product, max_days):
            valid_products.append(product)

    return valid_products

### To speed up the resolution of the issue, please provide the logs file.

_No response_

[Piero24]

Hi, As soon as I have time I will check better but I am quite busy this week. Try without the changes if the problem is still there (I assume not since there have been no problems so far) so we know if it is the new functions or not that are causing problems.

[peppees]

No worries, help me when you have some free time. Anyway, I already know that the issue is caused by the changes I made, that's why I was asking for your help

[peppees]

Hello, can I ask you if you had time to take a look at the code?

[Piero24]

Hi @peppees , sorry, but I’m quite busy this month and can’t keep up with it. However, I’ll try to take a look whenever I have time.
Could you explain better what result you’re trying to achieve and what’s going wrong? Ideally, with some examples, logs, or print statements if you have them… that way, I can fully understand before diving into the code.

[peppees]

honestly there is not much to say, simply with the changes I made to the database file, checking the asins already sent in the previous days via MAX_DAYS does not always seem to work. It doesn't give me any kind of error in the log or anything, just sometimes the check is done correctly, other times it just doesn't do it, as if it doesn't detect that that asin has already been sent in the previous days (it doesn't even start the verification)