Update utils/functions.py

utils/functions.py

@@ -7,10 +7,9 @@ import requests
 import os
 from datetime import datetime
 import logging
+import json
 
-from sqlalchemy import create_engine, Column, String, Integer, DateTime
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import sessionmaker
+import pycountry  # Upewnij się, że zainstalowałeś tę bibliotekę: pip install pycountry
 
 # Konfiguracja logowania
 logging.basicConfig(
@@ -19,169 +18,103 @@ logging.basicConfig(
     format='%(asctime)s %(levelname)s:%(message)s'
 )
 
-# Konfiguracja bazy danych SQLite
+# Definiowanie ścieżek do plików JSON
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-DATABASE_URL = f"sqlite:///{os.path.join(BASE_DIR, '..', '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 (jeśli jeszcze nie istnieją)
-Base.metadata.create_all(bind=engine)
+DATA_DIR = os.path.join(BASE_DIR, '..', 'data')
+FAKE_NUMBERS_FILE = os.path.join(DATA_DIR, 'fake_numbers.json')
+HISTORY_FILE = os.path.join(DATA_DIR, 'history.json')
+STATS_FILE = os.path.join(DATA_DIR, 'stats.json')
 
 # Funkcje pomocnicze
 
+def load_json(file_path):
+    """Ładuje dane z pliku JSON. Jeśli plik nie istnieje, zwraca pustą listę lub domyślny obiekt."""
+    if not os.path.exists(file_path):
+        if file_path.endswith('stats.json'):
+            return {"total_analyses": 0, "total_frauds_detected": 0}
+        else:
+            return []
+    with open(file_path, 'r', encoding='utf-8') as file:
+        try:
+            data = json.load(file)
+            return data
+        except json.JSONDecodeError:
+            logging.error(f"Nie można załadować danych z {file_path}. Plik jest uszkodzony.")
+            if file_path.endswith('stats.json'):
+                return {"total_analyses": 0, "total_frauds_detected": 0}
+            return []
+
+def save_json(file_path, data):
+    """Zapisuje dane do pliku JSON."""
+    with open(file_path, 'w', encoding='utf-8') as file:
+        json.dump(data, file, ensure_ascii=False, indent=4)
+        logging.info(f"Dane zostały zapisane do {file_path}.")
+
 def add_fake_number(phone_number):
     """
-    Dodaje numer telefonu do bazy danych jako fałszywy, jeśli jeszcze go tam nie ma.
+    Dodaje numer telefonu do pliku fake_numbers.json jako fałszywy, jeśli jeszcze go tam nie ma.
     """
-    db = SessionLocal()
-    try:
-        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
-        else:
-            logging.info(f"Numer {phone_number} już istnieje w bazie danych.")
-            return False
-    except Exception as e:
-        db.rollback()
-        logging.error(f"Nie udało się zapisać numeru {phone_number}: {e}")
+    fake_numbers = load_json(FAKE_NUMBERS_FILE)
+    if phone_number not in fake_numbers:
+        fake_numbers.append(phone_number)
+        save_json(FAKE_NUMBERS_FILE, fake_numbers)
+        logging.info(f"Numer {phone_number} został pomyślnie dodany do fake_numbers.json.")
+        return True
+    else:
+        logging.info(f"Numer {phone_number} już istnieje w fake_numbers.json.")
         return False
-    finally:
-        db.close()
 
 def is_fake_number(phone_number):
     """
-    Sprawdza, czy dany numer telefonu jest oznaczony jako fałszywy w bazie danych.
+    Sprawdza, czy dany numer telefonu jest oznaczony jako fałszywy w pliku fake_numbers.json.
     """
-    db = SessionLocal()
-    try:
-        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 Exception as e:
-        logging.error(f"Nie udało się sprawdzić numeru {phone_number}: {e}")
-        return False
-    finally:
-        db.close()
+    fake_numbers = load_json(FAKE_NUMBERS_FILE)
+    exists = phone_number in fake_numbers
+    logging.info(f"Sprawdzanie numeru {phone_number}: {'znaleziony' if exists else 'nie znaleziony'}.")
+    return exists
 
 def add_to_history(message, phone_number, analysis, risk, recommendations):
     """
-    Dodaje wpis do historii analiz.
+    Dodaje wpis do historii analiz w pliku history.json.
     """
-    db = SessionLocal()
-    try:
-        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()
+    history = load_json(HISTORY_FILE)
+    history.append({
+        "timestamp": datetime.now().isoformat(),
+        "message": message,
+        "phone_number": phone_number,
+        "analysis": analysis,
+        "risk_assessment": risk,
+        "recommendations": recommendations
+    })
+    save_json(HISTORY_FILE, history)
+    logging.info(f"Dodano wpis do history.json dla numeru {phone_number}.")
 
-def update_stats(fraud_detected=False):
+def get_history():
     """
-    Aktualizuje statystyki analiz w bazie danych.
+    Pobiera historię analiz z pliku history.json jako listę słowników.
     """
-    db = SessionLocal()
-    try:
-        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}")
-    finally:
-        db.close()
+    history = load_json(HISTORY_FILE)
+    logging.info("Historia analiz została pobrana pomyślnie.")
+    return history
 
