pax_global_header 0000666 0000000 0000000 00000000064 15107027250 0014511 g ustar 00root root 0000000 0000000 52 comment=c6afc1d2a280b6ab32e5b6fdc3f8e8be8f06664e
GetArgusFlow-argus-c6afc1d/ 0000775 0000000 0000000 00000000000 15107027250 0015752 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/.env.example 0000664 0000000 0000000 00000001504 15107027250 0020175 0 ustar 00root root 0000000 0000000 # ===============================================================
# Global Environment Variables (apply to ALL services)
# ===============================================================
# Service Configuration
# You can override the log level (e.g., "debug", "info", "warning")
# SERVICE__LOG_LEVEL="info"
# Docker Build Configuration
# For development builds (with live-reloading, debuggers, etc.),
# uncomment the following line in your .env file.
# Options: development and production
# By default, builds are optimized for production.
#
# SERVICE__ENVIRONMENT=development
# Database Credentials
# IMPORTANT: Replace these with your actual database connection details.
# DATABASE__HOST="localhost"
# DATABASE__USER="your_db_user"
# DATABASE__PASSWORD="your_secret_password"
# DATABASE__NAME="your_db_name"
# DATABASE__PORT="3306" GetArgusFlow-argus-c6afc1d/.gitignore 0000664 0000000 0000000 00000011623 15107027250 0017745 0 ustar 00root root 0000000 0000000 # Byte-compiled / optimized / DLL files
__pycache__/
*.py[codz]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py.cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
#poetry.toml
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
# https://pdm-project.org/en/latest/usage/project/#working-with-version-control
#pdm.lock
#pdm.toml
.pdm-python
.pdm-build/
# pixi
# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
#pixi.lock
# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
# in the .venv directory. It is recommended not to include this directory in version control.
.pixi
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.envrc
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Abstra
# Abstra is an AI-powered process automation framework.
# Ignore directories containing user credentials, local state, and settings.
# Learn more at https://abstra.io/docs
.abstra/
# Visual Studio Code
# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore
# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
# and can be added to the global gitignore or merged into this file. However, if you prefer,
# you could uncomment the following to ignore the entire vscode folder
# Ignore the entire VS Code directory by default
.vscode/
# But DO NOT ignore these specific shared configuration files
!.vscode/launch.json
!.vscode/extensions.json
!.vscode/tasks.json
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
# Cursor
# Cursor is an AI-powered code editor. `.cursorignore` specifies files/directories to
# exclude from AI features like autocomplete and code analysis. Recommended for sensitive data
# refer to https://docs.cursor.com/context/ignore-files
.cursorignore
.cursorindexingignore
# Marimo
marimo/_static/
marimo/_lsp/
__marimo__/
# Other
.DS_Store
# Argus
.history
logs/*
models/*
tools/*
.setup_complete
docker-compose.override.yml
docker-compose.pro.yml
custom_patterns.yml GetArgusFlow-argus-c6afc1d/LICENSE 0000664 0000000 0000000 00000002052 15107027250 0016756 0 ustar 00root root 0000000 0000000 MIT License
Copyright (c) 2025 ArgusFlow
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.
GetArgusFlow-argus-c6afc1d/Makefile 0000664 0000000 0000000 00000005736 15107027250 0017425 0 ustar 00root root 0000000 0000000 # argus/Makefile
# This Makefile delegates all commands to the service-level Makefiles.
# It does NOT run docker-compose commands itself.
SHELL := /bin/bash
.DEFAULT_GOAL := help
SETUP_MARKER := .setup_complete
# Automatic Service Detection
FOUND_MAKEFILES := $(wildcard services/*/Makefile)
SERVICE_DIRS := $(dir $(FOUND_MAKEFILES))
SERVICES := $(patsubst services/%/,%,$(SERVICE_DIRS))
.PHONY: help up up-dev down logs build build-dev test lint format clean install-pro setup check-setup _check_yq
# example: sudo dnf install yq
_check_yq:
@command -v $(YQ) >/dev/null 2>&1 || { \
echo "Error: 'yq' is not installed but is required by this Makefile."; \
echo "Please install it from: https://github.com/mikefarah/yq#install"; \
exit 1; \
}
help:
@echo "Usage: make [target]"
@echo "This is the global Makefile. Commands run from here affect ALL services."
@echo "To run a command on a single service, 'cd' into its directory (e..g., 'services/matcher')."
@echo ""
@echo "Global Commands:"
@echo " up, up-dev, down, logs, build, build-dev"
@echo " test, lint, format, clean, setup"
@echo ""
@echo "Available Services: $(SERVICES)"
check-setup:
@if [ ! -f "$(SETUP_MARKER)" ]; then \
echo "Project is not yet initialized."; \
echo " Please run 'make setup' once before you begin."; \
exit 1; \
fi
setup: _check_yq
@echo "Running '$@' on ALL services: $(SERVICES)"
@for s in $(SERVICES); do \
echo ""; \
echo "===== Running '$@' on service '$$s' ====="; \
$(MAKE) -C services/$$s $@; \
done; \
touch $(SETUP_MARKER);
@echo ""
@echo "Setup has finished!";
@echo ""
# Global Docker Commands
up up-dev down restart logs build build-dev test lint format clean: check-setup
@echo "Running '$@' on ALL services: $(SERVICES)"
@for s in $(SERVICES); do \
echo ""; \
echo "===== Running '$@' on service '$$s' ====="; \
$(MAKE) -C services/$$s $@; \
done; \
# Pro
-include .env
export
install-pro:
@echo "Preparing Argus Pro Installation"
@rm -rf ../argus-pro;
@mkdir -p ../argus-pro;
@read -p "Enter your Argus Pro license key: " LICENSE_KEY; \
if [ -z "$${LICENSE_KEY}" ]; then echo "Error: No token provided."; exit 1; fi; \
\
INSTALLER_URL="https://pro.argusflow.com"; \
\
mkdir -p ../argus-pro; \
echo "Verifying license and downloading Argus Pro from $${INSTALLER_URL}..."; \
if curl -sSLf "$${INSTALLER_URL}?key=$${LICENSE_KEY}" | tar -xzf - -C ../argus-pro --strip-components=1; then \
echo "Argus Pro downloaded successfully."; \
else \
echo "\nError: Installation failed. Please check your license key or network connection."; \
rm -rf ../argus-pro; \
exit 1; \
fi; \
chmod +x ../argus-pro/setup.sh; \
if [ -x "../argus-pro/setup.sh" ]; then \
echo "Running Pro setup script to configure services..."; \
../argus-pro/setup.sh .; \
echo "Argus Pro is now configured! Please restart your services with 'make down' then 'make up'."; \
else \
echo "Error: setup.sh not found in the downloaded package."; \
rm -rf ../argus-pro; \
exit 1; \
fi; GetArgusFlow-argus-c6afc1d/README.md 0000664 0000000 0000000 00000011047 15107027250 0017234 0 ustar 00root root 0000000 0000000 # ArgusFlow
**ArgusFlow** is a modular, microservice-based data extraction platform designed for **high performance, flexibility, and scalability**.
It provides a unified framework for extracting structured data from websites using a collection of independent yet interconnected services.
All services are containerized and orchestrated via **Docker Compose**, with a central **Makefile** providing full project automation.
---
## Table of Contents
- [Core Services](#core-services)
- [Prerequisites](#prerequisites)
- [Getting Started](#getting-started)
- [Main Commands](#main-commands)
- [Project Structure](#project-structure)
- [Argus Pro](#argus-pro)
---
## Core Services
ArgusFlow consists of four primary containerized microservices, each with a specific purpose:
| Service | Port | Description |
|----------|------|-------------|
| **Extractor** | `:8001` | Extracts structured data (price, brand, etc.) from product HTML using a modular, "scoreboard-based" parser system. |
| **LLM Parser** | `:8002` | Uses a local LLM and GBNF grammar to convert complex HTML snippets (e.g., specification tables) into strict JSON. |
| **Generalizer** | `:8003` | Transforms unstructured product titles into rich, structured JSON using local LLMs and dynamic language grammars. |
| **Matcher** | `:8004` | Finds similar products in your database using a fine-tuned Sentence Transformer and a high-speed FAISS index. |
---
## Prerequisites
Before you begin, ensure you have the following installed:
- **Docker**
- **Docker Compose**
- **make**
---
## Getting Started
### 1. Clone the Repository
```bash
mkdir argus && cd argus && curl -L https://get.argusflow.com | tar -xz --strip-components=1
```
Or via Git:
```bash
git clone https://github.com/getargusflow/argus.git
cd argus
```
---
### 2. Initial Project Setup
Run the setup command once to prepare the environment.
This will create `.env` files from the examples and execute each service’s one-time setup (e.g., downloading models):
```bash
make setup
```
---
### 3. Start the Services
To start all services in detached mode (builds images if needed):
```bash
make up
```
The services will start in the background.
Check their status with:
```bash
docker ps
```
---
## Main Commands
All project operations are managed through the **top-level Makefile** in the `argus/` directory.
### Project Setup
| Command | Description |
|----------|-------------|
| `make setup` | Initializes the project, creates `.env` files, and runs setup tasks for all services. |
### Production Environment
| Command | Description |
|----------|-------------|
| `make build` | Builds optimized Docker images for production. |
| `make up` | Starts all services in production mode. |
### Development Environment
| Command | Description |
|----------|-------------|
| `make build-dev` | Builds development images (with debuggers, live reload, etc.). |
| `make up-dev` | Starts all services in development mode. |
### General Commands
| Command | Description |
|----------|-------------|
| `make down` | Stops and removes all running service containers. |
| `make logs` | Displays live logs for all services. |
| `make test` | Runs all test suites. |
| `make lint` | Runs lint checks for code quality. |
| `make clean` | Cleans up temporary files and caches. |
---
## Targeting a Specific Service
To run a command (like logs or test) on a single service:
```bash
cd services/extractor
make logs
```
Or for the LLM Parser:
```bash
cd services/llm_parser
make test
```
---
## Project Structure
```
argus/
├── Makefile # Central control hub for the project
└── services/
├── extractor/ # Generic HTML Extractor microservice
├── llm_parser/ # LLM-based HTML Snippet Parser
├── generalizer/ # LLM-based Title Generalizer
└── matcher/ # Vector-based Product Matcher
```
---
## Argus Pro
**Argus Pro** is the closed-source commercial extension that unlocks advanced extraction capabilities for enterprise users.
### Why Upgrade?
- **Advanced Modules:**
Gain access to specialized modules for extracting complex data (EAN/SKU, Product Specifications, Original Price, Currency, etc.).
- **Optimized Performance:**
Enhanced pipelines and models for production-scale workloads.
### Installation
To install Argus Pro:
```bash
make install-pro
```
The script will prompt for your **license key** and automatically install the Pro extensions.
After installation, restart all services:
```bash
make down
make up
```
---
© ArgusFlow
GetArgusFlow-argus-c6afc1d/pyproject.toml 0000664 0000000 0000000 00000000204 15107027250 0020662 0 ustar 00root root 0000000 0000000 [tool.ruff]
exclude = [
".venv",
".git",
"__pycache__",
"build",
"dist",
"tools/json_schema_to_grammar.py"
] GetArgusFlow-argus-c6afc1d/services/ 0000775 0000000 0000000 00000000000 15107027250 0017575 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/common.mk 0000664 0000000 0000000 00000013403 15107027250 0021417 0 ustar 00root root 0000000 0000000 # argus/services/common.mk
# This Makefile is included by service-level Makefiles (e.g., matcher/Makefile)
# It assumes it's running from within a service directory (e.g., services/matcher/)
# Config
ROOT_DIR := ../..
# SERVICE_NAME must be defined in the Makefile that includes this one.
# Check docker version
COMPOSE_V1 := $(shell which docker-compose 2>/dev/null)
ifeq ($(COMPOSE_V1),)
COMPOSE_CMD := docker compose
else
COMPOSE_CMD := docker-compose
endif
# Add the local (service) compose file if it exists
BASE_FILES :=
ifneq ($(wildcard docker-compose.yml),)
BASE_FILES += -f docker-compose.yml
endif
# docker-compose.pro.yml
PRO_FILES :=
ifneq ($(wildcard docker-compose.pro.yml),)
PRO_FILES := -f docker-compose.pro.yml
endif
# PROD command (for 'make up', 'make build')
PROD_FILES := $(BASE_FILES)
ifneq ($(wildcard docker-compose.prod.yml),)
PROD_FILES += -f docker-compose.prod.yml
endif
# DEV for specific environment variables and services
DEV_FILES := $(BASE_FILES)
ifneq ($(wildcard docker-compose.dev.yml),)
DEV_FILES += -f docker-compose.dev.yml
endif
# TEST command (for 'make test')
# This uses dev files + a specific test override
TEST_FILES := $(DEV_FILES)
ifneq ($(wildcard docker-compose.test.yml),)
TEST_FILES += -f docker-compose.test.yml
endif
# Define commands
COMPOSE_CMD_PROD := $(COMPOSE_CMD) $(PROD_FILES) $(PRO_FILES)
COMPOSE_CMD_DEV := $(COMPOSE_CMD) $(DEV_FILES) $(PRO_FILES)
COMPOSE_CMD_TEST := $(COMPOSE_CMD) $(TEST_FILES) $(PRO_FILES)
COMPOSE_CMD_ALL := $(COMPOSE_CMD) $(PROD_FILES) $(DEV_FILES) $(TEST_FILES) $(PRO_FILES)
# Standard Targets
.PHONY: help _common_help up up-dev down logs build build-dev test lint format clean health ensure-env ensure-network
help: _common_help
_common_help:
@echo "Usage: make [target]"
@echo "This is a service-level Makefile for $(SERVICE_NAME)."
@echo "Run commands from here to affect ONLY this service."
@echo ""
@echo "Standard Targets:"
@echo " setup - Run $(SERVICE_NAME) setup."
@echo " up - Start $(SERVICE_NAME) in PRODUCTION mode."
@echo " up-dev - Start $(SERVICE_NAME) in DEVELOPMENT mode."
@echo " down - Stop and remove $(SERVICE_NAME) containers (and dev dependencies)."
@echo " logs - Show live logs for $(SERVICE_NAME)."
@echo " build - Build the PRODUCTION image for $(SERVICE_NAME)."
@echo " build-dev - Build the DEVELOPMENT image for $(SERVICE_NAME)."
@echo " test - Run tests for $(SERVICE_NAME)."
@echo " lint - Run the linter for $(SERVICE_NAME)."
@echo " format - Format code for $(SERVICE_NAME)."
@echo " clean - Clean Python cache files."
@echo " health - Check the /health endpoint of $(SERVICE_NAME)."
# Include the root .env file for shared vars
-include $(ROOT_DIR)/.env
# Include the local .env file for service-specific vars
-include .env
# Export all variables to sub-processes (like docker compose)
export
ensure-network:
@docker network inspect argus_network >/dev/null 2>&1 || docker network create argus_network
# Create .env file for services
ensure-env:
@if [ ! -f "../../.setup_complete"; then \
echo "Project is not yet initialized."; \
echo " Please run 'make setup' in the main Argus directory once before you begin."; \
exit 1; \
fi
@echo "Checking for required .env files..."
@if [ ! -f ".env" ] && [ -f ".env.example" ]; then \
echo "Creating main .env from .env.example..."; \
cp ".env.example" ".env"; \
fi
setup: ensure-env
@echo " $(SERVICE_NAME) setup..."
up: ensure-env ensure-network
@echo "Starting $(SERVICE_NAME) in PRODUCTION mode..."
$(COMPOSE_CMD_PROD) up --build -d $(SERVICE_NAME)
up-dev: ensure-env ensure-network
@echo "Starting $(SERVICE_NAME) in DEVELOPMENT mode..."
SERVICE__ENVIRONMENT=development $(COMPOSE_CMD_DEV) up --build -d $(SERVICE_NAME)
down: ensure-env
@echo "Stopping and removing $(SERVICE_NAME)..."
# --remove-orphans is key to removing dev-only containers (like db-demo)
$(COMPOSE_CMD_ALL) down --remove-orphans
logs: ensure-env
@echo "Tailing logs for $(SERVICE_NAME)..."
$(COMPOSE_CMD_ALL) logs -f $(SERVICE_NAME)
build: ensure-env
@echo "Building PRODUCTION image for $(SERVICE_NAME)..."
SERVICE__ENVIRONMENT=production $(COMPOSE_CMD_PROD) build $(SERVICE_NAME)
build-no-cache: ensure-env
@echo "Building PRODUCTION image for $(SERVICE_NAME)..."
SERVICE__ENVIRONMENT=production $(COMPOSE_CMD_PROD) build --no-cache $(SERVICE_NAME)
build-dev: ensure-env
@echo "Building DEVELOPMENT image for $(SERVICE_NAME)..."
SERVICE__ENVIRONMENT=development $(COMPOSE_CMD_DEV) build $(SERVICE_NAME)
build-dev-no-cache: ensure-env
@echo "Building DEVELOPMENT image for $(SERVICE_NAME)..."
SERVICE__ENVIRONMENT=development $(COMPOSE_CMD_DEV) build --no-cache $(SERVICE_NAME)
test: ensure-env ensure-network
ifneq ($(wildcard tests),)
@echo "Running tests for $(SERVICE_NAME) in a one-off container..."
@echo "This runs 'pytest' without starting the FastAPI service."
@( SERVICE__ENVIRONMENT=test $(COMPOSE_CMD_TEST) run --build --rm $(SERVICE_NAME) pytest tests ) ; \
EXIT_CODE=$$? ; \
@echo "Stopping test dependencies..." ; \
$(COMPOSE_CMD_TEST) down --remove-orphans ; \
(exit $$EXIT_CODE)
else
@echo "No 'tests' directory found for $(SERVICE_NAME). Skipping test run."
endif
lint: ensure-env
@echo "Running linter for $(SERVICE_NAME)..."
$(COMPOSE_CMD_DEV) run --rm $(SERVICE_NAME) ruff check .
format: ensure-env
@echo "Formatting code for $(SERVICE_NAME)..."
$(COMPOSE_CMD_DEV) run --rm $(SERVICE_NAME) ruff format .
clean:
@echo "Cleaning Python cache for $(SERVICE_NAME)..."
@find . -type f -name "*.py[co]" -delete
@find . -type d -name "__pycache__" -exec rm -rf {} +; \
echo "Cleanup complete."
health: ensure-env
@echo "- Checking service health at http://localhost:$(SERVICE_PORT)/health"; \
curl -f http://localhost:$(SERVICE_PORT)/health || (echo "Health check failed." && exit 1) GetArgusFlow-argus-c6afc1d/services/extractor/ 0000775 0000000 0000000 00000000000 15107027250 0021610 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/.env.example 0000664 0000000 0000000 00000001731 15107027250 0024035 0 ustar 00root root 0000000 0000000 # ===============================================================
# Generalizer Service Environment Variables
# ===============================================================
# These variables override the values in config/config.yml
# Use double underscores (__) to override nested settings.
# Service Settings
#SERVICE__HOST=0.0.0.0
#SERVICE__PORT=8001
# You can override the log level (e.g., "debug", "info", "warning")
# SERVICE__LOG_LEVEL="info"
# You can override the environment to production or development.
#SERVICE__ENVIRONMENT="development"
# Language Settings
#LANGUAGE__DEFAULT=en
#--------------------------------------------------------------------------
# AUTHENTICATION
#--------------------------------------------------------------------------
# This is the secret API key used to protect sensitive endpoints
# You MUST set this to a strong, random key in production.
# The default "default_dev_key" will work for local testing.
#AUTH__API_KEY="default_dev_key" GetArgusFlow-argus-c6afc1d/services/extractor/Dockerfile 0000664 0000000 0000000 00000007044 15107027250 0023607 0 ustar 00root root 0000000 0000000 # argus/services/extractor/Dockerfile
# =================================================================
# Stage 1: Base - The common foundation
# =================================================================
FROM python:3.11-slim AS base
WORKDIR /app
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
ENV PATH="/usr/local/bin:${PATH}"
RUN apt-get update && apt-get install -y --no-install-recommends \
libgomp1 \
# Playwright/Chromium dependencies
libglib2.0-0 \
libdbus-1-3 \
libnss3 \
libatk1.0-0 \
libatk-bridge2.0-0 \
libcups2 \
libdrm2 \
libgbm1 \
libxkbcommon0 \
libxcomposite1 \
libxdamage1 \
libxfixes3 \
libxrandr2 \
libxshmfence1 \
libasound2 \
libpangocairo-1.0-0 \
libfontconfig1 \
# end of Playwright dependencies
&& rm -rf /var/lib/apt/lists/*
# =================================================================
# Stage 2: Common Builder - Installs OS-level build tools
# Both prod and dev builders will use this as their base
# =================================================================
FROM base AS common-builder
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
curl \
git \
jq \
&& rm -rf /var/lib/apt/lists/*
# =================================================================
# Stage 3: Production Builder - Installs ONLY production dependencies
# =================================================================
FROM common-builder AS prod-builder
COPY services/extractor/requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
RUN playwright install --with-deps chromium
RUN python -m spacy download nl_core_news_lg
# =================================================================
# Stage 4: Development Builder - Installs ALL dependencies
# This builds ON TOP of the prod-builder for efficiency
# =================================================================
FROM prod-builder AS dev-builder
COPY services/extractor/requirements-dev.txt ./
COPY pyproject.toml ./
# This installs *only* the extra dev packages, since prod ones are inherited
RUN pip install --no-cache-dir -r requirements-dev.txt
# =================================================================
# Stage 5: Development - For local development with live-reload
# =================================================================
FROM dev-builder AS development
# Copy the entire project context
COPY services/extractor/ .
# This command still works, as dev-builder has debugpy
CMD ["python", "-Xfrozen_modules=off", "-m", "debugpy", "--listen", "0.0.0.0:5678", "--wait-for-client", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8001", "--reload"]
# =================================================================
# Stage 6: Production - A lean image for deployment
# =================================================================
FROM base AS production
# Copy only the lean production packages from the prod-builder
COPY --from=prod-builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=prod-builder /usr/local/bin /usr/local/bin
COPY --from=prod-builder /root/.cache/ms-playwright /root/.cache/ms-playwright
# Copy the app directory to /app/app to match the import path
COPY services/extractor/app /app/app
COPY services/extractor/config /app/config
# Expose the port the app runs on
EXPOSE 8001
# Command to run the app in production
CMD uvicorn app.main:app --host ${SERVICE__HOST:-0.0.0.0} --port ${SERVICE__PORT:-8001} --log-level ${SERVICE__LOG_LEVEL:-info} GetArgusFlow-argus-c6afc1d/services/extractor/Makefile 0000664 0000000 0000000 00000000166 15107027250 0023253 0 ustar 00root root 0000000 0000000 # ArgusProject/argus/services/extractor/Makefile
SERVICE_NAME := extractor
SERVICE_PORT := 8001
include ../common.mk GetArgusFlow-argus-c6afc1d/services/extractor/README.md 0000664 0000000 0000000 00000017074 15107027250 0023100 0 ustar 00root root 0000000 0000000 # Argus Extractor
Argus Extractor is a microservice designed to extract structured product information—such as price, brand, EAN, and specifications—from the HTML of a product page.
The service utilizes a modular architecture with various parsers and a “best-score-wins” mechanism to identify the most reliable data.
This project is fully containerized with Docker for easy deployment and consistent environments, following modern development standards.
-----
## Table of Contents
- [Features](#features)
- [Security](#security)
- [Architecture](#architecture)
- [Prerequisites](#prerequisites)
- [Quick Start (Development)](#quick-start-development)
- [Production Deployment](#production-deployment)
- [Stopping the Service](#stopping-the-service)
- [API Usage](#api-usage)
- [Multilingual Support & Customization](#multilingual-support--customization)
-----
## Features
- **FastAPI Backend** — A modern, high-performance web framework for building APIs.
- **Secure by Default** — **All API endpoints** (except `/health`) are protected by a mandatory `x-api-key` header.
- **Modular Parsers** — A scalable architecture where multiple, independent parsers can exist for each data field (e.g., price, brand).
- **Intelligent Data Selection** — A “scoreboard” system weighs results from all parsers and selects the data with the highest confidence score.
- **Advanced HTML Analysis** — Uses BeautifulSoup for HTML parsing and a spaCy NLP model for smarter text analysis.
- **Browser Automation** — Integrates with Playwright to retrieve dynamically loaded (JavaScript) content.
- **Fully Containerized** — Includes multi-stage Docker builds and docker-compose.yml files for both development and production.
- **Professional Tooling:**
- Configuration via YAML (`config/config.yml`) and `.env` files.
- Structured logging with Loguru.
- Automated tasks via a Makefile (linting, testing, etc.).
- Code quality enforcement with Ruff.
-----
## Security
All API endpoints (except for the `/health` check) require a valid API key to be passed in the `x-api-key` header.
- **Development**: The default key is `default_dev_key`.
- **Production**: You **must** override this by setting the `AUTH__API_KEY` environment variable to a strong, randomly generated key.
> **Production Warning**: This service is configured to listen on `0.0.0.0`. For production, it is strongly recommended to place this service behind a firewall or API gateway and restrict network access. Do not expose it directly to the public internet.
-----
## Architecture
Argus Extractor operates in several phases to extract structured data:
### 1. HTML Pre-processing
The raw HTML is cleaned by removing “noise” (headers, footers, sidebars) based on `noise_selectors` in `config/config.yml`.
### 2. Modules
Each parser module targets a specific field (e.g., title, price). Each parser returns a result with a score. General data extractors (like `json_ld` and `open_graph`) are stored in a shared context and immediately added to the scoreboard.
### 3. Enrichment
Extracted specifications are scanned for known aliases (e.g., `"Manufacturer" → "brand"`) to fill in missing fields.
### 4. Resolving
For each field, the result with the highest score is chosen as the final output.
-----
## Prerequisites
- Docker and Docker Compose
- `make` (available on Linux/macOS; installable on Windows via WSL or Chocolatey)
-----
## Quick Start (Development)
This setup is optimized for local development with live-reloading and debugging.
### 1. Configuration
From the `argus/services/extractor/` directory, create a local configuration file from the example template:
```bash
cp .env.example .env
```
Then open `.env` to override settings from `config/config.yml` if needed.
### 2. Start the Service
Ensure you are in the `argus/services/extractor/` directory, then run:
```bash
make build-dev
```
```bash
make up-dev
```
The service will be available at:
**http://localhost:8001**
Interactive API docs (Swagger UI):
**http://localhost:8001/docs**
-----
## Production Deployment
For production, the service uses optimized, minimal images.
### 1. Build the Production Image
From the `argus/services/extractor/` directory:
```bash
make build
```
### 2. Start the Production Service
```bash
make up
```
The service is now running in production mode.
-----
## Stopping the Service
To stop and remove all containers associated with this service (for both dev and prod):
```bash
make down
```
-----
## API Usage
### POST `/api/v1/extract`
(Requires API key) Extracts data from a URL or provided HTML content.
All requests must include a valid API key in the header:
`x-api-key: YOUR_API_KEY`
This endpoint has two modes:
- **Fetch Mode** — If `html_content` is null or empty, the service fetches content from the URL using Playwright.
- **Direct Mode** — If `html_content` is provided, the service parses it directly.
#### Request Body
```json
{
"url": "https://www.example.com/product/123",
"html_content": "...",
"use_llm": false
}
```
#### Example (Fetch Mode)
```bash
curl -X POST "http://localhost:8001/api/v1/extract" -H "Content-Type: application/json" -H "x-api-key: default_dev_key" -d '{
"url": "https://www.domain.nl/product/1234/product.html",
"html_content": null,
"use_llm": true
}'
```
#### Example (Direct Mode)
```bash
curl -X POST "http://localhost:8001/api/v1/extract" -H "Content-Type: application/json" -H "x-api-key: default_dev_key" -d '{
"url": "https://www.domain.nl/product/1234/product.html",
"html_content": "...",
"use_llm": false
}'
```
#### Success Response (200 OK)
```json
{
"data": {
"title": "An amazing product",
"brand": "ExampleBrand",
"price": 123.45,
"original_price": null,
"currency": "USD",
"specifications": []
},
"message": "Extraction successful"
}
```
-----
## Multilingual Support & Customization
Argus Extractor supports multilingual extraction through a modular file structure located in the `config/patterns/` directory.
### Directory Structure
1. **`config/patterns/default.yml`** Contains language-agnostic patterns and regex definitions (e.g., price formats, SKU regexes) that are **always** loaded.
2. **`config/patterns/[lang]_patterns.yml`** Contains keywords and specific logic for a supported language (e.g., `en_patterns.yml`, `nl_patterns.yml`).
3. **`config/patterns/custom/`** Reserved for your custom configurations. Files placed here take precedence over the base patterns of the same name.
### Language Loading Logic
The extractor uses a **cumulative** loading strategy to ensure maximum coverage:
1. **Default (Always Loaded):** The language defined in `config/config.yml` (e.g., `en`) is **always loaded**. This ensures that standard English terms (often used in HTML class names like `class="price"`) are recognized regardless of the page's language.
2. **Target (Added):** If a specific language is requested via API or detected (e.g., `nl`), its patterns are loaded **in addition** to the default.
### How to Customize
#### 1. Add a New Language
To add support for a new language (e.g., French), create a new file in the custom directory:
`config/patterns/custom/fr_patterns.yml`
#### 2. Override an Existing Language
To modify the patterns for an existing language (e.g., Dutch), create a file with the **exact same filename** in the custom directory:
`config/patterns/custom/nl_patterns.yml`
**Note:** This **replaces** the base file entirely, giving you full control over that language's definitions.
-----
© ArgusFlow
GetArgusFlow-argus-c6afc1d/services/extractor/app/ 0000775 0000000 0000000 00000000000 15107027250 0022370 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0024467 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/api/ 0000775 0000000 0000000 00000000000 15107027250 0023141 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/api/v1/ 0000775 0000000 0000000 00000000000 15107027250 0023467 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/api/v1/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0025566 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/api/v1/endpoints.py 0000664 0000000 0000000 00000014167 15107027250 0026055 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/api/v1/endpoints.py
from fastapi import APIRouter, Request, HTTPException, Depends
from loguru import logger
from typing import Optional, List
from playwright.async_api import (
async_playwright,
TimeoutError as PlaywrightTimeoutError,
)
from pathlib import Path
from urllib.parse import urlparse
import json
from app.api.v1.schemas import ExtractionRequest, ExtractionResponse
from app.core.analyzer import ProductPageAnalyzer
from app.config import settings
from .security import get_api_key
router = APIRouter(dependencies=[Depends(get_api_key)])
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36"
@router.post("/extract", response_model=ExtractionResponse)
async def extract_data(request: Request, payload: ExtractionRequest):
"""
(NEEDS KEY) Extracts structured data from a product page.
- If 'html_content' is provided (and not empty), it's used directly for analysis.
- If 'html_content' is null or empty, the 'url' parameter is used to fetch
the page content via Playwright, which is then analyzed.
"""
# Get the analyzer instance that was loaded on startup
analyzer: ProductPageAnalyzer = getattr(request.app.state, "analyzer", None)
if not analyzer:
logger.error("API: /extract called but analyzer is not available in app.state.")
raise HTTPException(
status_code=503, # Service Unavailable
detail="The analysis service is not ready. The analyzer failed to load on startup.",
)
logger.info(f"API: Received extraction request for URL: {payload.url}")
# Normalize the 'lang' parameter (which can be str, List[str], or None)
raw_lang = payload.lang
lang_list: Optional[List[str]] = None
if isinstance(raw_lang, str):
lang_list = [raw_lang]
elif isinstance(raw_lang, list):
lang_list = raw_lang
logger.info(
f"API: Received extraction request for URL: {payload.url}. Languages: {lang_list or 'FALLBACK'}"
)
html_content = payload.html_content
# Ensure url is a string for all operations
url = str(payload.url)
if payload.file:
try:
p = Path(payload.file)
if p.is_absolute() and p.is_file():
logger.info(
f"Dev mode: Detected absolute file path. Reading from: {url}"
)
html_content = p.read_text(encoding="utf-8")
except Exception as e:
logger.warning(f"Failed to read file path {url}. Error: {e}")
# If no HTML content is provided (falsy check for None or ""), fetch it
if not html_content:
logger.info(f"No html_content provided for {url}. Fetching with Playwright...")
try:
async with async_playwright() as p:
browser = await p.chromium.launch()
context = await browser.new_context(
viewport={"width": 1920, "height": 1080}, user_agent=USER_AGENT
)
page = await context.new_page()
logger.info(f"Navigating to {url} with Playwright...")
await page.goto(url, wait_until="networkidle", timeout=30000)
# Assign fetched content to html_content
html_content = await page.content()
await browser.close()
logger.info(
f"Successfully retrieved HTML for {url}. Content length: {len(html_content)}"
)
except PlaywrightTimeoutError:
logger.error(f"Playwright timed out while trying to load {url}")
raise HTTPException(
status_code=408,
detail=f"Timeout: The page at {url} took too long to load.",
)
except Exception as e:
logger.error(
f"An unexpected error occurred with Playwright for {url}: {e}",
exc_info=True,
)
raise HTTPException(
status_code=500,
detail=f"An internal error occurred while fetching the page content: {e}",
)
else:
logger.info(f"Using provided html_content for {url}.")
# At this point, html_content is populated either from the payload or Playwright
try:
# Call the analyze method on the shared analyzer instance
extracted_data = analyzer.analyze(
html_content=html_content,
url=url,
use_llm=payload.use_llm,
lang=lang_list,
)
# Log extracted data in development mode
# Check environment setting (adjust 'development' if your setting is different)
if settings.service.environment == "development":
try:
log_dir = Path(settings.service.log_dir)
log_dir.mkdir(parents=True, exist_ok=True)
sanitized_domain = urlparse(url).netloc.replace(".", "_")
# Save as .json, .html
filename_json = f"{sanitized_domain}.json"
filename_html = f"{sanitized_domain}.html"
file_path_json = log_dir / filename_json
file_path_html = log_dir / filename_html
# Serialize the Pydantic model to a JSON string
file_path_json.write_text(
json.dumps(extracted_data, indent=2), encoding="utf-8"
)
file_path_html.write_text(html_content)
# Update success message
logger.success(
f"Extracted data successfully saved to: {file_path_json}"
)
except Exception as e:
logger.warning(f"Could not save extracted_data to file. Reason: {e}")
# End save data
# Return the successful response
return ExtractionResponse(data=extracted_data, message="Extraction successful")
except Exception as e:
logger.error(
f"API: An unexpected error occurred during analysis for {url}: {e}",
exc_info=True,
)
raise HTTPException(
status_code=500, detail="An internal error occurred during analysis."
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/api/v1/schemas.py 0000664 0000000 0000000 00000001262 15107027250 0025465 0 ustar 00root root 0000000 0000000 # app/api/v1/schemas.py
from pydantic import BaseModel, HttpUrl, Field
from typing import Dict, Any, Optional
class ExtractionRequest(BaseModel):
"""The request body for the /extract endpoint."""
url: HttpUrl
html_content: Optional[str] = None
file: Optional[str] = None
lang: Optional[str] = None
use_llm: bool = Field(
default=False,
description="Enable (true) or disable (false) the LLM parser for specifications. Default: false.",
)
class ExtractionResponse(BaseModel):
"""
The response body for the /extract endpoint.
This can be expanded with more specific fields later.
"""
data: Dict[str, Any]
message: str
GetArgusFlow-argus-c6afc1d/services/extractor/app/api/v1/security.py 0000664 0000000 0000000 00000001155 15107027250 0025712 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/api/v1/security.py
from fastapi import Security, HTTPException, status
from fastapi.security.api_key import APIKeyHeader
from app.config import settings
api_key_header = APIKeyHeader(name="x-api-key", auto_error=False)
async def get_api_key(api_key: str = Security(api_key_header)):
"""
Checks if the provided API key in the 'x-api-key' header is valid.
"""
if api_key == settings.auth.api_key:
return api_key
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or missing API Key",
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/config.py 0000664 0000000 0000000 00000005255 15107027250 0024216 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/config.py
import yaml
import re
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import BaseModel, Field, field_validator
from pathlib import Path
from typing import List, Any, Tuple, Dict
# Define the project's root directory
BASE_DIR = Path(__file__).resolve().parent.parent
class PatternDetail(BaseModel):
description: str
examples: list[str]
pattern: str
@field_validator("pattern")
def valid_regex(cls, v):
try:
re.compile(v)
except re.error as e:
raise ValueError(f"'{v}' is not a valid regular expression: {e}")
return v
def yaml_config_settings_source() -> Dict[str, Any]:
"""
A custom settings source that loads variables from a YAML file.
It must accept 'settings_cls' as an argument to be compatible.
"""
yaml_config_path = BASE_DIR / "config/config.yml" # This path is correct
try:
with open(yaml_config_path, "r") as f:
return yaml.safe_load(f)
except FileNotFoundError:
print(
f"WARNING: YAML config file not found at {yaml_config_path}, using defaults and .env only."
)
return {}
# Pydantic models for nested structures
class ServiceSettings(BaseModel):
llm_parser_url: str
host: str
port: int
log_level: str
log_dir: str
environment: str
class HtmlPreprocessingSettings(BaseModel):
noise_selectors: List[str]
class ModelsSettings(BaseModel):
nlp: str
class LanguageSettings(BaseModel):
default: str
class AuthSettings(BaseModel):
api_key: str
# The main Settings class
class Settings(BaseSettings):
"""The main settings model using a custom YAML source."""
service: ServiceSettings
auth: AuthSettings
html_preprocessing: HtmlPreprocessingSettings
models: ModelsSettings
field_aliases: Dict[str, List[str]] = Field(default_factory=dict)
language: LanguageSettings
default_spec_category: str
model_config = SettingsConfigDict(
env_file=BASE_DIR / ".env",
env_file_encoding="utf-8",
env_nested_delimiter="__",
extra="ignore",
)
@classmethod
def settings_customise_sources(
cls,
settings_cls,
init_settings,
env_settings,
dotenv_settings,
file_secret_settings,
) -> Tuple[Any, ...]:
# Define the priority of settings sources
return (
env_settings,
dotenv_settings,
yaml_config_settings_source,
init_settings,
file_secret_settings,
)
# Create a single, global 'settings' instance for the application to use.
settings = Settings()
GetArgusFlow-argus-c6afc1d/services/extractor/app/core/ 0000775 0000000 0000000 00000000000 15107027250 0023320 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/core/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0025417 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/core/analyzer.py 0000664 0000000 0000000 00000030402 15107027250 0025516 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/core/analyzer.py
from typing import Dict, Any, List, Optional
from loguru import logger
from langdetect import detect
from bs4 import BeautifulSoup
from app.core.context import shared_context
from app.core.dependency_resolver import DependencyResolver
from app.core.types import ExtractionResult, FieldExtractionStatus
from app.core.models import get_product_data_model, _BaseProductData
from app.utils.html_processor import clean_html_for_extraction
from app.utils.shared_resources import get_resources
from app.core.module_loader import discover_and_load_modules
from app.config import settings
class ProductPageAnalyzer:
"""
Orchestrates the full product page analysis by dynamically discovering modules,
resolving their dependencies, and executing them in the correct order.
"""
def __init__(self):
# 1. Discover all available modules (both Free and Pro) at startup.
# This only happens once for efficiency.
self.free_modules = discover_and_load_modules("app.modules")
self.pro_modules = self._discover_pro_modules()
# Determine the status of the service at startup
self.is_pro_activated = bool(self.pro_modules)
# Combine all modules to build the complete data model
all_discovered_modules = self.free_modules | self.pro_modules
# 2. Now create the data model based on the discovered modules
self.ProductDataModel = get_product_data_model(all_discovered_modules)
# 3. The rest of the __init__ remains the same
self.alias_to_field_map = self._build_alias_map()
logger.info(
f"Analyzer initialized with {len(self.free_modules)} free modules discovered."
)
if self.is_pro_activated:
logger.info("Argus Pro package detected. Service is running in PRO mode.")
else:
logger.info("Running in FREE mode. Pro features are not available.")
def _discover_pro_modules(self) -> Dict[str, Any]:
"""Attempts to dynamically discover and load the Pro modules."""
try:
# Important: the path must match the structure in your argus-pro repo.
return discover_and_load_modules("pro.extractor.modules")
except ImportError:
# This is normal behavior if the Pro package is not installed.
return {}
def _build_alias_map(self) -> Dict[str, str]:
"""Builds the alias map from the settings for data enrichment."""
return {
alias.lower(): field
for field, aliases in settings.field_aliases.items()
for alias in aliases
}
def analyze(
self,
html_content: str,
url: str,
use_llm: bool,
lang: Optional[List[str]], # This remains the List from the API
) -> Dict[str, Any]:
"""
Performs the full analysis for a given user tier.
"""
tier = "pro" if self.is_pro_activated else "free"
logger.info(f"Analyzer: Starting analysis for URL: {url} (Tier: {tier})")
# STEP 1: Assemble the active modules based on the tier.
active_modules = self.free_modules.copy()
if tier == "pro" and self.pro_modules:
logger.info("Pro tier activated, merging Pro modules.")
active_modules.update(self.pro_modules)
# STEP 2: Use the DependencyResolver to determine the execution order.
try:
resolver = DependencyResolver(active_modules)
execution_order = resolver.sort()
logger.info(f"Module execution order resolved: {execution_order}")
except ValueError as e:
logger.error(
f"Module dependency error: {e}. Aborting analysis.", exc_info=True
)
raise # Stop the execution and propagate the error
# Initialize the data objects for this run
product_data = self.ProductDataModel()
# Initialize context, which now handles the new lang logic
initial_context = self._create_initial_context(html_content, url, use_llm, lang)
shared_context.initialize(initial_context)
# STEP 3: Execute the sorted modules.
self._run_modules(product_data, execution_order, active_modules)
# STEP 4: Enrich the data (optional, after the main analysis).
self._enrich_from_specifications(product_data)
# STEP 5: Choose the best results from the scoreboard.
self._resolve_best_results(product_data)
logger.success("Analyzer: Full analysis completed.")
# Get the final results dictionary
final_results = product_data.get_final_results()
# Create a new dictionary sorted alphabetically by key
return dict(sorted(final_results.items()))
def _create_initial_context(
self,
html_content: str,
url: str,
use_llm: bool,
lang: Optional[List[str]], # Lang from API
) -> Dict[str, Any]:
"""Creates the initial context for an analysis run."""
# PHASE 1: HTML Preprocessing
logger.info("Analyzer: PHASE 1: Starting HTML Preprocessing.")
preprocessed_soup = clean_html_for_extraction(html_content)
# This will be the final list of languages passed to the pattern_manager.
# It starts as the list from the API (which could be None).
final_requested_langs = lang
if final_requested_langs is None:
# No lang was provided by the user. Use langdetect as a fallback.
try:
detected_lang = detect(preprocessed_soup.get_text())
# Put the single detected lang into a list
final_requested_langs = [detected_lang]
logger.debug(
f"langdetect fallback: Detected language '{detected_lang}'"
)
except Exception:
# langdetect failed. 'final_requested_langs' remains None.
logger.debug("langdetect fallback: Failed to detect language.")
pass
return {
"raw_soup": BeautifulSoup(html_content, "lxml"),
"preprocessed_soup": preprocessed_soup,
"current_url": url,
"resources": get_resources(),
"requested_langs": final_requested_langs,
"use_llm": use_llm,
"processed_elements": set(),
}
def _run_modules(
self,
product_data: _BaseProductData,
execution_order: List[str],
active_modules: Dict[str, Any],
):
"""Executes all active modules in the correct sorted order."""
logger.info(f"Analyzer: Running modules in sorted order: {execution_order}")
for module_name in execution_order:
module = active_modules.get(module_name)
if not module:
logger.warning(
f"Module '{module_name}' was in execution order but could not be found."
)
continue
try:
# Call the 'extract' function of the module
extracted_data, selector, status, score = module.extract()
if extracted_data is None:
continue
# Step 1: ALWAYS put the module's primary result on the context
# This allows 'price' to depend on 'json_ld', etc.
shared_context.update(module_name, extracted_data)
# Step 2: Check if this module *is* a field on the data model
# (e.g., 'price', 'title', 'json_ld', 'open_graph')
if hasattr(product_data, module_name):
# If yes, add its main result to the scoreboard for that field
field_name = module_name
result = ExtractionResult(
value=extracted_data,
source=selector,
score=score,
status=status,
)
product_data.add_result(field_name, result)
# Step 3: If the data is a dict, ALSO loop through it
# to populate sub-fields (like 'title' from 'open_graph')
if isinstance(extracted_data, dict):
for field, value in extracted_data.items():
# Check if this sub-key is a field AND it's not the module's own name
# (We've already handled the 'open_graph' field itself in Step 2)
if (
field != module_name
and value is not None
and hasattr(product_data, field)
):
# This adds 'title' and 'image' from the 'open_graph' module
result = ExtractionResult(
value=value,
source=f"{module_name} ({selector})",
score=score,
status=status,
)
product_data.add_result(field, result)
except Exception as e:
logger.error(
f"Error during execution of module '{module_name}': {e}",
exc_info=True,
)
def _enrich_from_specifications(self, product_data: _BaseProductData):
"""
Scans the found specifications and adds results to the scoreboard
for fields that have not yet been found with high confidence.
"""
logger.info("Enrichment: Checking specifications for extra data...")
# Get the specifications from the snapshot of this run
specs_results = product_data.all_results.get("specifications")
if not specs_results:
logger.debug("Enrichment: No specifications found to enrich from.")
return
# We take the best specification extraction as the source
best_specs_result = max(specs_results, key=lambda r: r.score)
specifications = best_specs_result.value
if not specifications:
return
for spec_category in specifications:
details = spec_category.get("details")
if isinstance(details, dict):
for key, value in details.items():
key_lower = key.lower().strip()
# Check if this specification key is a known alias
target_field = self.alias_to_field_map.get(key_lower)
if target_field:
# We have a match! E.g., "Merk" -> "brand"
logger.debug(
f"Enrichment: Found potential '{target_field}' field from spec key '{key}' with value '{value}'."
)
# Add this result to the scoreboard
result = ExtractionResult(
value=str(value).strip(),
source=f"specifications (key: {key})",
# Give a solid, but not the highest, score
score=150,
status=FieldExtractionStatus.FOUND_IN_SPECS,
)
product_data.add_result(target_field, result)
def _resolve_best_results(self, product_data: _BaseProductData):
"""
Iterates over the scoreboard ('all_results') and selects the entry
with the highest score for each field to set the final value.
"""
logger.info("Resolving best results from scoreboard...")
for field_name, results in product_data.all_results.items():
if results:
# Find the result with the highest score
best_result = max(results, key=lambda r: r.score)
# Set the final value on the model
if hasattr(product_data, field_name):
setattr(product_data, field_name, best_result.value)
# Also update the status and selector for the winner
product_data.field_status[field_name] = best_result.status
product_data.selectors_used[field_name] = best_result.source
logger.debug(
f"Resolved field '{field_name}': chose value from '{best_result.source}' with score {best_result.score}."
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/core/context.py 0000664 0000000 0000000 00000002703 15107027250 0025360 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/core/context.py
from contextvars import ContextVar
from typing import Dict, Any
class SharedContextManager:
"""
A wrapper class that provides an intuitive and safe interface
for managing the request-scoped context (ContextVar).
"""
def __init__(self):
# The ContextVar is now a 'private' implementation detail.
self._context_var: ContextVar[Dict[str, Any]] = ContextVar(
"shared_context", default={}
)
def initialize(self, initial_data: Dict[str, Any]) -> None:
"""Sets the initial context for a new request."""
self._context_var.set(initial_data)
def get(self, key: str, default: Any = None) -> Any:
"""Retrieves one specific value from the context."""
return self._context_var.get().get(key, default)
def get_all(self) -> Dict[str, Any]:
"""Retrieves the complete context dictionary."""
return self._context_var.get()
def update(self, key: str, value: Any) -> None:
"""
The method you suggested: update the context safely.
This performs the 'get -> modify -> set' cycle internally.
"""
current_context = self._context_var.get().copy() # Work with a copy
current_context[key] = value
self._context_var.set(current_context)
# Create a single global instance of the manager.
# Other modules will import this instance.
shared_context = SharedContextManager()
GetArgusFlow-argus-c6afc1d/services/extractor/app/core/dependency_resolver.py 0000664 0000000 0000000 00000005455 15107027250 0027742 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/core/dependency_resolver.py
from collections import deque
from typing import Dict, Any, List
from loguru import logger
class DependencyResolver:
def __init__(self, modules: Dict[str, Any]):
"""
Initializes the resolver.
Args:
modules: A dictionary with the active modules for the current run.
"""
self.modules = modules
self.module_names = set(modules.keys())
# The 'raw' graph containing all dependencies, including invalid ones.
self.raw_dependencies = {
name: getattr(mod, "REQUIRES", []) for name, mod in modules.items()
}
def sort(self) -> List[str]:
"""
Performs a topological sort and returns the execution order.
Ignores dependencies that are not in the current set of active modules.
"""
# We build a 'clean' graph with only the dependencies that actually exist
# in the set of active modules for this tier.
dependencies = {}
for name, deps in self.raw_dependencies.items():
# Keep only the dependencies that are actually in self.module_names
valid_deps = [dep for dep in deps if dep in self.module_names]
if len(valid_deps) != len(deps):
invalid_deps = set(deps) - set(valid_deps)
logger.trace(
f"Module '{name}' has unmet dependencies which will be ignored: {invalid_deps}"
)
dependencies[name] = valid_deps
# Step 1: Calculate the "in-degree" for each module.
in_degree = {name: 0 for name in self.module_names}
graph = {name: [] for name in self.module_names}
for name, deps in dependencies.items():
for dep in deps:
# Because we filtered, 'dep' will always exist in self.module_names
graph[dep].append(name)
in_degree[name] += 1
# Step 2: Start the queue with all modules that have an in-degree of 0.
queue = deque([name for name in self.module_names if in_degree[name] == 0])
sorted_order = []
# Step 3: Process the queue.
while queue:
current_module = queue.popleft()
sorted_order.append(current_module)
for dependent_module in graph[current_module]:
in_degree[dependent_module] -= 1
if in_degree[dependent_module] == 0:
queue.append(dependent_module)
# Step 4: Check for circular dependencies.
if len(sorted_order) != len(self.module_names):
cycle_nodes = {name for name, degree in in_degree.items() if degree > 0}
raise ValueError(
f"Circular dependency detected! Modules in cycle: {cycle_nodes}"
)
return sorted_order
GetArgusFlow-argus-c6afc1d/services/extractor/app/core/models.py 0000664 0000000 0000000 00000004332 15107027250 0025157 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/core/models.py
from pydantic import BaseModel, Field, create_model
from typing import Dict, Any, Optional, List, Type
# Import the new building blocks from types.py
from app.core.types import FieldExtractionStatus, ExtractionResult
class _BaseProductData(BaseModel):
"""
The base class for our dynamic product data model.
It contains the core scoreboard and metadata functionality.
"""
all_results: Dict[str, List[ExtractionResult]] = Field(
default_factory=dict, exclude=True
)
field_status: Dict[str, FieldExtractionStatus] = Field(
default_factory=dict, exclude=True
)
selectors_used: Dict[str, str] = Field(default_factory=dict, exclude=True)
def add_result(self, field_name: str, result: ExtractionResult):
"""Adds a new found result to the scoreboard."""
if field_name not in self.all_results:
self.all_results[field_name] = []
self.all_results[field_name].append(result)
def get_final_results(self) -> Dict[str, Any]:
"""Returns the final model data, excluding metadata."""
# This helper function makes cleanup in the analyzer easier
return self.model_dump(
exclude={"all_results", "field_status", "selectors_used"}
)
# The Dynamic Model Factory
def get_product_data_model(
discovered_modules: Dict[str, Any],
) -> Type[_BaseProductData]:
"""
Dynamically creates the ProductData model by adding fields discovered
from the module files.
"""
dynamic_fields: Dict[str, Any] = {}
for name, module in discovered_modules.items():
if hasattr(module, "FIELD_TYPE"):
# Field definition: (Type Hint, Default Value)
# Example: title: (Optional[str], None)
dynamic_fields[name] = (getattr(module, "FIELD_TYPE"), None)
else:
# Fallback for modules without FIELD_TYPE
dynamic_fields[name] = (Optional[Any], None)
# Use Pydantic's create_model to build a new class
# that inherits from _BaseProductData and contains the dynamic fields.
ProductDataModel = create_model(
"ProductDataModel", __base__=_BaseProductData, **dynamic_fields
)
return ProductDataModel
GetArgusFlow-argus-c6afc1d/services/extractor/app/core/module_loader.py 0000664 0000000 0000000 00000003436 15107027250 0026513 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/core/module_loader.py
from typing import Dict, Any
from loguru import logger
import importlib
import pkgutil
def discover_and_load_modules(package_path: str) -> Dict[str, Any]:
"""
Scans a package path, dynamically discovers all submodules,
and loads them.
"""
loaded_modules = {}
try:
# Import the main package (e.g., app.modules.fields)
package = importlib.import_module(package_path)
except ImportError as e:
logger.error(f"Could not import main package at '{package_path}': {e}")
return {}
# Use pkgutil to find all modules in the package directory
for module_info in pkgutil.iter_modules(package.__path__, package_path + "."):
module_name = module_info.name.split(".")[-1]
# We are interested in the 'extract' submodule of each module
full_extract_path = f"{module_info.name}.extract"
try:
module = importlib.import_module(full_extract_path)
if hasattr(module, "extract") and callable(getattr(module, "extract")):
loaded_modules[module_name] = module
logger.debug(
f"Module '{module_name}' from '{package_path}' successfully loaded."
)
else:
logger.warning(
f"Module '{full_extract_path}' has no 'extract' function. Skipping."
)
except ImportError:
# This is normal if a module doesn't have an 'extract.py', so log at a lower level
logger.trace(f"No 'extract.py' found for module '{module_name}'. Skipping.")
except Exception as e:
logger.error(
f"Error loading module '{full_extract_path}': {e}", exc_info=True
)
return loaded_modules
GetArgusFlow-argus-c6afc1d/services/extractor/app/core/types.py 0000664 0000000 0000000 00000002177 15107027250 0025045 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/core/types.py
from pydantic import BaseModel
from enum import Enum
from typing import Dict, Any, List, Union
from functools import total_ordering
@total_ordering
class FieldExtractionStatus(Enum):
"""
Defines the reliability status of an extracted field.
"""
NOT_FOUND = 0
GENERIC_FALLBACK = 1
NLP_HEURISTIC = 2
MODULE_HEURISTIC = 3
MODULE_REGEX = 4
FOUND_IN_SPECS = 5
DOMAIN_MAPPING = 6
OPEN_GRAPH = 7
JSON_LD = 8
FOUND_IN_SHARED_CONTEXT = 9
def __lt__(self, other):
if not isinstance(other, FieldExtractionStatus):
return NotImplemented
return self.value < other.value
def __eq__(self, other):
if not isinstance(other, FieldExtractionStatus):
return NotImplemented
return self.value == other.value
class SpecificationCategory(BaseModel):
category: str
details: Union[Dict[str, Any], List[Any]]
class ExtractionResult(BaseModel):
"""A model to capture a single extraction result and its metadata."""
value: Any
source: str
score: int
status: FieldExtractionStatus
GetArgusFlow-argus-c6afc1d/services/extractor/app/main.py 0000664 0000000 0000000 00000003432 15107027250 0023670 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/main.py
import sys
from loguru import logger
from fastapi import FastAPI, Request, HTTPException
from contextlib import asynccontextmanager
from app.config import settings
from app.core.analyzer import ProductPageAnalyzer
from app.api.v1.endpoints import router as api_v1_router
from pathlib import Path
# Configure Loguru logger
log_level = settings.service.log_level.strip().upper()
log_dir = Path(settings.service.log_dir)
log_file = log_dir / "service.log"
logger.remove()
logger.add(sys.stderr, level=log_level)
logger.add(log_file, rotation="10 MB", level=log_level)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Code to run on startup
logger.info("Starting up service and loading analyzer...")
# Initialize the analyzer once and store it in the application state
app.state.analyzer = ProductPageAnalyzer()
logger.info("ProductPageAnalyzer loaded and stored in app.state.")
yield
# Code to run on shutdown
logger.info("Shutting down service...")
app.state.analyzer = None
app = FastAPI(
title="Argus Extractor Service",
description="A service to extract structured data from product pages.",
version="2.0.0",
lifespan=lifespan,
)
@app.get("/health", tags=["Monitoring"])
async def health_check(request: Request):
# Check if the analyzer is in the application state
analyzer_loaded = getattr(request.app.state, "analyzer", None) is not None
if not analyzer_loaded:
raise HTTPException(
status_code=503, detail="Service is unhealthy: Analyzer not loaded."
)
return {"status": "ok", "analyzer_loaded": True}
# Add the API routers
app.include_router(api_v1_router, prefix="/api/v1", tags=["Extraction"])
logger.info("Application setup complete. Awaiting requests...")
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/ 0000775 0000000 0000000 00000000000 15107027250 0024040 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0026137 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/ 0000775 0000000 0000000 00000000000 15107027250 0026512 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0030611 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/extract.py 0000664 0000000 0000000 00000006342 15107027250 0030543 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/availability/extract.py
from typing import Optional, Tuple, Any
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from .parsers.json_ld_parser import parse_json_ld
from .parsers.open_graph_parser import parse_open_graph
from .parsers.schema_parser import parse_schema
from .parsers.text_parser import parse_textual_indicators
from .parsers.meta_parser import parse_meta_tags
from .parsers.title_parser import parse_title
FIELD_TYPE = Optional[str]
REQUIRES = ["json_ld", "open_graph"]
def extract() -> Tuple[Any, str, FieldExtractionStatus, int]:
"""
The orchestrator function that extracts the availability status by calling a series of parsers.
"""
soup_to_use = shared_context.get("raw_soup")
logger.info(
"Availability Extractor (Main): Starting extraction of availability status."
)
if not soup_to_use:
logger.warning(
"Availability Extractor: No valid BeautifulSoup objects to work with."
)
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
# Priority 1: JSON-LD (highest reliability structured data)
availability, selector, status, score = parse_json_ld()
if availability:
logger.info(
f"Availability Extractor: Status successfully extracted with JSON-LD Parser. Score: {score}"
)
return availability, selector, status, score
# Priority 2: Open Graph (high reliability structured data)
availability, selector, status, score = parse_open_graph()
if availability:
logger.info(
f"Availability Extractor: Status successfully extracted with Open Graph Parser. Score: {score}"
)
return availability, selector, status, score
# Priority 3: Schema.org microdata (HTML fallback)
availability, selector, status, score = parse_schema(soup_to_use)
if availability:
logger.info(
f"Availability Extractor: Status successfully extracted with Schema Parser. Score: {score}"
)
return availability, selector, status, score
# Priority 4: Textual indicators in relevant DOM elements
availability, selector, status, score = parse_textual_indicators(soup_to_use)
if availability:
logger.info(
f"Availability Extractor: Status successfully extracted with Text Parser. Score: {score}"
)
return availability, selector, status, score
# Priority 5: General meta-tags
availability, selector, status, score = parse_meta_tags(soup_to_use)
if availability:
logger.info(
f"Availability Extractor: Status successfully extracted with Meta Parser. Score: {score}"
)
return availability, selector, status, score
# Priority 6: Page title (lowest reliability)
availability, selector, status, score = parse_title(soup_to_use)
if availability:
logger.info(
f"Availability Extractor: Status successfully extracted with Title Parser. Score: {score}"
)
return availability, selector, status, score
logger.info("Availability Extractor: No clear availability status found.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/parsers/ 0000775 0000000 0000000 00000000000 15107027250 0030171 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/parsers/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0032270 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/parsers/json_ld_parser.py 0000664 0000000 0000000 00000004106 15107027250 0033550 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/availability/parsers/json_ld_parser.py
from typing import Optional, Tuple
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.availability.utils import find_availability_status
def parse_json_ld() -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts availability from the 'json_ld' data in shared_context.
"""
logger.debug("JSON-LD Parser: Searching for availability in JSON-LD data.")
# Get the raw list of JSON-LD nodes
json_ld_data = shared_context.get("json_ld")
if not isinstance(json_ld_data, list):
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
for node in json_ld_data:
if not isinstance(node, dict):
continue
# Look for availability in an 'offers' node (dict or list)
offers = node.get("offers")
offer_list = []
if isinstance(offers, dict):
offer_list = [offers]
elif isinstance(offers, list):
offer_list = offers
for offer in offer_list:
if not isinstance(offer, dict):
continue
availability_raw = offer.get("availability")
if isinstance(availability_raw, str):
# The value is often a URL like 'http://schema.org/InStock'
# We just want the last part.
status_text = availability_raw.split("/")[-1]
status = find_availability_status(status_text)
if status:
logger.debug(
f"JSON-LD Parser: Found status '{status}' via 'offers.availability'."
)
return (
status,
"json_ld.offers.availability",
FieldExtractionStatus.JSON_LD,
200,
)
logger.debug("JSON-LD Parser: No availability found in JSON-LD data.")
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/parsers/meta_parser.py 0000664 0000000 0000000 00000002456 15107027250 0033054 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/availability/parsers/meta_parser.py
import re
from typing import Optional, Tuple
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.availability.utils import find_availability_status
def parse_meta_tags(
soup: BeautifulSoup,
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the availability status from general meta-tags.
"""
logger.debug("Meta Parser: Searching in general meta-tags.")
meta_name_avail = soup.find(
"meta",
attrs={
"name": re.compile(
r"availability|product:availability|item:availability", re.IGNORECASE
)
},
)
if meta_name_avail and meta_name_avail.get("content"):
content = meta_name_avail["content"].lower()
status = find_availability_status(content)
if status:
logger.debug(
f"Meta Parser: Found status '{status}' via meta-tag: {content}"
)
return (
status,
f'meta[name="{meta_name_avail["name"]}"]',
FieldExtractionStatus.MODULE_HEURISTIC,
80,
)
return None, "meta_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/parsers/open_graph_parser.py 0000664 0000000 0000000 00000002530 15107027250 0034241 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/availability/parsers/open_graph_parser.py
from typing import Optional, Tuple
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.availability.utils import find_availability_status
def parse_open_graph() -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts availability from the 'open_graph' data in shared_context.
"""
logger.debug("Open Graph Parser: Searching for availability in OG data.")
og_data = shared_context.get("open_graph")
if not isinstance(og_data, dict):
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
# Standard keys are 'og:availability' or 'product:availability'
availability_raw = og_data.get("availability") or og_data.get(
"product:availability"
)
if isinstance(availability_raw, str):
status = find_availability_status(availability_raw)
if status:
logger.debug(
f"Open Graph Parser: Found status '{status}' via 'og:availability'."
)
return status, "og:availability", FieldExtractionStatus.OPEN_GRAPH, 190
logger.debug("Open Graph Parser: No availability found in OG data.")
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/parsers/schema_parser.py 0000664 0000000 0000000 00000004052 15107027250 0033360 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/availability/parsers/schema_parser.py
import re
from typing import Optional, Tuple
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.availability.utils import find_availability_status
def parse_schema(
soup: BeautifulSoup,
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the availability status from Schema.org microdata.
"""
logger.debug("Schema Parser: Searching for itemprop='availability'.")
# Find the link tag with itemprop="availability"
avail_link_tag = soup.find("link", itemprop="availability")
if avail_link_tag and avail_link_tag.get("href"):
href = avail_link_tag["href"].lower()
if "instock" in href:
return (
"In Stock",
'link[itemprop="availability"]',
FieldExtractionStatus.MODULE_HEURISTIC,
100,
)
if "outofstock" in href:
return (
"Out of Stock",
'link[itemprop="availability"]',
FieldExtractionStatus.MODULE_HEURISTIC,
100,
)
if "preorder" in href:
return (
"Pre-order",
'link[itemprop="availability"]',
FieldExtractionStatus.MODULE_HEURISTIC,
100,
)
# Find other tags (span, div) with itemprop="availability"
avail_tag = soup.find(re.compile(r"span|div|p"), itemprop="availability")
if avail_tag:
text = avail_tag.get_text(strip=True)
content = avail_tag.get("content", "")
combined_text = f"{text} {content}".strip()
status = find_availability_status(combined_text)
if status:
return (
status,
f'{avail_tag.name}[itemprop="availability"]',
FieldExtractionStatus.MODULE_HEURISTIC,
98,
)
return None, "schema_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/parsers/text_parser.py 0000664 0000000 0000000 00000003417 15107027250 0033110 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/availability/parsers/text_parser.py
import re
from typing import Optional, Tuple
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.availability.utils import find_availability_status
from app.utils.pattern_manager import pattern_manager
def parse_textual_indicators(
soup: BeautifulSoup,
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the availability status based on text in relevant elements.
"""
logger.debug("Text Parser: Searching for textual indicators in relevant sections.")
check_tags = soup.find_all(
re.compile(r"div|span|p|button|a"),
# Use the dynamic regex from PatternManager instead of a hardcoded one
class_=pattern_manager.get_compiled_regex("availability_class_regex"),
limit=10,
)
for tag in check_tags:
text = (
tag.get_text(strip=True).lower()
or tag.get("title", "").lower()
or tag.get("value", "").lower()
)
if not text:
continue
status = find_availability_status(text)
if status:
selector_detail = f"{tag.name}"
if tag.get("class"):
selector_detail += f".{'.'.join(tag['class'])}"
elif tag.get("id"):
selector_detail += f"#{tag['id']}"
logger.debug(
f"Text Parser: Found status '{status}' via textual indicator: {text[:50]}"
)
return (
status,
f"{selector_detail} (textual)",
FieldExtractionStatus.MODULE_HEURISTIC,
90,
)
return None, "text_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/parsers/title_parser.py 0000664 0000000 0000000 00000001712 15107027250 0033241 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/availability/parsers/title_parser.py
from typing import Optional, Tuple
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.availability.utils import find_availability_status
def parse_title(
soup: BeautifulSoup,
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the availability status from the page title as a last resort.
"""
logger.debug("Title Parser: Checking the title as a last resort.")
title_tag = soup.find("title")
if title_tag:
title_text = title_tag.get_text(strip=True).lower()
status = find_availability_status(title_text)
if status:
logger.debug(f"Title Parser: Found status '{status}' in the title.")
return status, "title", FieldExtractionStatus.MODULE_HEURISTIC, 50
return None, "title_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/availability/utils.py 0000664 0000000 0000000 00000003343 15107027250 0030227 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/availability/utils.py
from typing import Dict, List, Optional
from loguru import logger
from app.utils.pattern_manager import pattern_manager
def get_status_map() -> Dict[str, List[str]]:
"""Builds the status map dynamically from the PatternManager."""
status_map = {
"In Stock": pattern_manager.get_keyword_list("availability_in_stock"),
"Out of Stock": pattern_manager.get_keyword_list("availability_out_of_stock"),
"Pre-order": pattern_manager.get_keyword_list("availability_pre_order"),
}
# Add common schema.org terms (and their lowercase versions)
# in case they are not in the patterns.yml. This fixes the
# bug where 'InStock' from JSON-LD was not being found.
status_map["In Stock"].extend(["InStock", "instock", "in stock"])
status_map["Out of Stock"].extend(["OutOfStock", "outofstock", "out of stock"])
status_map["Pre-order"].extend(["PreOrder", "preorder", "pre order"])
return status_map
def find_availability_status(text: str) -> Optional[str]:
"""
Searches for a valid availability status in the given text.
Returns the status-key ('In Stock', 'Out of Stock', 'Pre-order') or None.
"""
cleaned_text = text.lower()
# Get the dynamic map for this request's language(s)
STATUS_MAP = get_status_map()
for status_key in ["Out of Stock", "Pre-order", "In Stock"]:
if any(k.lower() in cleaned_text for k in STATUS_MAP[status_key]):
if len(cleaned_text.split()) > 10:
logger.debug(
f"Availability validation: Text '{cleaned_text[:50]}...' is too long. Skipping."
)
continue
return status_key
return None
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/ 0000775 0000000 0000000 00000000000 15107027250 0025126 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0027225 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/extract.py 0000664 0000000 0000000 00000005205 15107027250 0027154 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/brand/extract.py
from typing import Optional, Tuple, Any
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from .parsers.json_ld_parser import parse_json_ld
from .parsers.open_graph_parser import parse_open_graph
from .parsers.meta_parser import parse_meta_tags
from .parsers.nlp_parser import parse_with_nlp
from .parsers.dom_parser import parse_with_dom_heuristics
from .parsers.title_parser import parse_from_title
from .parsers.general_fallback_parser import parse_general_fallback
FIELD_TYPE = Optional[str]
REQUIRES = ["json_ld", "open_graph"]
def extract() -> Tuple[Any, str, FieldExtractionStatus, int]:
"""
The orchestrator function that extracts the brand name by calling a series of parsers.
"""
soup_to_use = shared_context.get("raw_soup")
nlp_model = shared_context.get("resources", {}).get("nlp_model")
logger.info("Brand Extractor (Main): Starting brand name extraction.")
if not soup_to_use:
logger.warning("Brand Extractor: No valid BeautifulSoup objects to work with.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
# Priority 1: JSON-LD (highest reliability)
brand, selector, status, score = parse_json_ld(soup_to_use, nlp_model)
if brand:
return brand, selector, status, score
# Priority 2: Open Graph (high reliability)
brand, selector, status, score = parse_open_graph(soup_to_use, nlp_model)
if brand:
return brand, selector, status, score
# Priority 3: Meta tags and structured microdata (from HTML)
brand, selector, status, score = parse_meta_tags(soup_to_use, nlp_model)
if brand:
return brand, selector, status, score
# Priority 4: DOM-based heuristics
brand, selector, status, score = parse_with_dom_heuristics(soup_to_use, nlp_model)
if brand:
return brand, selector, status, score
# Priority 5: Pattern recognition in the title
brand, selector, status, score = parse_from_title(soup_to_use, nlp_model)
if brand:
return brand, selector, status, score
# Priority 6: NLP-based extraction (run after DOM/Title heuristics)
brand, selector, status, score = parse_with_nlp(soup_to_use, nlp_model)
if brand:
return brand, selector, status, score
# Priority 7: General fallback search
brand, selector, status, score = parse_general_fallback(soup_to_use, nlp_model)
if brand:
return brand, selector, status, score
logger.info("Brand Extractor: No brand name found via any method.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/parsers/ 0000775 0000000 0000000 00000000000 15107027250 0026605 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/parsers/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0030704 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/parsers/dom_parser.py 0000664 0000000 0000000 00000006741 15107027250 0031322 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/brand/parsers/dom_parser.py
import re
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.brand.utils import is_plausible_brand, check_brand_in_main_title
from app.utils.pattern_manager import pattern_manager
def parse_with_dom_heuristics(
soup: BeautifulSoup, nlp_model: Any
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the brand name using DOM-based heuristics (classes, IDs, labels).
"""
logger.debug("DOM Parser: Starting search based on DOM heuristics.")
# Proximity search based on keywords (Brand:, Manufacturer:, etc.)
# Get the regex from the PatternManager
brand_label_keywords_regex = pattern_manager.get_compiled_regex("brand_label_regex")
for label_tag in soup.find_all(
lambda tag: tag.name in ["span", "div", "p", "dt", "th"]
and brand_label_keywords_regex.search(tag.get_text(strip=True))
):
next_sibling = label_tag.find_next_sibling()
if next_sibling:
candidate_brand = next_sibling.get_text(strip=True)
if is_plausible_brand(
candidate_brand, nlp_model
) and check_brand_in_main_title(candidate_brand, soup):
logger.debug(
f"DOM Parser: Found via sibling of label '{label_tag.get_text(strip=True)}': {candidate_brand}"
)
return (
candidate_brand,
f"{label_tag.name} + next sibling",
FieldExtractionStatus.MODULE_HEURISTIC,
120,
)
child_elements = label_tag.find_all(re.compile(r"span|div|a|strong|b"), limit=2)
for child_elem in child_elements:
candidate_brand = child_elem.get_text(strip=True)
if is_plausible_brand(
candidate_brand, nlp_model
) and check_brand_in_main_title(candidate_brand, soup):
logger.debug(
f"DOM Parser: Found via child of label '{label_tag.get_text(strip=True)}': {candidate_brand}"
)
return (
candidate_brand,
f"{label_tag.name} > child",
FieldExtractionStatus.MODULE_HEURISTIC,
115,
)
# Direct search based on classes and IDs
direct_brand_elements = soup.find_all(
re.compile(r"a|span|div|p|strong|h[1-6]"),
# Get the class regex from the PatternManager
class_=pattern_manager.get_compiled_regex("brand_class_regex"),
)
# Get the generic text regex from the PatternManager
generic_text_regex = pattern_manager.get_compiled_regex("generic_brand_text_regex")
for elem in direct_brand_elements:
text_content = elem.get_text(strip=True)
# Skip generic links (e.g., "All brands")
if generic_text_regex.search(text_content):
continue
if is_plausible_brand(text_content, nlp_model) and check_brand_in_main_title(
text_content, soup
):
logger.debug(f"DOM Parser: Found via direct class/id: {text_content}")
return (
text_content,
f"{elem.name}.{'.'.join(elem.get('class', []))}",
FieldExtractionStatus.MODULE_HEURISTIC,
125,
)
return None, "dom_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/parsers/general_fallback_parser.py 0000664 0000000 0000000 00000004232 15107027250 0033770 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/brand/parsers/general_fallback_parser.py
import re
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.brand.utils import is_plausible_brand, check_brand_in_main_title
from app.utils.pattern_manager import pattern_manager
def parse_general_fallback(
soup: BeautifulSoup, nlp_model: Any
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
The last resort parser that searches in general elements.
"""
logger.debug("General Fallback Parser: Last resort search.")
brand_elements_fallback = soup.find_all(
re.compile(r"a|span|div|p"),
# Use the consistent name 'brand_class_regex'
class_=pattern_manager.get_compiled_regex("brand_class_regex"),
limit=5,
)
# Get the generic text regex from the PatternManager
generic_text_regex = pattern_manager.get_compiled_regex("generic_brand_text_regex")
for elem in brand_elements_fallback:
brand_name_candidate = elem.get_text(strip=True)
# Skip generic elements
if generic_text_regex.search(brand_name_candidate) or re.search(
r"filter|list|nav|menu|category",
" ".join(elem.get("class", [])).lower() + " " + elem.get("id", "").lower(),
re.IGNORECASE,
):
continue
if is_plausible_brand(
brand_name_candidate, nlp_model
) and check_brand_in_main_title(brand_name_candidate, soup):
logger.debug(
f"General Fallback Parser: Found via class/id (fallback) validated by title: {brand_name_candidate}"
)
selector = f"{elem.name}"
if elem.get("class"):
selector += f".{'.'.join(elem['class'])}"
elif elem.get("id"):
selector += f"#{elem['id']}"
return (
brand_name_candidate,
selector + " (fallback)",
FieldExtractionStatus.MODULE_HEURISTIC,
70,
)
return None, "general_fallback_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/parsers/json_ld_parser.py 0000664 0000000 0000000 00000004150 15107027250 0032163 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/brand/parsers/json_ld_parser.py
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.brand.utils import is_plausible_brand, check_brand_in_main_title
def parse_json_ld(
soup: BeautifulSoup, nlp_model: Any
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts brand from the 'json_ld' data in shared_context.
"""
logger.debug("JSON-LD Parser: Searching for brand in JSON-LD data.")
json_ld_data = shared_context.get("json_ld")
if not isinstance(json_ld_data, list):
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
for node in json_ld_data:
if not isinstance(node, dict):
continue
# Check for brand info, which can be a string or an object
for key in ["brand", "manufacturer"]:
brand_node = node.get(key)
brand_name_candidate = None
if isinstance(brand_node, str):
brand_name_candidate = brand_node
elif isinstance(brand_node, dict):
brand_name_candidate = brand_node.get("name")
if brand_name_candidate:
brand_name_candidate = brand_name_candidate.strip()
# Validate it for plausibility and check if it's in the title
if is_plausible_brand(
brand_name_candidate, nlp_model
) and check_brand_in_main_title(brand_name_candidate, soup):
logger.debug(
f"JSON-LD Parser: Found validated brand '{brand_name_candidate}' via 'json_ld.{key}'."
)
return (
brand_name_candidate,
f"json_ld.{key}",
FieldExtractionStatus.JSON_LD,
250,
)
logger.debug("JSON-LD Parser: No brand found in JSON-LD data.")
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/parsers/meta_parser.py 0000664 0000000 0000000 00000006434 15107027250 0031470 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/brand/parsers/meta_parser.py
import re
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.brand.utils import is_plausible_brand, check_brand_in_main_title
def parse_meta_tags(
soup: BeautifulSoup, nlp_model: Any
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the brand name from tags and itemprop attributes.
(Open Graph logic has been moved to open_graph_parser.py)
"""
logger.debug("Meta Parser: Searching in tags and itemprop attributes.")
# Priority 1: Amazon-specific 'byline' link
byline_link = soup.select_one("#bylineInfo, #brand")
if byline_link:
brand_name_candidate = byline_link.get_text(strip=True)
# Sometimes it says "Bezoek de XXXXX Store" (Visit the XXXXX Store)
match_store = re.search(
r"Bezoek de\s*(.*?)\s*Store", brand_name_candidate, re.IGNORECASE
)
if match_store:
brand_name_candidate = match_store.group(1).strip()
if is_plausible_brand(
brand_name_candidate, nlp_model
) and check_brand_in_main_title(brand_name_candidate, soup):
logger.debug(
f"Meta Parser: Found via byline link ({byline_link.get('id') or byline_link.name}): {brand_name_candidate}"
)
return (
brand_name_candidate,
f"#{byline_link.get('id') or byline_link.name}",
FieldExtractionStatus.MODULE_HEURISTIC,
200,
)
# Priority 2: General itemprop-meta tag
brand_itemprop_meta = soup.find("meta", itemprop="brand")
if brand_itemprop_meta and brand_itemprop_meta.get("content"):
brand_name_candidate = brand_itemprop_meta["content"].strip()
if is_plausible_brand(
brand_name_candidate, nlp_model
) and check_brand_in_main_title(brand_name_candidate, soup):
logger.debug(
f"Meta Parser: Found via meta[itemprop='brand']: {brand_name_candidate}"
)
return (
brand_name_candidate,
'meta[itemprop="brand"]',
FieldExtractionStatus.MODULE_HEURISTIC,
190,
)
# Priority 3: itemprop-tag (span, div, etc.)
brand_itemprop_tag = soup.find(
re.compile(r"span|div|p|a|strong|b"), itemprop="brand"
)
if brand_itemprop_tag and brand_itemprop_tag.get_text(strip=True):
brand_name_candidate = brand_itemprop_tag.get_text(strip=True)
if is_plausible_brand(
brand_name_candidate, nlp_model
) and check_brand_in_main_title(brand_name_candidate, soup):
logger.debug(
f"Meta Parser: Found via {brand_itemprop_tag.name}[itemprop='brand']: {brand_name_candidate}"
)
return (
brand_name_candidate,
f'{brand_itemprop_tag.name}[itemprop="brand"]',
FieldExtractionStatus.MODULE_HEURISTIC,
185,
)
# REMOVED OPEN GRAPH PARSING
# The 'og:brand' logic was here, but is now handled by open_graph_parser.py
return None, "meta_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/parsers/nlp_parser.py 0000664 0000000 0000000 00000003077 15107027250 0031333 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/brand/parsers/nlp_parser.py
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.types import FieldExtractionStatus
from app.modules.brand.utils import is_plausible_brand, get_main_title_content
def parse_with_nlp(
soup: BeautifulSoup, nlp_model: Any
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts a brand name using the spaCy NER model.
"""
if not nlp_model:
logger.debug("NLP Parser: spaCy model not available. Skipping.")
return None, "nlp_parser", FieldExtractionStatus.NOT_FOUND, 0
main_title_text, main_title_selector = get_main_title_content(soup)
if not main_title_text:
return None, "nlp_parser", FieldExtractionStatus.NOT_FOUND, 0
logger.debug("NLP Parser: Running NLP extraction with spaCy.")
doc = nlp_model(main_title_text)
for ent in doc.ents:
# We look for organizations, as brands are typically classified as such.
if ent.label_ == "ORG" and is_plausible_brand(ent.text.strip(), nlp_model):
logger.info(
f"NLP Parser: Brand found via NLP in title: '{ent.text.strip()}' (Label: {ent.label_})"
)
return (
ent.text.strip(),
f"{main_title_selector} (NLP - {ent.label_})",
FieldExtractionStatus.NLP_HEURISTIC,
75,
)
logger.debug("NLP Parser: No valid brand found with spaCy NER.")
return None, "nlp_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/parsers/open_graph_parser.py 0000664 0000000 0000000 00000003157 15107027250 0032663 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/brand/parsers/open_graph_parser.py
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.brand.utils import is_plausible_brand, check_brand_in_main_title
def parse_open_graph(
soup: BeautifulSoup, nlp_model: Any
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts brand from the 'open_graph' data in shared_context.
"""
logger.debug("Open Graph Parser: Searching for brand in OG data.")
og_data = shared_context.get("open_graph")
if not isinstance(og_data, dict):
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
# Standard keys are 'og:brand' or 'product:brand'
brand_name_candidate = og_data.get("brand") or og_data.get("product:brand")
if isinstance(brand_name_candidate, str):
brand_name_candidate = brand_name_candidate.strip()
# Validate it
if is_plausible_brand(
brand_name_candidate, nlp_model
) and check_brand_in_main_title(brand_name_candidate, soup):
logger.debug(
f"Open Graph Parser: Found validated brand '{brand_name_candidate}' via 'og:brand'."
)
return (
brand_name_candidate,
"og:brand",
FieldExtractionStatus.OPEN_GRAPH,
240,
)
logger.debug("Open Graph Parser: No brand found in OG data.")
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/parsers/title_parser.py 0000664 0000000 0000000 00000011256 15107027250 0031661 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/brand/parsers/title_parser.py
import re
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.types import FieldExtractionStatus
from app.modules.brand.utils import (
is_plausible_brand,
get_main_title_content,
find_explicit_brands,
)
def parse_from_title(
soup: BeautifulSoup, nlp_model: Any
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the brand name based on patterns in the product title.
"""
logger.debug("Title Parser: Checking title for brand patterns.")
main_title_text, selector_title = get_main_title_content(soup)
if not main_title_text:
return None, "title_parser", FieldExtractionStatus.NOT_FOUND, 0
# First, find any explicitly mentioned brands on the page to use for validation.
explicit_brands_on_page = find_explicit_brands(soup)
if not explicit_brands_on_page:
logger.debug(
"Title Parser: No explicit brands found on page for validation. Skipping Heuristic 1."
)
else:
# Heuristic 1: Check N-grams (word combinations) from the title against explicit brands.
# We iterate from the longest possible word combination (the whole title)
# down to single words, returning the first (and therefore longest)
# plausible match that is also in the explicit_brands_on_page set.
words = main_title_text.split(" ")
n_words = len(words)
for length in range(n_words, 0, -1): # From n_words down to 1
for i in range(n_words - length + 1): # Sliding window start index
j = i + length # Sliding window end index
# Construct the candidate phrase
candidate_phrase = " ".join(words[i:j])
# Basic cleaning: remove leading/trailing punctuation and whitespace
# This handles cases like " (Brand) " or "Brand,"
candidate_phrase = candidate_phrase.strip().strip(".,;:|()[]{}")
if not candidate_phrase: # Skip if stripping left an empty string
continue
candidate_lower = candidate_phrase.lower()
# VALIDATION: Check if this phrase is one of the explicit brands
if candidate_lower in explicit_brands_on_page:
# PLAUSIBILITY: Check if it's a plausible brand name
if is_plausible_brand(candidate_phrase, nlp_model):
logger.debug(
f"Title Parser: Found via title (n-gram, validated): '{candidate_phrase}'"
)
return (
candidate_phrase,
f"{selector_title} (n-gram match)",
FieldExtractionStatus.MODULE_HEURISTIC,
110,
)
else:
logger.debug(
f"Title Parser: N-gram candidate '{candidate_phrase}' matched explicit brand, but failed plausibility. Continuing search..."
)
# We continue, because a shorter sub-phrase might be plausible
# e.g., "Douwe Egberts Junk" might match but fail plausibility,
# while "Douwe Egberts" will match and pass later.
logger.debug(
"Title Parser: Heuristic 1 (n-gram match) found no validated brand."
)
# Heuristic 2: Pattern "Brand - Product"
match = re.match(r"(.+?)\s*[-–—]\s*(.+)", main_title_text)
if match:
candidate_brand = match.group(1).strip()
if is_plausible_brand(candidate_brand, nlp_model):
logger.debug(
f"Title Parser: Found via title pattern 'Brand - Product': {candidate_brand}"
)
return (
candidate_brand,
f'{selector_title} (pattern "Brand - Product")',
FieldExtractionStatus.MODULE_HEURISTIC,
100,
)
# Heuristic 3: Pattern "Product (Brand)"
match = re.search(r"\(([^)]+)\)$", main_title_text)
if match:
candidate_brand = match.group(1).strip()
if is_plausible_brand(candidate_brand, nlp_model):
logger.debug(
f"Title Parser: Found via title pattern 'Product (Brand)': {candidate_brand}"
)
return (
candidate_brand,
f'{selector_title} (pattern "Product (Brand)")',
FieldExtractionStatus.MODULE_HEURISTIC,
95,
)
return None, "title_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/brand/utils.py 0000664 0000000 0000000 00000012160 15107027250 0026640 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/brand/utils.py
import re
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
GENERIC_STOP_WORDS = {
"the",
"a",
"an",
"and",
"or",
"for",
"with",
"plus",
"pro",
"max",
"mini",
"ultra",
"lite",
"premium",
"standard",
"original",
"official",
"generic",
"compatible",
"universal",
"charger",
"adapter",
"cable",
"power",
"output",
"input",
"watt",
"volt",
"amp",
"store",
"website",
"visit",
"page",
"product",
}
def _is_likely_model_number(text: str) -> bool:
"""
A specific heuristic to check if a string looks like a model number.
A model number typically contains a mix of letters and digits.
"""
# Rule 1: Must contain at least one digit.
if not any(char.isdigit() for char in text):
return False
# Rule 2: Must contain at least one letter.
if not any(char.isalpha() for char in text):
return False
# Rule 3: Check for common model number patterns (e.g., ABC-123, 123-ABC, X1)
# This regex is more specific: it looks for sequences of letters/numbers mixed together.
if re.fullmatch(r"([A-Z0-9]+[-./]?[A-Z0-9]+)+", text, re.IGNORECASE):
return True
return False
def is_plausible_brand(candidate_brand: Optional[str], nlp_model: Any) -> bool:
"""
Validates if a candidate string could plausibly be a brand name using a series
of linguistic and rule-based heuristics.
"""
if not candidate_brand:
return False
cleaned_brand = candidate_brand.strip()
# Rule-based Heuristics (Fast Checks)
# 1. Check length
if not (2 <= len(cleaned_brand) <= 50):
return False
# 2. NEW: Check if it's likely a model number using our smarter function
if _is_likely_model_number(cleaned_brand):
logger.debug(
f"Brand validation: '{cleaned_brand}' looks like a model number. Rejected."
)
return False
# 3. Reject if purely numeric
if cleaned_brand.isdigit():
return False
# 4. Reject if it's a generic stop word
if cleaned_brand.lower() in GENERIC_STOP_WORDS:
return False
# 5. Reject if it's too many words
if len(cleaned_brand.split()) > 4:
return False
# Linguistic Heuristics (Smarter Checks)
if not nlp_model:
logger.warning(
"Brand validation: NLP model not provided, skipping linguistic checks."
)
return True # Fallback to true if NLP model is missing
doc = nlp_model(cleaned_brand)
# A plausible brand name must contain at least one Proper Noun (PROPN).
if not any(token.pos_ == "PROPN" for token in doc):
logger.debug(
f"Brand validation: '{cleaned_brand}' rejected: contains no proper nouns."
)
return False
return True
def get_main_title_content(soup: BeautifulSoup) -> Tuple[Optional[str], str]:
"""
Finds the main title of the page (H1 or tag).
"""
title_selectors = [
'h1[itemprop="name"]',
"h1.product-title",
"h1.item-name",
"h1.product__title",
"h1.pdp-title",
"h1",
]
for selector in title_selectors:
element = soup.select_one(selector)
if element:
return element.get_text(strip=True), selector
# Fallback to the main <title> tag
if soup.title and soup.title.string:
return soup.title.get_text(strip=True), "title"
return None, "NO_TITLE_FOUND"
def check_brand_in_main_title(candidate_brand: str, soup: BeautifulSoup) -> bool:
"""
Checks if the candidate brand name (approximately) appears in the H1 or TITLE tag.
"""
title_text_content, _ = get_main_title_content(soup)
if not title_text_content:
return False
# Create a regex pattern to find the brand as a whole word
pattern = r"\b" + re.escape(candidate_brand) + r"\b"
if re.search(pattern, title_text_content, re.IGNORECASE):
logger.debug(
f"Brand validation: Candidate '{candidate_brand}' validated in page title."
)
return True
return False
def find_explicit_brands(soup: BeautifulSoup) -> set[str]:
"""
Finds all potential brand names from elements that are explicitly
marked with brand-related classes or itemprop attributes.
Returns a set of cleaned, lowercased brand candidates.
"""
explicit_brands = set()
# Search for elements with itemprop="brand"
brand_itemprop_tags = soup.find_all(itemprop="brand")
for tag in brand_itemprop_tags:
text = tag.get("content") or tag.get_text(strip=True)
if text:
explicit_brands.add(text.strip().lower())
# Search for common brand-related class names
brand_class_regex = re.compile(
r"brand|manufacturer|vendor|product-brand", re.IGNORECASE
)
brand_elements = soup.find_all(class_=brand_class_regex)
for element in brand_elements:
text = element.get_text(strip=True)
if text:
explicit_brands.add(text.strip().lower())
return explicit_brands
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/ 0000775 0000000 0000000 00000000000 15107027250 0026331 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0030430 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/extract.py 0000664 0000000 0000000 00000005337 15107027250 0030365 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/breadcrumbs/extract.py
from typing import Optional, Tuple, Any, List
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.breadcrumbs.parsers.json_ld_parser import parse_from_json_ld
from app.modules.breadcrumbs.parsers.itemprop_parser import parse_itemprop_schema
from app.modules.breadcrumbs.parsers.heuristic_parser import parse_with_heuristics
from app.modules.breadcrumbs.parsers.regex_parser import parse_with_regex
FIELD_TYPE = Optional[List[str]]
REQUIRES = ["json_ld"]
def extract() -> Tuple[Any, str, FieldExtractionStatus, int]:
"""
The orchestrator function that extracts the breadcrumb navigation by calling a series of parsers.
"""
logger.info("Breadcrumbs Extractor (Main): Starting breadcrumbs extraction.")
# Priority 1: JSON-LD (from shared_context)
# This parser reads the results from the 'json_ld' module,
# so it does not depend on the soup object.
breadcrumbs, selector, status, score = parse_from_json_ld()
if breadcrumbs:
logger.info(
f"Breadcrumbs Extractor: Breadcrumbs successfully extracted with JSON-LD Parser. Score: {score}"
)
return breadcrumbs, selector, status, score
# HTML parsers (itemprop, heuristic, regex) need the soup
soup_to_use = shared_context.get("raw_soup")
if not soup_to_use:
logger.warning(
"Breadcrumbs Extractor: No valid BeautifulSoup objects to work with (and JSON-LD failed)."
)
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
# Priority 2: Schema.org itemprop-microdata
breadcrumbs, selector, status, score = parse_itemprop_schema(soup_to_use)
if breadcrumbs:
logger.info(
f"Breadcrumbs Extractor: Breadcrumbs successfully extracted with Itemprop Parser. Score: {score}"
)
return breadcrumbs, selector, status, score
# Priority 3: General classes and IDs
breadcrumbs, selector, status, score = parse_with_heuristics(soup_to_use)
if breadcrumbs:
logger.info(
f"Breadcrumbs Extractor: Breadcrumbs successfully extracted with Heuristics Parser. Score: {score}"
)
return breadcrumbs, selector, status, score
# Priority 4: Regex on separators
breadcrumbs, selector, status, score = parse_with_regex(soup_to_use)
if breadcrumbs:
logger.info(
f"Breadcrumbs Extractor: Breadcrumbs successfully extracted with Regex Parser. Score: {score}"
)
return breadcrumbs, selector, status, score
logger.info("Breadcrumbs Extractor: No suitable breadcrumbs found.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/parsers/ 0000775 0000000 0000000 00000000000 15107027250 0030010 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/parsers/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0032107 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/parsers/heuristic_parser.py 0000664 0000000 0000000 00000010023 15107027250 0033731 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/breadcrumbs/parsers/heuristic_parser.py
from typing import Optional, Tuple, List
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.breadcrumbs.utils import clean_and_filter_breadcrumbs
from app.utils.pattern_manager import pattern_manager
def is_breadcrumb_container(tag: Tag) -> bool:
"""
Checks if a tag is a likely breadcrumb container by
scanning attributes for keywords or standard ARIA roles/labels.
"""
if tag.name not in ["nav", "ul", "ol", "div"]:
return False
# Priority 1: Check standard ARIA accessibility attributes
aria_label = tag.get("aria-label", "").lower()
if "breadcrumb" in aria_label:
return True
if tag.get("role") == "navigation" and "breadcrumb" in aria_label:
return True
# Priority 2: Check all attribute values (class, id, etc.)
all_attribute_values = []
for attr_value in tag.attrs.values():
if isinstance(attr_value, list):
all_attribute_values.extend(attr_value)
else:
all_attribute_values.append(str(attr_value))
attributes_string = " ".join(all_attribute_values).lower()
# Use the dynamic regex from PatternManager
container_regex = pattern_manager.get_compiled_regex("breadcrumb_container_regex")
return bool(container_regex.search(attributes_string))
def _extract_text_from_li(item: Tag) -> Optional[str]:
"""
Smarter text extraction from an <li> item.
It prioritizes <a> tags to avoid grabbing separators.
"""
# Try to find a link first, this is the most common
link = item.find("a")
if link:
return link.get_text(strip=True)
# Fallback for non-link items (often the last item)
# Try to find a span, which might also hold the item
span = item.find("span")
if span:
return span.get_text(strip=True)
# Last resort: get the whole text of the li
return item.get_text(strip=True)
def parse_with_heuristics(
soup: BeautifulSoup,
) -> Tuple[Optional[List[str]], str, FieldExtractionStatus, int]:
"""
Extracts breadcrumbs using heuristics to find the container and the items,
regardless of the specific HTML tags used.
"""
containers = soup.find_all(is_breadcrumb_container, limit=5)
for container in containers:
found_breadcrumbs = []
# Heuristic A: Find <li> items (most common structure)
list_items = container.find_all("li")
if len(list_items) > 1:
for item in list_items:
text = _extract_text_from_li(item)
if text:
found_breadcrumbs.append(text)
# Heuristic B: No <li>? Find direct <a> children
if not found_breadcrumbs:
links = container.find_all("a", recursive=False)
if len(links) > 1:
for link in links:
found_breadcrumbs.append(link.get_text(strip=True))
else:
# Try recursive search if no direct children
links = container.find_all("a")
if len(links) > 1:
for link in links:
found_breadcrumbs.append(link.get_text(strip=True))
# Heuristic C: No <li> or <a>? Find direct children (e.g., <span>)
if not found_breadcrumbs:
children = [child for child in container.children if isinstance(child, Tag)]
if len(children) > 1:
for child in children:
text = child.get_text(strip=True)
if len(text) > 1:
found_breadcrumbs.append(text)
cleaned = clean_and_filter_breadcrumbs(found_breadcrumbs)
if cleaned:
selector = f"{container.name} (heuristic search: aria/class/id)"
logger.debug(f"Heuristic Parser: Breadcrumbs found: {cleaned}")
return cleaned, selector, FieldExtractionStatus.MODULE_HEURISTIC, 100
return None, "heuristic_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/parsers/itemprop_parser.py 0000664 0000000 0000000 00000007151 15107027250 0033601 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/breadcrumbs/parsers/itemprop_parser.py
import re
from typing import Optional, Tuple, List
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.breadcrumbs.utils import clean_and_filter_breadcrumbs
def _find_breadcrumb_container(soup: BeautifulSoup) -> Optional[Tag]:
"""
Finds a breadcrumb container by checking both itemprop and itemtype attributes.
"""
# Find tags with 'itemtype' containing 'BreadcrumbList'
# This is a very reliable indicator.
itemtype_containers = soup.find_all(
re.compile(r"ol|ul|nav|div"),
itemtype=re.compile(r"BreadcrumbList", re.IGNORECASE),
)
if itemtype_containers:
return itemtype_containers[0] # Return the first match
# Fallback: Find tags with 'itemprop' containing 'breadcrumb'
# This is less common for the list itself but still valid.
itemprop_containers = soup.find_all(
re.compile(r"ol|ul|nav|div"),
itemprop=re.compile(r"breadcrumb|itemList", re.IGNORECASE),
)
if itemprop_containers:
return itemprop_containers[0]
return None
def parse_itemprop_schema(
soup: BeautifulSoup,
) -> Tuple[Optional[List[str]], str, FieldExtractionStatus, int]:
"""
Extracts breadcrumbs from Schema.org (BreadcrumbList) microdata,
checking both itemprop and itemtype.
"""
breadcrumb_list_container = _find_breadcrumb_container(soup)
if not breadcrumb_list_container:
return None, "itemprop_parser", FieldExtractionStatus.NOT_FOUND, 0
found_breadcrumbs = []
# Find all elements marked as 'itemListElement' or 'item'
list_items = breadcrumb_list_container.find_all(
itemprop=re.compile(r"itemListElement|item", re.IGNORECASE)
)
if not list_items:
# Fallback: sometimes items are direct children without itemprop
list_items = breadcrumb_list_container.find_all("li", recursive=False)
if list_items:
for item in list_items:
# Find the name of the item (the text)
name_tag = item.find(itemprop="name")
if name_tag and name_tag.get_text(strip=True):
found_breadcrumbs.append(name_tag.get_text(strip=True))
else:
# Fallback: search in an <a> tag
link_tag = item.find("a")
if link_tag and link_tag.get_text(strip=True):
found_breadcrumbs.append(link_tag.get_text(strip=True))
# Fallback: get text of the item itself, excluding sub-items
elif item.get_text(strip=True):
# Try to get text only from the item, not nested tags
main_text = "".join(
item.find_all(string=True, recursive=False)
).strip()
if main_text:
found_breadcrumbs.append(main_text)
elif link_tag is None: # Only if no link was found
found_breadcrumbs.append(item.get_text(strip=True))
# Clean up and validate the found list
cleaned_breadcrumbs = clean_and_filter_breadcrumbs(found_breadcrumbs)
if cleaned_breadcrumbs:
selector = (
f'{breadcrumb_list_container.name}[itemprop/itemtype="BreadcrumbList"]'
)
logger.debug(f"Itemprop Parser: Found breadcrumbs: {cleaned_breadcrumbs}")
return (
cleaned_breadcrumbs,
selector,
FieldExtractionStatus.MODULE_HEURISTIC,
120,
)
return None, "itemprop_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/parsers/json_ld_parser.py 0000664 0000000 0000000 00000006212 15107027250 0033367 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/breadcrumbs/parsers/json_ld_parser.py
from typing import Optional, Tuple, List
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.breadcrumbs.utils import clean_and_filter_breadcrumbs
def parse_from_json_ld() -> Tuple[Optional[List[str]], str, FieldExtractionStatus, int]:
"""
Parses breadcrumbs from the raw 'json_ld' list in shared_context.
It finds the 'BreadcrumbList' node and extracts the items.
"""
# 1. Get the RAW LIST of nodes from the 'json_ld' module
json_ld_data_list = shared_context.get("json_ld")
if not isinstance(json_ld_data_list, list):
logger.debug("Breadcrumbs JSON-LD Parser: No 'json_ld' list found in context.")
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
found_breadcrumbs = []
# 2. Find the BreadcrumbList node in the raw list
for node in json_ld_data_list:
if not isinstance(node, dict):
continue
if node.get("@type", "").lower() == "breadcrumblist":
item_list_element = node.get("itemListElement")
if not isinstance(item_list_element, list):
continue
# Sort by position to ensure correct order
sorted_item_list = sorted(
item_list_element,
key=lambda x: x.get("position", 99) if isinstance(x, dict) else 99,
)
# 3. Extract the name from each item
for item_data in sorted_item_list:
crumb_name = None
if isinstance(item_data, dict):
# Item can be a nested object
item_node = item_data.get("item")
if isinstance(item_node, dict):
crumb_name = item_node.get("name")
# Or the name can be directly on the list element
elif "name" in item_data:
crumb_name = item_data.get("name")
if crumb_name and isinstance(crumb_name, str):
found_breadcrumbs.append(crumb_name.strip())
# If we found a list, stop processing other nodes
if found_breadcrumbs:
break
if not found_breadcrumbs:
logger.debug(
"Breadcrumbs JSON-LD Parser: No 'BreadcrumbList' node found in 'json_ld' data."
)
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
logger.debug(
f"Breadcrumbs JSON-LD Parser: Found raw list: {found_breadcrumbs}. Applying strict validation..."
)
# 4. Apply the SHARED, strict validation
cleaned = clean_and_filter_breadcrumbs(found_breadcrumbs)
if cleaned:
logger.debug(f"Breadcrumbs JSON-LD Parser: Validation passed: {cleaned}")
return (
cleaned,
"json-ld(@type=BreadcrumbList)",
FieldExtractionStatus.JSON_LD,
150,
)
logger.debug(
"Breadcrumbs JSON-LD Parser: Raw list failed strict validation (e.g., too short)."
)
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/parsers/regex_parser.py 0000664 0000000 0000000 00000003431 15107027250 0033051 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/breadcrumbs/parsers/regex_parser.py
import re
from typing import Optional, Tuple, List
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.breadcrumbs.utils import clean_and_filter_breadcrumbs, is_unwanted_text
# Expanded regex to include '|' and '\' as separators
BREADCRUMB_SEP_REGEX = re.compile(r"\s*>\s*|\s*»\s*|\s*/\s*|\s*\|\s*|\\")
def parse_with_regex(
soup: BeautifulSoup,
) -> Tuple[Optional[List[str]], str, FieldExtractionStatus, int]:
"""
Finds breadcrumbs using a regex pattern based on separators,
while safely ignoring script and style content.
"""
# Search for text nodes containing any of the separators
all_text_nodes = soup.body.find_all(string=BREADCRUMB_SEP_REGEX)
for text_node in all_text_nodes:
# Avoid extracting from noisy or irrelevant tags
if text_node.parent.name in ["script", "style", "title", "head", "a", "option"]:
continue
text = text_node.strip()
# Check for minimal length and unwanted patterns
if not text or len(text) < 5 or is_unwanted_text(text):
continue
# Split the text based on the separators
parts = re.split(BREADCRUMB_SEP_REGEX, text)
cleaned_breadcrumbs = clean_and_filter_breadcrumbs(parts)
if cleaned_breadcrumbs:
selector_used = f"{text_node.parent.name} (text pattern)"
logger.debug(f"Regex Parser: Found breadcrumbs: {cleaned_breadcrumbs}")
return (
cleaned_breadcrumbs,
selector_used,
FieldExtractionStatus.MODULE_HEURISTIC,
80,
)
return None, "regex_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/breadcrumbs/utils.py 0000664 0000000 0000000 00000003241 15107027250 0030043 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/breadcrumbs/utils.py
from typing import List, Optional
import re
from loguru import logger
from app.utils.pattern_manager import pattern_manager
# FILTER_UNWANTED_CRUMBS constant is removed
def clean_and_filter_breadcrumbs(breadcrumbs: List[str]) -> Optional[List[str]]:
"""
Cleans up a list of breadcrumbs by removing empty, short, or generic items.
Returns None if the resulting list is invalid.
"""
# Get the dynamic filter list from PatternManager
unwanted_crumbs = pattern_manager.get_keyword_list("breadcrumb_filter_keywords")
unwanted_crumbs_lower = {k.lower() for k in unwanted_crumbs}
# Filter empty strings and generic terms
filtered = [
b.strip()
for b in breadcrumbs
if b and b.strip() and b.strip().lower() not in unwanted_crumbs_lower
]
# Deduplicate the list while preserving order
deduplicated_breadcrumbs = []
for item in filtered:
if not deduplicated_breadcrumbs or item != deduplicated_breadcrumbs[-1]:
deduplicated_breadcrumbs.append(item)
# A valid set of breadcrumbs must have at least two items
if len(deduplicated_breadcrumbs) >= 2:
return deduplicated_breadcrumbs
logger.debug(
f"Breadcrumb validation: The found list ({deduplicated_breadcrumbs}) is too short or invalid."
)
return None
def is_unwanted_text(text: str) -> bool:
"""
Checks if a piece of text is likely not a breadcrumb (e.g., address, date).
"""
return bool(
re.search(
r"\d{4}\s*[a-zA-Z]{2}|\d{2}-\d{2}-\d{4}|email|tel:|fax:",
text,
re.IGNORECASE,
)
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/ 0000775 0000000 0000000 00000000000 15107027250 0026363 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0030462 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/extract.py 0000664 0000000 0000000 00000006513 15107027250 0030414 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/description/extract.py
from typing import Optional, Tuple, Any, List
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from .parsers.json_ld_parser import parse_json_ld
from .parsers.open_graph_parser import parse_open_graph
from .parsers.amazon_parser import parse_amazon_sections
from .parsers.meta_parser import parse_meta_tags
from .parsers.general_parser import parse_general_sections
from .parsers.fallback_parser import parse_generic_fallback
FIELD_TYPE = Optional[str]
REQUIRES = ["json_ld", "open_graph"]
def extract() -> Tuple[Any, str, FieldExtractionStatus, int]:
"""
The orchestrator function that extracts the product description by calling a series
of parsers, collecting all results, and returning the one with the highest score.
"""
soup_to_use = shared_context.get("raw_soup")
nlp_model = shared_context.get("resources", {}).get("nlp_model")
if not soup_to_use:
logger.warning(
"Description Extractor: No valid BeautifulSoup objects to work with."
)
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
logger.info(
"Description Extractor (Main): Starting product description extraction."
)
# Step 1: Initialize a list to store all found results.
results: List[Tuple[Any, str, FieldExtractionStatus, int]] = []
# Step 2: Call all parsers and add their results to the list.
# Parser 1: JSON-LD (highest reliability)
description, selector, status, score = parse_json_ld(nlp_model)
if description:
results.append((description, selector, status, score))
# Parser 2: Open Graph (high reliability)
description, selector, status, score = parse_open_graph(nlp_model)
if description:
results.append((description, selector, status, score))
# Parser 3: Amazon-specific sections (high score HTML)
description, selector, status, score = parse_amazon_sections(soup_to_use, nlp_model)
if description:
results.append((description, selector, status, score))
# Parser 4: Meta-tags (Schema.org / name="description")
description, selector, status, score = parse_meta_tags(soup_to_use, nlp_model)
if description:
results.append((description, selector, status, score))
# Parser 5: General sections on the page
description, selector, status, score = parse_general_sections(
soup_to_use, nlp_model
)
if description:
results.append((description, selector, status, score))
# Parser 6: Fallback (large text blocks)
description, selector, status, score = parse_generic_fallback(
soup_to_use, nlp_model
)
if description:
results.append((description, selector, status, score))
# Step 7: Determine the best result from the list.
if not results:
logger.info(
"Description Extractor: No parser could find a suitable description."
)
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
# Use max() to find the result (tuple) with the highest score (the 4th element, index 3).
best_result = max(results, key=lambda item: item[3])
logger.success(
f"Description Extractor: Best description found via '{best_result[1]}' with score {best_result[3]}."
)
return best_result
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/parsers/ 0000775 0000000 0000000 00000000000 15107027250 0030042 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/parsers/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0032141 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/parsers/amazon_parser.py 0000664 0000000 0000000 00000005420 15107027250 0033256 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/description/parsers/amazon_parser.py
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.description.utils import clean_and_validate_description
def parse_amazon_sections(
soup: BeautifulSoup, nlp_model: Any = None
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the description from Amazon-specific sections.
"""
logger.debug("Amazon Parser: Searching in Amazon-specific description sections.")
# 1. Product Description Block
product_description_div = soup.select_one("#productDescription")
if product_description_div:
description_text = product_description_div.get_text(separator=" ", strip=True)
cleaned_desc = clean_and_validate_description(description_text, nlp_model)
if cleaned_desc:
logger.debug(
f"Amazon Parser: Found via #productDescription: {cleaned_desc[:100]}..."
)
return (
cleaned_desc,
"#productDescription",
FieldExtractionStatus.MODULE_HEURISTIC,
250,
)
# 2. Feature Bullets
feature_bullets_ul = soup.select_one("#feature-bullets ul")
if feature_bullets_ul:
bullets = [
li.get_text(strip=True)
for li in feature_bullets_ul.find_all("li")
if li.get_text(strip=True)
]
if bullets:
combined_description = "\n".join(bullets)
cleaned_desc = clean_and_validate_description(
combined_description, nlp_model
)
if cleaned_desc:
logger.debug(
f"Amazon Parser: Found via #feature-bullets: {cleaned_desc[:100]}..."
)
return (
cleaned_desc,
"#feature-bullets ul",
FieldExtractionStatus.MODULE_HEURISTIC,
220,
)
# 3. 'About this item' section
about_item_section = soup.select_one(
"#productOverview_feature_div .a-section.a-spacing-small"
)
if about_item_section:
about_text = about_item_section.get_text(separator="\n", strip=True)
cleaned_desc = clean_and_validate_description(about_text, nlp_model)
if cleaned_desc:
logger.debug(
f"Amazon Parser: Found via 'About this item' section: {cleaned_desc[:100]}..."
)
return (
cleaned_desc,
"#productOverview_feature_div",
FieldExtractionStatus.MODULE_HEURISTIC,
210,
)
return None, "amazon_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/parsers/fallback_parser.py 0000664 0000000 0000000 00000003676 15107027250 0033543 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/description/parsers/fallback_parser.py
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.context import shared_context
from app.core.models import FieldExtractionStatus
from app.modules.description.utils import clean_and_validate_description
def parse_generic_fallback(
soup: BeautifulSoup, nlp_model: Any = None
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts a description by searching for large, general text blocks.
"""
logger.debug("Fallback Parser: Last fallback: Large text blocks.")
main_content_area = soup.find("div", id="dp") or soup.find("body")
if main_content_area:
potential_texts = main_content_area.find_all(["p", "div", "span"], string=True)
for p_tag in potential_texts:
# Skip elements that are likely already processed
if p_tag.parent and (
p_tag.parent.name == "li"
or "feature-bullets" in (p_tag.parent.get("id") or "")
or "feature-bullets" in " ".join(p_tag.parent.get("class", []))
):
continue
text_content = p_tag.get_text(strip=True)
cleaned_desc = clean_and_validate_description(text_content, nlp_model)
# Prevent duplication of other extracted fields
if cleaned_desc and cleaned_desc not in [
shared_context.get("title"),
shared_context.get("price"),
]:
logger.debug(
f"Fallback Parser: Found via general text block: {cleaned_desc[:100]}..."
)
return (
cleaned_desc,
f"{p_tag.name} (generic text block)",
FieldExtractionStatus.GENERIC_FALLBACK,
80,
)
return None, "fallback_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/parsers/general_parser.py 0000664 0000000 0000000 00000002735 15107027250 0033414 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/description/parsers/general_parser.py
import re
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.description.utils import clean_and_validate_description
def parse_general_sections(
soup: BeautifulSoup, nlp_model: Any = None
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the description from general sections related to the description.
"""
logger.debug("General Parser: Searching for general description sections.")
general_desc_elements = soup.find_all(
["div", "p", "section", "span"],
class_=re.compile(r"description|details|info|content|main-text", re.IGNORECASE),
id=re.compile(r"description|details|info|content|main-text", re.IGNORECASE),
)
for elem in general_desc_elements:
text_content = elem.get_text(separator=" ", strip=True)
cleaned_desc = clean_and_validate_description(text_content, nlp_model)
if cleaned_desc:
logger.debug(
f"General Parser: Found via general element: {cleaned_desc[:100]}..."
)
selector = (
f"{elem.name}#{elem.get('id') or ''} (class={elem.get('class', [])})"
)
return cleaned_desc, selector, FieldExtractionStatus.MODULE_HEURISTIC, 150
return None, "general_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/parsers/json_ld_parser.py 0000664 0000000 0000000 00000003130 15107027250 0033415 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/description/parsers/json_ld_parser.py
from typing import Optional, Tuple, Any
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.description.utils import clean_and_validate_description
def parse_json_ld(
nlp_model: Any = None,
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts description from the 'json_ld' data in shared_context.
"""
logger.debug("JSON-LD Parser: Searching for description in JSON-LD data.")
json_ld_data = shared_context.get("json_ld")
if not isinstance(json_ld_data, list):
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
for node in json_ld_data:
if not isinstance(node, dict):
continue
# The description can be at the top level of any node
desc_text = node.get("description")
if isinstance(desc_text, str):
cleaned_desc = clean_and_validate_description(desc_text, nlp_model)
if cleaned_desc:
logger.debug(
f"JSON-LD Parser: Found via json_ld.description: {cleaned_desc[:100]}..."
)
# Give this the highest score
return (
cleaned_desc,
"json_ld.description",
FieldExtractionStatus.JSON_LD,
300,
)
logger.debug("JSON-LD Parser: No description found in JSON-LD data.")
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/parsers/meta_parser.py 0000664 0000000 0000000 00000003321 15107027250 0032715 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/description/parsers/meta_parser.py
from typing import Optional, Tuple, Any
from bs4 import BeautifulSoup
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.description.utils import clean_and_validate_description
def parse_meta_tags(
soup: BeautifulSoup, nlp_model: Any = None
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the description from Open Graph and Schema.org meta-tags.
"""
logger.debug("Meta Parser: Searching in Open Graph / Schema.org meta tags.")
# 1. Schema.org
schema_description = soup.find("meta", itemprop="description")
if schema_description and schema_description.get("content"):
desc_text = schema_description["content"].strip()
cleaned_desc = clean_and_validate_description(desc_text, nlp_model)
if cleaned_desc:
logger.debug(
f"Meta Parser: Found itemprop='description': {cleaned_desc[:100]}..."
)
return (
cleaned_desc,
'meta[itemprop="description"]',
FieldExtractionStatus.MODULE_HEURISTIC,
175,
)
# 2. Meta name
meta_description = soup.find("meta", attrs={"name": "description"})
if meta_description and meta_description.get("content"):
desc_text = meta_description["content"].strip()
logger.debug(f"Meta Parser: Found via name='description': {desc_text[:100]}...")
return (
desc_text,
'meta[name="description"]',
FieldExtractionStatus.MODULE_HEURISTIC,
100,
)
return None, "meta_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/parsers/open_graph_parser.py 0000664 0000000 0000000 00000002561 15107027250 0034116 0 ustar 00root root 0000000 0000000 # app/modules/description/parsers/open_graph_parser.py
from typing import Optional, Tuple, Any
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.description.utils import clean_and_validate_description
def parse_open_graph(
nlp_model: Any = None,
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts description from the 'open_graph' data in shared_context.
"""
logger.debug("Open Graph Parser: Searching for description in OG data.")
og_data = shared_context.get("open_graph")
if not isinstance(og_data, dict):
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
# The open_graph module populates the 'description' key
desc_text = og_data.get("description")
if isinstance(desc_text, str):
cleaned_desc = clean_and_validate_description(desc_text, nlp_model)
if cleaned_desc:
logger.debug(
f"Open Graph Parser: Found via og:description: {cleaned_desc[:100]}..."
)
# Give this a very high score, just under JSON-LD
return cleaned_desc, "og:description", FieldExtractionStatus.OPEN_GRAPH, 275
logger.debug("Open Graph Parser: No description found in OG data.")
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/description/utils.py 0000664 0000000 0000000 00000004422 15107027250 0030077 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/description/utils.py
import re
from typing import Optional, Any
from loguru import logger
from app.utils.pattern_manager import pattern_manager
def clean_and_validate_description(text: str, nlp_model: Any = None) -> Optional[str]:
"""
Cleans up the description text and validates it generically based
on the density of descriptive words (nouns and adjectives).
"""
cleaned_text = text.strip()
# Use the dynamic regex from PatternManager
undesired_phrases_regex = pattern_manager.get_compiled_regex(
"description_filter_regex"
)
cleaned_text = undesired_phrases_regex.sub("", cleaned_text).strip()
cleaned_text = re.sub(r"\n\s*\n", "\n", cleaned_text)
cleaned_text = re.sub(r"\s+", " ", cleaned_text).strip()
if len(cleaned_text) < 50:
logger.debug(
f"Description validation: Text is too short ({len(cleaned_text)} characters). Skipping."
)
return None
if len(cleaned_text) > 5000:
logger.debug(
f"Description validation: Text is too long ({len(cleaned_text)} characters). Cutting text."
)
cleaned_text = cleaned_text[:5000]
if not nlp_model:
logger.warning(
"Description validation: NLP model not available, cannot perform generic validation. Skipping."
)
return cleaned_text
try:
doc = nlp_model(cleaned_text)
noun_adj_count = sum(1 for token in doc if token.pos_ in ["NOUN", "ADJ"])
total_tokens = len([token for token in doc if token.is_alpha])
if total_tokens == 0:
return None
density = noun_adj_count / total_tokens
MIN_DENSITY_THRESHOLD = 0.25
MIN_ABSOLUTE_COUNT = 2
logger.debug(
f"Description validation: Noun/Adjective density={density:.2f}, count={noun_adj_count}"
)
if density >= MIN_DENSITY_THRESHOLD or noun_adj_count >= MIN_ABSOLUTE_COUNT:
return cleaned_text
else:
logger.debug(
"Description validation: Density and absolute count are too low. Skipping."
)
return None
except Exception as e:
logger.warning(f"Description validation: NLP analysis failed: {e}")
return cleaned_text
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/ 0000775 0000000 0000000 00000000000 15107027250 0025122 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0027221 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/extract.py 0000664 0000000 0000000 00000006447 15107027250 0027161 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/image/extract.py
from typing import Optional, Tuple, Any
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from .parsers.json_ld_parser import parse_json_ld
from .parsers.open_graph_parser import parse_open_graph
from .parsers.meta_parser import parse_meta_tags
from .parsers.amazon_parser import parse_amazon_selectors
from .parsers.context_parser import parse_from_product_context
from .parsers.fallback_parser import parse_largest_image_fallback
FIELD_TYPE = Optional[str]
REQUIRES = ["json_ld", "open_graph"]
def extract() -> Tuple[Any, str, FieldExtractionStatus, int]:
"""
The orchestrator function that extracts the image URL by calling a series of parsers.
"""
soup_to_use = shared_context.get("raw_soup")
processed_elements = shared_context.get("processed_elements", set())
if not soup_to_use:
logger.warning("Image Extractor: No valid BeautifulSoup objects to work with.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
logger.info("Image Extractor (Main): Starting image URL extraction.")
# Priority 1: JSON-LD (highest reliability)
image_url, selector, status, score = parse_json_ld()
if image_url:
logger.info(
f"Image Extractor: Image URL successfully extracted with JSON-LD Parser. Score: {score}"
)
return image_url, selector, status, score
# Priority 2: Open Graph (high reliability)
image_url, selector, status, score = parse_open_graph()
if image_url:
logger.info(
f"Image Extractor: Image URL successfully extracted with Open Graph Parser. Score: {score}"
)
return image_url, selector, status, score
# Priority 3: Schema.org Microdata (HTML)
image_url, selector, status, score = parse_meta_tags(
soup_to_use, processed_elements
)
if image_url:
logger.info(
f"Image Extractor: Image URL successfully extracted with Meta Parser. Score: {score}"
)
return image_url, selector, status, score
# Priority 4: Amazon-specific selectors (HTML)
image_url, selector, status, score = parse_amazon_selectors(
soup_to_use, processed_elements
)
if image_url:
logger.info(
f"Image Extractor: Image URL successfully extracted with Amazon Parser. Score: {score}"
)
return image_url, selector, status, score
# Priority 5: Image in product context (HTML)
image_url, selector, status, score = parse_from_product_context(
soup_to_use, processed_elements
)
if image_url:
logger.info(
f"Image Extractor: Image URL successfully extracted with Context Parser. Score: {score}"
)
return image_url, selector, status, score
# Priority 6: Fallback (largest image, HTML)
image_url, selector, status, score = parse_largest_image_fallback(
soup_to_use, processed_elements
)
if image_url:
logger.info(
f"Image Extractor: Image URL successfully extracted with Fallback Parser. Score: {score}"
)
return image_url, selector, status, score
logger.info("Image Extractor: No suitable image URL found.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/parsers/ 0000775 0000000 0000000 00000000000 15107027250 0026601 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/parsers/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0030700 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/parsers/amazon_parser.py 0000664 0000000 0000000 00000004202 15107027250 0032012 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/image/parsers/amazon_parser.py
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.image.utils import is_valid_image_url
def parse_amazon_selectors(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the image from Amazon-specific HTML elements.
"""
logger.debug("Amazon Parser: Searching in Amazon-specific selectors.")
amazon_image_selectors = [
"#landingImage",
"#imgTagWrapperId img",
"#main-image-container img",
"#mainImage",
".a-dynamic-image",
"#imgBlkFront",
]
for selector_str in amazon_image_selectors:
img_element = soup.select_one(selector_str)
if img_element and img_element.get("src"):
if img_element in processed_elements:
continue
src = img_element["src"].strip()
hires_src = img_element.get("data-old-hires") or img_element.get(
"data-a-image-src"
)
# Try the high-resolution image first
if hires_src and is_valid_image_url(hires_src.strip()):
logger.debug(
f"Amazon Parser: Found via Amazon selector '{selector_str}' (hires): {hires_src}"
)
processed_elements.add(img_element)
return (
hires_src.strip(),
selector_str + " (hires)",
FieldExtractionStatus.MODULE_HEURISTIC,
180,
)
# Validate the standard src if there is no high-resolution version
elif is_valid_image_url(src):
logger.debug(
f"Amazon Parser: Found via Amazon selector '{selector_str}': {src}"
)
processed_elements.add(img_element)
return src, selector_str, FieldExtractionStatus.MODULE_HEURISTIC, 175
return None, "amazon_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/parsers/context_parser.py 0000664 0000000 0000000 00000004740 15107027250 0032220 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/image/parsers/context_parser.py
import re
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.image.utils import is_valid_image_url
def parse_from_product_context(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the first reasonable image from product-related sections.
"""
logger.debug("Context Parser: Searching in general product-related sections.")
product_section_elements = soup.select(
'div[id*="product"], div[class*="product"], main, article'
)
for section in product_section_elements:
if section in processed_elements:
continue
# We search for all images, not just the first one, to have a better chance
image_tags = section.find_all("img", src=re.compile(r"https?://"))
for img_tag in image_tags:
img_src = img_tag.get("src")
if not img_src or not is_valid_image_url(img_src):
continue
width = img_tag.get("width")
height = img_tag.get("height")
# If we have dimensions, we validate them.
if width and height:
try:
# Remove 'px' etc. and convert to int
w = int(re.sub(r"\D", "", width))
h = int(re.sub(r"\D", "", height))
# If the image is PROVEN to be too small, we skip it.
if w < 50 or h < 50:
logger.debug(
f"Context Parser: Image skipped due to being too small: {w}x{h}"
)
continue # Continue to the next image in the section
except (ValueError, TypeError):
# Dimensions are not a valid number, so we ignore them and proceed.
pass
# If the image is large enough, OR if there are no dimensions,
# we accept it and stop searching.
logger.debug(f"Context Parser: Found in product section: {img_src}")
processed_elements.add(img_tag)
return (
img_src,
"img_in_product_section",
FieldExtractionStatus.MODULE_HEURISTIC,
100,
)
return None, "context_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/parsers/fallback_parser.py 0000664 0000000 0000000 00000003145 15107027250 0032271 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/image/parsers/fallback_parser.py
import re
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.image.utils import is_valid_image_url
def parse_largest_image_fallback(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Searches for the largest image in the body as a last resort.
"""
logger.debug("Fallback Parser: Searching for largest image in body.")
all_images = soup.find_all("img", src=re.compile(r"https?://"))
best_image_url = None
max_area = 0
for img in all_images:
if img in processed_elements:
continue
src = img.get("src")
if not is_valid_image_url(src):
continue
width = img.get("width")
height = img.get("height")
if width and height:
try:
w = int(width)
h = int(height)
area = w * h
if area > max_area and area > 10000: # Minimum 100x100 pixels
max_area = area
best_image_url = src
except ValueError:
pass
if best_image_url:
logger.debug(f"Fallback Parser: Found via largest image: {best_image_url}")
return (
best_image_url,
"largest_img_in_body",
FieldExtractionStatus.GENERIC_FALLBACK,
50,
)
return None, "fallback_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/parsers/json_ld_parser.py 0000664 0000000 0000000 00000003175 15107027250 0032165 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/image/parsers/json_ld_parser.py
from typing import Optional, Tuple
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.image.utils import is_valid_image_url
def parse_json_ld() -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the image URL from the 'json_ld' data in shared_context.
"""
logger.debug("JSON-LD Parser: Searching for image in JSON-LD data.")
json_ld_data = shared_context.get("json_ld")
if not isinstance(json_ld_data, list):
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
for node in json_ld_data:
if not isinstance(node, dict):
continue
image_data = node.get("image")
image_url = None
if isinstance(image_data, str):
image_url = image_data
elif isinstance(image_data, dict):
image_url = image_data.get("url")
elif isinstance(image_data, list) and image_data:
if isinstance(image_data[0], str):
image_url = image_data[0]
elif isinstance(image_data[0], dict):
image_url = image_data[0].get("url")
if image_url and is_valid_image_url(image_url):
logger.debug(f"JSON-LD Parser: Found via json_ld.image: {image_url}")
# Give this the highest score
return image_url, "json_ld.image", FieldExtractionStatus.JSON_LD, 200
logger.debug("JSON-LD Parser: No image found in JSON-LD data.")
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/parsers/meta_parser.py 0000664 0000000 0000000 00000002434 15107027250 0031460 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/image/parsers/meta_parser.py
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.image.utils import is_valid_image_url
def parse_meta_tags(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the image from Schema.org microdata meta-tags.
(Open Graph logic is now handled by open_graph_parser.py)
"""
logger.debug("Meta Parser: Searching in Schema.org meta tags.")
# Look for Schema.org microdata tag
schema_image = soup.find("meta", itemprop="image")
if (
schema_image
and schema_image.get("content")
and is_valid_image_url(schema_image["content"])
):
logger.debug(
f"Meta Parser: Found via itemprop='image': {schema_image['content']}"
)
processed_elements.add(schema_image)
# This is now the highest-priority HTML-based meta tag
return (
schema_image["content"],
'meta[itemprop="image"]',
FieldExtractionStatus.MODULE_HEURISTIC,
180,
)
return None, "meta_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/parsers/open_graph_parser.py 0000664 0000000 0000000 00000002255 15107027250 0032655 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/image/parsers/open_graph_parser.py
from typing import Optional, Tuple
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.image.utils import is_valid_image_url
def parse_open_graph() -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the image URL from the 'open_graph' data in shared_context.
"""
logger.debug("Open Graph Parser: Searching for image in OG data.")
og_data = shared_context.get("open_graph")
if not isinstance(og_data, dict):
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
# The open_graph module populates the 'image' key from 'og:image'
image_url = og_data.get("image")
if image_url and is_valid_image_url(image_url):
logger.debug(f"Open Graph Parser: Found via og:image: {image_url}")
# Give this a high score, just under JSON-LD
return image_url, "og:image", FieldExtractionStatus.OPEN_GRAPH, 190
logger.debug("Open Graph Parser: No image found in OG data.")
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/image/utils.py 0000664 0000000 0000000 00000001403 15107027250 0026632 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/image/utils.py
import re
from typing import Optional
from loguru import logger
def is_valid_image_url(url: Optional[str]) -> bool:
"""
Validates a URL as a valid image URL.
"""
if not url:
return False
# Basic validation of the URL, must start with http(s) and have a common extension
if not re.match(r"^https?://.*\.(jpg|jpeg|png|gif|webp|bmp)$", url, re.IGNORECASE):
logger.debug(
f"Image validation: URL '{url}' does not look like a valid image URL."
)
return False
# URLs that are too short are often invalid or data URIs
if len(url) < 10:
logger.debug(f"Image validation: URL '{url}' is too short.")
return False
return True
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/ 0000775 0000000 0000000 00000000000 15107027250 0025470 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0027567 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/extract.py 0000664 0000000 0000000 00000003072 15107027250 0027516 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/json_ld/extract.py
from typing import Dict, Any, Optional, Tuple, List
from loguru import logger
from app.core.context import shared_context
from app.core.models import FieldExtractionStatus
from app.modules.json_ld.utils import parse_json_ld_scripts
REQUIRES = []
FIELD_TYPE = Optional[List[Dict[str, Any]]]
def extract() -> Tuple[Optional[List[Dict[str, Any]]], str, FieldExtractionStatus, int]:
"""
Extracts all raw JSON-LD data from the page.
This module acts as a data provider. Other modules (like 'price',
'title', 'reviews') will depend on this module and consume this
raw data from the shared_context.
"""
soup_to_use = shared_context.get("raw_soup")
selector = "script[type='application/ld+json']"
if not soup_to_use:
logger.warning("JSON_LD Extractor: No raw_soup found in context.")
return None, selector, FieldExtractionStatus.NOT_FOUND, 0
# Use the utility to find and parse all scripts
all_parsed_json_data = parse_json_ld_scripts(soup_to_use)
if not all_parsed_json_data:
logger.info("JSON_LD Extractor: No valid JSON-LD data found on page.")
return None, selector, FieldExtractionStatus.NOT_FOUND, 0
logger.info(
f"JSON_LD Extractor: Found {len(all_parsed_json_data)} raw JSON-LD nodes."
)
# Return the raw list.
# The Analyzer will place this list on the shared_context['json_ld']
# AND add it to the final 'json_ld' key in the API response.
return all_parsed_json_data, selector, FieldExtractionStatus.JSON_LD, 200
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/parsers/ 0000775 0000000 0000000 00000000000 15107027250 0027147 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/parsers/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0031246 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/parsers/breadcrumb_parser.py 0000664 0000000 0000000 00000004513 15107027250 0033206 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/json_ld/parsers/breadcrumb_parser.py
from typing import Dict, Any
from loguru import logger
from app.modules.json_ld.utils import is_title_matching_breadcrumb
def parse_breadcrumbs(node: Dict[str, Any], extracted_fields: Dict[str, Any]):
"""
Extracts breadcrumbs from a BreadcrumbList node.
This populates the 'breadcrumbs' key within the 'json_ld' result object.
It performs minimal filtering; the main 'breadcrumbs' module
will re-validate this data for the final top-level field.
"""
if node.get("@type", "").lower() != "breadcrumblist":
return
item_list_element = node.get("itemListElement")
if not isinstance(item_list_element, list):
return
breadcrumbs = []
# CRITICAL: Sort by position first to ensure correct order
sorted_item_list = sorted(
item_list_element,
key=lambda x: x.get("position", 99) if isinstance(x, dict) else 99,
)
for item_data in sorted_item_list:
crumb_name = None
if (
isinstance(item_data, dict)
and "item" in item_data
and isinstance(item_data["item"], dict)
):
crumb_name = item_data["item"].get("name")
elif isinstance(item_data, dict) and "name" in item_data:
crumb_name = item_data.get("name")
if crumb_name and isinstance(crumb_name, str):
cleaned_crumb_name = crumb_name.strip()
# Apply the original simple "home" filter
if cleaned_crumb_name and cleaned_crumb_name.lower() != "home":
breadcrumbs.append(cleaned_crumb_name)
if breadcrumbs:
# This logic was in the original file, so we keep it for the json_ld field
extracted_title = extracted_fields.get("title")
if extracted_title and is_title_matching_breadcrumb(
extracted_title, breadcrumbs[-1]
):
logger.debug(
f"JSON-LD Breadcrumb Parser: Last crumb '{breadcrumbs[-1]}' matches title. Filtering for 'json_ld' field."
)
breadcrumbs.pop()
if breadcrumbs: # Check again if list is not empty
extracted_fields["breadcrumbs"] = breadcrumbs
logger.debug(
f"JSON-LD Breadcrumb Parser: Found raw breadcrumbs for 'json_ld' field: {breadcrumbs}"
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/parsers/offer_parser.py 0000664 0000000 0000000 00000005477 15107027250 0032213 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/json_ld/parsers/offer_parser.py
from typing import Dict, Any, List, Union
from loguru import logger
from app.modules.json_ld.utils import extract_value
from app.utils.data_utils import clean_price_text
def parse_offer_details(node: Dict[str, Any], extracted_fields: Dict[str, Any]):
"""
Extracts price, currency, and availability from an Offer node or nested offers.
Modifies the 'extracted_fields' dictionary in-place.
"""
offers_data: Union[Dict[str, Any], List[Dict[str, Any]], None]
if node.get("@type", "").lower() == "offer":
offers_data = node
else:
offers_data = node.get("offers")
offers_list: List[Dict[str, Any]] = []
if isinstance(offers_data, dict):
offers_list = [offers_data]
elif isinstance(offers_data, list):
offers_list = offers_data
if not offers_list:
logger.debug("Offer Parser: No offer data found in this node.")
return
# Process the first valid offer in the list
offer = offers_list[0]
# Price
price_raw = offer.get("lowPrice") or offer.get("price")
if price_raw is None:
price_raw = extract_value(
offer, "priceSpecification.minPrice"
) or extract_value(offer, "priceSpecification.price")
if price_raw is not None and "price" not in extracted_fields:
try:
price_str_candidate = str(price_raw)
price_float = clean_price_text(price_str_candidate)
if price_float is not None:
extracted_fields["price"] = price_float
logger.debug(f"Offer Parser: Price found: {price_float}")
except (ValueError, TypeError):
logger.debug(
f"Offer Parser: Error converting/cleaning price '{price_raw}'."
)
# Currency
currency = offer.get("priceCurrency")
if currency and "currency" not in extracted_fields:
extracted_fields["currency"] = currency
logger.debug(f"Offer Parser: Currency found: {currency}")
# Availability
availability_value = offer.get("availability")
if isinstance(availability_value, str) and "availability" not in extracted_fields:
availability_status = availability_value.split("/")[-1]
valid_statuses = [
"InStock",
"OutOfStock",
"PreOrder",
"LimitedAvailability",
"Discontinued",
"SoldOut",
"OnlineOnly",
"InStoreOnly",
]
if availability_status in valid_statuses:
extracted_fields["availability"] = availability_status
logger.debug(f"Offer Parser: Availability found: {availability_status}")
else:
logger.warning(
f"Offer Parser: Unknown availability status in JSON-LD: '{availability_value}'."
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/parsers/product_parser.py 0000664 0000000 0000000 00000004665 15107027250 0032570 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/json_ld/parsers/product_parser.py
from typing import Dict, Any
from loguru import logger
from app.modules.json_ld.utils import extract_value
from app.utils.data_utils import is_valid_ean_checksum
def parse_product_details(node: Dict[str, Any], extracted_fields: Dict[str, Any]):
"""
Extracts product-specific details from a Product or ProductGroup node.
Modifies the 'extracted_fields' dictionary in-place.
"""
node_type = node.get("@type", "").lower()
logger.debug(f"Product Parser: Processing node type: {node_type}")
if node_type not in ["product", "productgroup"]:
return
# Title
title = node.get("name")
if title:
extracted_fields["title"] = title
# Description
description = node.get("description")
if description:
extracted_fields["description"] = description
# Image
image_data = node.get("image")
if isinstance(image_data, str):
extracted_fields["image"] = image_data
elif isinstance(image_data, dict) and "url" in image_data:
extracted_fields["image"] = image_data["url"]
elif isinstance(image_data, list) and image_data:
if isinstance(image_data[0], str):
extracted_fields["image"] = image_data[0]
elif isinstance(image_data[0], dict) and "url" in image_data[0]:
extracted_fields["image"] = image_data[0]["url"]
# Brand
brand = extract_value(node, "brand.name")
if not brand:
brand = extract_value(node, "manufacturer.name")
if not brand:
direct_brand = node.get("brand")
if isinstance(direct_brand, str):
brand = direct_brand
if brand:
extracted_fields["brand"] = brand
# SKU / Product ID
sku = node.get("sku") or node.get("productID")
if sku:
extracted_fields["sku"] = str(sku)
# EAN / GTIN
ean_candidate = (
node.get("gtin13")
or node.get("gtin")
or node.get("gtin8")
or node.get("gtin14")
or node.get("ean")
)
if ean_candidate:
ean_str = str(ean_candidate)
if len(ean_str) == 13 and ean_str.isdigit() and is_valid_ean_checksum(ean_str):
extracted_fields["ean"] = ean_str
logger.debug(f"Product Parser: Valid EAN (gtin13) found: {ean_str}")
else:
logger.debug(
f"Product Parser: Found EAN/GTIN '{ean_str}' in JSON-LD is invalid. Skipping."
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/parsers/review_parser.py 0000664 0000000 0000000 00000005057 15107027250 0032405 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/json_ld/parsers/review_parser.py
from typing import Dict, Any, List, Optional
from loguru import logger
from app.utils.data_utils import clean_text_and_remove_unicode
def _parse_single_review(review_node: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Parses a single review node and extracts the relevant fields."""
if not isinstance(review_node, dict):
return None
extracted_review = {}
# Extract the author's name
author_node = review_node.get("author")
if isinstance(author_node, dict):
extracted_review["name"] = clean_text_and_remove_unicode(
author_node.get("name")
)
elif isinstance(author_node, str):
extracted_review["name"] = clean_text_and_remove_unicode(author_node)
# Extract the rating
rating_node = review_node.get("reviewRating")
if isinstance(rating_node, dict):
rating = rating_node.get("ratingValue")
if rating is not None:
extracted_review["rating"] = float(rating)
# Extract the review text and date
review_text = review_node.get("reviewBody")
if review_text:
extracted_review["review_text"] = clean_text_and_remove_unicode(review_text)
date_published = review_node.get("datePublished")
if date_published:
extracted_review["date"] = clean_text_and_remove_unicode(date_published)
# Only return if there is at least a review text or a rating
if (
extracted_review.get("review_text")
or extracted_review.get("rating") is not None
):
return extracted_review
return None
def parse_reviews(json_ld_data: List[Dict[str, Any]], extracted_fields: Dict[str, Any]):
"""
Finds and parses review data from a list of JSON-LD nodes.
This function processes both single Review objects and nested lists of reviews.
"""
all_reviews = []
for node in json_ld_data:
reviews = node.get("review")
if isinstance(reviews, dict):
# A single review node
parsed_review = _parse_single_review(reviews)
if parsed_review:
all_reviews.append(parsed_review)
elif isinstance(reviews, list):
# A list of review nodes
for review_item in reviews:
parsed_review = _parse_single_review(review_item)
if parsed_review:
all_reviews.append(parsed_review)
if all_reviews:
extracted_fields["reviews"] = all_reviews
logger.info(f"JSON_LD Reviews Parser: {len(all_reviews)} reviews extracted.")
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/json_ld/utils.py 0000664 0000000 0000000 00000006256 15107027250 0027213 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/json_ld/utils.py
import json
import re
from bs4 import BeautifulSoup
from typing import Dict, Any, Optional, List
from loguru import logger
def parse_json_ld_scripts(soup: BeautifulSoup) -> List[Dict[str, Any]]:
"""
Finds and parses all JSON-LD scripts on the page.
Returns a list of all parsed dictionaries.
"""
all_parsed_json_data: List[Dict[str, Any]] = []
json_ld_scripts = soup.find_all("script", type="application/ld+json")
if not json_ld_scripts:
logger.debug("JSON_LD Utils: No JSON-LD scripts found.")
return all_parsed_json_data
logger.debug(f"JSON_LD Utils: {len(json_ld_scripts)} JSON-LD scripts found.")
for script in json_ld_scripts:
json_string_content: Optional[str] = None
try:
json_string_content = script.string
if json_string_content:
json_content = json.loads(json_string_content)
if isinstance(json_content, list):
all_parsed_json_data.extend(json_content)
elif isinstance(json_content, dict):
all_parsed_json_data.append(json_content)
except (json.JSONDecodeError, TypeError) as e:
logger.warning(f"JSON_LD Utils: Error parsing JSON-LD script: {e}")
return all_parsed_json_data
def extract_value(data: Dict[str, Any], key: str, default: Any = None) -> Any:
"""
Retrieves a value from a dictionary, with support for nested paths.
"""
keys = key.split(".")
current_data = data
for k in keys:
if isinstance(current_data, dict) and k in current_data:
current_data = current_data[k]
elif isinstance(current_data, list):
try:
index = int(k)
if 0 <= index < len(current_data):
current_data = current_data[index]
else:
return default
except (ValueError, IndexError):
return default
else:
return default
return current_data
def is_title_matching_breadcrumb(
title: str, breadcrumb: str, fuzziness_threshold: float = 0.8
) -> bool:
"""
Checks if a breadcrumb (the last one) matches the product title.
"""
title_lower = title.lower()
breadcrumb_lower = breadcrumb.lower()
if title_lower == breadcrumb_lower:
return True
clean_title = re.sub(r"[^a-z0-9\s]", "", title_lower).strip()
clean_breadcrumb = re.sub(r"[^a-z0-9\s]", "", breadcrumb_lower).strip()
if not clean_title or not clean_breadcrumb:
return False
if clean_title == clean_breadcrumb:
return True
if clean_title in clean_breadcrumb or clean_breadcrumb in clean_title:
return True
title_tokens = set(clean_title.split())
breadcrumb_tokens = set(clean_breadcrumb.split())
if not title_tokens or not breadcrumb_tokens:
return False
intersection = len(title_tokens.intersection(breadcrumb_tokens))
union = len(title_tokens.union(breadcrumb_tokens))
jaccard_similarity = intersection / union if union > 0 else 0
if jaccard_similarity >= fuzziness_threshold:
return True
return False
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/ 0000775 0000000 0000000 00000000000 15107027250 0026162 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0030261 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/extract.py 0000664 0000000 0000000 00000002740 15107027250 0030211 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/open_graph/extract.py
from typing import Dict, Any, Tuple, Optional
from loguru import logger
from app.core.context import shared_context
from app.core.types import FieldExtractionStatus
from .utils import find_og_tags
REQUIRES = []
FIELD_TYPE = Optional[Dict[str, Any]]
def extract() -> Tuple[Optional[Dict[str, Any]], str, FieldExtractionStatus, int]:
"""
Extracts all raw Open Graph (og:) data from the page's meta tags.
This module acts as a data provider. Other modules (like 'price',
'title', 'image') will depend on this module and consume this
raw dictionary from the shared_context.
"""
raw_soup = shared_context.get("raw_soup")
selector_used = "meta[property^='og:']"
if not raw_soup:
logger.warning("Open Graph Extractor: No raw_soup found in context.")
return None, selector_used, FieldExtractionStatus.NOT_FOUND, 0
# Call the utility function to find all tags
og_tags = find_og_tags(raw_soup)
if not og_tags:
logger.info("Open Graph Extractor: No Open Graph tags found.")
return None, selector_used, FieldExtractionStatus.NOT_FOUND, 0
logger.info(f"Open Graph Extractor: Found {len(og_tags)} raw OG tags.")
# Return the raw dictionary.
# The Analyzer will place this dict on the shared_context['open_graph']
# AND add it to the final 'open_graph' key in the API response.
return og_tags, selector_used, FieldExtractionStatus.OPEN_GRAPH, 200
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/parsers/ 0000775 0000000 0000000 00000000000 15107027250 0027641 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/parsers/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0031740 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/parsers/availability_parser.py 0000664 0000000 0000000 00000002014 15107027250 0034236 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/open_graph/parsers/availability_parser.py
from typing import Dict, Any
from loguru import logger
from app.utils.data_utils import normalize_availability
def parse_availability(og_tags: Dict[str, Any], extracted_fields: Dict[str, Any]):
"""
Extracts and normalizes the availability status from the Open Graph tags.
Modifies the 'extracted_fields' dictionary in-place.
"""
logger.debug("Availability Parser: Starting availability extraction.")
availability_og = og_tags.get("availability")
if availability_og:
normalized_availability = normalize_availability(availability_og)
if normalized_availability:
extracted_fields["availability"] = normalized_availability
logger.debug(
f"Availability Parser: Availability '{normalized_availability}' found."
)
else:
logger.debug(
f"Availability Parser: Could not normalize availability '{availability_og}'."
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/parsers/brand_parser.py 0000664 0000000 0000000 00000001051 15107027250 0032652 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/open_graph/parsers/brand_parser.py
from typing import Dict, Any
from loguru import logger
def parse_brand(og_tags: Dict[str, Any], extracted_fields: Dict[str, Any]):
"""
Extracts the brand from the non-standard og:brand tag.
Modifies the 'extracted_fields' dictionary in-place.
"""
logger.debug("Brand Parser: Starting brand extraction.")
brand = og_tags.get("brand")
if brand:
extracted_fields["brand"] = brand
logger.debug(f"Brand Parser: Brand '{brand}' found.")
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/parsers/metadata_parser.py 0000664 0000000 0000000 00000001741 15107027250 0033352 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/open_graph/parsers/metadata_parser.py
from typing import Dict, Any
from loguru import logger
def parse_metadata(og_tags: Dict[str, Any], extracted_fields: Dict[str, Any]):
"""
Extracts title, description, and image from the Open Graph tags.
Modifies the 'extracted_fields' dictionary in-place.
"""
logger.debug(
"Metadata Parser: Starting extraction of title, description, and image."
)
# Title
title = og_tags.get("title")
if title:
extracted_fields["title"] = title
logger.debug(f"Metadata Parser: Title '{title}' found.")
# Description
description = og_tags.get("description")
if description:
extracted_fields["description"] = description
logger.debug("Metadata Parser: Description found.")
# Image
image = og_tags.get("image")
if image:
extracted_fields["image"] = image
logger.debug(f"Metadata Parser: Image '{image}' found.")
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/parsers/pricing_parser.py 0000664 0000000 0000000 00000002143 15107027250 0033222 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/open_graph/parsers/pricing_parser.py
from typing import Dict, Any
from loguru import logger
from app.utils.data_utils import clean_price_text
def parse_pricing(og_tags: Dict[str, Any], extracted_fields: Dict[str, Any]):
"""
Extracts price and currency from the Open Graph tags.
Modifies the 'extracted_fields' dictionary in-place.
"""
logger.debug("Pricing Parser: Starting price and currency extraction.")
# Price
price_amount_str = og_tags.get("price:amount")
if price_amount_str:
cleaned_price = clean_price_text(price_amount_str)
if cleaned_price is not None:
extracted_fields["price"] = cleaned_price
logger.debug(f"Pricing Parser: Price {cleaned_price} found.")
else:
logger.debug(
f"Pricing Parser: Could not convert price '{price_amount_str}' to float."
)
# Currency
currency = og_tags.get("price:currency")
if currency:
extracted_fields["currency"] = currency
logger.debug(f"Pricing Parser: Currency '{currency}' found.")
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/open_graph/utils.py 0000664 0000000 0000000 00000001230 15107027250 0027670 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/open_graph/utils.py
from typing import Dict, Any
from bs4 import BeautifulSoup
def find_og_tags(soup: BeautifulSoup) -> Dict[str, Any]:
"""
Finds all Open Graph meta-tags and returns them as a dictionary.
The 'og:' prefix is removed from the keys.
"""
og_tags: Dict[str, Any] = {}
for tag in soup.find_all("meta", property=True):
prop = tag.get("property")
content = tag.get("content")
if prop and content and prop.startswith("og:"):
# 'og:title' -> 'title'
key = prop.replace("og:", "")
og_tags[key] = content.strip()
return og_tags
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/ 0000775 0000000 0000000 00000000000 15107027250 0025142 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0027241 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/extract.py 0000664 0000000 0000000 00000006337 15107027250 0027177 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/price/extract.py
from typing import Optional, Tuple, Any
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from .parsers.json_ld_parser import parse_json_ld
from .parsers.open_graph_parser import parse_open_graph
from .parsers.itemprop_parser import parse_itemprop
from .parsers.class_parser import parse_price_classes
from .parsers.regex_section_parser import parse_regex_in_sections
from .parsers.regex_body_parser import parse_regex_in_body
FIELD_TYPE = Optional[float]
REQUIRES = ["json_ld", "open_graph"]
def extract() -> Tuple[Any, str, FieldExtractionStatus, int]:
"""
The main extractor function that extracts the price by calling a series of parsers.
"""
soup_to_use = shared_context.get("raw_soup")
if not soup_to_use:
logger.warning("Price Extractor: No valid BeautifulSoup objects to work with.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
logger.info("Price Extractor (Main): Starting price extraction.")
processed_elements = shared_context.get("processed_elements", set())
# Priority 1: JSON-LD (highest reliability)
price, selector, status, score = parse_json_ld()
if price is not None:
logger.info(
f"Price Extractor: Price successfully extracted with JSON-LD Parser. Score: {score}"
)
return price, selector, status, score
# Priority 2: Open Graph (high reliability)
price, selector, status, score = parse_open_graph()
if price is not None:
logger.info(
f"Price Extractor: Price successfully extracted with Open Graph Parser. Score: {score}"
)
return price, selector, status, score
# Priority 3: itemprop="price"
price, selector, status, score = parse_itemprop(soup_to_use, processed_elements)
if price is not None:
logger.info(
f"Price Extractor: Price successfully extracted with itemprop parser. Score: {score}"
)
return price, selector, status, score
# Priority 4: Price-related classes/IDs
price, selector, status, score = parse_price_classes(
soup_to_use, processed_elements
)
if price is not None:
logger.info(
f"Price Extractor: Price successfully extracted with Class Parser. Score: {score}"
)
return price, selector, status, score
# Priority 5: Regex in specific sections
price, selector, status, score = parse_regex_in_sections(
soup_to_use, processed_elements
)
if price is not None:
logger.info(
f"Price Extractor: Price successfully extracted with Regex Section Parser. Score: {score}"
)
return price, selector, status, score
# Priority 6: Regex in the whole body (last resort)
price, selector, status, score = parse_regex_in_body(
soup_to_use, processed_elements
)
if price is not None:
logger.info(
f"Price Extractor: Price successfully extracted with Regex Body Parser. Score: {score}"
)
return price, selector, status, score
logger.info("Price Extractor: No suitable price found.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/parsers/ 0000775 0000000 0000000 00000000000 15107027250 0026621 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/parsers/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0030720 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/parsers/class_parser.py 0000664 0000000 0000000 00000007404 15107027250 0031661 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/price/parsers/class_parser.py
import re
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.price.utils import clean_price_text
def _reconstruct_price_from_fragments(tag: BeautifulSoup) -> Optional[str]:
"""
Finds all text fragments within a tag, validates them, and combines them
into a correct decimal number. Only accepts digits and currency symbols.
"""
# Find all text nodes, including in nested tags
all_strings = tag.find_all(string=True)
number_parts = []
for s in all_strings:
cleaned_s = s.strip()
if not cleaned_s:
continue
# Check if the fragment only contains digits, dots, commas, or currency symbols.
# If it contains other letters (like in "other text"), the price is invalid.
if re.search(r"[a-zA-Z]", cleaned_s) and not re.fullmatch(
r"[\s\d.,€$£]*", cleaned_s, re.IGNORECASE
):
logger.trace(
f"Price reconstructor: Invalid text fragment found: '{cleaned_s}'"
)
return None # Invalid text found
# Get only the digits from the fragment
digits = re.findall(r"\d+", cleaned_s)
if digits:
number_parts.extend(digits)
if not number_parts:
return None
# Join the numeric parts. If there is more than one part,
# the last one is considered the fraction.
if len(number_parts) == 2:
# e.g., ['1', '25'] becomes '1.25'
return f"{'.'.join(number_parts[:-1])}.{number_parts[-1]}"
else:
# e.g., ['1.25'] or ['1,25'] remains itself
return number_parts[0]
def parse_price_classes(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[float], str, FieldExtractionStatus, int]:
"""
Extracts the price based on relevant CSS classes and IDs,
with special logic for split prices.
"""
price_tags = soup.find_all(
re.compile(r"div|span|p|b|section"), class_=re.compile(r"price", re.IGNORECASE)
)
for tag in price_tags:
if tag in processed_elements:
continue
# First, try to intelligently reconstruct the price from fragments
text = _reconstruct_price_from_fragments(tag)
# If that fails, fall back to the old, simple method
if text is None:
text = tag.get_text(strip=True)
if not text:
continue
# The rest of your logic for checking 'old price' remains the same
is_old_price = False
parent_with_old_price = tag.find_parent(
class_=re.compile(
r"old-price|original-price|price-old|compare-price", re.IGNORECASE
)
)
if parent_with_old_price:
logger.debug(
f"Class Parser: Skipped tag inside old price container: {text[:50]}"
)
is_old_price = True
if not is_old_price:
numerical_price = clean_price_text(text)
if numerical_price is not None:
selector_detail = (
f'{tag.name}[class*="{tag.get("class")[0] if tag.get("class") else ""}"]'
if tag.get("class")
else tag.name
)
logger.debug(
f"Class Parser: Found via class '{selector_detail}': {numerical_price}"
)
processed_elements.add(tag)
return (
numerical_price,
selector_detail,
FieldExtractionStatus.MODULE_HEURISTIC,
100,
)
return None, "class_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/parsers/itemprop_parser.py 0000664 0000000 0000000 00000003636 15107027250 0032416 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/price/parsers/itemprop_parser.py
import re
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.price.utils import clean_price_text
def parse_itemprop(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[float], str, FieldExtractionStatus, int]:
"""
Extracts the price using itemprop="price" microdata.
"""
# Priority 1a: meta tag
price_meta = soup.find("meta", itemprop="price")
if price_meta and price_meta.get("content"):
price_val = clean_price_text(price_meta["content"].strip())
if price_val is not None:
logger.debug(
f"Itemprop Parser: Found via meta[itemprop='price'] (content): {price_val}"
)
processed_elements.add(price_meta)
return (
price_val,
'meta[itemprop="price"]',
FieldExtractionStatus.MODULE_HEURISTIC,
120,
)
# Priority 1b: other tags
price_tag = soup.find(re.compile(r"span|div|b|p"), itemprop="price")
if price_tag and price_tag.get_text(strip=True):
price_val = clean_price_text(price_tag.get_text(strip=True))
if price_val is not None:
logger.debug(
f"Itemprop Parser: Found via span/div[itemprop='price'] (text): {price_val}"
)
selector_detail = f'{price_tag.name}[itemprop="price"]'
if price_tag.get("class"):
selector_detail += f".{'.'.join(price_tag['class'])}"
processed_elements.add(price_tag)
return (
price_val,
selector_detail,
FieldExtractionStatus.MODULE_HEURISTIC,
115,
)
return None, "itemprop_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/parsers/json_ld_parser.py 0000664 0000000 0000000 00000004263 15107027250 0032204 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/price/parsers/json_ld_parser.py
from typing import Optional, Tuple
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.price.utils import clean_price_text
def parse_json_ld() -> Tuple[Optional[float], str, FieldExtractionStatus, int]:
"""
Extracts price from the 'json_ld' data in shared_context.
"""
logger.debug("JSON-LD Parser: Searching for price in JSON-LD data.")
json_ld_data = shared_context.get("json_ld")
if not isinstance(json_ld_data, list):
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
for node in json_ld_data:
if not isinstance(node, dict):
continue
# Look for price in an 'offers' node (dict or list)
offers = node.get("offers")
offer_list = []
if isinstance(offers, dict):
offer_list = [offers]
elif isinstance(offers, list):
offer_list = offers
for offer in offer_list:
if not isinstance(offer, dict):
continue
# Find the price, looking at 'lowPrice' first, then 'price'
price_raw = offer.get("lowPrice") or offer.get("price")
# If not found, check a nested 'priceSpecification'
if price_raw is None:
price_spec = offer.get("priceSpecification")
if isinstance(price_spec, dict):
price_raw = price_spec.get("minPrice") or price_spec.get("price")
if price_raw is not None:
price_float = clean_price_text(str(price_raw))
if price_float is not None:
logger.debug(
f"JSON-LD Parser: Found price '{price_float}' via 'json_ld.offers'."
)
return (
price_float,
"json_ld.offers.price",
FieldExtractionStatus.JSON_LD,
200,
)
logger.debug("JSON-LD Parser: No price found in JSON-LD data.")
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/parsers/open_graph_parser.py 0000664 0000000 0000000 00000002552 15107027250 0032675 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/price/parsers/open_graph_parser.py
from typing import Optional, Tuple
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.price.utils import clean_price_text
def parse_open_graph() -> Tuple[Optional[float], str, FieldExtractionStatus, int]:
"""
Extracts price from the 'open_graph' data in shared_context.
"""
logger.debug("Open Graph Parser: Searching for price in OG data.")
og_data = shared_context.get("open_graph")
if not isinstance(og_data, dict):
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
# Standard keys are 'og:price:amount' or 'product:price:amount'
# We assume the open_graph module provides these as 'price:amount'
price_raw = og_data.get("price:amount") or og_data.get("product:price:amount")
if price_raw is not None:
price_float = clean_price_text(str(price_raw))
if price_float is not None:
logger.debug(
f"Open Graph Parser: Found price '{price_float}' via 'og:price:amount'."
)
return price_float, "og:price:amount", FieldExtractionStatus.OPEN_GRAPH, 190
logger.debug("Open Graph Parser: No price found in OG data.")
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/parsers/regex_body_parser.py 0000664 0000000 0000000 00000003015 15107027250 0032675 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/price/parsers/regex_body_parser.py
import re
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.price.utils import clean_price_text
from app.utils.pattern_manager import pattern_manager
def parse_regex_in_body(
soup: BeautifulSoup,
processed_elements: Set[Tag],
) -> Tuple[Optional[float], str, FieldExtractionStatus, int]:
"""
Searches for price formats with regex in the entire body of the page.
"""
body_text = soup.body.get_text(strip=True) if soup.body else ""
if len(body_text) > 10000:
body_text = body_text[:10000]
# Use dynamic regex for price format
price_regex = pattern_manager.get_compiled_regex("price_format_regex")
matches = re.finditer(price_regex, body_text, re.IGNORECASE)
for match in matches:
for group in match.groups():
if group:
numerical_price = clean_price_text(group)
if numerical_price is not None:
logger.debug(
f"Regex Body Parser: Found via general regex in body: {numerical_price}"
)
return (
numerical_price,
"body (regex heuristic)",
FieldExtractionStatus.MODULE_REGEX,
50,
)
return None, "regex_body_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/parsers/regex_section_parser.py 0000664 0000000 0000000 00000004147 15107027250 0033413 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/price/parsers/regex_section_parser.py
import re
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.price.utils import clean_price_text
from app.utils.pattern_manager import pattern_manager
def parse_regex_in_sections(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[float], str, FieldExtractionStatus, int]:
"""
Searches for price formats with regex in specific sections of the page.
"""
potential_price_sections = soup.find_all(
["div", "span", "section"],
# Use dynamic regex for classes
class_=pattern_manager.get_compiled_regex("price_section_class_regex"),
limit=5,
)
# Use dynamic regex for price format
price_regex = pattern_manager.get_compiled_regex("price_format_regex")
for section in potential_price_sections:
if section in processed_elements:
continue
section_text = section.get_text(strip=True)
if len(section_text) > 1000:
section_text = section_text[:1000]
matches = re.finditer(price_regex, section_text, re.IGNORECASE)
for match in matches:
for group in match.groups():
if group:
numerical_price = clean_price_text(group)
if numerical_price is not None:
selector_detail = f'{section.name}[class*="{section.get("class")[0] if section.get("class") else ""}"] (regex)'
logger.debug(
f"Regex Section Parser: Found via general regex in specific section: {numerical_price}"
)
processed_elements.add(section)
return (
numerical_price,
selector_detail,
FieldExtractionStatus.MODULE_REGEX,
70,
)
return None, "regex_section_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/price/utils.py 0000664 0000000 0000000 00000002435 15107027250 0026660 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/price/utils.py
import re
from typing import Optional
def clean_price_text(text: str) -> Optional[float]:
"""
Cleans up the price string and converts it to a float.
Supports different notations (with commas or periods as decimal separators).
"""
if not text:
return None
cleaned_text = re.sub(
r"[^\d.,]+", "", text
) # Remove everything except digits, periods, and commas
cleaned_text = cleaned_text.replace("\u200e", "").replace("\u200f", "").strip()
if not cleaned_text:
return None
# First, check if the last comma or period is the decimal separator.
# If there is more than one period, and a comma, then the comma is the decimal separator.
if "," in cleaned_text and "." in cleaned_text:
if cleaned_text.rfind(",") > cleaned_text.rfind("."):
# Format: 1.234,56
cleaned_text = cleaned_text.replace(".", "").replace(",", ".")
else:
# Format: 1,234.56
cleaned_text = cleaned_text.replace(",", "")
elif "," in cleaned_text:
# Format: 1234,56 or 1,234
cleaned_text = cleaned_text.replace(",", ".")
try:
price = float(cleaned_text)
return price
except ValueError:
return None
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/ 0000775 0000000 0000000 00000000000 15107027250 0025161 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0027260 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/extract.py 0000664 0000000 0000000 00000005766 15107027250 0027223 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/title/extract.py
from typing import Optional, Tuple, Any, List
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from .parsers.json_ld_parser import parse_json_ld
from .parsers.open_graph_parser import parse_open_graph
from .parsers.h1_parser import parse_h1_tags
from .parsers.title_tag_parser import parse_title_tag
from .parsers.meta_parser import parse_meta_tags
from .parsers.fallback_parser import parse_generic_fallback
FIELD_TYPE = Optional[str]
REQUIRES = ["json_ld", "open_graph"]
def extract() -> Tuple[Any, str, FieldExtractionStatus, int]:
"""
The main extractor function that extracts the title by calling a series of parsers,
collects all results, and returns the one with the highest score.
"""
soup_to_use = shared_context.get("raw_soup")
processed_elements = shared_context.get("processed_elements", set())
logger.info("Title Extractor (Main): Starting product title extraction.")
if not soup_to_use:
logger.warning("Title Extractor: No valid BeautifulSoup objects to work with.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
# Step 1: Initialize a list to store all found results.
results: List[Tuple[Any, str, FieldExtractionStatus, int]] = []
# Step 2: Call all parsers and add their results to the list.
# Parser 1: JSON-LD (highest reliability)
title, selector, status, score = parse_json_ld()
if title:
results.append((title, selector, status, score))
# Parser 2: Open Graph (high reliability)
title, selector, status, score = parse_open_graph()
if title:
results.append((title, selector, status, score))
# Parser 3: H1 tags
title, selector, status, score = parse_h1_tags(soup_to_use, processed_elements)
if title:
results.append((title, selector, status, score))
# Parser 4: <title> tag
title, selector, status, score = parse_title_tag(soup_to_use, processed_elements)
if title:
results.append((title, selector, status, score))
# Parser 5: Meta tags (itemprop='name')
title, selector, status, score = parse_meta_tags(soup_to_use, processed_elements)
if title:
results.append((title, selector, status, score))
# Parser 6: Generic fallback
title, selector, status, score = parse_generic_fallback(
soup_to_use, processed_elements
)
if title:
results.append((title, selector, status, score))
# Step 3: Determine the best result from the list.
if not results:
logger.info("Title Extractor: No suitable product title found.")
return None, "NOT_FOUND", FieldExtractionStatus.NOT_FOUND, 0
# Use max() to find the result (tuple) with the highest score (the 4th element, index 3).
best_result = max(results, key=lambda item: item[3])
logger.success(
f"Title Extractor: Best title found via '{best_result[1]}' with score {best_result[3]}."
)
return best_result
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/parsers/ 0000775 0000000 0000000 00000000000 15107027250 0026640 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/parsers/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0030737 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/parsers/fallback_parser.py 0000664 0000000 0000000 00000002601 15107027250 0032324 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/title/parsers/fallback_parser.py
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.title.utils import clean_title
def parse_generic_fallback(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Searches for the title in general elements as a last resort.
"""
potential_title_elements = soup.select(
'span[id*="title"], div[id*="title"], strong[id*="title"], h2.title, p.title'
)
for elem in potential_title_elements:
if elem in processed_elements:
continue
extracted_title = elem.get_text(strip=True)
cleaned_title = clean_title(extracted_title)
if cleaned_title and 10 < len(cleaned_title) < 200:
logger.debug(
f"Fallback Parser: Found via general elements: {cleaned_title}"
)
processed_elements.add(elem)
selector = (
f"{elem.name}#{elem.get('id')}"
if elem.get("id")
else f"{elem.name}.{elem.get('class', [''])[0]}"
)
return cleaned_title, selector, FieldExtractionStatus.GENERIC_FALLBACK, 80
return None, "fallback_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/parsers/h1_parser.py 0000664 0000000 0000000 00000006464 15107027250 0031110 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/title/parsers/h1_parser.py
import re
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.title.utils import clean_title
def parse_h1_tags(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the title from <h1> tags, including specific patterns and a generic fallback.
"""
# Priority 1a: Specific Amazon H1 tag
amazon_title_h1 = soup.select_one("#productTitle")
if amazon_title_h1:
extracted_title = amazon_title_h1.get_text(strip=True)
cleaned_title = clean_title(extracted_title)
if cleaned_title:
logger.debug(f"H1 Parser: Found via Amazon #productTitle: {cleaned_title}")
processed_elements.add(amazon_title_h1)
return (
cleaned_title,
"#productTitle",
FieldExtractionStatus.MODULE_HEURISTIC,
250,
)
# Priority 1b: General H1 tags with specific attributes
h1_with_attrs = soup.find("h1", itemprop="name") or soup.find(
"h1",
class_=re.compile(
r"product-title|item-name|title|product__title", re.IGNORECASE
),
)
if h1_with_attrs and h1_with_attrs not in processed_elements:
extracted_title = h1_with_attrs.get_text(strip=True)
cleaned_title = clean_title(extracted_title)
if cleaned_title and len(cleaned_title) > 5:
logger.debug(
f"H1 Parser: Found via H1 with specific attributes: {cleaned_title}"
)
processed_elements.add(h1_with_attrs)
selector = (
f'{h1_with_attrs.name}[itemprop="name"]'
if h1_with_attrs.get("itemprop")
else f"{h1_with_attrs.name}[class]"
)
return cleaned_title, selector, FieldExtractionStatus.MODULE_HEURISTIC, 220
# Priority 1c: Generic <h1> tag fallback
all_h1s = [
h
for h in soup.find_all("h1")
if h not in processed_elements and h.get_text(strip=True)
]
if len(all_h1s) == 1:
# High confidence if there's only one H1 on the page
h1_tag = all_h1s[0]
cleaned_title = clean_title(h1_tag.get_text(strip=True))
if cleaned_title:
logger.debug(f"H1 Parser: Found single, unambiguous H1: {cleaned_title}")
processed_elements.add(h1_tag)
return (
cleaned_title,
"h1 (single)",
FieldExtractionStatus.MODULE_HEURISTIC,
200,
)
elif len(all_h1s) > 1:
# Lower confidence if there are multiple H1s; we take the first one.
h1_tag = all_h1s[0]
cleaned_title = clean_title(h1_tag.get_text(strip=True))
if cleaned_title:
logger.debug(
f"H1 Parser: Found multiple H1s, taking first one: {cleaned_title}"
)
processed_elements.add(h1_tag)
return (
cleaned_title,
"h1 (first of many)",
FieldExtractionStatus.MODULE_HEURISTIC,
150,
)
return None, "h1_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/parsers/json_ld_parser.py 0000664 0000000 0000000 00000002600 15107027250 0032214 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/title/parsers/json_ld_parser.py
from typing import Optional, Tuple
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.title.utils import clean_title
def parse_json_ld() -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the title (name) from the 'json_ld' data in shared_context.
"""
logger.debug("JSON-LD Parser: Searching for title (name) in JSON-LD data.")
json_ld_data = shared_context.get("json_ld")
if not isinstance(json_ld_data, list):
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
for node in json_ld_data:
if not isinstance(node, dict):
continue
# The 'name' key is used for title in Product, Offer, etc.
title_text = node.get("name")
if isinstance(title_text, str):
cleaned_title = clean_title(title_text)
if cleaned_title and len(cleaned_title) > 5:
logger.debug(f"JSON-LD Parser: Found via json_ld.name: {cleaned_title}")
# Give this the highest score
return cleaned_title, "json_ld.name", FieldExtractionStatus.JSON_LD, 300
logger.debug("JSON-LD Parser: No title (name) found in JSON-LD data.")
return None, "json_ld_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/parsers/meta_parser.py 0000664 0000000 0000000 00000002435 15107027250 0031520 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/title/parsers/meta_parser.py
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.title.utils import clean_title
def parse_meta_tags(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the title from Schema.org microdata meta-tags.
(Open Graph logic is now handled by open_graph_parser.py)
"""
# Look for Schema.org microdata tag
schema_name = soup.find("meta", itemprop="name")
if (
schema_name
and schema_name.get("content")
and schema_name not in processed_elements
):
extracted_title = schema_name["content"].strip()
cleaned_title = clean_title(extracted_title)
if cleaned_title and len(cleaned_title) > 5:
logger.debug(f"Meta Parser: Found via itemprop='name': {cleaned_title}")
processed_elements.add(schema_name)
return (
cleaned_title,
'meta[itemprop="name"]',
FieldExtractionStatus.MODULE_HEURISTIC,
140,
)
return None, "meta_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/parsers/open_graph_parser.py 0000664 0000000 0000000 00000002414 15107027250 0032711 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/title/parsers/open_graph_parser.py
from typing import Optional, Tuple
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.core.context import shared_context
from app.modules.title.utils import clean_title
def parse_open_graph() -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the title from the 'open_graph' data in shared_context.
"""
logger.debug("Open Graph Parser: Searching for title in OG data.")
og_data = shared_context.get("open_graph")
if not isinstance(og_data, dict):
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
# The open_graph module populates the 'title' key from 'og:title'
title_text = og_data.get("title")
if isinstance(title_text, str):
cleaned_title = clean_title(title_text)
if cleaned_title and len(cleaned_title) > 5:
logger.debug(f"Open Graph Parser: Found via og:title: {cleaned_title}")
# Give this a high score, just under JSON-LD
return cleaned_title, "og:title", FieldExtractionStatus.OPEN_GRAPH, 275
logger.debug("Open Graph Parser: No title found in OG data.")
return None, "open_graph_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/parsers/title_tag_parser.py 0000664 0000000 0000000 00000002073 15107027250 0032544 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/title/parsers/title_tag_parser.py
from typing import Optional, Tuple, Set
from bs4 import BeautifulSoup, Tag
from loguru import logger
from app.core.models import FieldExtractionStatus
from app.modules.title.utils import clean_title
def parse_title_tag(
soup: BeautifulSoup, processed_elements: Set[Tag]
) -> Tuple[Optional[str], str, FieldExtractionStatus, int]:
"""
Extracts the title from the <title> tag in the HTML head.
"""
title_tag = soup.find("title")
if title_tag:
extracted_title = title_tag.get_text(strip=True)
cleaned_title = clean_title(extracted_title)
if cleaned_title and len(cleaned_title) > 5:
logger.debug(f"Title Tag Parser: Found via <title> tag: {cleaned_title}")
processed_elements.add(title_tag)
return (
cleaned_title,
"title (tag)",
FieldExtractionStatus.MODULE_HEURISTIC,
180,
)
return None, "title_tag_parser", FieldExtractionStatus.NOT_FOUND, 0
GetArgusFlow-argus-c6afc1d/services/extractor/app/modules/title/utils.py 0000664 0000000 0000000 00000001574 15107027250 0026702 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/title/utils.py
import re
from app.utils.pattern_manager import pattern_manager
def clean_title(title_text: str) -> str:
"""Cleans up the title from common website-specific additions."""
if not title_text:
return ""
# Get dynamic regex patterns to remove site names, etc.
clean_patterns = pattern_manager.get_keyword_list("title_clean_patterns_regex")
for pattern in clean_patterns:
title_text = re.sub(pattern, "", title_text, flags=re.IGNORECASE).strip()
# Get dynamic regex patterns to remove filter phrases
undesired_phrases = pattern_manager.get_keyword_list("title_filter_phrases_regex")
for phrase_regex in undesired_phrases:
title_text = re.sub(phrase_regex, "", title_text, flags=re.IGNORECASE).strip()
title_text = re.sub(r"\s+", " ", title_text).strip()
return title_text
GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/ 0000775 0000000 0000000 00000000000 15107027250 0023530 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0025627 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/cache_utils.py 0000664 0000000 0000000 00000006141 15107027250 0026367 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/utils/cache_utils.py
import os
import json
import hashlib
import re
from urllib.parse import urlparse
from typing import Dict, Any
from bs4 import BeautifulSoup
from loguru import logger
def get_url_hash(url: str) -> str:
"""Generates a SHA256 hash of a URL for caching."""
return hashlib.sha256(url.encode("utf-8")).hexdigest()
def load_json_file(file_path: str) -> Dict[str, Any]:
"""Loads a JSON file, with error handling."""
if not os.path.exists(file_path):
logger.debug(
f"Cache Utils: File not found: '{file_path}'. Returning empty dict."
)
return {}
try:
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
logger.debug(f"Cache Utils: JSON file loaded: '{file_path}'.")
return data
except json.JSONDecodeError as e:
logger.error(f"Cache Utils: Error decoding JSON '{file_path}': {e}.")
return {}
except Exception as e:
logger.error(f"Cache Utils: Unexpected error loading JSON '{file_path}': {e}.")
return {}
def save_json_file(data: Dict[str, Any], file_path: str):
"""Saves data to a JSON file, with error handling."""
try:
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
logger.info(f"Cache Utils: Data saved to '{file_path}'.")
except Exception as e:
logger.error(f"Cache Utils: Error saving data to '{file_path}': {e}")
def save_preprocessed_html(
soup: BeautifulSoup,
source_identifier: str,
output_dir: str = "preprocessed_html_output",
):
"""
Saves a BeautifulSoup object as an HTML file with a unique, safe filename.
The filename is based on the URL or another identifier.
"""
if not soup:
logger.warning("Cache Utils: Cannot save empty HTML.")
return
# Generate a safe and unique filename
if source_identifier.startswith("http"):
parsed_url = urlparse(source_identifier)
# Remove disallowed characters for filenames
safe_name = re.sub(
r"[^a-zA-Z0-9_\-.]", "_", parsed_url.netloc + parsed_url.path
)
# Limit the filename length to avoid OS limits
if len(safe_name) > 150:
safe_name = safe_name[:150]
# Add a short hash to ensure uniqueness
url_hash = hashlib.md5(source_identifier.encode("utf-8")).hexdigest()[:8]
file_name = f"{safe_name}_{url_hash}.html"
else:
file_name = f"{source_identifier}.html"
output_path = os.path.join(output_dir, file_name)
try:
os.makedirs(output_dir, exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
f.write(str(soup))
logger.info(f"Cache Utils: Preprocessed HTML saved in '{output_path}'.")
except IOError as e:
logger.error(f"Cache Utils: Error writing to file '{output_path}': {e}")
except Exception as e:
logger.error(
f"Cache Utils: Unexpected error saving HTML to '{output_path}': {e}"
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/data_utils.py 0000664 0000000 0000000 00000011704 15107027250 0026236 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/utils/data_utils.py
import re
import unicodedata
from typing import Union, Optional
from loguru import logger
from app.utils.pattern_manager import pattern_manager
def clean_price_text(price_str: Optional[Union[str, float]]) -> Optional[float]:
"""
Cleans and converts a price string to a float.
Supports various European notations (with period or comma as decimal separator).
"""
if price_str is None:
return None
if isinstance(price_str, (int, float)):
return float(price_str)
s = unicodedata.normalize("NFKC", str(price_str)).strip()
s = re.sub(r"[^\d.,]", "", s)
if "," in s:
if "." in s and s.rfind(",") > s.rfind("."):
# Format: 1.234,56
s = s.replace(".", "").replace(",", ".")
else:
# Format: 1,234.56 (or 1234,56)
s = s.replace(",", ".")
try:
return float(s)
except ValueError:
logger.debug(f"Data Utils: Could not parse price '{price_str}' to a float.")
return None
def normalize_ean_candidate(ean_candidate: str) -> str:
"""
Normalizes an EAN candidate by removing all non-numeric characters
and padding to standard GTIN lengths if necessary.
"""
if not isinstance(ean_candidate, str):
ean_candidate = str(ean_candidate)
ean_candidate = re.sub(r"\D", "", ean_candidate).strip()
length = len(ean_candidate)
if length == 12:
return "0" + ean_candidate
elif 8 < length < 13:
return ean_candidate.zfill(13)
elif 13 < length < 14:
return ean_candidate.zfill(14)
else:
return ean_candidate
def is_valid_ean_checksum(ean_candidate: str) -> bool:
"""
Validates an EAN (GTIN-8, -12, -13, -14) using the checksum calculation.
The input is assumed to be already normalized.
"""
ean_candidate = ean_candidate.strip()
length = len(ean_candidate)
if length not in [8, 12, 13, 14] or not ean_candidate.isdigit():
return False
digits = [int(d) for d in ean_candidate]
check_digit = digits[-1]
total_sum = 0
for i in range(length - 1):
digit = digits[length - 2 - i]
total_sum += digit * (3 if (i % 2) == 0 else 1)
calculated_check_digit = (10 - (total_sum % 10)) % 10
return calculated_check_digit == check_digit
def normalize_availability(availability_str: Optional[str]) -> Optional[str]:
"""
Normalizes availability strings to a standard format ("In Stock", etc.)
by checking against keywords from the PatternManager.
"""
if not availability_str:
return None
lower_val = availability_str.lower()
if "/" in lower_val:
# Handle strings like "Discontinued/Out of Stock"
lower_val = lower_val.split("/")[-1]
# 1. Get the lists from the pattern manager.
# These lists are already merged (e.g., default + en + nl)
in_stock_words = pattern_manager.get_keyword_list("availability_in_stock")
out_of_stock_words = pattern_manager.get_keyword_list("availability_out_of_stock")
pre_order_words = pattern_manager.get_keyword_list("availability_pre_order")
# 2. Check against the lists
# We use 'any' to check if any keyword is present in the string
if any(word in lower_val for word in in_stock_words):
return "In Stock"
if any(word in lower_val for word in out_of_stock_words):
return "Out of Stock"
if any(word in lower_val for word in pre_order_words):
return "Pre-order"
# Return the original string if no match is found
return availability_str
def clean_text_and_remove_unicode(text: str) -> str:
"""
Cleans text by normalizing unicode, removing invisible format characters,
collapsing all whitespace, stripping, and removing trailing
non-alphanumeric characters.
"""
if not isinstance(text, str):
return ""
# 1. Normalize unicode (NFKC)
# This handles composite characters and compatibility variants.
try:
text = unicodedata.normalize("NFKC", text)
except Exception:
# Fallback for any unexpected normalization errors
pass
# 2. Remove Unicode "format" characters (like \u200f RLM)
# These are invisible and cause issues. 'Cf' is the category.
# We leave 'Cc' (control chars like \n, \t) and 'Zs' (space separators like \xa0)
# for the regex (\s+) to handle.
text = "".join(c for c in text if unicodedata.category(c) != "Cf")
# 3. Collapse all whitespace characters (\n, \t, \xa0, ' ', etc.) to a single space
# This is an improvement on the original, as \s+ handles all of them.
text = re.sub(r"\s+", " ", text)
# 4. Trim on space (removes leading/trailing spaces)
text = text.strip()
# 5. Remove last non-alphanumeric characters on the right
# [^a-zA-Z0-9] matches anything NOT a letter or number.
# + matches one or more of them.
# $ anchors the match to the end of the string.
# text = re.sub(r"[^a-zA-Z0-9]+$", "", text)
return text
GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/hf_utils.py 0000664 0000000 0000000 00000003300 15107027250 0025713 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/utils/hf_utils.py
import torch
from transformers import pipeline
from typing import Any, Optional
import threading
from loguru import logger
_hf_classifier: Optional[Any] = None
_hf_lock = threading.Lock()
def get_hf_classifier(model_name: str) -> Optional[Any]:
"""
Loads and caches a Hugging Face zero-shot classification pipeline.
"""
global _hf_classifier
if _hf_classifier is not None:
return _hf_classifier
with _hf_lock:
if _hf_classifier is not None:
return _hf_classifier
logger.info(f"HF_Utils: Loading Hugging Face model '{model_name}'...")
try:
device = 0 if torch.cuda.is_available() else -1
if device != -1:
logger.info(f"HF_Utils: GPU (CUDA) detected. Using device {device}.")
else:
logger.warning("HF_Utils: No GPU (CUDA) detected. Falling back to CPU.")
_hf_classifier = pipeline(
"zero-shot-classification", model=model_name, device=device
)
logger.info(
f"HF_Utils: Hugging Face model '{model_name}' loaded successfully."
)
except Exception as e:
logger.error(
f"HF_Utils: Error loading Hugging Face model '{model_name}': {e}",
exc_info=True,
)
_hf_classifier = None
return _hf_classifier
def unload_hf_classifier():
"""Unloads the Hugging Face classifier to free up memory."""
global _hf_classifier
with _hf_lock:
if _hf_classifier is not None:
logger.info("HF_Utils: Unloading Hugging Face classifier.")
_hf_classifier = None
GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/html_processor.py 0000664 0000000 0000000 00000006173 15107027250 0027154 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/utils/html_processor.py
from bs4 import BeautifulSoup, Comment
from loguru import logger
from app.config import settings
import re
def clean_html_for_extraction(raw_html_content: str) -> BeautifulSoup:
"""
Preprocesses HTML by removing noise and normalizing whitespace,
without polluting the logic with I/O operations.
"""
if not raw_html_content:
logger.warning("HTML Processor: Received empty HTML content.")
return BeautifulSoup("", "lxml")
soup = BeautifulSoup(raw_html_content, "lxml")
logger.info("HTML Processor: Starting HTML cleanup.")
# Step 1: Remove known noise tags
for tag_name in [
"script",
"style",
"noscript",
"link",
"template",
"svg",
"iframe",
"button",
]:
for element in soup.find_all(tag_name):
element.decompose()
# Step 2: Remove HTML comments
for comment in soup.find_all(string=lambda text: isinstance(text, Comment)):
comment.extract()
logger.debug("HTML Processor: All HTML comments removed.")
# Step 3: Remove noise sections based on selectors, including the function selector
for selector in settings.html_preprocessing.noise_selectors:
try:
if isinstance(selector, str):
for element in soup.select(selector):
logger.debug(
f"HTML Processor: Removing element by selector '{selector}'."
)
element.decompose()
elif callable(selector):
# If it's a function, call it to find the elements
for element in soup.find_all(selector):
logger.debug(
f"HTML Processor: Removing element by function '{selector}'."
)
element.decompose()
else:
logger.warning(
f"HTML Processor: Skipping invalid selector type: {type(selector)}."
)
except Exception as e:
logger.error(
f"HTML Processor: Error during removal with selector '{selector}': {e}",
exc_info=True,
)
# Step 4: Normalize whitespace and remove empty tags
if soup.body:
# Loop through the tags in reverse order, from inside out
for tag in soup.body.find_all(True, reverse=False):
if tag.name in ["img", "br", "hr", "input", "meta"]:
continue
# Use get_text() to get all nested text
raw_text = tag.get_text(strip=True)
# Remove all whitespace (spaces, newlines, tabs) and check if anything remains
if not re.sub(r"\s+", "", raw_text):
# The tag is empty, remove it unless it's an excluded tag
tag.decompose()
for tag in soup.find_all():
if (
not tag.contents
and not tag.get_text(strip=True)
and tag.name not in ["img", "br", "hr", "meta", "input"]
):
tag.decompose()
logger.info("HTML Processor: Cleanup complete.")
return soup
GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/html_utils.py 0000664 0000000 0000000 00000004204 15107027250 0026266 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/utils/html_utils.py
from bs4 import BeautifulSoup, Comment
from loguru import logger
from typing import List
def preprocess_html_for_extraction(
html_content: str, hidden_selectors: List[str]
) -> BeautifulSoup:
"""
Removes unnecessary tags and comments from HTML to simplify the extraction
of structured data. Returns a BeautifulSoup object.
"""
if not html_content:
return BeautifulSoup("", "lxml")
soup = BeautifulSoup(html_content, "lxml")
# Remove tags that cause noise
for selector in ["script", "style", "noscript", "meta", "link", "template"]:
for element in soup.find_all(selector):
element.decompose()
# Remove HTML comments
for comment in soup.find_all(string=lambda text: isinstance(text, Comment)):
comment.extract()
# Remove hidden elements based on CSS classes or attributes
for selector in hidden_selectors:
for element in soup.select(selector):
element.decompose()
logger.debug("HTML Utils: HTML successfully preprocessed for extraction.")
return soup
def preprocess_html_for_llm(
html_content: str, noise_selectors: List[str], max_chars: int
) -> str:
"""
Converts HTML to clean, truncated text for use with an LLM.
Removes an extensive list of selectors considered to be noise.
"""
if not html_content:
return ""
soup = BeautifulSoup(html_content, "lxml")
for selector in noise_selectors:
if callable(selector): # To handle lambda functions for comments
# This part of bs4 logic is a bit different for callables
for element in soup.find_all(string=selector):
element.extract()
else:
for element in soup.select(selector):
element.decompose()
clean_text = " ".join(soup.get_text(separator=" ", strip=True).split())
if len(clean_text) > max_chars:
logger.warning(
f"HTML Utils: Text truncated for LLM: from {len(clean_text)} to {max_chars} chars."
)
clean_text = clean_text[:max_chars] + " [TRUNCATED FOR LLM]"
return clean_text
GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/module_sorter.py 0000664 0000000 0000000 00000003140 15107027250 0026763 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/utils/module_sorter.py
from typing import Dict, List, Set
from collections import defaultdict
from loguru import logger
def topological_sort_modules(dependencies: Dict[str, List[str]]) -> List[str]:
"""
Sorts modules topologically based on their dependencies.
"""
graph: Dict[str, List[str]] = defaultdict(list)
in_degree: Dict[str, int] = defaultdict(int)
nodes: Set[str] = set()
# Build the graph and calculate the 'in-degree' of each node
for module, deps in dependencies.items():
nodes.add(module)
for dep in deps:
nodes.add(dep)
graph[dep].append(module)
in_degree[module] += 1
# Add all modules without dependencies to the queue
queue = [module for module in nodes if in_degree[module] == 0]
result: List[str] = []
while queue:
module = queue.pop(0)
result.append(module)
for dependent_module in graph[module]:
in_degree[dependent_module] -= 1
if in_degree[dependent_module] == 0:
queue.append(dependent_module)
if len(result) != len(nodes):
# There is a cycle in the dependencies
all_modules = sorted(list(nodes))
processed_modules = set(result)
unprocessed_modules = sorted(list(set(all_modules) - processed_modules))
error_message = (
f"Topological sort failed: A cycle exists in the dependencies. "
f"Unprocessed modules: {unprocessed_modules}"
)
logger.error(error_message)
raise ValueError(error_message)
return result
GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/nlp_utils.py 0000664 0000000 0000000 00000002663 15107027250 0026122 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/utils/nlp_utils.py
import spacy
from typing import Optional
import threading
from loguru import logger
SpacyModel = spacy.language.Language
_nlp_model: Optional[SpacyModel] = None
_nlp_lock = threading.Lock()
def get_nlp_model(model_name: str) -> Optional[SpacyModel]:
"""
Loads the specified spaCy model if needed and returns it.
"""
global _nlp_model
if _nlp_model is not None:
return _nlp_model
with _nlp_lock:
if _nlp_model is not None:
return _nlp_model
logger.info(f"NLP_Utils: Loading spaCy model '{model_name}'...")
try:
_nlp_model = spacy.load(model_name)
logger.info(f"NLP_Utils: spaCy model '{model_name}' loaded successfully.")
except OSError:
logger.error(
f"NLP_Utils: SpaCy model '{model_name}' not found. "
f"Run 'python -m spacy download {model_name}'."
)
_nlp_model = None
except Exception as e:
logger.error(
f"NLP_Utils: Unexpected error loading spaCy model: {e}", exc_info=True
)
_nlp_model = None
return _nlp_model
def unload_nlp_model():
"""Unloads the spaCy model to free up memory."""
global _nlp_model
with _nlp_lock:
if _nlp_model is not None:
logger.info("NLP_Utils: Unloading spaCy NLP model.")
_nlp_model = None
GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/pattern_manager.py 0000664 0000000 0000000 00000017447 15107027250 0027266 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/modules/pattern_manager.py
import yaml
import re
from pathlib import Path
from typing import Dict, Pattern, List, Any
from loguru import logger
from app.core.context import shared_context
from app.config import settings
# Define the new directory paths
CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config"
BASE_PATTERNS_DIR = CONFIG_DIR / "patterns"
CUSTOM_PATTERNS_DIR = BASE_PATTERNS_DIR / "custom"
DEFAULT_CONFIG_PATH = BASE_PATTERNS_DIR / "default.yml"
class PatternManager:
def __init__(self, base_dir: Path, custom_dir: Path, default_path: Path):
self._all_patterns: Dict[str, Dict[str, Any]] = {}
self._compiled_regex_cache: Dict[tuple, Pattern] = {}
self._compiled_list_cache: Dict[tuple, List[str]] = {}
self._load_all_patterns(base_dir, custom_dir, default_path)
def _load_all_patterns(self, base_dir: Path, custom_dir: Path, default_path: Path):
"""
Loads the default patterns, all base language-specific patterns,
and all custom language-specific patterns, handling overrides.
"""
# 1. Load default patterns (default.yml)
try:
with open(default_path, "r", encoding="utf-8") as f:
self._all_patterns["default"] = yaml.safe_load(f) or {}
logger.success(
f"PatternManager: Loaded default patterns from {default_path.name}."
)
except FileNotFoundError:
logger.error(
f"PatternManager FATAL: Default config file not found at {default_path}"
)
self._all_patterns["default"] = {}
except Exception as e:
logger.error(f"Error loading default patterns: {e}")
# 2. Discover all base and custom pattern files
base_files = {f.name: f for f in base_dir.glob("*_patterns.yml")}
custom_files = {f.name: f for f in custom_dir.glob("*_patterns.yml")}
# Get all unique file names from both directories
all_lang_file_names = set(base_files.keys()) | set(custom_files.keys())
loaded_langs_count = 0
# 3. Load language patterns with custom override logic
for filename in all_lang_file_names:
lang_code = filename.split("_")[0]
if not lang_code:
continue
file_to_load = None
source_type = ""
# Check for custom file first (override)
if filename in custom_files:
file_to_load = custom_files[filename]
source_type = "custom"
# Fallback to base file
elif filename in base_files:
file_to_load = base_files[filename]
source_type = "base"
if not file_to_load:
continue
# Load the chosen file
try:
with open(file_to_load, "r", encoding="utf-8") as f:
lang_patterns = yaml.safe_load(f) or {}
# Store patterns under the language code
self._all_patterns[lang_code] = lang_patterns
loaded_langs_count += 1
logger.debug(
f"PatternManager: Loaded language '{lang_code}' from {source_type} file: {file_to_load.name}"
)
except Exception as e:
logger.error(
f"Error loading language patterns from {file_to_load}: {e}"
)
logger.info(
f"PatternManager initialized with 'default' and {loaded_langs_count} language(s)."
)
def _get_languages_to_load(self) -> List[str]:
"""
Gets the list of language codes to use for this request.
It ALWAYS includes 'default' and the configured default language.
"""
# 1. Get the configured default language
default_lang = settings.language.default
# 2. ALWAYS start with 'default' and the configured default language
langs_to_load = ["default", default_lang]
# 3. Get the list from the analyzer (e.g., ['nl'] or ['en', 'nl'] or None)
requested_langs = shared_context.get("requested_langs")
if isinstance(requested_langs, list):
for lang in requested_langs:
if lang in self._all_patterns:
# Add the language if we have a file for it
if lang not in langs_to_load:
langs_to_load.append(lang)
else:
# Log a warning if a requested lang (from API or langdetect)
# does not have a matching pattern file.
logger.warning(
f"PatternManager: Requested language '{lang}' has no matching "
f"'{lang}_patterns.yml' file in base or custom directories."
)
# 4. De-duplicate the list (e.g., if default='en' and request=None, list is ['default', 'en'])
# (e.g., if default='en' and request=['en', 'nl'], list is ['default', 'en', 'nl'])
final_list = list(dict.fromkeys(langs_to_load))
logger.debug(f"PatternManager: Loading patterns for: {final_list}")
return final_list
def get_keyword_list(self, pattern_name: str) -> List[str]:
"""
Gets a combined list of keywords for all active languages (request-specific).
"""
langs = self._get_languages_to_load()
cache_key = (tuple(sorted(langs)), pattern_name)
if cache_key in self._compiled_list_cache:
return self._compiled_list_cache[cache_key]
combined_list = []
for lang in langs:
lang_patterns = self._all_patterns.get(lang, {})
keywords = lang_patterns.get(pattern_name, [])
if isinstance(keywords, list):
combined_list.extend(keywords)
elif keywords:
logger.warning(
f"PatternManager: Pattern '{pattern_name}' for lang '{lang}' is not a list."
)
# De-duplicate the list
final_list = list(dict.fromkeys(combined_list))
self._compiled_list_cache[cache_key] = final_list
return final_list
def get_compiled_regex(self, pattern_name: str) -> Pattern:
"""
Gets a combined, compiled regex pattern for all active languages (request-specific).
"""
langs = self._get_languages_to_load()
cache_key = (tuple(sorted(langs)), pattern_name)
if cache_key in self._compiled_regex_cache:
return self._compiled_regex_cache[cache_key]
combined_pattern_parts = []
for lang in langs:
lang_patterns = self._all_patterns.get(lang, {})
pattern_str = lang_patterns.get(pattern_name)
if pattern_str and isinstance(pattern_str, str):
combined_pattern_parts.append(
f"({pattern_str})"
) # Group each lang's pattern
elif pattern_str:
logger.warning(
f"PatternManager: Pattern '{pattern_name}' for lang '{lang}' is not a string."
)
if not combined_pattern_parts:
logger.warning(
f"No regex pattern found for '{pattern_name}' in active langs {langs}. This regex will not match anything."
)
# Return a regex that never matches
return re.compile(r"a^")
# Join all patterns with an OR operator
final_pattern_str = "|".join(combined_pattern_parts)
compiled_regex = re.compile(final_pattern_str, re.IGNORECASE)
self._compiled_regex_cache[cache_key] = compiled_regex
return compiled_regex
# Create a single global instance with the new directory structure
pattern_manager = PatternManager(
base_dir=BASE_PATTERNS_DIR,
custom_dir=CUSTOM_PATTERNS_DIR,
default_path=DEFAULT_CONFIG_PATH,
)
GetArgusFlow-argus-c6afc1d/services/extractor/app/utils/shared_resources.py 0000664 0000000 0000000 00000002126 15107027250 0027443 0 ustar 00root root 0000000 0000000 # argus/services/extractor/app/utils/shared_resources.py
import threading
from typing import Dict, Any
from loguru import logger
from app.utils.nlp_utils import get_nlp_model, unload_nlp_model
from app.config import settings
_shared_resources: Dict[str, Any] = {}
_resources_lock = threading.Lock()
def get_resources() -> Dict[str, Any]:
global _shared_resources
if _shared_resources:
return _shared_resources
with _resources_lock:
if _shared_resources:
return _shared_resources
logger.info("Shared Resources: Loading all global resources...")
_shared_resources["nlp_model"] = get_nlp_model(settings.models.nlp)
logger.info("Shared Resources: Global resources loaded.")
return _shared_resources
def unload_resources():
global _shared_resources
with _resources_lock:
if not _shared_resources:
return
logger.info("Shared Resources: Unloading all global resources...")
unload_nlp_model()
_shared_resources.clear()
logger.info("Shared Resources: Global resources unloaded.")
GetArgusFlow-argus-c6afc1d/services/extractor/config/ 0000775 0000000 0000000 00000000000 15107027250 0023055 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/config/config.yml 0000664 0000000 0000000 00000004076 15107027250 0025054 0 ustar 00root root 0000000 0000000 #--------------------------------------------------------------------------
# Main Configuration for the Extractor Service
#--------------------------------------------------------------------------
# Service-level settings
service:
host: "0.0.0.0"
port: 8001
log_level: "info"
log_dir: "logs"
environment: "production"
llm_parser_url: "http://argus-llm-parser:8002/api/v1/parse"
auth:
api_key: "default_dev_key"
models:
# The name of the spaCy model to download and use
nlp: "nl_core_news_lg"
default_spec_category: "General Specifications"
language:
default: "en"
field_aliases:
brand:
- "merk"
- "fabrikant"
- "manufacturer"
- "producent"
- "leverancier"
- "maker"
- "vendor"
- "merknaam"
- "product_brand"
ean:
- "ean"
- "gtin"
- "streepjescode"
- "barcode"
sku:
- "sku"
- "artikelnummer"
- "productcode"
- "partnummer"
- "modelnummer"
- "itemnummer"
- "asin"
- "mpn"
# HTML preprocessing settings
html_preprocessing:
# CSS selectors to identify and remove noise from the HTML before analysis
noise_selectors:
- 'footer'
- 'aside'
- '#main-footer'
- '#sidebar'
- '.sidebar'
- '.site-info'
- '.copyright'
- '.related-products'
- '.upsells'
- '.cross-sells'
- '.recommendations'
- '.also-bought'
- '.similar-items'
- '.other-products'
- '.widget-area'
- '.promo-banner'
- '.newsletter-signup'
- '.cookie-consent'
- '.social-share-buttons'
- '.back-to-top'
- '[class*="breadcrumb"]'
- '[itemtype="https://schema.org/BreadcrumbList"]'
- '[role="complementary"]'
- '.review-form'
- '.review-pagination'
- '.comment-form'
- '.comment-navigation'
- '#comments'
- '.comments-area'
- '.customer-reviews__header'
- '.pagination'
- '.pager'
- '.ad'
- '.advertisement'
- '.modal'
- '.overlay'
- '.hidden'
- '.visuallyhidden'
- '.print-only'
- '[aria-hidden="true"]'
- '[style*="display: none"]'
- '[style*="visibility: hidden"]'
GetArgusFlow-argus-c6afc1d/services/extractor/config/patterns/ 0000775 0000000 0000000 00000000000 15107027250 0024715 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/config/patterns/default.yml 0000664 0000000 0000000 00000001770 15107027250 0027071 0 ustar 00root root 0000000 0000000 #--------------------------------------------------------------------------
# Default patterns for Argus Extractor.
# These patterns are always loaded as the base and are language-agnostic.
#--------------------------------------------------------------------------
#------------------------------------------------------------------------
# Pro: Original Price
#------------------------------------------------------------------------
screenreader_class_regex: 'screenreader-only|sr-only'
price_number_format_regex: '\d{1,3}(?:[.,]\d{3})*(?:[.,]\d{1,2})?'
#------------------------------------------------------------------------
# Pro: SKU
#------------------------------------------------------------------------
sku_format_regex: '[a-zA-Z0-9\-\_\.]+'
asin_format_regex: '\b(B0[0-9A-Z]{8})\b'
#------------------------------------------------------------------------
# Pro: Specifications
#------------------------------------------------------------------------
kv_separator_regex: '\s*:\s*|\s+-\s*|\s+–\s*' GetArgusFlow-argus-c6afc1d/services/extractor/config/patterns/en_patterns.yml 0000664 0000000 0000000 00000011072 15107027250 0027763 0 ustar 00root root 0000000 0000000 #--------------------------------------------------------------------------
# English (en) patterns for Argus Extractor.
#--------------------------------------------------------------------------
# Availability
availability_in_stock:
- 'in stock'
- 'available'
- 'add to cart'
- 'buy now'
- 'add to basket'
- 'available from stock'
availability_out_of_stock:
- 'out of stock'
- 'not available'
- 'sold out'
- 'not in stock'
- 'product unavailable'
- 'temporarily unavailable'
availability_pre_order:
- 'pre-order'
- 'preorder'
- 'expected'
- 'coming soon'
- 'delivery from'
- 'reserve now'
availability_class_regex: 'availability|stock|status|add-to-cart-button|buy-button|product-status|stock-status|stock__status|qty-status|buy-now|in-stock|out-of-stock|pre-order-button'
# Brand
brand_label_regex: '\b(brand|manufacturer|vendor)\b'
brand_class_regex: 'brand|manufacturer|vendor|by-brand|item-brand|product-brand'
generic_brand_text_regex: 'all\s*brands|view\s*all|store|website|shop\s*by\s*brand|filter|list|nav|menu|category'
# Breadcrumbs
breadcrumb_filter_keywords:
- 'home'
- 'start'
- 'back'
- 'products'
- '«'
- '»'
- '>'
- '/'
- '>'
- '<'
- ''
breadcrumb_container_regex: 'breadcrumb|crumb|path'
# Description
description_filter_regex: 'Total price:\s*.*|Add all three items to wishlist|Try again\.|Details Added to cart|Add all \d+ to cart|Product details|Technical Details|Additional information'
# Price
price_section_class_regex: 'product-info|price-section|buy-box|add-to-cart-section'
price_format_regex: '(?:$|USD)\s*(\d{1,3}(?:[,\.]\d{3})*(?:[,\.]\d{2})?)\b|\b(\d{1,3}(?:[,\.]\d{3})*(?:[,\.]\d{2})?)\s*(?:usd|dollar)\b'
currency_format_regex: '(\$|USD|dollar|£|GBP)'
# Title
title_clean_patterns_regex:
- '\s*[:|]?\s*(?:Amazon\.(?:com|co\.uk)|BestBuy|Walmart|Target)[:|\s].*$'
- '\s*[:|]\s*(?:Electronics|Books|Clothing|Home & Garden)\s*$'
- '\s*(\s?-\s?.*)?\s*\|.*$'
title_filter_phrases_regex:
- 'Official website of'
- 'Compare prices at'
- 'Buy now'
- 'Best offers'
- 'Newest model'
- 'Free shipping'
- 'Product details'
#------------------------------------------------------------------------
# Pro: Original Price
#------------------------------------------------------------------------
original_price_class_regex: '(?:old|original|list|was|compare|regular|normal|promo)[_-]?price|price[_-]?(?:old|original|list|was|compare|regular|normal|promo)'
current_price_class_regex: '(?:current|sale|discount|special|final)[_-]?price|price[_-]?(?:current|sale|discount|special|final)'
original_price_keywords_regex: 'from|original|list price|before|old price|was'
current_price_keywords_regex: '\b(?:now|sale|discount|special|final|current)\b'
#------------------------------------------------------------------------
# Pro: SKU
#------------------------------------------------------------------------
sku_class_regex: 'sku|product-code|item-number|model|part-no|product-id|mpn|asin'
sku_label_regex: 'SKU|Article Number|Product Code|Item Number|Model Number|Part Number|ID|MPN|ASIN'
#------------------------------------------------------------------------
# Pro: Specifications
#------------------------------------------------------------------------
spec_concept_keywords:
- "specifications"
- "specs"
- "properties"
- "characteristics"
- "attributes"
- "nutrition"
- "nutritional values"
- "ingredients"
filter_class_keywords:
- "nav"
- "menu"
- "sidebar"
- "footer"
- "category"
- "breadcrumb"
common_non_spec_keywords:
- "url"
- "return"
- "contact"
- "disclaimer"
- "copyright"
- "order"
- "cart"
- "choose"
- "select"
- "more information"
- "read more"
- "view all"
- "view"
- "shipping"
- "product description"
- "overview"
- "user manual"
- "manual"
- "download"
- "documents"
- "login"
- "sign in"
- "view more"
- "show details"
- "hide details"
- "expand"
- "collapse"
- "description"
- "information"
- "product information"
- "content"
- "navigation"
- "footer"
- "header"
- "gift"
- "tooltip"
- "customer service"
- "reviews"
- "questions"
- "faq"
- "what's in the box"
- "related products"
- "accessories"
- "offers"
- "product advantages"
- "highlights"
- "key features"
- "item"
- "total"
- "price"
- "total price"
- "payment"
- "discount"
- "offer"
- "pros"
- "cons"
- "advantages"
- "disadvantages"
- "rating"
- "review"
- "customer review"
- "summary"
- "warranty"
irrelevant_spec_keys:
- "date first available"
- "customer reviews"
- "guaranteed software updates until" GetArgusFlow-argus-c6afc1d/services/extractor/config/patterns/nl_patterns.yml 0000664 0000000 0000000 00000011362 15107027250 0027774 0 ustar 00root root 0000000 0000000 #--------------------------------------------------------------------------
# Dutch (nl) patterns for Argus Extractor.
#--------------------------------------------------------------------------
# Availability
availability_in_stock:
- 'op voorraad'
- 'direct leverbaar'
- 'leverbaar'
- 'beschikbaar'
- 'in winkelmand'
- 'nu kopen'
- 'aan winkelwagen toevoegen'
- 'uit voorraad leverbaar'
availability_out_of_stock:
- 'uitverkocht'
- 'niet leverbaar'
- 'tijdelijk niet op voorraad'
- 'niet beschikbaar'
- 'geen voorraad'
- 'niet meer leverbaar'
availability_pre_order:
- 'pre-order'
- 'verwacht'
- 'binnenkort leverbaar'
- 'voorraad verwacht'
- 'levering vanaf'
- 'verwachte leverdatum'
- 'reserveer nu'
availability_class_regex: 'beschikbaarheid|voorraad|status|koopknop|levertijd|in-voorraad|uit-voorraad'
# Brand
brand_label_regex: '\b(merk|fabrikant|producent|leverancier)\b'
brand_class_regex: 'merknaam|product__supplier|fabrikant'
generic_brand_text_regex: 'alle\s*merken|bekijk\s*alle|winkel|website|shop\s*per\s*merk'
# Breadcrumbs
breadcrumb_filter_keywords:
- 'terug'
- 'producten'
# Description
description_filter_regex: 'Totaalprijs:\s*€.*|Alle drie de items op verlanglijst zetten|Probeer het nog eens\.|Gegevens In winkelwagen geplaatst|Alle \d+ aan winkelwagen toevoegen|Productgegevens|Technische Details|Aanvullende informatie'
# Price
price_section_class_regex: 'product-informatie|prijs-sectie|koopblok|winkelwagen-sectie'
price_format_regex: '(?:€|EUR)\s*(\d{1,3}(?:[\.,]\d{3})*(?:[\.,]\d{2})?)\b|\b(\d{1,3}(?:[\.,]\d{3})*(?:[\.,]\d{2})?)\s*(?:eur|euro)\b'
currency_format_regex: '(€|EUR|euro)'
# Title
title_clean_patterns_regex:
- '\s*[:|]?\s*(?:Amazon\.(?:nl|de)|bol\.com|coolblue\.nl|mediamarkt\.nl|Kieskeurig)[:|\s].*$'
- '\s*[:|]\s*(?:Elektronica|Boeken|Kleding|Huis & Tuin)\s*$'
title_filter_phrases_regex:
- 'Officiële website van'
- 'Vergelijk prijzen bij'
- 'Koop nu'
- 'Beste aanbiedingen'
- 'Nieuwste model'
- 'Gratis verzending'
- 'Product details'
#------------------------------------------------------------------------
# Pro: Original Price
#------------------------------------------------------------------------
original_price_class_regex: '(?:oud|origineel|lijst|was|vergelijk|regulier|normaal|promo)[_-]?prijs|prijs[_-]?(?:oud|origineel|lijst|was|vergelijk|regulier|normaal|promo)'
current_price_class_regex: '(?:huidig|verkoop|korting|speciaal|definitief)[_-]?prijs|prijs[_-]?(?:huidig|verkoop|korting|speciaal|definitief)'
original_price_keywords_regex: 'van|oorspronkelijk|adviesprijs|voor|oude prijs|was'
current_price_keywords_regex: '\b(?:nu|aanbieding|actie|korting|promo|deal)\b'
#------------------------------------------------------------------------
# Pro: SKU
#------------------------------------------------------------------------
sku_class_regex: 'sku|product-code|item-nummer|model|part-no|product-id|mpn|asin|artikelnummer|productcode'
sku_label_regex: 'SKU|Artikelnummer|Productcode|Itemnummer|Modelnummer|Partnummer|ID|MPN|ASIN'
#------------------------------------------------------------------------
# Pro: Specifications
#------------------------------------------------------------------------
spec_concept_keywords:
- "specificaties"
- "eigenschappen"
- "details"
- "kenmerken"
- "voedingswaarden"
- "ingrediënten"
filter_class_keywords:
- "nav"
- "menu"
- "sidebar"
- "footer"
- "category"
- "breadcrumb"
- "navigatie"
- "voettekst"
- "categorie"
- "broodkruimel"
common_non_spec_keywords:
- "url"
- "afbeelding"
- "retour"
- "contact"
- "algemene voorwaarden"
- "disclaimer"
- "copyright"
- "bestellen"
- "winkelwagen"
- "kies"
- "selecteer"
- "besteleenheid"
- "meer informatie"
- "lees meer"
- "bekijk alle"
- "bekijk"
- "verzenden"
- "productomschrijving"
- "overzicht"
- "gebruikershandleiding"
- "handleiding"
- "download"
- "documenten"
- "inloggen"
- "meld je aan"
- "bekijk meer"
- "toon details"
- "toon geen details"
- "omschrijving"
- "informatie"
- "productinformatie"
- "navigatie"
- "voettekst"
- "kop"
- "cadeau"
- "tooltip"
- "klantenservice"
- "reviews"
- "beoordelingen"
- "verzending"
- "vragen"
- "faq"
- "in de doos"
- "gerelateerde producten"
- "accessoires"
- "aanbiedingen"
- "product voordelen"
- "highlights"
- "belangrijkste kenmerken"
- "algemeen"
- "item"
- "totaal"
- "prijs"
- "totaalprijs"
- "betaling"
- "korting"
- "aanbieding"
- "pluspunten"
- "minpunten"
- "voordelen"
- "nadelen"
- "beoordeling"
- "review"
- "klantbeoordeling"
- "samenvatting"
- "garantie"
- "recensie"
irrelevant_spec_keys:
- "datum eerste beschikbaarheid"
- "klantenrecensies"
- "gegarandeerde software-updates tot" GetArgusFlow-argus-c6afc1d/services/extractor/docker-compose.prod.yml 0000664 0000000 0000000 00000000213 15107027250 0026204 0 ustar 00root root 0000000 0000000 # ArgusProject/argus/services/extractor/docker-compose.prod.yml
services:
extractor:
build:
target: production
volumes: [] GetArgusFlow-argus-c6afc1d/services/extractor/docker-compose.test.yml 0000664 0000000 0000000 00000000221 15107027250 0026216 0 ustar 00root root 0000000 0000000 # argus/services/extractor/docker-compose.test.yml
services:
extractor:
command: python -m uvicorn app.main:app --host 0.0.0.0 --port 8001 GetArgusFlow-argus-c6afc1d/services/extractor/docker-compose.yml 0000664 0000000 0000000 00000000753 15107027250 0025252 0 ustar 00root root 0000000 0000000 # argus/services/extractor/docker-compose.yml
services:
extractor:
build:
context: ../..
dockerfile: services/extractor/Dockerfile
target: development
container_name: argus-extractor
ports:
- "8001:8001"
- "5678:5678"
volumes:
- ./app:/app/app
- ../../logs:/app/logs
- ../../tools:/app/tools
working_dir: /app
restart: unless-stopped
networks:
- argus_network
networks:
argus_network:
external: true GetArgusFlow-argus-c6afc1d/services/extractor/requirements-dev.txt 0000664 0000000 0000000 00000000550 15107027250 0025650 0 ustar 00root root 0000000 0000000 # ArgusProject/argus/services/extractor/requirements-dev.txt
# Include production requirements
-r requirements.txt
# Development Tools
deepdiff # For tests results
ruff # For linting and formatting
pytest # For running tests
httpx # For making requests in tests
yq # For parsing YAML in the Makefile during the Docker build
debugpy>=1.8.0 GetArgusFlow-argus-c6afc1d/services/extractor/requirements.txt 0000664 0000000 0000000 00000000451 15107027250 0025074 0 ustar 00root root 0000000 0000000 # ArgusProject/argus/services/extractor/requirements.txt
# Core Application
fastapi
uvicorn[standard]
loguru
pydantic-settings
PyYAML
# HTTP Client
httpx
playwright
# HTML & Data Processing
beautifulsoup4
lxml
fuzzywuzzy[speedup]
tenacity
# Language detection
langdetect
# NLP Libraries
spacy GetArgusFlow-argus-c6afc1d/services/extractor/tests/ 0000775 0000000 0000000 00000000000 15107027250 0022752 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/tests/__init__.py 0000664 0000000 0000000 00000000000 15107027250 0025051 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/tests/test_cases/ 0000775 0000000 0000000 00000000000 15107027250 0025107 5 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/tests/test_cases/01_simple_product/ 0000775 0000000 0000000 00000000000 15107027250 0030440 5 ustar 00root root 0000000 0000000 expected_output.json 0000664 0000000 0000000 00000000507 15107027250 0034477 0 ustar 00root root 0000000 0000000 GetArgusFlow-argus-c6afc1d/services/extractor/tests/test_cases/01_simple_product {
"availability": null,
"brand": "BrandCo",
"breadcrumbs": null,
"description": "The best widget you will ever buy. High quality and durable.",
"image": null,
"json_ld": null,
"open_graph": {
"title": "Awesome Test Widget",
"price:amount": "99.95"
},
"price": 99.95,
"title": "Awesome Test Widget"
} GetArgusFlow-argus-c6afc1d/services/extractor/tests/test_cases/01_simple_product/input.html 0000664 0000000 0000000 00000000760 15107027250 0032470 0 ustar 00root root 0000000 0000000 <!DOCTYPE html>
<html lang="en">
<head>
<title>Test Product Page
Awesome Test Widget BrandCo
€99,95
BrandCo