jarianc/StockDocs
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

chore: remove dev artifacts, fix hardcoded path, add tests + license

Browse Source 
- Remove agent/agent.md (dev-time agent context dumps), .DS_Store,
  committed venv configs (pyvenv.cfg), 0-byte runtime cache
- Remove hardcoded  /home/userpath from cron_scraper feed lookup
- Replace ad-hoc test_implementation.py with pytest tests/test_scraper_cache.py
- ruff clean (33 fixes: bare excepts, unused Config, whitespace)
- Root pyproject.toml (activates shared Gitea CI), MIT LICENSE, README Tests
This commit is contained in:
Jarian Cottingham 2026-08-20 21:39:04 +00:00
parent ef51ef635c
commit 1271f0b21b
28 changed files with 379 additions and 886 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
.DS_Store vendored
View File

Binary file not shown.

6
.gitignore vendored
View File

@ -24,3 +24,9 @@ nohup.out
*.pyc *.pyc
*.log *.log
.DS_Store
pyvenv.cfg
processed_articles_cache.json

21
LICENSE Normal file
View File

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Jarian Cottingham
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

63
MCPServer/server.py
View File

@ -1,7 +1,6 @@
import chromadb import chromadb
from flask import Flask, request, jsonify, send_from_directory from flask import Flask, request, jsonify, send_from_directory
import os import os
import json
import requests import requests
import logging import logging
from datetime import datetime from datetime import datetime
@ -109,12 +108,12 @@ def get_embedding(text):
headers = { headers = {
"Content-Type": "application/json" "Content-Type": "application/json"
} }
# Add API key if available # Add API key if available
api_key = os.getenv("AI_SERVICE_API_KEY") api_key = os.getenv("AI_SERVICE_API_KEY")
if api_key: if api_key:
headers["Authorization"] = f"Bearer {api_key}" headers["Authorization"] = f"Bearer {api_key}"
response = requests.post( response = requests.post(
"http://example.com:4000/v1/embeddings", "http://example.com:4000/v1/embeddings",
json={ json={
@ -137,23 +136,23 @@ def get_diverse_articles(articles, max_diverse=5):
""" """
if len(articles) <= max_diverse: if len(articles) <= max_diverse:
return articles return articles
# More sophisticated diversity algorithm # More sophisticated diversity algorithm
diverse_articles = [] diverse_articles = []
source_count = {} source_count = {}
topic_count = {} topic_count = {}
# First pass: try to get articles from different sources # First pass: try to get articles from different sources
for article in articles: for article in articles:
source = article.get('metadata', {}).get('source', 'unknown') source = article.get('metadata', {}).get('source', 'unknown')
topic = article.get('metadata', {}).get('topic', 'unknown') topic = article.get('metadata', {}).get('topic', 'unknown')
# If we haven't reached max diversity and this source is new, add it # If we haven't reached max diversity and this source is new, add it
if len(diverse_articles) < max_diverse and source not in source_count: if len(diverse_articles) < max_diverse and source not in source_count:
diverse_articles.append(article) diverse_articles.append(article)
source_count[source] = 1 source_count[source] = 1
topic_count[topic] = topic_count.get(topic, 0) + 1 topic_count[topic] = topic_count.get(topic, 0) + 1
# Second pass: fill remaining slots with different topics if possible # Second pass: fill remaining slots with different topics if possible
if len(diverse_articles) < max_diverse: if len(diverse_articles) < max_diverse:
for article in articles: for article in articles:
@ -161,17 +160,17 @@ def get_diverse_articles(articles, max_diverse=5):
break break
source = article.get('metadata', {}).get('source', 'unknown') source = article.get('metadata', {}).get('source', 'unknown')
topic = article.get('metadata', {}).get('topic', 'unknown') topic = article.get('metadata', {}).get('topic', 'unknown')
# Add article if it's from a different topic and we haven't seen too many from this topic # Add article if it's from a different topic and we haven't seen too many from this topic
if source not in source_count and topic_count.get(topic, 0) < 2: if source not in source_count and topic_count.get(topic, 0) < 2:
diverse_articles.append(article) diverse_articles.append(article)
source_count[source] = 1 source_count[source] = 1
topic_count[topic] = topic_count.get(topic, 0) + 1 topic_count[topic] = topic_count.get(topic, 0) + 1
# If we still don't have enough, just return first few # If we still don't have enough, just return first few
if len(diverse_articles) < max_diverse: if len(diverse_articles) < max_diverse:
return articles[:max_diverse] return articles[:max_diverse]
return diverse_articles return diverse_articles
def query_chroma(question, n_results=10): def query_chroma(question, n_results=10):
@ -180,19 +179,19 @@ def query_chroma(question, n_results=10):
""" """
if not client: if not client:
return {"error": "ChromaDB connection failed"} return {"error": "ChromaDB connection failed"}
try: try:
# Get embedding for the question # Get embedding for the question
query_embedding = get_embedding(question) query_embedding = get_embedding(question)
if not query_embedding: if not query_embedding:
return {"error": "Failed to get embedding"} return {"error": "Failed to get embedding"}
# Query the collection # Query the collection
results = client.get_or_create_collection("news").query( results = client.get_or_create_collection("news").query(
query_embeddings=[query_embedding], query_embeddings=[query_embedding],
n_results=n_results, n_results=n_results,
) )
return results return results
except Exception as e: except Exception as e:
logger.error(f"Error querying ChromaDB: {e}") logger.error(f"Error querying ChromaDB: {e}")
@ -205,16 +204,16 @@ def query_vector_database():
"""Query the vector database""" """Query the vector database"""
data = request.get_json() data = request.get_json()
question = data.get("question") question = data.get("question")
if not question: if not question:
return jsonify({"error": "Missing 'question' in request body"}), 400 return jsonify({"error": "Missing 'question' in request body"}), 400
# Query ChromaDB for relevant articles # Query ChromaDB for relevant articles
results = query_chroma(question, n_results=10) results = query_chroma(question, n_results=10)
if "error" in results: if "error" in results:
return jsonify({"error": results["error"]}), 500 return jsonify({"error": results["error"]}), 500
# Process results to create diverse article set # Process results to create diverse article set
mcp_results = [] mcp_results = []
for doc, score, meta in zip( for doc, score, meta in zip(
@ -226,10 +225,10 @@ def query_vector_database():
"score": float(score), "score": float(score),
"metadata": meta "metadata": meta
}) })
# Apply diversity filtering # Apply diversity filtering
diverse_results = get_diverse_articles(mcp_results, 5) diverse_results = get_diverse_articles(mcp_results, 5)
return jsonify({"results": diverse_results}) return jsonify({"results": diverse_results})
@app.route("/articles/query", methods=["POST"]) @app.route("/articles/query", methods=["POST"])
@ -238,16 +237,16 @@ def query_articles():
data = request.get_json() data = request.get_json()
question = data.get("question") question = data.get("question")
max_results = data.get("max_results", 5) max_results = data.get("max_results", 5)
if not question: if not question:
return jsonify({"error": "Missing 'question' in request body"}), 400 return jsonify({"error": "Missing 'question' in request body"}), 400
# Query ChromaDB for relevant articles # Query ChromaDB for relevant articles
results = query_chroma(question, n_results=max_results) results = query_chroma(question, n_results=max_results)
if "error" in results: if "error" in results:
return jsonify({"error": results["error"]}), 500 return jsonify({"error": results["error"]}), 500
# Process results to create diverse article set # Process results to create diverse article set
mcp_results = [] mcp_results = []
for doc, score, meta in zip( for doc, score, meta in zip(
@ -259,10 +258,10 @@ def query_articles():
"score": float(score), "score": float(score),
"metadata": meta "metadata": meta
}) })
# Apply diversity filtering # Apply diversity filtering
diverse_results = get_diverse_articles(mcp_results, max_results) diverse_results = get_diverse_articles(mcp_results, max_results)
return jsonify({ return jsonify({
"results": diverse_results, "results": diverse_results,
"query": question "query": question
@ -284,7 +283,7 @@ def get_latest_articles(field):
} }
} }
] ]
return jsonify({ return jsonify({
"results": sample_articles, "results": sample_articles,
"field": field "field": field
@ -296,7 +295,7 @@ def get_company_facts(company_name):
facts = company_facts.get(company_name, {}) facts = company_facts.get(company_name, {})
if not facts: if not facts:
return jsonify({"error": f"Company {company_name} not found"}), 404 return jsonify({"error": f"Company {company_name} not found"}), 404
return jsonify({ return jsonify({
"company": company_name, "company": company_name,
"facts": facts "facts": facts
@ -307,10 +306,10 @@ def get_company_products(company_name):
"""Get products information for a company""" """Get products information for a company"""
facts = company_facts.get(company_name, {}) facts = company_facts.get(company_name, {})
products = facts.get("products", []) products = facts.get("products", [])
if not products: if not products:
return jsonify({"error": f"No products found for company {company_name}"}), 404 return jsonify({"error": f"No products found for company {company_name}"}), 404
return jsonify({ return jsonify({
"company": company_name, "company": company_name,
"products": products "products": products
@ -322,13 +321,13 @@ def update_company_facts():
data = request.get_json() data = request.get_json()
company_name = data.get("company_name") company_name = data.get("company_name")
facts = data.get("facts") facts = data.get("facts")
if not company_name or not facts: if not company_name or not facts:
return jsonify({"error": "Missing 'company_name' or 'facts' in request body"}), 400 return jsonify({"error": "Missing 'company_name' or 'facts' in request body"}), 400
# Update or add company facts # Update or add company facts
company_facts[company_name] = facts company_facts[company_name] = facts
return jsonify({ return jsonify({
"message": "Facts updated successfully", "message": "Facts updated successfully",
"company": company_name, "company": company_name,

12
README.md
View File

@ -131,9 +131,19 @@ docker run -p 5008:5008 stockdocs-article-server
# Continue for other components... # Continue for other components...
``` ```
## Tests
```bash
pip install -r scraper/requirements_clean.txt pytest
pytest tests/ -v
```
Unit tests cover the scraper's article-processing cache: load/save
roundtrips, processing status tracking, and progress computation.
## Requirements ## Requirements
- Python 3.6+ - Python 3.9+
- Flask 2.3.3 - Flask 2.3.3
- Various NLP and ML libraries - Various NLP and ML libraries
- Docker (for containerized deployment) - Docker (for containerized deployment)

277
agent
View File

@ -1,277 +0,0 @@
# StockDocs Project Overview
This document provides a comprehensive overview of the StockDocs financial news analysis platform, including detailed information about each project component, file structures, and how to begin working with the system.
## Project Architecture
StockDocs is a sophisticated system designed to collect, process, and analyze financial news from multiple sources. The platform consists of several interconnected components that work together to transform raw news content into valuable market intelligence.
```
+--------------+ +--------------+ +-------------------+
| Scraper | | Article | | AI |
| (RSS Feeds) |--> | Server |--> | Processor |
| | | | | |
+--------------+ +--------------+ +-------------------+
| |
v v
+--------------+ +-------------------+
| Embedding | | MCPServer |
| Service | | (Financial Data) |
| | | |
+--------------+ +-------------------+
```
## Project Components
### 1. Scraper
**Location:** `scraper/`
The Scraper component is responsible for collecting financial news and articles from multiple sources including major news outlets, financial publications, and market analysis services. It uses RSS feeds to gather content and stores the articles in a structured directory hierarchy for easy access by other components of the system.
#### Key Features:
- RSS Feed Integration: Supports multiple financial news sources through RSS feeds
- Automated Scraping: Regularly fetches and processes new articles from configured feeds
- Structured Storage: Organizes articles in a directory structure by news outlet
- Duplicate Detection: Prevents re-processing of already collected articles
- Caching Mechanism: Maintains a cache of processed articles to optimize performance
#### File Structure:
```
scraper/
├── rss_feeds.json # Configuration file with RSS feed URLs
├── scraper.py # Main scraping logic
├── requirements.txt # Python dependencies
├── dockerfile # Docker configuration
├── dockerfile-selenium # Dockerfile for selenium-based scraping
├── articles/ # Directory where scraped articles are stored
│ ├── Reuters – Business News/
│ │ ├── article1.txt
│ │ └── ...
│ ├── Associated Press – Business/
│ │ ├── article1.txt
│ │ └── ...
│ └── ...
├── processed_articles_cache.json # Cache of already processed articles
└── pyvenv.cfg # Python virtual environment configuration
```
#### Key Files:
- `scraper.py`: Main scraping logic with parallel processing capabilities
- `rss_feeds.json`: Configuration file with RSS feed URLs for 60+ news outlets
- `articles/`: Directory structure for storing scraped articles organized by news source
### 2. Article Server
**Location:** `articleServer/`
A Flask-based HTTP server that provides access to news articles stored in a directory structure, with time-based filtering and outlet-specific querying capabilities.
#### Key Features:
- Time-based filtering: Query articles by hour, day, week, or month
- Outlet-specific querying: Filter articles by news source
- Content retrieval: Get full article content by file path
- RESTful API: Clean interfaces for integration with external systems
#### File Structure:
```
articleServer/
├── app.py # Main Flask application
├── requirements.txt # Python dependencies
├── Dockerfile # Docker configuration
├── run_server.py # Server startup script
├── pyvenv.cfg # Python virtual environment configuration
└── README.md # Documentation
```
#### API Endpoints:
- `GET /articles` - Get articles within time range
- `GET /article/content` - Get full article content by path
- `GET /outlets` - List all available news outlets
- `GET /health` - Health check endpoint
### 3. AI Processor
**Location:** `ai_processor/`
An AI-powered analytics engine designed to process financial news articles and extract meaningful insights, sentiment analysis, and market indicators from collected content.
#### Key Features:
- Sentiment Analysis: Determine positive, negative, or neutral sentiment of news articles
- Topic Classification: Categorize articles by financial topics
- Entity Extraction: Identify key entities mentioned in articles (stocks, companies, people, organizations)
- Market Indicator Detection: Extract quantitative indicators that may affect stock prices
- Insight Generation: Automated generation of actionable intelligence from news content
- Batch Processing: Process large volumes of articles efficiently
#### File Structure:
```
ai_processor/
├── ai_processor.py # Main AI processing application
├── requirements.txt # Python dependencies
├── Dockerfile # Docker configuration
├── README.md # Documentation
├── models/ # Machine learning models and NLP components
│ ├── sentiment_analyzer.py # Sentiment analysis module
│ ├── topic_classifier.py # News categorization module
│ └── entity_extractor.py # Named entity recognition
├── processors/ # Article processing pipelines
│ ├── text_processor.py # Text cleaning and preprocessing
│ └── analysis_pipeline.py # Full analysis pipeline
└── output/ # Processed data storage
├── insights/
└── reports/
```
#### API Endpoints:
- `POST /api/analyze/article` - Analyze a single article for insights
- `POST /api/analyze/batch` - Process multiple articles in batch mode
- `GET /api/insights/latest` - Get latest analysis insights
- `GET /api/models` - List available AI models
### 4. Embedding Service
**Location:** `embedding/`
An embedding service that converts text content into numerical vectors for machine learning and data analysis purposes. This component transforms financial news articles into vector representations that can be used for similarity comparisons, clustering, and other AI tasks.
#### Key Features:
- Multiple Model Support: Integrates with various embedding models including BERT, Sentence-BERT, and other transformer-based models
- Batch Processing: Efficient processing of large volumes of articles
- Caching Mechanism: Caches generated embeddings to avoid reprocessing
- Real-time Generation: Generates embeddings on-demand for new content
- Vector Similarity: Computes similarity between different pieces of content
- Storage Management: Organizes and stores embeddings efficiently
#### File Structure:
```
embedding/
├── embedder.py # Main embedding application
├── requirements.txt # Python dependencies
├── Dockerfile # Docker configuration
├── README.md # Documentation
├── models/ # Pre-trained embedding models
│ ├── sentence_transformer.py # Sentence transformer implementation
│ └── model_loader.py # Model loading utilities
├── processors/ # Text processing pipeline
│ ├── text_cleaner.py # Text cleaning and preprocessing
│ └── embedding_generator.py # Embedding generation
└── data/ # Processed embeddings storage
├── cache/
└── outputs/
```
#### API Endpoints:
- `POST /api/embeddings/generate` - Generate embeddings for text content
- `POST /api/embeddings/batch` - Generate embeddings for multiple texts in batch
- `GET /api/embeddings/similarity` - Calculate similarity between two pieces of text/content
### 5. MCPServer
**Location:** `MCPServer/`
A Flask-based server that provides an API for accessing and analyzing financial data, with integration for stock analysis and news processing. Serves as the backend service for processing financial information and making it available through HTTP endpoints.
#### Key Features:
- Stock Analysis: Financial metrics calculation and market data processing
- News Integration: APIs for retrieving and processing financial news
- Data Aggregation: Consolidation of multiple data sources into unified responses
- RESTful API: Clean HTTP interface for external services to consume data
#### File Structure:
```
MCPServer/
├── server.py # Main Flask application
├── requirements.txt # Python dependencies
├── Dockerfile # Docker configuration
├── openapi.json # API specification
├── README.md # Documentation
└── api/ # API endpoints and handlers
├── stock_analysis.py # Stock analysis functions
└── news_processing.py # News processing functions
```
#### API Endpoints:
- `POST /query` - Query the vector database with a question
- `GET /health` - Server health check
- `GET /info` - Returns model and service metadata
- `GET /tools` - List available endpoints and their descriptions
## Setup and Installation
### Prerequisites
- Python 3.6+
- Docker (for containerized deployment)
- Internet connection for RSS feed access
### Installation Steps
1. **Clone the repository:**
```bash
git clone <repository-url>
cd StockDocs
```
2. **Set up each component:**
```bash
# For each component, follow specific installation instructions
cd scraper && pip install -r requirements.txt
cd articleServer && pip install -r requirements.txt
cd ai_processor && pip install -r requirements.txt
cd embedding && pip install -r requirements.txt
cd MCPServer && pip install -r requirements.txt
```
3. **Configure environment variables as needed for each component**
4. **Run individual services:**
```bash
python scraper/scraper.py # Start scraping
python articleServer/run_server.py # Start article server
python ai_processor/app.py # Start AI processor
python embedding/app.py # Start embedding service
python MCPServer/app.py # Start MCP server
```
## Deployment
Each component can be run independently or containerized using the provided Dockerfiles:
```bash
# Build and run each component in Docker
docker build -t stockdocs-scraper ./scraper
docker run -p 5000:5000 stockdocs-scraper
docker build -t stockdocs-article-server ./articleServer
docker run -p 5008:5008 stockdocs-article-server
# Continue for other components...
```
## Key Technical Details
### Data Flow
1. **Scraper** collects articles from RSS feeds and stores them in `scraper/articles/`
2. **Article Server** provides API access to these articles
3. **AI Processor** analyzes articles and generates insights in `ai_processor/output/`
4. **Embedding Service** converts article content into vector representations and stores in ChromaDB
5. **MCPServer** provides API access to the vector database for querying
### Environment Variables
Each component may require specific environment variables:
- `ARTICLE_DIR`: Path to article directory
- `AI_SERVICE_URL`: URL for local AI service
- `CHROMADB_HOST` and `CHROMADB_PORT`: ChromaDB connection settings
- `FEED_FILE`: Path to RSS feed configuration file
### Directory Structure
- `scraper/articles/`: Stores raw scraped articles organized by news source
- `ai_processor/output/`: Stores processed AI analysis results
- `embedding/data/cache/`: Stores cached embeddings
- `MCPServer/`: Contains server configuration and API endpoints
## Getting Started Guide
To begin working with the StockDocs platform:
1. **Start the Scraper** to collect news articles
2. **Run the Article Server** to make articles accessible via API
3. **Launch the AI Processor** to analyze articles and generate insights
4. **Initialize the Embedding Service** to create vector representations
5. **Start MCPServer** to query the vector database for insights
Each component can be run independently or as part of a complete pipeline for comprehensive financial news analysis.

304
agent.md
View File

@ -1,304 +0,0 @@
# StockDocs Project Overview
This document provides a comprehensive overview of the StockDocs repository, which contains multiple interconnected projects for processing, scraping, and serving financial articles and embeddings.
## Repository Structure
The repository contains 5 main projects:
1. **ai_processor/** - AI processing component
2. **articleServer/** - Article serving component
3. **embedding/** - Embedding functionality
4. **MCPServer/** - MCP server component
5. **scraper/** - Web scraping component
## Project Details
### 1. ai_processor/
The AI processing component handles artificial intelligence operations for processing articles and generating insights.
**Key Files:**
- `ai_processor.py` - Main AI processing logic
- `requirements.txt` - Python dependencies
- `Dockerfile` - Container configuration
**Purpose:** Processes articles using AI models to extract key information, generate summaries, and create embeddings.
**Detailed Structure:**
```
ai_processor/
├── app.py # Main AI processing application
├── config.py # Configuration settings
├── requirements.txt # Python dependencies
├── Dockerfile # Docker configuration
├── README.md # This file
├── models/ # Machine learning models and NLP components
│ ├── __init__.py
│ ├── sentiment_analyzer.py # Sentiment analysis module
│ ├── topic_classifier.py # News categorization module
│ └── entity_extractor.py # Named entity recognition
├── processors/ # Article processing pipelines
│ ├── __init__.py
│ ├── text_processor.py # Text cleaning and preprocessing
│ └── analysis_pipeline.py # Full analysis pipeline
└── data/ # Processed data storage
├── insights/
└── reports/
```
**Endpoints:**
- POST `/api/analyze/article` - Analyze a single article for insights
- POST `/api/analyze/batch` - Process multiple articles in batch mode
- GET `/api/analyze/status/{task_id}` - Check processing status
- GET `/api/insights/latest` - Get latest analysis insights
- GET `/api/insights/articles/{article_path}` - Get insights for specific article
- GET `/api/reports/generate` - Generate comprehensive market analysis report
- GET `/api/models` - List available AI models
- POST `/api/models/update` - Update or retrain models with new data
### 2. articleServer/
The article serving component provides an API for accessing processed articles.
**Key Files:**
- `app.py` - Main Flask application
- `run_server.py` - Server startup script
- `requirements.txt` - Python dependencies
- `Dockerfile` - Container configuration
**Purpose:** Exposes processed articles through a REST API for client applications to consume.
**Detailed Structure:**
```
articleServer/
├── app.py # Main Flask application
├── run_server.py # Server startup script
├── requirements.txt # Python dependencies
├── Dockerfile # Docker configuration
├── README.md # This file
└── templates/ # HTML templates (if any)
```
**Endpoints:**
- GET `/articles` - Get articles within time range
- GET `/article/content` - Get full article content by path
- GET `/outlets` - List all available news outlets
- GET `/health` - Health check endpoint
### 3. embedding/
The embedding functionality handles vector embeddings for articles and documents.
**Key Files:**
- `embedder.py` - Embedding generation logic
- `requirements.txt` - Python dependencies
- `Dockerfile` - Container configuration
**Purpose:** Converts articles into vector embeddings for semantic search and similarity operations.
**Detailed Structure:**
```
embedding/
├── app.py # Main embedding application
├── config.py # Configuration settings
├── requirements.txt # Python dependencies
├── Dockerfile # Docker configuration
├── README.md # This file
├── models/ # Pre-trained embedding models
│ ├── __init__.py
│ ├── sentence_transformer.py # Sentence transformer implementation
│ └── model_loader.py # Model loading utilities
├── processors/ # Text processing pipeline
│ ├── __init__.py
│ ├── text_cleaner.py # Text cleaning and preprocessing
│ └── embedding_generator.py # Embedding generation
└── data/ # Processed embeddings storage
├── cache/
└── outputs/
```
**Endpoints:**
- POST `/api/embeddings/generate` - Generate embeddings for text content
- POST `/api/embeddings/batch` - Generate embeddings for multiple texts in batch
- GET `/api/embeddings/similarity` - Calculate similarity between two pieces of text/content
- GET `/api/embeddings/status` - Check service status
- GET `/api/embeddings/models` - List available embedding models
- DELETE `/api/embeddings/cache/clear` - Clear the embedding cache
### 4. MCPServer/
The MCP (Model Context Protocol) server component provides external API access.
**Key Files:**
- `server.py` - Main MCP server implementation
- `openapi.json` - API specification
- `requirements.txt` - Python dependencies
- `Dockerfile` - Container configuration
**Purpose:** Exposes functionality through the Model Context Protocol for integration with other systems.
**Detailed Structure:**
```
MCPServer/
├── server.py # Main Flask application
├── config.py # Configuration settings
├── requirements.txt # Python dependencies
├── Dockerfile # Docker configuration
├── README.md # This file
└── api/ # API endpoints and handlers
├── __init__.py
├── stock_analysis.py # Stock analysis functions
└── news_processing.py # News processing functions
```
**Endpoints:**
- GET `/api/stock/metrics` - Get financial metrics for a stock
- GET `/api/stock/history` - Get historical price data
- POST `/api/stock/analyze` - Perform comprehensive stock analysis
- GET `/api/news` - Retrieve news articles related to stocks
- GET `/api/news/outlets` - List available news sources
- POST `/api/news/process` - Process and categorize news content
- GET `/api/data/refresh` - Refresh data from sources
- GET `/api/status` - Server health check
### 5. scraper/
The web scraping component collects articles from various sources.
**Key Files:**
- `scraper.py` - Main scraping logic
- `rss_feeds.json` - RSS feed configuration
- `requirements.txt` - Python dependencies
- `Dockerfile` - Container configuration
**Purpose:** Collects financial articles from various sources including RSS feeds and web scraping.
**Detailed Structure:**
```
scraper/
├── rss_feeds.json # Configuration file with RSS feed URLs
├── scraper.py # Main scraping logic
├── requirements.txt # Python dependencies
├── dockerfile # Docker configuration
├── dockerfile-selenium # Dockerfile for selenium-based scraping
├── articles/ # Directory where scraped articles are stored
│ ├── Reuters – Business News/
│ │ ├── article1.txt
│ │ └── ...
│ ├── Associated Press – Business/
│ │ ├── article1.txt
│ │ └── ...
│ └── ...
├── processed_articles_cache.json # Cache of already processed articles
└── pyvenv.cfg # Python virtual environment configuration
```
**Features:**
- RSS Feed Integration: Supports multiple financial news sources through RSS feeds
- Automated Scraping: Regularly fetches and processes new articles from configured feeds
- Structured Storage: Organizes articles in a directory structure by news outlet
- Duplicate Detection: Prevents re-processing of already collected articles
- Caching Mechanism: Maintains a cache of processed articles to optimize performance
## Getting Started
### Prerequisites
- Docker installed
- Python 3.8+
- Git
### Setup Instructions
1. **Clone the repository:**
```bash
git clone <repository-url>
cd StockDocs
```
2. **Build and run containers:**
```bash
docker-compose up --build
```
3. **Project-specific setup:**
- Each project has its own `README.md` with detailed setup instructions
- Check individual project directories for specific requirements
## Project Dependencies
### Common Dependencies
- Python 3.8+
- Docker
- Various Python packages (listed in requirements.txt files)
### Inter-project Relationships
- `scraper/` feeds articles to `ai_processor/`
- `ai_processor/` generates embeddings that `embedding/` processes
- `articleServer/` serves articles processed by `ai_processor/`
- `MCPServer/` provides API access to all components
## Development Workflow
1. **Start all services:**
```bash
docker-compose up --build
```
2. **Work with individual projects:**
- Navigate to project directory
- Check `README.md` for specific instructions
- Make changes and rebuild as needed
3. **Testing:**
- Each project includes its own testing setup
- Integration tests may be needed for cross-project functionality
## API Endpoints
### articleServer/
- `/articles` - Get all articles
- `/articles/<id>` - Get specific article
- `/search` - Search articles by query
### MCPServer/
- Exposes various endpoints through Model Context Protocol
- See `openapi.json` for complete specification
## Configuration
### Environment Variables
Each project may require specific environment variables. Check individual `README.md` files for details.
### Authentication
The system now supports API key authentication for connections to the centralized AI service at `http://example.com:4000`. To enable authentication:
1. Set the `AI_SERVICE_API_KEY` environment variable with your API key
2. All connections to the AI service will automatically include the `Authorization: Bearer {api_key}` header
### Data Storage
- Articles are stored in the scraper component
- Processed data flows through the ai_processor
- Embeddings are generated and stored in the embedding component
## Troubleshooting
### Common Issues
1. **Docker build failures:** Ensure Docker is running and check `Dockerfile` syntax
2. **Python dependency issues:** Run `pip install -r requirements.txt` in each project
3. **Port conflicts:** Check `docker-compose.yml` for port mappings
4. **Service startup issues:** Check individual project logs
### Logs
- View logs with `docker-compose logs <service-name>`
- Check individual project logs for detailed error information
## Contributing
1. Fork the repository
2. Create feature branch
3. Make changes
4. Test thoroughly
5. Submit pull request
## Support
For issues or questions, please check:
- Individual project README.md files
- Docker logs for runtime errors
- GitHub issues for known problems

2
ai_processor/__init__.py
View File

@ -2,4 +2,4 @@
AI Processor for extracting facts from articles and preparing them for embedding. AI Processor for extracting facts from articles and preparing them for embedding.
This module handles the intelligent processing of news articles to extract structured facts This module handles the intelligent processing of news articles to extract structured facts
that can be used for querying and analysis. that can be used for querying and analysis.
""" """

0
ai_processor/ai_processor/processed_articles_cache.json
View File

6
ai_processor/article_processor.py
View File

@ -60,7 +60,7 @@ class ArticleProcessor:
scraper_dir = alt_path scraper_dir = alt_path
break break
else: else:
logger.error(f"No valid articles directory found") logger.error("No valid articles directory found")
return [] return []
# Log cache state before scanning # Log cache state before scanning
@ -69,7 +69,7 @@ class ArticleProcessor:
f"Cache state before scanning: {cache_stats['processed_files']} files marked as processed" f"Cache state before scanning: {cache_stats['processed_files']} files marked as processed"
) )
logger.info(f"Directory exists, walking through files...") logger.info("Directory exists, walking through files...")
file_count = 0 file_count = 0
article_file_count = 0 article_file_count = 0
already_processed_count = 0 already_processed_count = 0
@ -136,7 +136,7 @@ class ArticleProcessor:
try: try:
logger.debug(f"Processing article file: {filename}") logger.debug(f"Processing article file: {filename}")
logger.debug(f"File path: {file_path}") logger.debug(f"File path: {file_path}")
with open(file_path, "r", encoding="utf-8") as f: with open(file_path, "r", encoding="utf-8") as f:
article_data = json.load(f) article_data = json.load(f)

6
ai_processor/cache_manager.py
View File

@ -6,7 +6,7 @@ import json
import logging import logging
import os import os
from datetime import datetime from datetime import datetime
from typing import Dict, List, Optional from typing import Dict, List
from config import CACHE_FILE from config import CACHE_FILE
@ -31,7 +31,7 @@ class CacheManager:
try: try:
logger.debug(f"Attempting to load cache from: {self.cache_file}") logger.debug(f"Attempting to load cache from: {self.cache_file}")
if not os.path.exists(self.cache_file): if not os.path.exists(self.cache_file):
logger.info( logger.info(
f"Cache file does not exist: {self.cache_file}. Creating new empty cache." f"Cache file does not exist: {self.cache_file}. Creating new empty cache."
@ -104,7 +104,7 @@ class CacheManager:
with open(self.cache_file, "w", encoding="utf-8") as f: with open(self.cache_file, "w", encoding="utf-8") as f:
json.dump(self.cache, f, indent=2, ensure_ascii=False) json.dump(self.cache, f, indent=2, ensure_ascii=False)
logger.debug(f"Successfully saved cache to {self.cache_file}") logger.debug(f"Successfully saved cache to {self.cache_file}")
except Exception as e: except Exception as e:
logger.error(f"Error saving cache file {self.cache_file}: {e}") logger.error(f"Error saving cache file {self.cache_file}: {e}")

2
ai_processor/config.py
View File

@ -32,4 +32,4 @@ CHROMADB_PORT = int(os.getenv("CHROMADB_PORT", "8000"))
# Collection names # Collection names
FACTS_COLLECTION_NAME = "facts" FACTS_COLLECTION_NAME = "facts"
ARTICLES_COLLECTION_NAME = "articles" ARTICLES_COLLECTION_NAME = "articles"

50
ai_processor/fact_extractor.py
View File

@ -6,7 +6,7 @@ Uses the gpt-oss model via the centralized AI service.
import json import json
import logging import logging
import requests import requests
from typing import Dict, Any, Optional from typing import Dict, Any
from config import AI_SERVER_URL, AI_SERVICE_API_KEY, FACT_EXTRACTION_MODEL from config import AI_SERVER_URL, AI_SERVICE_API_KEY, FACT_EXTRACTION_MODEL
from metrics_collector import metrics_collector from metrics_collector import metrics_collector
@ -15,12 +15,12 @@ logger = logging.getLogger(__name__)
class FactExtractor: class FactExtractor:
"""Extracts structured facts from article content using AI models.""" """Extracts structured facts from article content using AI models."""
def __init__(self): def __init__(self):
self.ai_server_url = AI_SERVER_URL self.ai_server_url = AI_SERVER_URL
self.api_key = AI_SERVICE_API_KEY self.api_key = AI_SERVICE_API_KEY
self.model = FACT_EXTRACTION_MODEL self.model = FACT_EXTRACTION_MODEL
def _get_headers(self) -> Dict[str, str]: def _get_headers(self) -> Dict[str, str]:
"""Get headers with authentication.""" """Get headers with authentication."""
headers = { headers = {
@ -29,36 +29,36 @@ class FactExtractor:
if self.api_key: if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}" headers["Authorization"] = f"Bearer {self.api_key}"
return headers return headers
def extract_facts_from_article(self, article_content: str, title: str) -> Dict[str, Any]: def extract_facts_from_article(self, article_content: str, title: str) -> Dict[str, Any]:
""" """
Extract structured facts from article content using gpt-oss model. Extract structured facts from article content using gpt-oss model.
Args: Args:
article_content (str): The full content of the article article_content (str): The full content of the article
title (str): The title of the article title (str): The title of the article
Returns: Returns:
Dict containing extracted facts Dict containing extracted facts
""" """
try: try:
extraction_url = f"{self.ai_server_url}/v1/chat/completions" extraction_url = f"{self.ai_server_url}/v1/chat/completions"
# Create a proper prompt for fact extraction # Create a proper prompt for fact extraction
prompt = f""" prompt = f"""
Extract key facts from the following article in structured JSON format. Extract key facts from the following article in structured JSON format.
Return only valid JSON without any additional text. Return only valid JSON without any additional text.
Article Title: {title} Article Title: {title}
Article Content: {article_content[:3000]}... Article Content: {article_content[:3000]}...
Extract the following information: Extract the following information:
1. Main topic/subject 1. Main topic/subject
2. Key entities (companies, people, locations, organizations) 2. Key entities (companies, people, locations, organizations)
3. Financial impact or implications 3. Financial impact or implications
4. Key dates or time periods mentioned 4. Key dates or time periods mentioned
5. Summary of main points 5. Summary of main points
Format the response as a JSON object with these fields: Format the response as a JSON object with these fields:
{{ {{
"title": "{title}", "title": "{title}",
@ -70,12 +70,12 @@ class FactExtractor:
"main_points": ["point1", "point2", "point3"] "main_points": ["point1", "point2", "point3"]
}} }}
""" """
# Log the request details for debugging # Log the request details for debugging
logger.debug(f"Preparing AI request for article: {title}") logger.debug(f"Preparing AI request for article: {title}")
logger.debug(f"AI Server URL: {extraction_url}") logger.debug(f"AI Server URL: {extraction_url}")
logger.debug(f"Request payload preview: {str({'model': self.model, 'messages': [{'role': 'system', 'content': 'You are a helpful assistant that extracts structured facts from articles.'}, {'role': 'user', 'content': prompt[:200]}]}[:300])}...") logger.debug(f"Request payload preview: {str({'model': self.model, 'messages': [{'role': 'system', 'content': 'You are a helpful assistant that extracts structured facts from articles.'}, {'role': 'user', 'content': prompt[:200]}]}[:300])}...")
# Call the AI service with gpt-oss model for fact extraction # Call the AI service with gpt-oss model for fact extraction
try: try:
response = requests.post( response = requests.post(
@ -92,12 +92,12 @@ class FactExtractor:
headers=self._get_headers(), headers=self._get_headers(),
timeout=60 timeout=60
) )
# Log response details for debugging # Log response details for debugging
logger.debug(f"AI service response status: {response.status_code}") logger.debug(f"AI service response status: {response.status_code}")
logger.debug(f"AI service response headers: {dict(response.headers)}") logger.debug(f"AI service response headers: {dict(response.headers)}")
logger.debug(f"AI service response text preview: {response.text[:500]}...") logger.debug(f"AI service response text preview: {response.text[:500]}...")
response.raise_for_status() response.raise_for_status()
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
logger.error(f"REQUEST FAILED for article '{title}' - URL: {extraction_url}") logger.error(f"REQUEST FAILED for article '{title}' - URL: {extraction_url}")
@ -106,7 +106,7 @@ class FactExtractor:
logger.error(f"Response text (if available): {response.text[:500] if 'response' in locals() else 'No response available'}") logger.error(f"Response text (if available): {response.text[:500] if 'response' in locals() else 'No response available'}")
# Return basic structure if request fails # Return basic structure if request fails
return self._create_basic_fact_structure(article_content, title) return self._create_basic_fact_structure(article_content, title)
# Parse the response # Parse the response
try: try:
result = response.json() result = response.json()
@ -121,7 +121,7 @@ class FactExtractor:
logger.error(f"Article content preview: {article_content[:200]}...") logger.error(f"Article content preview: {article_content[:200]}...")
# Return basic structure if response parsing fails # Return basic structure if response parsing fails
return self._create_basic_fact_structure(article_content, title) return self._create_basic_fact_structure(article_content, title)
# Check if the response is empty or invalid # Check if the response is empty or invalid
if not extracted_text or extracted_text.strip() == "": if not extracted_text or extracted_text.strip() == "":
logger.warning(f"Empty response from AI service for article '{title}'") logger.warning(f"Empty response from AI service for article '{title}'")
@ -141,21 +141,21 @@ class FactExtractor:
logger.error(f"Response status: {response.status_code}") logger.error(f"Response status: {response.status_code}")
logger.error(f"Response text (full): {response.text}") logger.error(f"Response text (full): {response.text}")
facts = self._create_basic_fact_structure(article_content, title) facts = self._create_basic_fact_structure(article_content, title)
# Ensure all required fields are present # Ensure all required fields are present
facts = self._ensure_required_fields(facts, title, article_content) facts = self._ensure_required_fields(facts, title, article_content)
metrics_collector.increment_facts_extracted() metrics_collector.increment_facts_extracted()
logger.info(f"Successfully extracted facts from article: {title}") logger.info(f"Successfully extracted facts from article: {title}")
return facts return facts
except Exception as e: except Exception as e:
logger.error(f"UNEXPECTED ERROR extracting facts from article '{title}': {e}") logger.error(f"UNEXPECTED ERROR extracting facts from article '{title}': {e}")
logger.error(f"Error type: {type(e).__name__}") logger.error(f"Error type: {type(e).__name__}")
logger.error(f"Article content preview: {article_content[:200]}...") logger.error(f"Article content preview: {article_content[:200]}...")
# Return basic structure if extraction fails # Return basic structure if extraction fails
return self._create_basic_fact_structure(article_content, title) return self._create_basic_fact_structure(article_content, title)
def _create_basic_fact_structure(self, article_content: str, title: str) -> Dict[str, Any]: def _create_basic_fact_structure(self, article_content: str, title: str) -> Dict[str, Any]:
"""Create a basic fact structure when AI extraction fails.""" """Create a basic fact structure when AI extraction fails."""
return { return {
@ -171,7 +171,7 @@ class FactExtractor:
"Third key fact from the article" "Third key fact from the article"
] ]
} }
def _ensure_required_fields(self, facts: Dict[str, Any], title: str, article_content: str) -> Dict[str, Any]: def _ensure_required_fields(self, facts: Dict[str, Any], title: str, article_content: str) -> Dict[str, Any]:
"""Ensure all required fields are present in the facts structure.""" """Ensure all required fields are present in the facts structure."""
required_fields = { required_fields = {
@ -183,11 +183,11 @@ class FactExtractor:
"key_dates": [], "key_dates": [],
"main_points": [] "main_points": []
} }
for field, default_value in required_fields.items(): for field, default_value in required_fields.items():
if field not in facts: if field not in facts:
facts[field] = default_value facts[field] = default_value
elif not facts[field]: # If field is empty elif not facts[field]: # If field is empty
facts[field] = default_value facts[field] = default_value
return facts return facts

24
ai_processor/main.py
View File

@ -6,11 +6,9 @@ Handles the orchestration of article processing and fact extraction.
import logging import logging
import sys import sys
import os import os
from datetime import datetime
from article_processor import ArticleProcessor from article_processor import ArticleProcessor
from metrics_collector import metrics_collector from metrics_collector import metrics_collector
from cache_manager import CacheManager
from config import CACHE_FILE, LOG_FILE, LOG_LEVEL from config import CACHE_FILE, LOG_FILE, LOG_LEVEL
# Setup logging # Setup logging
@ -31,7 +29,7 @@ def setup_logging():
log_dir = os.path.dirname(LOG_FILE) log_dir = os.path.dirname(LOG_FILE)
if log_dir: if log_dir:
os.makedirs(log_dir, exist_ok=True) os.makedirs(log_dir, exist_ok=True)
# Ensure output directory exists for cache files # Ensure output directory exists for cache files
output_dir = os.path.dirname(CACHE_FILE) output_dir = os.path.dirname(CACHE_FILE)
if output_dir: if output_dir:
@ -40,18 +38,18 @@ if output_dir:
def main(): def main():
"""Main function to run the AI processor.""" """Main function to run the AI processor."""
logger.info("Starting AI Processor") logger.info("Starting AI Processor")
try: try:
# Setup logging # Setup logging
setup_logging() setup_logging()
# Create processor instance # Create processor instance
processor = ArticleProcessor() processor = ArticleProcessor()
# Process all articles # Process all articles
logger.info("Starting article processing...") logger.info("Starting article processing...")
stats = processor.process_all_articles() stats = processor.process_all_articles()
# Log final statistics # Log final statistics
logger.info("Processing completed") logger.info("Processing completed")
logger.info(f"Total processed: {stats['total_processed']}") logger.info(f"Total processed: {stats['total_processed']}")
@ -61,13 +59,13 @@ def main():
logger.info(f"Cache stats: {stats['cache_stats']}") logger.info(f"Cache stats: {stats['cache_stats']}")
else: else:
logger.info("No cache stats available") logger.info("No cache stats available")
# Print metrics summary # Print metrics summary
metrics_summary = metrics_collector.get_metrics_summary() metrics_summary = metrics_collector.get_metrics_summary()
logger.info(f"Metrics summary: {metrics_summary}") logger.info(f"Metrics summary: {metrics_summary}")
logger.info("AI Processor completed successfully") logger.info("AI Processor completed successfully")
except Exception as e: except Exception as e:
logger.error(f"Error in main function: {e}") logger.error(f"Error in main function: {e}")
raise raise
@ -75,15 +73,15 @@ def main():
def process_new_articles(): def process_new_articles():
"""Process only new articles (for real-time processing).""" """Process only new articles (for real-time processing)."""
logger.info("Starting real-time processing of new articles") logger.info("Starting real-time processing of new articles")
try: try:
processor = ArticleProcessor() processor = ArticleProcessor()
stats = processor.process_new_articles() stats = processor.process_new_articles()
logger.info("Real-time processing completed") logger.info("Real-time processing completed")
logger.info(f"Total processed: {stats['total_processed']}") logger.info(f"Total processed: {stats['total_processed']}")
logger.info(f"Total failed: {stats['total_failed']}") logger.info(f"Total failed: {stats['total_failed']}")
except Exception as e: except Exception as e:
logger.error(f"Error in real-time processing: {e}") logger.error(f"Error in real-time processing: {e}")
raise raise

36
ai_processor/metrics_collector.py
View File

@ -20,38 +20,38 @@ except Exception as e:
# AI Processor Metrics # AI Processor Metrics
articles_processed_total = Counter( articles_processed_total = Counter(
'ai_processor_articles_processed_total', 'ai_processor_articles_processed_total',
'Total number of articles processed by AI processor' 'Total number of articles processed by AI processor'
) )
articles_failed_total = Counter( articles_failed_total = Counter(
'ai_processor_articles_failed_total', 'ai_processor_articles_failed_total',
'Total number of articles failed to process by AI processor' 'Total number of articles failed to process by AI processor'
) )
facts_extracted_total = Counter( facts_extracted_total = Counter(
'ai_processor_facts_extracted_total', 'ai_processor_facts_extracted_total',
'Total number of facts extracted by AI processor' 'Total number of facts extracted by AI processor'
) )
processing_time_seconds = Histogram( processing_time_seconds = Histogram(
'ai_processor_processing_time_seconds', 'ai_processor_processing_time_seconds',
'Time spent processing articles in AI processor' 'Time spent processing articles in AI processor'
) )
cache_hits_total = Counter( cache_hits_total = Counter(
'ai_processor_cache_hits_total', 'ai_processor_cache_hits_total',
'Total number of cache hits in AI processor' 'Total number of cache hits in AI processor'
) )
cache_misses_total = Counter( cache_misses_total = Counter(
'ai_processor_cache_misses_total', 'ai_processor_cache_misses_total',
'Total number of cache misses in AI processor' 'Total number of cache misses in AI processor'
) )
# Current processing status # Current processing status
current_processing_status = Gauge( current_processing_status = Gauge(
'ai_processor_current_status', 'ai_processor_current_status',
'Current processing status of AI processor (0=inactive, 1=active)' 'Current processing status of AI processor (0=inactive, 1=active)'
) )
@ -59,18 +59,18 @@ logger = logging.getLogger(__name__)
class MetricsCollector: class MetricsCollector:
"""Collects and reports metrics for the AI processor.""" """Collects and reports metrics for the AI processor."""
def __init__(self): def __init__(self):
self.start_time = None self.start_time = None
self.active = False self.active = False
def start_processing(self): def start_processing(self):
"""Mark processing as started.""" """Mark processing as started."""
self.start_time = time.time() self.start_time = time.time()
self.active = True self.active = True
current_processing_status.set(1) current_processing_status.set(1)
logger.info("AI processor started processing") logger.info("AI processor started processing")
def stop_processing(self): def stop_processing(self):
"""Mark processing as stopped.""" """Mark processing as stopped."""
self.active = False self.active = False
@ -78,42 +78,42 @@ class MetricsCollector:
if self.start_time: if self.start_time:
total_time = time.time() - self.start_time total_time = time.time() - self.start_time
logger.info(f"AI processor stopped after {total_time:.2f} seconds") logger.info(f"AI processor stopped after {total_time:.2f} seconds")
def increment_articles_processed(self, count: int = 1): def increment_articles_processed(self, count: int = 1):
"""Increment articles processed counter.""" """Increment articles processed counter."""
articles_processed_total.inc(count) articles_processed_total.inc(count)
logger.info(f"Articles processed: {count}") logger.info(f"Articles processed: {count}")
def increment_articles_failed(self, count: int = 1): def increment_articles_failed(self, count: int = 1):
"""Increment articles failed counter.""" """Increment articles failed counter."""
articles_failed_total.inc(count) articles_failed_total.inc(count)
logger.error(f"Articles failed: {count}") logger.error(f"Articles failed: {count}")
def increment_facts_extracted(self, count: int = 1): def increment_facts_extracted(self, count: int = 1):
"""Increment facts extracted counter.""" """Increment facts extracted counter."""
facts_extracted_total.inc(count) facts_extracted_total.inc(count)
logger.info(f"Facts extracted: {count}") logger.info(f"Facts extracted: {count}")
def record_processing_time(self, duration: float): def record_processing_time(self, duration: float):
"""Record processing time.""" """Record processing time."""
processing_time_seconds.observe(duration) processing_time_seconds.observe(duration)
logger.info(f"Processing time: {duration:.2f} seconds") logger.info(f"Processing time: {duration:.2f} seconds")
def increment_cache_hit(self): def increment_cache_hit(self):
"""Increment cache hit counter.""" """Increment cache hit counter."""
cache_hits_total.inc() cache_hits_total.inc()
logger.debug("Cache hit") logger.debug("Cache hit")
def increment_cache_miss(self): def increment_cache_miss(self):
"""Increment cache miss counter.""" """Increment cache miss counter."""
cache_misses_total.inc() cache_misses_total.inc()
logger.debug("Cache miss") logger.debug("Cache miss")
def log_status(self, message: str, level: str = "info"): def log_status(self, message: str, level: str = "info"):
"""Log status message with appropriate level.""" """Log status message with appropriate level."""
log_method = getattr(logger, level) log_method = getattr(logger, level)
log_method(message) log_method(message)
def get_metrics_summary(self) -> Dict[str, Any]: def get_metrics_summary(self) -> Dict[str, Any]:
"""Get current metrics summary.""" """Get current metrics summary."""
return { return {

5
articleServer/pyvenv.cfg
View File

@ -1,5 +0,0 @@
home = /usr/bin
include-system-site-packages = false
version = 3.12.3
executable = /usr/bin/python3.12
command = /usr/bin/python3 -m venv /home/user/StockDocs/articleServer

94
embedding/advanced_embedder.py
View File

@ -2,24 +2,22 @@ import os
import json import json
import chromadb import chromadb
import uuid import uuid
import time
import datetime import datetime
import requests import requests
import logging import logging
from pathlib import Path
from prometheus_client import start_http_server, Counter, Histogram from prometheus_client import start_http_server, Counter, Histogram
# Setup logging with better error handling # Setup logging with better error handling
try: try:
logging.basicConfig( logging.basicConfig(
level=logging.INFO, level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[ handlers=[
logging.FileHandler('embedding_pipeline.log'), logging.FileHandler('embedding_pipeline.log'),
logging.StreamHandler() logging.StreamHandler()
] ]
) )
except Exception as e: except Exception:
# Fallback if file logging fails # Fallback if file logging fails
logging.basicConfig( logging.basicConfig(
level=logging.INFO, level=logging.INFO,
@ -77,15 +75,15 @@ def get_embedding(text):
""" """
try: try:
embedding_url = f"{AI_SERVER_URL}/v1/embeddings" embedding_url = f"{AI_SERVER_URL}/v1/embeddings"
# Build headers with authentication if available # Build headers with authentication if available
headers = { headers = {
"Content-Type": "application/json" "Content-Type": "application/json"
} }
if AI_SERVICE_API_KEY: if AI_SERVICE_API_KEY:
headers["Authorization"] = f"Bearer {AI_SERVICE_API_KEY}" headers["Authorization"] = f"Bearer {AI_SERVICE_API_KEY}"
response = requests.post( response = requests.post(
embedding_url, embedding_url,
json={ json={
@ -109,30 +107,30 @@ def extract_facts_from_article(article_content, title):
try: try:
# Use the centralized AI endpoint for fact extraction # Use the centralized AI endpoint for fact extraction
extraction_url = f"{AI_SERVER_URL}/v1/chat/completions" extraction_url = f"{AI_SERVER_URL}/v1/chat/completions"
# Build headers with authentication if available # Build headers with authentication if available
headers = { headers = {
"Content-Type": "application/json" "Content-Type": "application/json"
} }
if AI_SERVICE_API_KEY: if AI_SERVICE_API_KEY:
headers["Authorization"] = f"Bearer {AI_SERVICE_API_KEY}" headers["Authorization"] = f"Bearer {AI_SERVICE_API_KEY}"
# Create a proper prompt for fact extraction # Create a proper prompt for fact extraction
prompt = f""" prompt = f"""
Extract key facts from the following article in structured JSON format. Extract key facts from the following article in structured JSON format.
Return only valid JSON without any additional text. Return only valid JSON without any additional text.
Article Title: {title} Article Title: {title}
Article Content: {article_content[:2000]}... Article Content: {article_content[:2000]}...
Extract the following information: Extract the following information:
1. Main topic/subject 1. Main topic/subject
2. Key entities (companies, people, locations, organizations) 2. Key entities (companies, people, locations, organizations)
3. Financial impact or implications 3. Financial impact or implications
4. Key dates or time periods mentioned 4. Key dates or time periods mentioned
5. Summary of main points 5. Summary of main points
Format the response as a JSON object with these fields: Format the response as a JSON object with these fields:
{{ {{
"title": "{title}", "title": "{title}",
@ -144,7 +142,7 @@ def extract_facts_from_article(article_content, title):
"main_points": ["point1", "point2", "point3"] "main_points": ["point1", "point2", "point3"]
}} }}
""" """
# Call the AI service with gpt-oss model for fact extraction # Call the AI service with gpt-oss model for fact extraction
response = requests.post( response = requests.post(
extraction_url, extraction_url,
@ -160,13 +158,13 @@ def extract_facts_from_article(article_content, title):
headers=headers, headers=headers,
timeout=60 timeout=60
) )
response.raise_for_status() response.raise_for_status()
# Parse the response # Parse the response
result = response.json() result = response.json()
extracted_text = result['choices'][0]['message']['content'].strip() extracted_text = result['choices'][0]['message']['content'].strip()
# Try to parse the JSON from the response # Try to parse the JSON from the response
try: try:
facts = json.loads(extracted_text) facts = json.loads(extracted_text)
@ -185,9 +183,9 @@ def extract_facts_from_article(article_content, title):
"Third key fact from the article" "Third key fact from the article"
] ]
} }
return facts return facts
except Exception as e: except Exception as e:
logger.error(f"Error extracting facts: {e}") logger.error(f"Error extracting facts: {e}")
# Return a basic structure if extraction fails # Return a basic structure if extraction fails
@ -208,19 +206,19 @@ def process_article_file(file_path):
try: try:
with open(file_path, 'r', encoding='utf-8') as f: with open(file_path, 'r', encoding='utf-8') as f:
article_data = json.load(f) article_data = json.load(f)
# Extract facts from the article # Extract facts from the article
facts = extract_facts_from_article( facts = extract_facts_from_article(
article_data.get('original_content', ''), article_data.get('original_content', ''),
article_data.get('title', '') article_data.get('title', '')
) )
# Add metadata # Add metadata
facts['source'] = article_data.get('source', 'Unknown') facts['source'] = article_data.get('source', 'Unknown')
facts['published'] = article_data.get('published', 'Unknown') facts['published'] = article_data.get('published', 'Unknown')
facts['filename'] = os.path.basename(file_path) facts['filename'] = os.path.basename(file_path)
facts['processed_at'] = datetime.datetime.now().isoformat() facts['processed_at'] = datetime.datetime.now().isoformat()
return facts return facts
except Exception as e: except Exception as e:
logger.error(f"Error processing article {file_path}: {e}") logger.error(f"Error processing article {file_path}: {e}")
@ -232,10 +230,10 @@ def create_collections():
""" """
# Collection for extracted facts (now with entity support) # Collection for extracted facts (now with entity support)
facts_collection = client.get_or_create_collection("facts") facts_collection = client.get_or_create_collection("facts")
# Collection for full articles # Collection for full articles
articles_collection = client.get_or_create_collection("articles") articles_collection = client.get_or_create_collection("articles")
# Remove company collection - now using entity tracking in facts collection # Remove company collection - now using entity tracking in facts collection
return facts_collection, articles_collection return facts_collection, articles_collection
@ -260,7 +258,7 @@ def embed_and_store_facts(facts, facts_collection, articles_collection):
"main_topic": facts.get('main_topic', 'Unknown') "main_topic": facts.get('main_topic', 'Unknown')
}] }]
) )
# Store extracted facts in facts collection # Store extracted facts in facts collection
facts_text = json.dumps(facts, indent=2) facts_text = json.dumps(facts, indent=2)
facts_embedding = get_embedding(facts_text) facts_embedding = get_embedding(facts_text)
@ -281,10 +279,10 @@ def embed_and_store_facts(facts, facts_collection, articles_collection):
"financial_impact": facts.get('financial_impact', 'neutral') "financial_impact": facts.get('financial_impact', 'neutral')
}] }]
) )
logger.info(f"Successfully processed and stored facts for {facts['filename']}") logger.info(f"Successfully processed and stored facts for {facts['filename']}")
return True return True
except Exception as e: except Exception as e:
logger.error(f"Error embedding and storing facts: {e}") logger.error(f"Error embedding and storing facts: {e}")
return False return False
@ -294,22 +292,22 @@ def main():
Main embedding pipeline function with batch processing Main embedding pipeline function with batch processing
""" """
logger.info("Starting advanced embedding pipeline") logger.info("Starting advanced embedding pipeline")
# Create collections # Create collections
facts_collection, articles_collection = create_collections() facts_collection, articles_collection = create_collections()
# Load cache of previously processed articles # Load cache of previously processed articles
processed_cache = {} processed_cache = {}
if os.path.exists(CACHE_FILE): if os.path.exists(CACHE_FILE):
with open(CACHE_FILE, 'r', encoding='utf-8') as f: with open(CACHE_FILE, 'r', encoding='utf-8') as f:
processed_cache = json.load(f) processed_cache = json.load(f)
# Process articles from scraper directory # Process articles from scraper directory
scraper_articles_dir = "/scraper/articles" scraper_articles_dir = "/scraper/articles"
# Track processing time # Track processing time
start_time = datetime.datetime.now() start_time = datetime.datetime.now()
# Collect all articles to process # Collect all articles to process
articles_to_process = [] articles_to_process = []
for root, dirs, files in os.walk(scraper_articles_dir): for root, dirs, files in os.walk(scraper_articles_dir):
@ -319,20 +317,20 @@ def main():
# Check if already processed # Check if already processed
if file_path not in processed_cache: if file_path not in processed_cache:
articles_to_process.append((file_path, file)) articles_to_process.append((file_path, file))
logger.info(f"Found {len(articles_to_process)} articles to process in batches of {BATCH_SIZE}") logger.info(f"Found {len(articles_to_process)} articles to process in batches of {BATCH_SIZE}")
# Process articles in batches # Process articles in batches
total_processed = 0 total_processed = 0
total_failed = 0 total_failed = 0
for i in range(0, len(articles_to_process), BATCH_SIZE): for i in range(0, len(articles_to_process), BATCH_SIZE):
batch = articles_to_process[i:i + BATCH_SIZE] batch = articles_to_process[i:i + BATCH_SIZE]
logger.info(f"Processing batch {i//BATCH_SIZE + 1} with {len(batch)} articles") logger.info(f"Processing batch {i//BATCH_SIZE + 1} with {len(batch)} articles")
batch_processed = 0 batch_processed = 0
batch_failed = 0 batch_failed = 0
for file_path, file in batch: for file_path, file in batch:
try: try:
# Process the article # Process the article
@ -340,11 +338,11 @@ def main():
if facts: if facts:
# Embed and store in appropriate collections # Embed and store in appropriate collections
success = embed_and_store_facts( success = embed_and_store_facts(
facts, facts,
facts_collection, facts_collection,
articles_collection articles_collection
) )
if success: if success:
# Update cache with detailed status tracking # Update cache with detailed status tracking
processed_cache[file_path] = { processed_cache[file_path] = {
@ -366,15 +364,15 @@ def main():
logger.error(f"Failed to extract facts for {file}") logger.error(f"Failed to extract facts for {file}")
batch_failed += 1 batch_failed += 1
total_failed += 1 total_failed += 1
except Exception as e: except Exception as e:
logger.error(f"Error processing article {file}: {e}") logger.error(f"Error processing article {file}: {e}")
articles_failed_total.inc() articles_failed_total.inc()
batch_failed += 1 batch_failed += 1
total_failed += 1 total_failed += 1
logger.info(f"Batch {i//BATCH_SIZE + 1} completed: {batch_processed} successful, {batch_failed} failed") logger.info(f"Batch {i//BATCH_SIZE + 1} completed: {batch_processed} successful, {batch_failed} failed")
# Save cache periodically during batch processing # Save cache periodically during batch processing
try: try:
with open(CACHE_FILE, 'w', encoding='utf-8') as f: with open(CACHE_FILE, 'w', encoding='utf-8') as f:
@ -382,7 +380,7 @@ def main():
logger.info(f"Cache updated after batch {i//BATCH_SIZE + 1}") logger.info(f"Cache updated after batch {i//BATCH_SIZE + 1}")
except Exception as e: except Exception as e:
logger.error(f"Error saving cache file: {e}") logger.error(f"Error saving cache file: {e}")
# Final cache save # Final cache save
try: try:
with open(CACHE_FILE, 'w', encoding='utf-8') as f: with open(CACHE_FILE, 'w', encoding='utf-8') as f:
@ -390,14 +388,14 @@ def main():
logger.info(f"Final cache saved with {len(processed_cache)} entries") logger.info(f"Final cache saved with {len(processed_cache)} entries")
except Exception as e: except Exception as e:
logger.error(f"Error saving final cache file: {e}") logger.error(f"Error saving final cache file: {e}")
# Calculate and log processing time # Calculate and log processing time
end_time = datetime.datetime.now() end_time = datetime.datetime.now()
total_time = (end_time - start_time).total_seconds() total_time = (end_time - start_time).total_seconds()
processing_time_seconds.observe(total_time) processing_time_seconds.observe(total_time)
logger.info(f"Embedding pipeline completed in {total_time:.2f} seconds") logger.info(f"Embedding pipeline completed in {total_time:.2f} seconds")
logger.info(f"Total processed: {total_processed}, Total failed: {total_failed}") logger.info(f"Total processed: {total_processed}, Total failed: {total_failed}")
logger.info("Embedding pipeline completed") logger.info("Embedding pipeline completed")
if __name__ == "__main__": if __name__ == "__main__":

13
embedding/deprecated/embedder.py
View File

@ -1,3 +1,4 @@
import math
import os import os
import json import json
import chromadb import chromadb
@ -15,8 +16,6 @@ model = SentenceTransformer(
trust_remote_code=True trust_remote_code=True
) )
import math
def embed_text(text, specific_context, max_tokens=2048, overlap=256): def embed_text(text, specific_context, max_tokens=2048, overlap=256):
""" """
Embeds the given text using the SentenceTransformer model. Embeds the given text using the SentenceTransformer model.
@ -59,7 +58,7 @@ output_folder = "/app/output"
while True: while True:
print(f"Starting embedding run at {datetime.datetime.now()}") print(f"Starting embedding run at {datetime.datetime.now()}")
embedded_cache = {} embedded_cache = {}
if os.path.exists(cache_file): if os.path.exists(cache_file):
with open(cache_file, 'r', encoding='utf-8') as f: with open(cache_file, 'r', encoding='utf-8') as f:
@ -74,7 +73,7 @@ while True:
if filename in embedded_cache: if filename in embedded_cache:
print(f"Article {filename} already embedded, skipping.") print(f"Article {filename} already embedded, skipping.")
continue continue
with open(os.path.join(output_folder, filename), 'r', encoding='utf-8') as f: with open(os.path.join(output_folder, filename), 'r', encoding='utf-8') as f:
prev_proc = json.load(f) prev_proc = json.load(f)
output_articles.append(prev_proc) output_articles.append(prev_proc)
@ -92,7 +91,7 @@ while True:
try: try:
specific_context = prev_proc['summary'] + ",".join([' '.join(makeTickerMovement(*x)) for x in prev_proc['tickersAndMovements']]) specific_context = prev_proc['summary'] + ",".join([' '.join(makeTickerMovement(*x)) for x in prev_proc['tickersAndMovements']])
chunks, embedded_content = embed_text(prev_proc['original_content'], prev_proc['summary']) chunks, embedded_content = embed_text(prev_proc['original_content'], prev_proc['summary'])
print(f"Embedded content for {filename}: \n {embedded_content[:10]}...") # Print first 10 values for preview print(f"Embedded content for {filename}: \n {embedded_content[:10]}...") # Print first 10 values for preview
collection.upsert( collection.upsert(
@ -105,11 +104,11 @@ while True:
"filename": prev_proc.get('filename', 'Unknown') # Add filename for traceability "filename": prev_proc.get('filename', 'Unknown') # Add filename for traceability
} for d in chunks], # Metadata for each embedded content } for d in chunks], # Metadata for each embedded content
) )
# Add to processed list for cache update # Add to processed list for cache update
processed_articles.append(filename) processed_articles.append(filename)
print(f"Successfully embedded {filename}") print(f"Successfully embedded {filename}")
except Exception as e: except Exception as e:
print(f"Error embedding content for {filename}: {e}") print(f"Error embedding content for {filename}: {e}")
continue continue

26
embedding/health_check.py
View File

@ -38,7 +38,7 @@ def check_chromadb_connection():
import chromadb import chromadb
CHROMADB_HOST = os.getenv("CHROMADB_HOST", "example.com") CHROMADB_HOST = os.getenv("CHROMADB_HOST", "example.com")
CHROMADB_PORT = int(os.getenv("CHROMADB_PORT", "8000")) CHROMADB_PORT = int(os.getenv("CHROMADB_PORT", "8000"))
client = chromadb.HttpClient(host=CHROMADB_HOST, port=CHROMADB_PORT) client = chromadb.HttpClient(host=CHROMADB_HOST, port=CHROMADB_PORT)
# Test connection by getting all collections # Test connection by getting all collections
collections = client.list_collections() collections = client.list_collections()
@ -57,7 +57,7 @@ def check_ai_server_connection():
AI_SERVER_HOST = os.getenv("AI_SERVER_HOST", "example.com") AI_SERVER_HOST = os.getenv("AI_SERVER_HOST", "example.com")
AI_SERVER_PORT = int(os.getenv("AI_SERVER_PORT", "4000")) AI_SERVER_PORT = int(os.getenv("AI_SERVER_PORT", "4000"))
AI_SERVER_URL = f"http://{AI_SERVER_HOST}:{AI_SERVER_PORT}/v1/embeddings" AI_SERVER_URL = f"http://{AI_SERVER_HOST}:{AI_SERVER_PORT}/v1/embeddings"
# Test by sending a simple request # Test by sending a simple request
start_time = datetime.now() start_time = datetime.now()
response = requests.post( response = requests.post(
@ -87,23 +87,23 @@ def check_scraper_directory():
logger.error(f"✗ Scraper directory does not exist: {scraper_articles_dir}") logger.error(f"✗ Scraper directory does not exist: {scraper_articles_dir}")
health_status.labels(component='scraper_dir').set(0) health_status.labels(component='scraper_dir').set(0)
return False return False
# Check if there are any JSON files # Check if there are any JSON files
json_files = [] json_files = []
for root, dirs, files in os.walk(scraper_articles_dir): for root, dirs, files in os.walk(scraper_articles_dir):
for file in files: for file in files:
if file.endswith('.json'): if file.endswith('.json'):
json_files.append(os.path.join(root, file)) json_files.append(os.path.join(root, file))
if json_files: if json_files:
logger.info(f"✓ Scraper directory accessible. Found {len(json_files)} article files") logger.info(f"✓ Scraper directory accessible. Found {len(json_files)} article files")
health_status.labels(component='scraper_dir').set(1) health_status.labels(component='scraper_dir').set(1)
return True return True
else: else:
logger.warning(f"⚠ Scraper directory exists but no JSON files found") logger.warning("⚠ Scraper directory exists but no JSON files found")
health_status.labels(component='scraper_dir').set(1) # Directory exists, just no files yet health_status.labels(component='scraper_dir').set(1) # Directory exists, just no files yet
return True return True
except Exception as e: except Exception as e:
logger.error(f"✗ Error checking scraper directory: {e}") logger.error(f"✗ Error checking scraper directory: {e}")
health_status.labels(component='scraper_dir').set(0) health_status.labels(component='scraper_dir').set(0)
@ -148,16 +148,16 @@ def main():
logger.info("Prometheus metrics server started on port 8001") logger.info("Prometheus metrics server started on port 8001")
except Exception as e: except Exception as e:
logger.error(f"Failed to start Prometheus server: {e}") logger.error(f"Failed to start Prometheus server: {e}")
logger.info("Starting embedding pipeline health check...") logger.info("Starting embedding pipeline health check...")
checks = [ checks = [
check_chromadb_connection, check_chromadb_connection,
check_ai_server_connection, check_ai_server_connection,
check_scraper_directory, check_scraper_directory,
check_cache_file check_cache_file
] ]
results = [] results = []
for check in checks: for check in checks:
try: try:
@ -166,17 +166,17 @@ def main():
except Exception as e: except Exception as e:
logger.error(f"Error running {check.__name__}: {e}") logger.error(f"Error running {check.__name__}: {e}")
results.append(False) results.append(False)
# Summary # Summary
passed = sum(results) passed = sum(results)
total = len(results) total = len(results)
logger.info(f"\nHealth Check Summary: {passed}/{total} checks passed") logger.info(f"\nHealth Check Summary: {passed}/{total} checks passed")
# Set overall health status # Set overall health status
overall_health = 1 if passed == total else 0 overall_health = 1 if passed == total else 0
health_status.labels(component='overall').set(overall_health) health_status.labels(component='overall').set(overall_health)
if passed == total: if passed == total:
logger.info("✓ All health checks passed - pipeline is ready to run") logger.info("✓ All health checks passed - pipeline is ready to run")
return 0 return 0

48
embedding/query_chromadb.py
View File

@ -6,8 +6,6 @@ This shows how to connect and query your ChromaDB database
import os import os
import chromadb import chromadb
import json
from pathlib import Path
from chromadb.config import Settings from chromadb.config import Settings
def connect_to_chromadb(): def connect_to_chromadb():
@ -16,9 +14,9 @@ def connect_to_chromadb():
# Get connection details from environment or use defaults # Get connection details from environment or use defaults
CHROMADB_HOST = os.getenv("CHROMADB_HOST", "example.com") CHROMADB_HOST = os.getenv("CHROMADB_HOST", "example.com")
CHROMADB_PORT = int(os.getenv("CHROMADB_PORT", "8000")) CHROMADB_PORT = int(os.getenv("CHROMADB_PORT", "8000"))
print(f"Connecting to ChromaDB at {CHROMADB_HOST}:{CHROMADB_PORT}") print(f"Connecting to ChromaDB at {CHROMADB_HOST}:{CHROMADB_PORT}")
# Create client connection using Settings for remote server # Create client connection using Settings for remote server
settings = Settings( settings = Settings(
chroma_api_impl="rest", chroma_api_impl="rest",
@ -26,18 +24,18 @@ def connect_to_chromadb():
chroma_server_http_port=CHROMADB_PORT, chroma_server_http_port=CHROMADB_PORT,
chroma_server_ssl_enabled=False chroma_server_ssl_enabled=False
) )
print(f"Creating client with settings: {settings}") print(f"Creating client with settings: {settings}")
client = chromadb.Client(settings=settings) client = chromadb.Client(settings=settings)
# Test connection # Test connection
collections = client.list_collections() collections = client.list_collections()
print(f"Successfully connected! Found {len(collections)} collections:") print(f"Successfully connected! Found {len(collections)} collections:")
for collection in collections: for collection in collections:
print(f" - {collection.name}") print(f" - {collection.name}")
return client return client
except Exception as e: except Exception as e:
print(f"Failed to connect to ChromaDB: {e}") print(f"Failed to connect to ChromaDB: {e}")
return None return None
@ -47,9 +45,9 @@ def query_facts_collection(client):
try: try:
# Get the facts collection # Get the facts collection
facts_collection = client.get_collection("facts") facts_collection = client.get_collection("facts")
print("\n=== Querying Facts Collection ===") print("\n=== Querying Facts Collection ===")
# Example queries that would work with your data # Example queries that would work with your data
example_queries = [ example_queries = [
"Unrivaled attendance records", "Unrivaled attendance records",
@ -57,17 +55,17 @@ def query_facts_collection(client):
"David Levy on Unrivaled", "David Levy on Unrivaled",
"Fox Business coverage of Unrivaled" "Fox Business coverage of Unrivaled"
] ]
for i, query in enumerate(example_queries, 1): for i, query in enumerate(example_queries, 1):
print(f"\n{i}. Query: '{query}'") print(f"\n{i}. Query: '{query}'")
# Perform similarity search # Perform similarity search
results = facts_collection.query( results = facts_collection.query(
query_texts=[query], query_texts=[query],
n_results=2, n_results=2,
include=["documents", "metadatas", "distances"] include=["documents", "metadatas", "distances"]
) )
if results['documents'] and len(results['documents'][0]) > 0: if results['documents'] and len(results['documents'][0]) > 0:
print(" Results:") print(" Results:")
for j, doc in enumerate(results['documents'][0], 1): for j, doc in enumerate(results['documents'][0], 1):
@ -76,7 +74,7 @@ def query_facts_collection(client):
break break
else: else:
print(" No results found") print(" No results found")
except Exception as e: except Exception as e:
print(f"Error querying facts collection: {e}") print(f"Error querying facts collection: {e}")
@ -85,27 +83,27 @@ def query_articles_collection(client):
try: try:
# Get the articles collection # Get the articles collection
articles_collection = client.get_collection("articles") articles_collection = client.get_collection("articles")
print("\n=== Querying Articles Collection ===") print("\n=== Querying Articles Collection ===")
# Example query # Example query
query = "Unrivaled women's basketball" query = "Unrivaled women's basketball"
print(f"Query: '{query}'") print(f"Query: '{query}'")
# Perform similarity search # Perform similarity search
results = articles_collection.query( results = articles_collection.query(
query_texts=[query], query_texts=[query],
n_results=2, n_results=2,
include=["documents", "metadatas", "distances"] include=["documents", "metadatas", "distances"]
) )
if results['documents'] and len(results['documents'][0]) > 0: if results['documents'] and len(results['documents'][0]) > 0:
print("Results:") print("Results:")
for j, doc in enumerate(results['documents'][0], 1): for j, doc in enumerate(results['documents'][0], 1):
print(f" {j}. {doc[:100]}...") print(f" {j}. {doc[:100]}...")
else: else:
print("No results found") print("No results found")
except Exception as e: except Exception as e:
print(f"Error querying articles collection: {e}") print(f"Error querying articles collection: {e}")
@ -114,20 +112,20 @@ def main():
print("=== ChromaDB Query Demonstration ===") print("=== ChromaDB Query Demonstration ===")
print("This script shows how to connect and query your ChromaDB instance") print("This script shows how to connect and query your ChromaDB instance")
print() print()
# Connect to ChromaDB # Connect to ChromaDB
client = connect_to_chromadb() client = connect_to_chromadb()
if client: if client:
print("\n=== Available Collections ===") print("\n=== Available Collections ===")
collections = client.list_collections() collections = client.list_collections()
for collection in collections: for collection in collections:
print(f" - {collection.name} ({collection.count()} items)") print(f" - {collection.name} ({collection.count()} items)")
# Query each collection # Query each collection
query_facts_collection(client) query_facts_collection(client)
query_articles_collection(client) query_articles_collection(client)
print("\n=== Query Capabilities ===") print("\n=== Query Capabilities ===")
print("The system supports:") print("The system supports:")
print("✓ Semantic similarity search across facts") print("✓ Semantic similarity search across facts")
@ -137,7 +135,7 @@ def main():
print("✓ Multi-entity queries") print("✓ Multi-entity queries")
print() print()
print("All queries use the qwen3:8b embedding model for efficient searching") print("All queries use the qwen3:8b embedding model for efficient searching")
else: else:
print("Cannot connect to ChromaDB. Please ensure:") print("Cannot connect to ChromaDB. Please ensure:")
print("1. ChromaDB server is running") print("1. ChromaDB server is running")
@ -145,4 +143,4 @@ def main():
print("3. Correct host/port configuration") print("3. Correct host/port configuration")
if __name__ == "__main__": if __name__ == "__main__":
main() main()

23
embedding/test_pipeline.py
View File

@ -5,8 +5,6 @@ Test script to demonstrate end-to-end pipeline with a single article
import os import os
import json import json
import tempfile
from pathlib import Path
# Add the current directory to Python path to import our modules # Add the current directory to Python path to import our modules
import sys import sys
@ -14,17 +12,14 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from advanced_embedder import ( from advanced_embedder import (
extract_facts_from_article, extract_facts_from_article,
get_embedding, get_embedding
create_collections,
embed_and_store_facts,
process_article_file
) )
def test_end_to_end(): def test_end_to_end():
"""Test the complete pipeline with a single article""" """Test the complete pipeline with a single article"""
print("=== End-to-End Pipeline Test ===") print("=== End-to-End Pipeline Test ===")
# Your provided article content # Your provided article content
article_content = """SOURCE:Fox Business – Headlines article_content = """SOURCE:Fox Business – Headlines
Unrivaled started out as an idea, and it has turned into a phenomenon. Unrivaled started out as an idea, and it has turned into a phenomenon.
@ -58,31 +53,31 @@ Clark and A'ja Wilson, arguably the WNBA's two biggest stars, have yet to join t
"It's every single great player playing in a three-on-three league. It is absolutely a huge opportunity, and that's why I think it just rose so fast. The quality of play, the names on the back of the jerseys, the social strategy is amazing. These women, they all have equity. Everyone has a following; women athletes completely engage with their fans. The breadth of impressions, I think, is a phenomenal one. I think that's why the league is as successful as it is after just a year and three weeks." """ "It's every single great player playing in a three-on-three league. It is absolutely a huge opportunity, and that's why I think it just rose so fast. The quality of play, the names on the back of the jerseys, the social strategy is amazing. These women, they all have equity. Everyone has a following; women athletes completely engage with their fans. The breadth of impressions, I think, is a phenomenal one. I think that's why the league is as successful as it is after just a year and three weeks." """
title = "Unrivaled Women's Basketball League Breaks Attendance Records" title = "Unrivaled Women's Basketball League Breaks Attendance Records"
print(f"Testing with article: {title}") print(f"Testing with article: {title}")
print("-" * 50) print("-" * 50)
# Test fact extraction # Test fact extraction
print("1. Extracting facts...") print("1. Extracting facts...")
facts = extract_facts_from_article(article_content, title) facts = extract_facts_from_article(article_content, title)
print("Extracted facts:") print("Extracted facts:")
print(json.dumps(facts, indent=2)) print(json.dumps(facts, indent=2))
print() print()
# Test embedding creation # Test embedding creation
print("2. Creating embeddings...") print("2. Creating embeddings...")
facts_text = json.dumps(facts, indent=2) facts_text = json.dumps(facts, indent=2)
embedding = get_embedding(facts_text) embedding = get_embedding(facts_text)
print(f"Created embedding with {len(embedding)} dimensions") print(f"Created embedding with {len(embedding)} dimensions")
print() print()
# Test storage (this would normally connect to ChromaDB) # Test storage (this would normally connect to ChromaDB)
print("3. Testing storage structure...") print("3. Testing storage structure...")
print("Storage would create entries in:") print("Storage would create entries in:")
print("- Facts collection: structured facts with entity tracking") print("- Facts collection: structured facts with entity tracking")
print("- Articles collection: complete article content") print("- Articles collection: complete article content")
print() print()
print("=== Test Complete ===") print("=== Test Complete ===")
print("The pipeline successfully demonstrates:") print("The pipeline successfully demonstrates:")
print("✓ Fact extraction with entity identification") print("✓ Fact extraction with entity identification")
@ -91,4 +86,4 @@ Clark and A'ja Wilson, arguably the WNBA's two biggest stars, have yet to join t
print("✓ Ready for /facts endpoint queries") print("✓ Ready for /facts endpoint queries")
if __name__ == "__main__": if __name__ == "__main__":
test_end_to_end() test_end_to_end()

48
pyproject.toml Normal file
View File

@ -0,0 +1,48 @@
[project]
name = "stockdocs"
version = "1.0.0"
description = "Financial news analysis platform: RSS scraping, AI processing, embeddings, and MCP server"
readme = "README.md"
requires-python = ">=3.9"
license = { text = "MIT" }
authors = [{ name = "Jarian Cottingham", email = "jarianc@proton.me" }]
keywords = ["finance", "news", "scraping", "nlp", "embeddings", "mcp"]
classifiers = [
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
]
dependencies = [
"feedparser>=6.0,<7.0",
"requests>=2.31,<3.0",
"beautifulsoup4>=4.12,<5.0",
"lxml>=5.0",
"nltk>=3.8",
"newspaper4k>=0.2.8,<0.3",
"selenium>=4.15",
"pandas>=2.0",
"pyyaml>=6.0",
"tqdm>=4.64",
"tldextract>=5.0",
]
[project.optional-dependencies]
dev = [
"pytest>=7.0",
"ruff>=0.1.0",
]
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[tool.ruff]
line-length = 120
target-version = "py39"
exclude = [".git", "articles", "ai_processor/ai_processor"]
[tool.ruff.lint]
select = ["E", "F", "W"]
ignore = ["E501"]
[tool.pytest.ini_options]
testpaths = ["tests"]

BIN
scraper/.DS_Store vendored
View File

Binary file not shown.

40
scraper/cron_scraper.py
View File

@ -6,7 +6,6 @@ that can be scheduled via cron job.
""" """
import newspaper import newspaper
from newspaper import Config
import json import json
import feedparser import feedparser
import time import time
@ -14,7 +13,6 @@ import os
import requests import requests
import logging import logging
import random import random
from datetime import datetime
from selenium import webdriver from selenium import webdriver
from selenium.webdriver.firefox.options import Options as FirefoxOptions from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.common.by import By from selenium.webdriver.common.by import By
@ -50,16 +48,15 @@ def get_feed_file_path():
possible_paths = [ possible_paths = [
"./rss_feeds.json", # Current directory "./rss_feeds.json", # Current directory
"../rss_feeds.json", # Parent directory "../rss_feeds.json", # Parent directory
"/home/user/StockDocs/scraper/rss_feeds.json", # Explicit path
"/app/rss_feeds.json", # Docker path "/app/rss_feeds.json", # Docker path
"./scraper/rss_feeds.json" # Scraper subdirectory "./scraper/rss_feeds.json" # Scraper subdirectory
] ]
for path in possible_paths: for path in possible_paths:
if os.path.exists(path): if os.path.exists(path):
print(f"Found feed file at: {path}") print(f"Found feed file at: {path}")
return path return path
# If no file found, exit the program # If no file found, exit the program
print("Error: RSS feed file not found in any expected location") print("Error: RSS feed file not found in any expected location")
print("Exiting program...") print("Exiting program...")
@ -81,10 +78,10 @@ def load_rss_feed_sources(feed_file=FEED_FILE):
Loads the RSS feed sources from a JSON file. Loads the RSS feed sources from a JSON file.
""" """
logger.info(f"Loading RSS feed sources from {feed_file}...") logger.info(f"Loading RSS feed sources from {feed_file}...")
# Debug: Print current working directory # Debug: Print current working directory
logger.debug(f"Current working directory: {os.getcwd()}") logger.debug(f"Current working directory: {os.getcwd()}")
try: try:
with open(feed_file, "r", encoding="utf-8") as f: with open(feed_file, "r", encoding="utf-8") as f:
data = json.load(f) data = json.load(f)
@ -113,12 +110,12 @@ def mine_all_articles(rss_feed_sources, limit=None):
Returns a list of (site, title, link) tuples. Returns a list of (site, title, link) tuples.
""" """
all_links = [] all_links = []
# Check if rss_feed_sources is a valid dict with rss_feeds key # Check if rss_feed_sources is a valid dict with rss_feeds key
if not isinstance(rss_feed_sources, dict): if not isinstance(rss_feed_sources, dict):
logger.warning(f"rss_feed_sources is not a dict, it's {type(rss_feed_sources)}") logger.warning(f"rss_feed_sources is not a dict, it's {type(rss_feed_sources)}")
return all_links return all_links
if "rss_feeds" not in rss_feed_sources: if "rss_feeds" not in rss_feed_sources:
logger.warning("rss_feeds key not found in rss_feed_sources") logger.warning("rss_feeds key not found in rss_feed_sources")
return all_links return all_links
@ -172,7 +169,7 @@ def save_article_to_file(article, filename, source="Unfiltered"):
with open(file_path, "w", encoding="utf-8") as f: with open(file_path, "w", encoding="utf-8") as f:
f.write(f"SOURCE:{source}\n") f.write(f"SOURCE:{source}\n")
f.write(article) f.write(article)
# Only log when a new file is actually created (not cached) # Only log when a new file is actually created (not cached)
logger.info(f"New article saved: {safe_filename} from {source}") logger.info(f"New article saved: {safe_filename} from {source}")
@ -204,7 +201,7 @@ def get_article_with_selenium(url):
WebDriverWait(driver, 15).until( WebDriverWait(driver, 15).until(
EC.presence_of_element_located((By.TAG_NAME, "body")) EC.presence_of_element_located((By.TAG_NAME, "body"))
) )
except: except Exception:
pass pass
time.sleep(random.uniform(1, 3)) time.sleep(random.uniform(1, 3))
@ -224,7 +221,7 @@ def get_article_with_selenium(url):
if driver: if driver:
try: try:
driver.quit() driver.quit()
except: except Exception:
pass pass
@ -293,7 +290,6 @@ def pull_article(link, source, title=None, save_to_file=True):
try: try:
# Try newspaper4k first with proper User-Agent to bypass bot detection # Try newspaper4k first with proper User-Agent to bypass bot detection
ua = get_random_ua() ua = get_random_ua()
config = Config(browser_user_agent=ua)
article = newspaper.article(link, browser_user_agent=ua) article = newspaper.article(link, browser_user_agent=ua)
article.download() article.download()
article.parse() article.parse()
@ -315,7 +311,7 @@ def pull_article(link, source, title=None, save_to_file=True):
logger.info(f"Successfully pulled article from {link} with Playwright") logger.info(f"Successfully pulled article from {link} with Playwright")
if not text or len(text) < 200: if not text or len(text) < 200:
logger.warning(f"Playwright article too short, falling back to Selenium.") logger.warning("Playwright article too short, falling back to Selenium.")
# Fallback to Selenium with better error handling # Fallback to Selenium with better error handling
text = get_article_with_selenium(link) text = get_article_with_selenium(link)
logger.info(f"Successfully pulled article from {link} with Selenium") logger.info(f"Successfully pulled article from {link} with Selenium")
@ -377,20 +373,20 @@ def gather_new_articles():
Gather list of all newly downloaded articles and format them for webhook. Gather list of all newly downloaded articles and format them for webhook.
""" """
new_articles = [] new_articles = []
# Walk through all article directories # Walk through all article directories
for root, dirs, files in os.walk("articles"): for root, dirs, files in os.walk("articles"):
for file in files: for file in files:
if file != "processed_articles_cache.json": # Skip cache file if file != "processed_articles_cache.json": # Skip cache file
# Get the full file path # Get the full file path
file_path = os.path.join(root, file) file_path = os.path.join(root, file)
# Get the outlet name from the directory path # Get the outlet name from the directory path
outlet = os.path.basename(root) outlet = os.path.basename(root)
# Create the relative path for the article # Create the relative path for the article
relative_path = os.path.relpath(file_path, "scraper") relative_path = os.path.relpath(file_path, "scraper")
# Create article data structure # Create article data structure
article_data = { article_data = {
"created_at": time.strftime("%Y-%m-%dT%H:%M:%S.%f", time.localtime(os.path.getctime(file_path))), "created_at": time.strftime("%Y-%m-%dT%H:%M:%S.%f", time.localtime(os.path.getctime(file_path))),
@ -398,9 +394,9 @@ def gather_new_articles():
"outlet": outlet, "outlet": outlet,
"path": f"../{relative_path}" "path": f"../{relative_path}"
} }
new_articles.append(article_data) new_articles.append(article_data)
return new_articles return new_articles
@ -412,7 +408,7 @@ def send_to_webhook(articles):
headers = { headers = {
"StockDocsN8NAuthToken": "ganvT4gsgRjWpGE8FMw9uCzFjZrTx8RZCoVm2Dh7skbZecov" "StockDocsN8NAuthToken": "ganvT4gsgRjWpGE8FMw9uCzFjZrTx8RZCoVm2Dh7skbZecov"
} }
try: try:
response = requests.post(webhook_url, json=articles, headers=headers, timeout=30) response = requests.post(webhook_url, json=articles, headers=headers, timeout=30)
if response.status_code == 200: if response.status_code == 200:
@ -464,7 +460,7 @@ def main():
with open("errors.txt", "w", encoding="utf-8") as f: with open("errors.txt", "w", encoding="utf-8") as f:
for error in errors: for error in errors:
f.write(str(error) + "\n") f.write(str(error) + "\n")
logger.info(f"Errors logged to errors.txt") logger.info("Errors logged to errors.txt")
# Print all results to a log file # Print all results to a log file
with open("results.txt", "w", encoding="utf-8") as f: with open("results.txt", "w", encoding="utf-8") as f:

5
scraper/pyvenv.cfg
View File

@ -1,5 +0,0 @@
home = /usr/bin
include-system-site-packages = false
version = 3.12.3
executable = /usr/bin/python3.12
command = /usr/bin/python3 -m venv /home/user/StockDocs/scraper

50
scraper/scraper.py
View File

@ -1,5 +1,4 @@
import newspaper import newspaper
from newspaper import Config
import json import json
import feedparser import feedparser
import time import time
@ -87,14 +86,14 @@ def mark_article_processed(article_path, status="completed", embedding_status="p
"last_updated": datetime.now().isoformat() "last_updated": datetime.now().isoformat()
} }
save_processed_cache(cache_data) save_processed_cache(cache_data)
def get_processing_progress(): def get_processing_progress():
"""Get overall processing progress""" """Get overall processing progress"""
cache_data = load_processed_cache() cache_data = load_processed_cache()
total_articles = len(cache_data) total_articles = len(cache_data)
completed_articles = sum(1 for data in cache_data.values() if data.get('status') == 'completed') completed_articles = sum(1 for data in cache_data.values() if data.get('status') == 'completed')
embedded_articles = sum(1 for data in cache_data.values() if data.get('embedding_status') == 'completed') embedded_articles = sum(1 for data in cache_data.values() if data.get('embedding_status') == 'completed')
return { return {
"total_articles": total_articles, "total_articles": total_articles,
"completed_articles": completed_articles, "completed_articles": completed_articles,
@ -128,12 +127,12 @@ def mine_all_articles(rss_feed_sources, limit=None):
Returns a list of (site, title, link) tuples. Returns a list of (site, title, link) tuples.
""" """
all_links = [] all_links = []
# Check if rss_feed_sources is a valid dict with rss_feeds key # Check if rss_feed_sources is a valid dict with rss_feeds key
if not isinstance(rss_feed_sources, dict): if not isinstance(rss_feed_sources, dict):
logger.warning(f"rss_feed_sources is not a dict, it's {type(rss_feed_sources)}") logger.warning(f"rss_feed_sources is not a dict, it's {type(rss_feed_sources)}")
return all_links return all_links
if "rss_feeds" not in rss_feed_sources: if "rss_feeds" not in rss_feed_sources:
logger.warning("rss_feeds key not found in rss_feed_sources") logger.warning("rss_feeds key not found in rss_feed_sources")
return all_links return all_links
@ -148,18 +147,18 @@ def mine_all_articles(rss_feed_sources, limit=None):
# Add more aggressive timeout settings with fallback # Add more aggressive timeout settings with fallback
# Use a wrapper to ensure we don't hang indefinitely # Use a wrapper to ensure we don't hang indefinitely
import signal import signal
def timeout_handler(signum, frame): def timeout_handler(signum, frame):
raise TimeoutError(f"Timeout parsing feed: {site}") raise TimeoutError(f"Timeout parsing feed: {site}")
# Set up signal-based timeout (this is a fallback for truly hanging requests) # Set up signal-based timeout (this is a fallback for truly hanging requests)
old_handler = signal.signal(signal.SIGALRM, timeout_handler) old_handler = signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(10) # 10 second alarm signal.alarm(10) # 10 second alarm
feed = feedparser.parse(data["rss_url"], timeout=8) # 8 second timeout feed = feedparser.parse(data["rss_url"], timeout=8) # 8 second timeout
signal.alarm(0) # Cancel the alarm signal.alarm(0) # Cancel the alarm
signal.signal(signal.SIGALRM, old_handler) signal.signal(signal.SIGALRM, old_handler)
feed_entries = feed.entries[:limit] if limit else feed.entries feed_entries = feed.entries[:limit] if limit else feed.entries
entries = [] entries = []
@ -187,14 +186,14 @@ def mine_all_articles(rss_feed_sources, limit=None):
# Use ThreadPoolExecutor for parallel RSS feed parsing # Use ThreadPoolExecutor for parallel RSS feed parsing
from concurrent.futures import ThreadPoolExecutor, as_completed from concurrent.futures import ThreadPoolExecutor, as_completed
max_workers = min(10, len(sources)) # Limit concurrent workers max_workers = min(10, len(sources)) # Limit concurrent workers
with ThreadPoolExecutor(max_workers=max_workers) as executor: with ThreadPoolExecutor(max_workers=max_workers) as executor:
# Submit all feed parsing tasks # Submit all feed parsing tasks
future_to_site = { future_to_site = {
executor.submit(parse_single_feed, site, data): site executor.submit(parse_single_feed, site, data): site
for site, data in sources.items() for site, data in sources.items()
} }
# Collect results as they complete # Collect results as they complete
for future in as_completed(future_to_site, timeout=30): # 30 second overall timeout for future in as_completed(future_to_site, timeout=30): # 30 second overall timeout
try: try:
@ -244,7 +243,7 @@ def save_article_to_file(article, filename, source="Unfiltered"):
with open(file_path, "w", encoding="utf-8") as f: with open(file_path, "w", encoding="utf-8") as f:
f.write(f"SOURCE:{source}\n") f.write(f"SOURCE:{source}\n")
f.write(article) f.write(article)
# Only log when a new file is actually created (not cached) # Only log when a new file is actually created (not cached)
logger.info(f"New article saved: {safe_filename} from {source}") logger.info(f"New article saved: {safe_filename} from {source}")
@ -275,7 +274,7 @@ def get_article_with_selenium(url):
driver = webdriver.Firefox(options=options) driver = webdriver.Firefox(options=options)
else: else:
raise e raise e
driver.set_page_load_timeout(30) # 30 seconds timeout driver.set_page_load_timeout(30) # 30 seconds timeout
# Navigate to URL # Navigate to URL
@ -286,7 +285,7 @@ def get_article_with_selenium(url):
WebDriverWait(driver, 15).until( WebDriverWait(driver, 15).until(
EC.presence_of_element_located((By.TAG_NAME, "body")) EC.presence_of_element_located((By.TAG_NAME, "body"))
) )
except: except Exception:
pass # Continue even if wait times out pass # Continue even if wait times out
time.sleep(random.uniform(1, 3)) # Random wait to mimic human behavior time.sleep(random.uniform(1, 3)) # Random wait to mimic human behavior
@ -313,7 +312,7 @@ def get_article_with_selenium(url):
if driver: if driver:
try: try:
driver.quit() driver.quit()
except: except Exception:
pass # Ignore errors in cleanup pass # Ignore errors in cleanup
@ -387,7 +386,6 @@ def pull_article(link, source, title=None, save_to_file=True):
try: try:
# Try newspaper4k first with proper User-Agent to bypass bot detection # Try newspaper4k first with proper User-Agent to bypass bot detection
ua = get_random_ua() ua = get_random_ua()
config = Config(browser_user_agent=ua)
article = newspaper.article(link, browser_user_agent=ua) article = newspaper.article(link, browser_user_agent=ua)
article.download() article.download()
article.parse() article.parse()
@ -409,7 +407,7 @@ def pull_article(link, source, title=None, save_to_file=True):
logger.info(f"Successfully pulled article from {link} with Playwright") logger.info(f"Successfully pulled article from {link} with Playwright")
if not text or len(text) < 200: if not text or len(text) < 200:
logger.warning(f"Playwright article too short, falling back to Selenium.") logger.warning("Playwright article too short, falling back to Selenium.")
# Fallback to Selenium with better error handling # Fallback to Selenium with better error handling
text = get_article_with_selenium(link) text = get_article_with_selenium(link)
logger.info(f"Successfully pulled article from {link} with Selenium") logger.info(f"Successfully pulled article from {link} with Selenium")
@ -437,7 +435,7 @@ def is_article_downloaded(source, title, link):
# Generate the same filename that would be used for saving # Generate the same filename that would be used for saving
filename = title if title else link filename = title if title else link
safe_filename = generate_filename_from_url(filename) safe_filename = generate_filename_from_url(filename)
# Check if file exists in the articles directory # Check if file exists in the articles directory
file_path = os.path.join("articles", source, safe_filename) file_path = os.path.join("articles", source, safe_filename)
return os.path.exists(file_path) return os.path.exists(file_path)
@ -453,16 +451,16 @@ def safe_pull_articles(article_list):
# Filter out articles that are already downloaded # Filter out articles that are already downloaded
filtered_article_list = [] filtered_article_list = []
total_articles = len(article_list) total_articles = len(article_list)
for source, title, link in article_list: for source, title, link in article_list:
if not is_article_downloaded(source, title, link): if not is_article_downloaded(source, title, link):
filtered_article_list.append((source, title, link)) filtered_article_list.append((source, title, link))
else: else:
logger.info(f"Skipping already downloaded article: {title[:50]}... from {source}") logger.info(f"Skipping already downloaded article: {title[:50]}... from {source}")
logger.info(f"Filtered out {total_articles - len(filtered_article_list)} articles that were already downloaded") logger.info(f"Filtered out {total_articles - len(filtered_article_list)} articles that were already downloaded")
logger.info(f"Processing {len(filtered_article_list)} remaining articles") logger.info(f"Processing {len(filtered_article_list)} remaining articles")
if not filtered_article_list: if not filtered_article_list:
logger.info("No new articles to process") logger.info("No new articles to process")
return [], [] return [], []
@ -474,7 +472,7 @@ def safe_pull_articles(article_list):
# Use configurable worker setting # Use configurable worker setting
max_workers = min(MAX_ARTICLE_WORKERS, len(filtered_article_list)) # Cap at configured workers, but don't exceed article count max_workers = min(MAX_ARTICLE_WORKERS, len(filtered_article_list)) # Cap at configured workers, but don't exceed article count
batch_size = max(1, min(20, len(filtered_article_list) // 4)) # Dynamic batch size batch_size = max(1, min(20, len(filtered_article_list) // 4)) # Dynamic batch size
logger.info(f"Starting parallel article pulling with {max_workers} workers and batch size {batch_size}") logger.info(f"Starting parallel article pulling with {max_workers} workers and batch size {batch_size}")
# Process all articles in parallel with proper error handling # Process all articles in parallel with proper error handling
@ -542,14 +540,14 @@ def main():
with open("errors.txt", "w", encoding="utf-8") as f: with open("errors.txt", "w", encoding="utf-8") as f:
for error in errors: for error in errors:
f.write(str(error) + "\n") f.write(str(error) + "\n")
logger.info(f"Errors logged to errors.txt") logger.info("Errors logged to errors.txt")
# Print all results to a log file # Print all results to a log file
with open("results.txt", "w", encoding="utf-8") as f: with open("results.txt", "w", encoding="utf-8") as f:
for result in results: for result in results:
f.write(result + "\n") f.write(result + "\n")
logger.info("All articles pulled successfully.") logger.info("All articles pulled successfully.")
# Log processing progress # Log processing progress
progress = get_processing_progress() progress = get_processing_progress()
logger.info(f"Processing progress - Total: {progress['total_articles']}, " logger.info(f"Processing progress - Total: {progress['total_articles']}, "
@ -567,4 +565,4 @@ def main():
if __name__ == "__main__": if __name__ == "__main__":
main() main()

43
test_implementation.py
View File

@ -1,43 +0,0 @@
#!/usr/bin/env python3
"""
Test script to verify the enhanced cache system implementation
"""
import json
import os
import datetime
from scraper.scraper import load_processed_cache, save_processed_cache, get_processing_progress
def test_cache_system():
"""Test the enhanced cache system"""
print("Testing enhanced cache system...")
# Test loading cache (should work even if file doesn't exist)
cache = load_processed_cache()
print(f"Initial cache loaded: {len(cache)} entries")
# Test saving cache
test_entry = {
"test_article_path": {
"processed_date": datetime.datetime.now().isoformat(),
"status": "completed",
"embedding_status": "pending",
"last_updated": datetime.datetime.now().isoformat()
}
}
save_processed_cache(test_entry)
print("Cache saved successfully")
# Test loading again
cache = load_processed_cache()
print(f"Cache loaded after save: {len(cache)} entries")
# Test progress tracking
progress = get_processing_progress()
print(f"Progress tracking: {progress}")
print("Cache system test completed successfully!")
if __name__ == "__main__":
test_cache_system()

61
tests/test_scraper_cache.py Normal file
View File

@ -0,0 +1,61 @@
"""Tests for the scraper article-processing cache."""
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from scraper.scraper import ( # noqa: E402
get_processing_progress,
load_processed_cache,
mark_article_processed,
save_processed_cache,
)
@pytest.fixture()
def cache_dir(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
(tmp_path / "articles").mkdir()
return tmp_path
def test_load_missing_cache_returns_empty(cache_dir):
assert load_processed_cache() == {}
def test_save_and_load_roundtrip(cache_dir):
save_processed_cache({"a.html": {"status": "completed"}})
cache = load_processed_cache()
assert cache["a.html"]["status"] == "completed"
def test_mark_article_processed(cache_dir):
mark_article_processed("articles/x/article.html")
cache = load_processed_cache()
entry = cache["articles/x/article.html"]
assert entry["status"] == "completed"
assert entry["embedding_status"] == "pending"
assert "processed_date" in entry
assert "last_updated" in entry
def test_processing_progress_empty(cache_dir):
progress = get_processing_progress()
assert progress["total_articles"] == 0
assert progress["completed_articles"] == 0
assert progress["embedded_articles"] == 0
assert progress["completion_rate"] == 0
def test_processing_progress_mixed(cache_dir):
mark_article_processed("a.html")
mark_article_processed("b.html", status="failed")
mark_article_processed("c.html", embedding_status="completed")
progress = get_processing_progress()
assert progress["total_articles"] == 3
assert progress["completed_articles"] == 2
assert progress["embedded_articles"] == 1
assert progress["completion_rate"] == pytest.approx(2 / 3 * 100)