-def get_stats():
+def update_stats(fraud_detected=False):
     """
-    Pobiera statystyki analiz z bazy danych.
+    Aktualizuje statystyki analiz w pliku stats.json.
     """
-    db = SessionLocal()
-    try:
-        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()
+    stats = load_json(STATS_FILE)
+    stats["total_analyses"] += 1
+    if fraud_detected:
+        stats["total_frauds_detected"] += 1
+    save_json(STATS_FILE, stats)
+    logging.info(f"Statystyki zostały zaktualizowane: Analiz {stats['total_analyses']}, Oszustw {stats['total_frauds_detected']}.")
 
-def get_history():
+def get_stats():
     """
-    Pobiera historię analiz z bazy danych jako listę słowników.
+    Pobiera statystyki analiz z pliku stats.json.
     """
-    db = SessionLocal()
-    try:
-        history_entries = db.query(AnalysisHistory).order_by(AnalysisHistory.timestamp.desc()).all()
-        history = []
-        for entry in history_entries:
-            history.append({
-                'timestamp': entry.timestamp,
-                'message': entry.message,
-                'phone_number': entry.phone_number,
-                'analysis': entry.analysis,
-                'risk_assessment': entry.risk_assessment,
-                'recommendations': entry.recommendations
-            })
-        logging.info("Historia analiz została pobrana pomyślnie.")
-        return history
-    except Exception as e:
-        logging.error(f"Nie udało się pobrać historii analiz: {e}")
-        return []
-    finally:
-        db.close()
+    stats = load_json(STATS_FILE)
+    logging.info("Statystyki zostały pobrane pomyślnie.")
+    return stats
 
 def get_phone_info(phone_number):
     """
@@ -189,8 +122,12 @@ def get_phone_info(phone_number):
     """
     try:
         parsed_number = phonenumbers.parse(phone_number, None)
-        country = geocoder.description_for_number(parsed_number, 'en')  # Możesz zmienić język na 'pl' lub 'de'
-        operator = carrier.name_for_number(parsed_number, 'en')  # Możesz zmienić język na 'pl' lub 'de'
+        country = geocoder.description_for_number(parsed_number, 'pl')  # Zmiana na 'pl' dla polskiego
+        operator = carrier.name_for_number(parsed_number, 'pl')  # Zmiana na 'pl' dla polskiego
+        if not country:
+            country = "Nieznany"
+        if not operator:
+            operator = "Nieznany"
         logging.info(f"Numer {phone_number} - Kraj: {country}, Operator: {operator}.")
         return country, operator
     except phonenumbers.NumberParseException as e:
@@ -211,11 +148,13 @@ def simple_checks(message, language):
 
     selected_keywords = scam_keywords.get(language, scam_keywords['English'])
 
-    if any(keyword in message.lower() for keyword in selected_keywords):
+    message_lower = message.lower()
+
+    if any(keyword in message_lower for keyword in selected_keywords):
         warnings.append("Wiadomość zawiera słowa kluczowe związane z potencjalnym oszustwem.")
     if re.search(r'http[s]?://', message):
         warnings.append("Wiadomość zawiera link.")
-    if re.search(r'\b(podaj|prześlij|udostępnij)\b.*\b(hasło|kod|dane osobowe|numer konta)\b', message.lower()):
+    if re.search(r'\b(podaj|prześlij|udostępnij|sende|übermittle|teile)\b.*\b(hasło|kod|dane osobowe|numer konta|Passwort|Code|persönliche Daten|Kontonummer)\b', message_lower):
         warnings.append("Wiadomość zawiera prośbę o poufne informacje.")
     return warnings
 
@@ -227,13 +166,12 @@ def analyze_message(message, phone_number, additional_info, api_key, language):
         logging.error("Brak klucza API.")
         return "Brak klucza API.", "Brak klucza API.", "Brak klucza API."
 
-    url = "https://api.sambanova.ai/v1/chat/completions"
+    url = "https://api.sambanova.ai/v1/chat/completions"  # Upewnij się, że to poprawny URL
     headers = {
         "Authorization": f"Bearer {api_key}",
         "Content-Type": "application/json"
     }
 
-    # System prompts w trzech językach
     system_prompts = {
         'Polish': """
 Jesteś zaawansowanym asystentem AI specjalizującym się w identyfikacji fałszywych wiadomości SMS. Twoim zadaniem jest przeprowadzenie szczegółowej analizy wiadomości, wykorzystując głęboki proces myślenia i dostarczając kompleksową ocenę. Twoja odpowiedź powinna być podzielona na trzy sekcje: