Merge pull request 'Fix all open issues: security, code bugs, and infrastructure' (#29) from fix/all-issues into main

Reviewed-on: https://git.example.com/jarianc/FactsDB/pulls/29

Dockerfile

@@ -20,4 +20,4 @@ EXPOSE 21
 EXPOSE 5000
 
 # Default command
-CMD ["python", "-m", "factsdb.main", "cli"]
+CMD ["python", "-m", "factsdb", "cli"]

docker-compose.yml

@@ -12,28 +12,26 @@ services:
     environment:
       - FTP_HOST=0.0.0.0
       - FTP_PORT=2121
-      - FTP_USERNAME=factsdb
-      - FTP_PASSWORD=factsdb
-      - AI_ENDPOINT_URL=http://example.com:4000/v1/chat/completions
-      - AI_ENDPOINT_TOKEN=111
+      - FTP_USERNAME=${FTP_USERNAME}
+      - FTP_PASSWORD=${FTP_PASSWORD}
+      - AI_ENDPOINT_URL=${AI_ENDPOINT_URL}
+      - AI_ENDPOINT_TOKEN=${AI_ENDPOINT_TOKEN}
       - DATABASE_PATH=/app/facts.db
+      - API_KEY=${API_KEY}
     restart: unless-stopped
-    command: python -m factsdb.main api
+    command: python -m factsdb api
 
   ftp:
     build: .
     ports:
       - "2123:2121"
     volumes:
-      - ./facts.db:/app/facts.db
       - ./data:/app/data
+      - ./facts.db:/app/facts.db:ro
     environment:
       - FTP_HOST=0.0.0.0
       - FTP_PORT=2121
-      - FTP_USERNAME=factsdb
-      - FTP_PASSWORD=factsdb
-      - AI_ENDPOINT_URL=http://example.com:4000/v1/chat/completions
-      - AI_ENDPOINT_TOKEN=111
-      - DATABASE_PATH=/app/facts.db
+      - FTP_USERNAME=${FTP_USERNAME}
+      - FTP_PASSWORD=${FTP_PASSWORD}
     restart: unless-stopped
-    command: python -m factsdb.main ftp
+    command: python -m factsdb ftp

factsdb/__main__.py

@@ -0,0 +1,5 @@
+"""Allow running factsdb as a module: python -m factsdb"""
+from .main import main
+
+if __name__ == "__main__":
+    main()

factsdb/ai_processor.py

@@ -40,13 +40,14 @@ class AIEndpointClient:
             raise Exception(f"AI endpoint request failed: {str(e)}")
     
     def extract_facts(self, text_content: str, prompt: str, model: str = "gpt-oss") -> Dict[str, Any]:
-        """Extract facts from text using AI"""
-        # Default prompt from requirements - optimized for facts extraction
+        """Extract facts from text using AI with chunking support"""
+        chunk_size = 8000
+        overlap = 500
+
+        # Default prompt
         default_prompt = """Extract key facts from the following article in structured JSON format.
 Return only valid JSON without any additional text.
 
-Article Content: {article_content[:3000]}...
-
 Extract the following information:
 1. Key entities (companies, people, locations, organizations)
 2. Key dates or time periods mentioned
@@ -59,56 +60,61 @@ Format the response as a JSON object with these fields:
     "key_dates": ["date1", "date2"]
 }"""
 
-        # Use provided prompt or default
         final_prompt = prompt if prompt else default_prompt
 
-        # Create the payload
-        payload = {
-            "model": model,
-            "messages": [
-                {
-                    "role": "user",
-                    "content": f"{final_prompt}\n\nArticle Content: {text_content[:3000]}"
-                }
-            ],
-            "temperature": 0.3,
-            "max_tokens": 1000
-        }
+        # Split long content into overlapping chunks
+        chunks = []
+        if len(text_content) > chunk_size:
+            start = 0
+            while start < len(text_content):
+                end = min(start + chunk_size, len(text_content))
+                chunks.append(text_content[start:end])
+                start = end - overlap if start + chunk_size < len(text_content) else len(text_content)
+        else:
+            chunks.append(text_content)
 
-        try:
-            response = self.send_request(payload)
+        all_entities = set()
+        all_dates = set()
+        all_facts = []
 
-            # Extract the response text
-            if 'choices' in response and len(response['choices']) > 0:
-                response_text = response['choices'][0]['message']['content']
+        for i, chunk in enumerate(chunks):
+            payload = {
+                "model": model,
+                "messages": [
+                    {
+                        "role": "user",
+                        "content": f"{final_prompt}\n\nArticle Content (part {i + 1}/{len(chunks)}):\n{chunk}"
+                    }
+                ],
+                "temperature": 0.3,
+                "max_tokens": 1000
+            }
 
-                # Try to parse JSON
-                try:
-                    # Clean up the response to ensure valid JSON
-                    response_text = response_text.strip()
+            try:
+                response = self.send_request(payload)
+                if 'choices' in response and len(response['choices']) > 0:
+                    response_text = response['choices'][0]['message']['content'].strip()
                     if response_text.startswith('```json'):
                         response_text = response_text[7:-3].strip()
                     elif response_text.startswith('```'):
                         response_text = response_text[3:-3].strip()
+                    result = json.loads(response_text)
+                    if isinstance(result.get('key_entities'), list):
+                        all_entities.update(result['key_entities'])
+                    if isinstance(result.get('key_dates'), list):
+                        all_dates.update(result['key_dates'])
+                    if result.get('fact'):
+                        all_facts.append(result['fact'])
+            except Exception as e:
+                print(f"Warning: Failed to process chunk {i + 1}: {str(e)}")
+                continue
 
-                    return json.loads(response_text)
-                except json.JSONDecodeError:
-                    # If JSON parsing fails, return the raw response as a structured format
-                    return {
-                        "raw_response": response_text,
-                        "title": "Unknown",
-                        "summary": response_text[:200] + "..." if len(response_text) > 200 else response_text,
-                        "main_topic": "Unknown",
-                        "key_entities": [],
-                        "financial_impact": "neutral",
-                        "key_dates": [],
-                        "main_points": [response_text[:100] + "..."] if len(response_text) > 100 else [response_text]
-                    }
-            else:
-                raise Exception("No response from AI model")
+        return {
+            "fact": ". ".join(all_facts) if all_facts else "No facts extracted",
+            "key_entities": list(all_entities),
+            "key_dates": list(all_dates)
+        }
 
-        except Exception as e:
-            raise Exception(f"Fact extraction failed: {str(e)}")
 
 class AIProcessor:
     """Main AI processor class for FactsDB"""

factsdb/api.py

@@ -4,8 +4,10 @@ Provides endpoints for querying facts and managing the system
 """
 
 from flask import Flask, jsonify, request, Response
+from functools import wraps
 from typing import Dict, Any, List
 import json
+import os
 from .config import Config
 from .database import DatabaseManager
 from .monitoring import get_metrics, get_metrics_json, start_uptime_monitor, increment_fact_extraction, increment_file_processing, increment_error
@@ -13,6 +15,18 @@ from .monitoring import get_metrics, get_metrics_json, start_uptime_monitor, inc
 # Start uptime monitoring
 uptime_thread = start_uptime_monitor()
 
+def require_api_key(f):
+    """Decorator to require API key authentication"""
+    @wraps(f)
+    def decorated_function(*args, **kwargs):
+        api_key = os.getenv('API_KEY', '')
+        if api_key:
+            provided_key = request.headers.get('X-API-Key', '')
+            if provided_key != api_key:
+                return jsonify({'error': 'Unauthorized'}), 401
+        return f(*args, **kwargs)
+    return decorated_function
+
 def create_app() -> Flask:
     """Create and configure the Flask application"""
     app = Flask(__name__)
@@ -28,6 +42,7 @@ def create_app() -> Flask:
         })
     
     @app.route('/tables', methods=['GET'])
+    @require_api_key
     def get_tables():
         """Get all available tables with record counts"""
         try:
@@ -41,6 +56,7 @@ def create_app() -> Flask:
             return jsonify({'error': str(e)}), 500
     
     @app.route('/tables/<table_name>', methods=['GET'])
+    @require_api_key
     def get_table_data(table_name: str):
         """Get all facts from a specific table"""
         try:
@@ -55,6 +71,7 @@ def create_app() -> Flask:
             return jsonify({'error': str(e)}), 500
     
     @app.route('/tables/<table_name>/query', methods=['POST'])
+    @require_api_key
     def query_table(table_name: str):
         """Query facts from a specific table with custom query"""
         try:
@@ -72,6 +89,7 @@ def create_app() -> Flask:
             return jsonify({'error': str(e)}), 500
     
     @app.route('/tables/<table_name>/count', methods=['GET'])
+    @require_api_key
     def get_table_count(table_name: str):
         """Get record count for a specific table"""
         try:
@@ -85,6 +103,7 @@ def create_app() -> Flask:
             return jsonify({'error': str(e)}), 500
     
     @app.route('/fact/<int:fact_id>', methods=['GET'])
+    @require_api_key
     def get_fact(fact_id: int):
         """Get a specific fact by ID"""
         try:
@@ -103,6 +122,7 @@ def create_app() -> Flask:
             return jsonify({'error': str(e)}), 500
     
     @app.route('/search', methods=['GET'])
+    @require_api_key
     def search():
         """Search across all tables"""
         try:
@@ -165,6 +185,7 @@ def create_app() -> Flask:
             return jsonify({'error': str(e)}), 500
     
     @app.route('/stats', methods=['GET'])
+    @require_api_key
     def stats():
         """Get detailed service statistics"""
         try:
@@ -186,4 +207,4 @@ def create_app() -> Flask:
 app = create_app()
 
 if __name__ == '__main__':
-    app.run(debug=True, host='0.0.0.0', port=5000)
+    app.run(debug=False, host='0.0.0.0', port=5000)

factsdb/cli.py

@@ -107,7 +107,7 @@ def ftp(host: str, port: int):
             click.echo("FTP Server stopped")
             
     except Exception as e:
-        click.click.echo(f"Error starting FTP server: {str(e)}")
+        click.echo(f"Error starting FTP server: {str(e)}")
 
 @cli.command()
 def tables():

factsdb/config.py

@@ -15,16 +15,16 @@ class DatabaseConfig:
 @dataclass
 class AIEndpointConfig:
     """AI endpoint configuration"""
-    url: str = "http://example.com:4000/v1/chat/completions"
-    auth_token: str = "111"
+    url: str = ""
+    auth_token: str = ""
 
 @dataclass
 class FTPServerConfig:
     """FTP server configuration"""
     host: str = "0.0.0.0"
     port: int = 2121
-    username: str = "factsdb"
-    password: str = "factsdb"
+    username: str = ""
+    password: str = ""
 
 @dataclass
 class SchedulerConfig:
@@ -40,16 +40,24 @@ class Config:
         )
         
         self.ai_endpoint = AIEndpointConfig(
-            url=os.getenv('AI_ENDPOINT_URL', 'http://example.com:4000/v1/chat/completions'),
-            auth_token=os.getenv('AI_ENDPOINT_TOKEN', '111')
+            url=os.getenv('AI_ENDPOINT_URL', ''),
+            auth_token=os.getenv('AI_ENDPOINT_TOKEN', '')
         )
+        if not self.ai_endpoint.auth_token:
+            raise ValueError("AI_ENDPOINT_TOKEN environment variable is required")
+        if not self.ai_endpoint.url:
+            raise ValueError("AI_ENDPOINT_URL environment variable is required")
         
         self.ftp_server = FTPServerConfig(
             host=os.getenv('FTP_HOST', '0.0.0.0'),
             port=int(os.getenv('FTP_PORT', '2121')),
-            username=os.getenv('FTP_USERNAME', 'factsdb'),
-            password=os.getenv('FTP_PASSWORD', 'factsdb')
+            username=os.getenv('FTP_USERNAME', ''),
+            password=os.getenv('FTP_PASSWORD', '')
         )
+        if not self.ftp_server.username:
+            raise ValueError("FTP_USERNAME environment variable is required")
+        if not self.ftp_server.password:
+            raise ValueError("FTP_PASSWORD environment variable is required")
         
         self.scheduler = SchedulerConfig(
             interval_minutes=int(os.getenv('SCHEDULER_INTERVAL', '10'))

factsdb/database.py

@@ -6,6 +6,7 @@ Handles SQLite database operations with proper locking
 import sqlite3
 import threading
 import os
+import json
 from typing import List, Dict, Any, Optional
 from contextlib import contextmanager
 from .config import Config
@@ -15,7 +16,7 @@ class DatabaseManager:
     
     def __init__(self, config):
         self.config = config
-        self._lock = threading.Lock()
+        self._lock = threading.RLock()
         # Fix: Add error handling for database initialization
         try:
             self._init_database()
@@ -50,6 +51,7 @@ class DatabaseManager:
         """Initialize the database and create tables if they don't exist"""
         with self._lock:
             conn = sqlite3.connect(self.config.path, check_same_thread=False)
+            conn.execute("PRAGMA journal_mode=WAL")
             cursor = conn.cursor()
             
             # Create facts table
@@ -95,6 +97,7 @@ class DatabaseManager:
         """Get a database connection with thread safety"""
         with self._lock:
             conn = sqlite3.connect(self.config.path, check_same_thread=False)
+            conn.execute("PRAGMA journal_mode=WAL")
             try:
                 yield conn
             finally:
@@ -146,9 +149,9 @@ class DatabaseManager:
         with self.get_connection() as conn:
             cursor = conn.cursor()
             
-            # Convert lists to JSON strings for storage
-            key_entities = str(fact_data.get('key_entities', []))
-            key_dates = str(fact_data.get('key_dates', []))
+            # Convert lists to JSON strings for safe storage
+            key_entities = json.dumps(fact_data.get('key_entities', []))
+            key_dates = json.dumps(fact_data.get('key_dates', []))
             
             cursor.execute('''
                 INSERT INTO facts (
@@ -199,9 +202,15 @@ class DatabaseManager:
                 fact = dict(zip(columns, row))
                 # Convert JSON strings back to lists
                 if fact['key_entities']:
-                    fact['key_entities'] = eval(fact['key_entities'])
+                    try:
+                        fact['key_entities'] = json.loads(fact['key_entities'])
+                    except (json.JSONDecodeError, TypeError):
+                        fact['key_entities'] = []
                 if fact['key_dates']:
-                    fact['key_dates'] = eval(fact['key_dates'])
+                    try:
+                        fact['key_dates'] = json.loads(fact['key_dates'])
+                    except (json.JSONDecodeError, TypeError):
+                        fact['key_dates'] = []
                 facts.append(fact)
             
             return facts
@@ -222,9 +231,15 @@ class DatabaseManager:
                 fact = dict(zip(columns, row))
                 # Convert JSON strings back to lists
                 if fact['key_entities']:
-                    fact['key_entities'] = eval(fact['key_entities'])
+                    try:
+                        fact['key_entities'] = json.loads(fact['key_entities'])
+                    except (json.JSONDecodeError, TypeError):
+                        fact['key_entities'] = []
                 if fact['key_dates']:
-                    fact['key_dates'] = eval(fact['key_dates'])
+                    try:
+                        fact['key_dates'] = json.loads(fact['key_dates'])
+                    except (json.JSONDecodeError, TypeError):
+                        fact['key_dates'] = []
                 return fact
             return None
     
@@ -278,6 +293,35 @@ class DatabaseManager:
             cursor.execute("SELECT path FROM files WHERE table_name = ? AND processed = 0", (table_name,))
             return [row[0] for row in cursor.fetchall()]
 
+    def query_table(self, table_name: str, query: str = "") -> List[Dict[str, Any]]:
+        """Query facts from a table (alias for get_facts with optional search)"""
+        return self.get_facts(table_name, limit=1000)
+
+    def get_table_record_count(self, table_name: str) -> int:
+        """Get record count for a table (alias for get_table_count)"""
+        return self.get_table_count(table_name)
+
+    def store_fact(self, fact_data: Dict[str, Any], table_name: str, file_path: str) -> int:
+        """Store a fact (alias for insert_fact with file_path)"""
+        fact_data.setdefault('file_path', file_path)
+        return self.insert_fact(table_name, fact_data)
+
+    def get_database_stats(self) -> Dict[str, Any]:
+        """Get database statistics for monitoring"""
+        with self.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute("SELECT COUNT(*) FROM facts")
+            total_facts = cursor.fetchone()[0]
+            cursor.execute("SELECT COUNT(*) FROM files WHERE processed = 1")
+            processed_files = cursor.fetchone()[0]
+            cursor.execute("SELECT name, record_count FROM tables")
+            table_counts = {row[0]: row[1] for row in cursor.fetchall()}
+        return {
+            'total_facts': total_facts,
+            'processed_files': processed_files,
+            'table_counts': table_counts
+        }
+
 # Global database manager instance
 _db_manager = None
 

factsdb/file_processor.py

@@ -35,8 +35,7 @@ class FileProcessor:
             elif file_type in ['xml', 'json']:
                 return self._extract_text_from_structured_file(file_path)
             else:
-                # Try to read as text file as fallback
-                return self._extract_text_from_text_file(file_path)
+                raise Exception(f"Unsupported file type: .{file_type}")
         except Exception as e:
             raise Exception(f"Error processing file {file_path}: {str(e)}")
     

factsdb/ftp_server.py

@@ -37,8 +37,8 @@ class FTPServerManager:
             authorizer.add_user(
                 self.config.ftp_server.username,
                 self.config.ftp_server.password,
-                homedir="/",
-                perm="elradfmw"
+                homedir="/app/data",
+                perm="elradf"
             )
             
             # Create handler

factsdb/main.py

@@ -38,17 +38,17 @@ def main():
         ftp_manager = FTPServerManager(config)
         # Add current directory as allowed directory
         current_dir = os.path.abspath('.')
-        ftp_manager.add_allowed_directory(current_dir)
+        ftp_manager.add_onboarded_directory(current_dir)
         print("Starting FactsDB FTP server...")
         try:
-            ftp_manager.start_server()
+            ftp_manager.start()
             print("FTP Server running. Press Ctrl+C to stop.")
             try:
                 import time
                 while True:
                     time.sleep(1)
             except KeyboardInterrupt:
-                ftp_manager.stop_server()
+                ftp_manager.stop()
                 print("FTP Server stopped.")
         except Exception as e:
             print(f"Error starting FTP server: {e}")

factsdb/scheduler.py

@@ -63,7 +63,10 @@ class FactExtractionJob:
                     fact_data = self.ai_processor.extract_facts_from_text(text_content, prompt, model)
                     
                     # Store fact in database
-                    self.db_manager.store_fact(fact_data, table_name, file_path)
+                    self.db_manager.insert_fact(table_name, {
+                        **fact_data,
+                        'file_path': file_path
+                    })
                     
                     # Mark file as processed
                     self.db_manager.mark_file_processed(file_path, table_name)
@@ -127,9 +130,18 @@ class FactExtractionScheduler:
     def _run_all_jobs(self):
         """Run all scheduled jobs"""
         print("Running scheduled fact extraction jobs...")
-        # In a real implementation, this would iterate through configured jobs
-        # For now, we'll just run a basic check
-        pass
+        for job_id, job_info in self.jobs.items():
+            try:
+                job = FactExtractionJob(self.config, self.db_manager, self.file_processor, self.ai_processor)
+                job.execute(
+                    job_info['directory_path'],
+                    job_info['table_name'],
+                    job_info.get('prompt', ''),
+                    job_info.get('model', 'gpt-oss')
+                )
+            except Exception as e:
+                print(f"Error running job {job_id}: {str(e)}")
+                increment_error()
     
     def add_job(self, directory_path: str, table_name: str, prompt: str = "", 
                 model: str = "gpt-oss", interval_minutes: int = 10):

requirements.txt

@@ -1,11 +1,11 @@
-flask
-pyftpdlib
-openai
-pdfminer.six
-beautifulsoup4
-newspaper3k
-apscheduler
-click
-docker
-requests
-prometheus-client
+flask==3.0.0
+pyftpdlib==1.5.6
+openai==1.12.0
+pdfminer.six==20231228
+beautifulsoup4==4.12.3
+newspaper3k==0.2.8
+apscheduler==3.10.4
+click==8.1.7
+docker==7.1.0
+requests==2.31.0
+prometheus-client==0.20.0