Update utils/functions.py

utils/functions.py

@@ -5,74 +5,102 @@ from phonenumbers import geocoder, carrier
 import re
 import requests
 import os
-import json
 from datetime import datetime
 import logging
 
+from sqlalchemy import create_engine, Column, String, Integer, DateTime
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+from dotenv import load_dotenv
+
+# Załaduj zmienne środowiskowe z pliku .env (jeśli używasz)
+# W Streamlit Spaces Secrets są automatycznie dostępne
+load_dotenv()
+
 # Konfiguracja logowania
-logging.basicConfig(filename='app.log', level=logging.INFO, format='%(asctime)s %(levelname)s:%(message)s')
-
-# Inicjalizacja ścieżki bazowej
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-
-# Ścieżka do pliku JSON przechowującego fałszywe numery
-FAKE_NUMBERS_FILE = os.path.join(BASE_DIR, '..', 'fake_numbers.json')
-
-# Inicjalizacja pliku JSON przechowującego fałszywe numery
-def init_fake_numbers_file():
-    if not os.path.exists(FAKE_NUMBERS_FILE):
-        with open(FAKE_NUMBERS_FILE, 'w', encoding='utf-8') as f:
-            json.dump([], f, ensure_ascii=False, indent=4)
-        logging.info(f"Utworzono nowy plik {FAKE_NUMBERS_FILE}.")
-    else:
-        try:
-            with open(FAKE_NUMBERS_FILE, 'r', encoding='utf-8') as f:
-                json.load(f)
-            logging.info(f"Plik {FAKE_NUMBERS_FILE} został załadowany pomyślnie.")
-        except json.JSONDecodeError:
-            # Jeśli plik jest uszkodzony lub pusty, zresetuj go do pustej listy
-            with open(FAKE_NUMBERS_FILE, 'w', encoding='utf-8') as f:
-                json.dump([], f, ensure_ascii=False, indent=4)
-            logging.warning(f"Plik {FAKE_NUMBERS_FILE} był uszkodzony i został zresetowany.")
-
-# Dodanie numeru telefonu do pliku JSON
+logging.basicConfig(
+    filename='app.log',
+    level=logging.INFO,
+    format='%(asctime)s %(levelname)s:%(message)s'
+)
+
+# Konfiguracja bazy danych SQLite
+DATABASE_URL = "sqlite:///scam_detector.db"
+
+engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+Base = declarative_base()
+
+# Definicje modeli
+class FakeNumber(Base):
+    __tablename__ = "fake_numbers"
+    id = Column(Integer, primary_key=True, index=True)
+    phone_number = Column(String, unique=True, index=True, nullable=False)
+    reported_at = Column(DateTime, default=datetime.utcnow)
+
+class AnalysisHistory(Base):
+    __tablename__ = "analysis_history"
+    id = Column(Integer, primary_key=True, index=True)
+    timestamp = Column(DateTime, default=datetime.utcnow)
+    message = Column(String, nullable=False)
+    phone_number = Column(String, nullable=False)
+    analysis = Column(String, nullable=False)
+    risk_assessment = Column(String, nullable=False)
+    recommendations = Column(String, nullable=False)
+
+class Stats(Base):
+    __tablename__ = "stats"
+    id = Column(Integer, primary_key=True, index=True)
+    total_analyses = Column(Integer, default=0)
+    total_frauds_detected = Column(Integer, default=0)
+
+# Tworzenie tabel w bazie danych
+Base.metadata.create_all(bind=engine)
+
+# Funkcje pomocnicze
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+# Dodanie numeru telefonu do bazy danych
 def add_fake_number(phone_number):
-    logging.info(f"Próbuję dodać numer: {phone_number}")
+    db = SessionLocal()
     try:
-        with open(FAKE_NUMBERS_FILE, 'r', encoding='utf-8') as f:
-            fake_numbers = json.load(f)
-    except (json.JSONDecodeError, FileNotFoundError):
-        fake_numbers = []
-        logging.warning(f"{FAKE_NUMBERS_FILE} jest pusty lub uszkodzony. Inicjalizuję nową listę.")
-
-    if not any(entry["phone_number"] == phone_number for entry in fake_numbers):
-        fake_numbers.append({
-            "phone_number": phone_number,
-            "reported_at": datetime.now().isoformat()
-        })
-        try:
-            with open(FAKE_NUMBERS_FILE, 'w', encoding='utf-8') as f:
-                json.dump(fake_numbers, f, ensure_ascii=False, indent=4)
-            logging.info(f"Numer {phone_number} został pomyślnie dodany do fake_numbers.json.")
+        existing_number = db.query(FakeNumber).filter(FakeNumber.phone_number == phone_number).first()
+        if not existing_number:
+            new_number = FakeNumber(phone_number=phone_number)
+            db.add(new_number)
+            db.commit()
+            db.refresh(new_number)
+            logging.info(f"Numer {phone_number} został pomyślnie dodany do bazy danych.")
             return True
-        except Exception as e:
-            logging.error(f"Nie udało się zapisać numeru {phone_number}: {e}")
+        else:
+            logging.info(f"Numer {phone_number} już istnieje w bazie danych.")
             return False
-    else:
-        logging.info(f"Numer {phone_number} już istnieje w fake_numbers.json.")
-        return False  # Numer już istnieje
+    except Exception as e:
+        db.rollback()
+        logging.error(f"Nie udało się zapisać numeru {phone_number}: {e}")
+        return False
+    finally:
+        db.close()
 
-# Sprawdzenie, czy numer telefonu jest w pliku JSON
+# Sprawdzenie, czy numer telefonu jest w bazie danych
 def is_fake_number(phone_number):
+    db = SessionLocal()
     try:
-        with open(FAKE_NUMBERS_FILE, 'r', encoding='utf-8') as f:
-            fake_numbers = json.load(f)
-        exists = any(entry["phone_number"] == phone_number for entry in fake_numbers)
+        exists = db.query(FakeNumber).filter(FakeNumber.phone_number == phone_number).first() is not None
         logging.info(f"Sprawdzanie numeru {phone_number}: {'znaleziony' if exists else 'nie znaleziony'}.")
         return exists
-    except (json.JSONDecodeError, FileNotFoundError):
-        logging.error(f"Nie udało się załadować {FAKE_NUMBERS_FILE}.")
+    except Exception as e:
+        logging.error(f"Nie udało się sprawdzić numeru {phone_number}: {e}")
         return False
+    finally:
+        db.close()
 
 # Funkcja do weryfikacji numeru telefonu
 def get_phone_info(phone_number):
@@ -244,87 +272,87 @@ Provide your analysis and conclusions following the guidelines above."""
 
 # Inicjalizacja pliku statystyk
 def init_stats_file():
-    stats_file = os.path.join(BASE_DIR, '..', 'stats.json')
-    if not os.path.exists(stats_file):
-        with open(stats_file, 'w', encoding='utf-8') as f:
-            json.dump({"total_analyses": 0, "total_frauds_detected": 0}, f, ensure_ascii=False, indent=4)
-        logging.info(f"Utworzono nowy plik statystyk {stats_file}.")
+    db = SessionLocal()
+    try:
+        stats = db.query(Stats).first()
+        if not stats:
+            new_stats = Stats()
+            db.add(new_stats)
+            db.commit()
+            logging.info("Utworzono nowe statystyki.")
+    except Exception as e:
+        db.rollback()
+        logging.error(f"Nie udało się zainicjalizować statystyk: {e}")
+    finally:
+        db.close()
 
 # Pobranie statystyk
 def get_stats():
-    stats_file = os.path.join(BASE_DIR, '..', 'stats.json')
+    db = SessionLocal()
     try:
-        with open(stats_file, 'r', encoding='utf-8') as f:
-            stats = json.load(f)
-        logging.info("Statystyki zostały pobrane pomyślnie.")
-        return stats
-    except (json.JSONDecodeError, FileNotFoundError) as e:
-        logging.error(f"Nie udało się załadować statystyk: {e}")
+        stats = db.query(Stats).first()
+        if stats:
+            logging.info("Statystyki zostały pobrane pomyślnie.")
+            return {"total_analyses": stats.total_analyses, "total_frauds_detected": stats.total_frauds_detected}
+        else:
+            logging.info("Brak statystyk w bazie danych.")
+            return {"total_analyses": 0, "total_frauds_detected": 0}
+    except Exception as e:
+        logging.error(f"Nie udało się pobrać statystyk: {e}")
         return {"total_analyses": 0, "total_frauds_detected": 0}
+    finally:
+        db.close()
 
 # Aktualizacja statystyk analizy
 def update_stats(fraud_detected=False):
-    stats_file = os.path.join(BASE_DIR, '..', 'stats.json')
-    try:
-        with open(stats_file, 'r', encoding='utf-8') as f:
-            stats = json.load(f)
-    except (json.JSONDecodeError, FileNotFoundError):
-        stats = {"total_analyses": 0, "total_frauds_detected": 0}
-        logging.warning(f"{stats_file} jest pusty lub uszkodzony. Inicjalizuję nową listę.")
-
-    stats["total_analyses"] += 1
-    if fraud_detected:
-        stats["total_frauds_detected"] += 1
-
+    db = SessionLocal()
     try:
-        with open(stats_file, 'w', encoding='utf-8') as f:
-            json.dump(stats, f, ensure_ascii=False, indent=4)
-        logging.info(f"Statystyki zostały zaktualizowane: {stats}.")
+        stats = db.query(Stats).first()
+        if not stats:
+            stats = Stats()
+            db.add(stats)
+        
+        stats.total_analyses += 1
+        if fraud_detected:
+            stats.total_frauds_detected += 1
+
+        db.commit()
+        logging.info(f"Statystyki zostały zaktualizowane: Analiz {stats.total_analyses}, Oszustw {stats.total_frauds_detected}.")
     except Exception as e:
+        db.rollback()
         logging.error(f"Nie udało się zaktualizować statystyk: {e}")
-
-# Inicjalizacja pliku historii analiz
-def init_history_file():
-    history_file = os.path.join(BASE_DIR, '..', 'history.json')
-    if not os.path.exists(history_file):
-        with open(history_file, 'w', encoding='utf-8') as f:
-            json.dump([], f, ensure_ascii=False, indent=4)
-        logging.info(f"Utworzono nowy plik historii analiz {history_file}.")
+    finally:
+        db.close()
 
 # Dodanie wpisu do historii analiz
 def add_to_history(message, phone_number, analysis, risk, recommendations):
-    history_file = os.path.join(BASE_DIR, '..', 'history.json')
+    db = SessionLocal()
     try:
-        with open(history_file, 'r', encoding='utf-8') as f:
-            history = json.load(f)
-    except (json.JSONDecodeError, FileNotFoundError):
-        history = []
-        logging.warning(f"{history_file} jest pusty lub uszkodzony. Inicjalizuję nową listę.")
-
-    history.append({
-        "timestamp": datetime.now().isoformat(),
-        "message": message,
-        "phone_number": phone_number,
-        "analysis": analysis,
-        "risk_assessment": risk,
-        "recommendations": recommendations
-    })
-
-    try:
-        with open(history_file, 'w', encoding='utf-8') as f:
-            json.dump(history, f, ensure_ascii=False, indent=4)
+        new_entry = AnalysisHistory(
+            message=message,
+            phone_number=phone_number,
+            analysis=analysis,
+            risk_assessment=risk,
+            recommendations=recommendations
+        )
+        db.add(new_entry)
+        db.commit()
         logging.info(f"Dodano wpis do historii analiz dla numeru {phone_number}.")
     except Exception as e:
+        db.rollback()
         logging.error(f"Nie udało się zapisać historii analiz dla numeru {phone_number}: {e}")
+    finally:
+        db.close()
 
 # Pobranie historii analiz
 def get_history():
-    history_file = os.path.join(BASE_DIR, '..', 'history.json')
+    db = SessionLocal()
     try:
-        with open(history_file, 'r', encoding='utf-8') as f:
-            history = json.load(f)
+        history = db.query(AnalysisHistory).order_by(AnalysisHistory.timestamp.desc()).all()
         logging.info("Historia analiz została pobrana pomyślnie.")
         return history
-    except (json.JSONDecodeError, FileNotFoundError) as e:
-        logging.error(f"Nie udało się załadować historii analiz: {e}")
+    except Exception as e:
+        logging.error(f"Nie udało się pobrać historii analiz: {e}")
         return []
+    finally:
+        db.close()