diff --git a/.github/workflows/build-embedding-image.yml b/.github/workflows/build-embedding-image.yml new file mode 100644 index 0000000..1d3a892 --- /dev/null +++ b/.github/workflows/build-embedding-image.yml @@ -0,0 +1,68 @@ +name: Build embedding image + +on: + push: + branches: [ main, Phins-branch ] + pull_request: + branches: [ main ] + workflow_dispatch: + +jobs: + build: + runs-on: ubuntu-latest + env: + IMAGE_NAME: ghcr.io/${{ github.repository_owner }}/lexcam-embedding-service + MODEL_NAME: intfloat/multilingual-e5-small + MODEL_URL: https://github.com/${{ github.repository }}/releases/download/${{ vars.EMBEDDING_MODEL_RELEASE_TAG }}/model.onnx + MODEL_SHA256: ${{ vars.EMBEDDING_MODEL_SHA256 }} + + steps: + - name: Debug workflow environment + run: | + echo "GITHUB_REPOSITORY=$GITHUB_REPOSITORY" + echo "GITHUB_REPOSITORY_OWNER=$GITHUB_REPOSITORY_OWNER" + echo "IMAGE_NAME=$IMAGE_NAME" + echo "MODEL_NAME=$MODEL_NAME" + echo "MODEL_URL=$MODEL_URL" + echo "MODEL_SHA256=$MODEL_SHA256" + + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: '3.11' + + - name: Build Docker image + run: | + docker build \ + --tag $IMAGE_NAME:${{ github.sha }} \ + --build-arg MODEL_NAME=$MODEL_NAME \ + --build-arg MODEL_URL=$MODEL_URL \ + --build-arg MODEL_SHA256=$MODEL_SHA256 \ + services/embedding-service + + - name: Smoke test image + run: | + id=$(docker create --name tmp_test -p 8001:8000 $IMAGE_NAME:${{ github.sha }}) + docker start $id + sleep 5 + docker ps --filter "name=tmp_test" --format "{{.Names}} {{.Status}}" + docker exec tmp_test /bin/sh -c "curl -sS -f http://localhost:8000/api/v1/health || exit 1" + docker stop $id + docker rm $id + + - name: Publish to GHCR + if: github.event_name == 'push' + env: + CR_PAT: ${{ secrets.GHCR_PAT }} + run: | + if [ -z "$CR_PAT" ]; then + echo "GHCR_PAT not set; skipping publish" + exit 0 + fi + echo "$CR_PAT" | docker login ghcr.io -u ${{ github.repository_owner }} --password-stdin + docker tag $IMAGE_NAME:${{ github.sha }} $IMAGE_NAME:latest + docker push $IMAGE_NAME:${{ github.sha }} + docker push $IMAGE_NAME:latest diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..a3389f5 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,76 @@ +name: CI + +on: + push: + branches: [ main, Phins-branch ] + pull_request: + branches: [ main ] + +jobs: + test-and-build: + runs-on: ubuntu-latest + services: + postgres: + image: postgres:17-alpine + ports: ['5432:5432'] + env: + POSTGRES_USER: lexcam + POSTGRES_PASSWORD: lexcam_dev + POSTGRES_DB: postgres + options: >- + --health-cmd="pg_isready -U lexcam" --health-interval=10s --health-timeout=5s --health-retries=5 + redis: + image: redis:7-alpine + ports: ['6379:6379'] + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: '3.11' + + - name: Create service databases + env: + PGPASSWORD: lexcam_dev + run: | + psql -h localhost -U lexcam -d postgres -c "CREATE DATABASE lexcam_feedback;" || true + psql -h localhost -U lexcam -d postgres -c "CREATE DATABASE lexcam_notif;" || true + psql -h localhost -U lexcam -d postgres -c "CREATE DATABASE lexcam_scraping;" || true + + - name: Install dependencies, run migrations and tests (feedback) + working-directory: services/feedback-service + env: + DATABASE_URL: postgresql://lexcam:lexcam_dev@localhost:5432/lexcam_feedback + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + alembic -c alembic.ini upgrade head + pytest -q tests + + - name: Install dependencies, run migrations and tests (notification) + working-directory: services/notification-service + env: + DATABASE_URL: postgresql://lexcam:lexcam_dev@localhost:5432/lexcam_notif + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + alembic -c alembic.ini upgrade head + pytest -q tests + + - name: Install dependencies, run migrations and tests (scraper) + working-directory: services/scraper-service + env: + DATABASE_URL: postgresql://lexcam:lexcam_dev@localhost:5432/lexcam_scraping + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + alembic -c alembic.ini upgrade head + pytest -q tests + + - name: Build Docker images + run: | + docker build -t lexcam-feedback-service:ci services/feedback-service + docker build -t lexcam-notification-service:ci services/notification-service + docker build -t lexcam-scraper-service:ci services/scraper-service diff --git a/.gitignore b/.gitignore index d5edb14..9667a2b 100644 --- a/.gitignore +++ b/.gitignore @@ -53,5 +53,17 @@ coverage.xml # Docker *.tar +# Downloaded or generated embedding artifacts +services/embedding-service/model/*.onnx + # Helm charts/*.tgz + +# md files +build-plan.md +description.md +services-desccription.md + +# Local safe backups (onnx model artifacts) +_safe_backups/ + diff --git a/README.md b/README.md index c398468..6199700 100644 --- a/README.md +++ b/README.md @@ -1 +1,15 @@ # LexCamAI + +## Embedding model storage + +The embedding service now expects the ONNX model to live outside git, with a +GitHub Releases asset as the preferred source. + +Use a release tag such as `embedding-model-v1`, upload the model as +`model.onnx`, and set these repository variables for the build workflow: + +- `EMBEDDING_MODEL_RELEASE_TAG` +- `EMBEDDING_MODEL_SHA256` + +The image build downloads the asset from: +`https://github.com///releases/download//model.onnx` diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..a438f80 --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,8 @@ +from pathlib import Path + +# Shim package so pytest can import `app.main` from the embedding service +# when running from the repository root. +root = Path(__file__).resolve().parent +service_app = root.joinpath("..", "services", "embedding-service", "app").resolve() +if str(service_app) not in __path__: + __path__.insert(0, str(service_app)) \ No newline at end of file diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 9248996..b0d9581 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -59,9 +59,170 @@ services: volumes: - minio_data:/data + embedding-service: + build: + context: ./services/embedding-service + args: + - MODEL_NAME=intfloat/multilingual-e5-small + container_name: lexcam-embedding-service + environment: + SERVICE_NAME: embedding-service + API_PREFIX: /api/v1 + MODEL_NAME: intfloat/multilingual-e5-small + DEVICE: cpu + MAX_BATCH_SIZE: 64 + MAX_SEQ_LENGTH: 512 + NORMALIZE_EMBEDDINGS: "true" + LOG_LEVEL: INFO + REQUIRE_API_KEY: "false" + RATE_LIMIT_ENABLED: "true" + RATE_LIMIT: "120/minute" + ports: + - "8001:8000" + volumes: + - embedding_cache:/cache + + knowledge-base-service: + build: + context: ./services/knowledge-base-service + container_name: lexcam-knowledge-base-service + environment: + SERVICE_NAME: knowledge-base-service + API_PREFIX: /api/v1 + DATABASE_URL: postgresql+psycopg://lexcam:lexcam_dev@postgres:5432/lexcam_knowledge + QDRANT_URL: http://qdrant:6333 + EMBEDDING_SERVICE_URL: http://embedding-service:8000 + REDIS_URL: redis://redis:6379/0 + LOG_LEVEL: INFO + PLAIN_SUMMARY_CACHE_TTL_SECONDS: 604800 + MAX_SEARCH_RESULTS: 10 + ports: + - "8003:8000" + depends_on: + - postgres + - qdrant + - redis + - embedding-service + + rag-service: + build: + context: ./services/rag-service + container_name: lexcam-rag-service + environment: + SERVICE_NAME: rag-service + API_PREFIX: /api/v1 + KNOWLEDGE_BASE_URL: http://knowledge-base-service:8000 + EMBEDDING_SERVICE_URL: http://embedding-service:8000 + HF_API_KEY: ${HF_API_KEY} + HF_MODEL: ${HF_MODEL:-google/flan-t5-small} + LOG_LEVEL: INFO + ports: + - "8004:8000" + depends_on: + - knowledge-base-service + - embedding-service + + feedback-service: + build: + context: ./services/feedback-service + image: lexcam-feedback-service + container_name: lexcam-feedback-service + environment: + SERVICE_NAME: feedback-service + API_PREFIX: /v1 + DATABASE_URL: postgresql://lexcam:lexcam_dev@postgres:5432/lexcam_feedback + REDIS_URL: redis://redis:6379/0 + CELERY_BROKER_URL: amqp://lexcam:lexcam_dev@rabbitmq:5672/%2F + CELERY_RESULT_BACKEND: rpc:// + LOG_LEVEL: INFO + ports: + - "8010:8000" + depends_on: + - postgres + - redis + + notification-service: + build: + context: ./services/notification-service + image: lexcam-notification-service + container_name: lexcam-notification-service + environment: + SERVICE_NAME: notification-service + API_PREFIX: /v1 + DATABASE_URL: postgresql://lexcam:lexcam_dev@postgres:5432/lexcam_notif + REDIS_URL: redis://redis:6379/0 + CELERY_BROKER_URL: amqp://lexcam:lexcam_dev@rabbitmq:5672/%2F + CELERY_RESULT_BACKEND: rpc:// + SMTP_HOST: smtp.example.com + SMTP_PORT: 587 + SMTP_USER: smtp_user + SMTP_PASS: smtp_pass + LOG_LEVEL: INFO + ports: + - "8011:8000" + depends_on: + - postgres + - redis + + scraper-service: + build: + context: ./services/scraper-service + image: lexcam-scraper-service + container_name: lexcam-scraper-service + environment: + SERVICE_NAME: scraper-service + API_PREFIX: /v1 + DATABASE_URL: postgresql://lexcam:lexcam_dev@postgres:5432/lexcam_scraping + REDIS_URL: redis://redis:6379/0 + CELERY_BROKER_URL: amqp://lexcam:lexcam_dev@rabbitmq:5672/%2F + CELERY_RESULT_BACKEND: rpc:// + MINIO_URL: http://minio:9000 + MINIO_ACCESS_KEY: lexcam + MINIO_SECRET_KEY: lexcam_dev123 + LOG_LEVEL: INFO + ports: + - "8012:8000" + depends_on: + - postgres + - redis + + feedback-worker: + image: lexcam-feedback-service + container_name: lexcam-feedback-worker + depends_on: + - feedback-service + - redis + command: celery -A app.tasks.celery_app worker -l info -Q default + + notification-worker: + image: lexcam-notification-service + container_name: lexcam-notification-worker + depends_on: + - notification-service + - redis + command: celery -A app.tasks.celery_app worker -l info -Q default + + notification-consumer: + build: + context: ./services/notification-service + container_name: lexcam-notification-consumer + command: python -u -m app.events_consumer + depends_on: + - rabbitmq + - notification-service + + scraper-worker: + image: lexcam-scraper-service + container_name: lexcam-scraper-worker + depends_on: + - scraper-service + - redis + command: celery -A app.tasks.celery_app worker -l info -Q default + volumes: postgres_data: qdrant_data: redis_data: rabbitmq_data: minio_data: + embedding_cache: diff --git a/infrastructure/helm/embedding-service/Chart.yaml b/infrastructure/helm/embedding-service/Chart.yaml new file mode 100644 index 0000000..b76e214 --- /dev/null +++ b/infrastructure/helm/embedding-service/Chart.yaml @@ -0,0 +1,6 @@ +apiVersion: v2 +name: embedding-service +description: LexCam Embedding Service +type: application +version: 0.1.0 +appVersion: "0.1.0" diff --git a/infrastructure/helm/embedding-service/values.yaml b/infrastructure/helm/embedding-service/values.yaml new file mode 100644 index 0000000..4850c7c --- /dev/null +++ b/infrastructure/helm/embedding-service/values.yaml @@ -0,0 +1,45 @@ +replicaCount: 1 + +image: + repository: lexcam/embedding-service + tag: latest + pullPolicy: IfNotPresent + +service: + type: ClusterIP + port: 8000 + +resources: + requests: + cpu: 500m + memory: 1Gi + limits: + cpu: "1" + memory: 2Gi + +env: + SERVICE_NAME: embedding-service + API_PREFIX: /api/v1 + MODEL_NAME: intfloat/multilingual-e5-small + DEVICE: cpu + MAX_BATCH_SIZE: "64" + MAX_SEQ_LENGTH: "512" + NORMALIZE_EMBEDDINGS: "true" + LOG_LEVEL: INFO + REQUIRE_API_KEY: "true" + RATE_LIMIT_ENABLED: "true" + RATE_LIMIT: 120/minute + +secret: + API_KEY: change-me + +autoscaling: + enabled: false + minReplicas: 1 + maxReplicas: 2 + targetCPUUtilizationPercentage: 75 + +podAnnotations: {} + +nameOverride: "" +fullnameOverride: "" diff --git a/infrastructure/k8s/embedding-service/configmap.yaml b/infrastructure/k8s/embedding-service/configmap.yaml new file mode 100644 index 0000000..524c1d7 --- /dev/null +++ b/infrastructure/k8s/embedding-service/configmap.yaml @@ -0,0 +1,16 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: embedding-service-config +data: + SERVICE_NAME: embedding-service + API_PREFIX: /api/v1 + MODEL_NAME: intfloat/multilingual-e5-small + DEVICE: cpu + MAX_BATCH_SIZE: "64" + MAX_SEQ_LENGTH: "512" + NORMALIZE_EMBEDDINGS: "true" + LOG_LEVEL: INFO + REQUIRE_API_KEY: "true" + RATE_LIMIT_ENABLED: "true" + RATE_LIMIT: 120/minute diff --git a/infrastructure/k8s/embedding-service/deployment.yaml b/infrastructure/k8s/embedding-service/deployment.yaml new file mode 100644 index 0000000..feaa603 --- /dev/null +++ b/infrastructure/k8s/embedding-service/deployment.yaml @@ -0,0 +1,51 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: embedding-service +spec: + replicas: 1 + selector: + matchLabels: + app: embedding-service + template: + metadata: + labels: + app: embedding-service + spec: + containers: + - name: embedding-service + image: lexcam/embedding-service:latest + imagePullPolicy: IfNotPresent + ports: + - containerPort: 8000 + name: http + envFrom: + - configMapRef: + name: embedding-service-config + - secretRef: + name: embedding-service-secrets + volumeMounts: + - name: model-cache + mountPath: /cache + readinessProbe: + httpGet: + path: /api/v1/health + port: 8000 + initialDelaySeconds: 20 + periodSeconds: 10 + livenessProbe: + httpGet: + path: /api/v1/health + port: 8000 + initialDelaySeconds: 30 + periodSeconds: 20 + resources: + requests: + cpu: "500m" + memory: "1Gi" + limits: + cpu: "2" + memory: "4Gi" + volumes: + - name: model-cache + emptyDir: {} diff --git a/infrastructure/k8s/embedding-service/secret.yaml b/infrastructure/k8s/embedding-service/secret.yaml new file mode 100644 index 0000000..8c1553b --- /dev/null +++ b/infrastructure/k8s/embedding-service/secret.yaml @@ -0,0 +1,7 @@ +apiVersion: v1 +kind: Secret +metadata: + name: embedding-service-secrets +type: Opaque +stringData: + API_KEY: change-me diff --git a/infrastructure/k8s/embedding-service/service.yaml b/infrastructure/k8s/embedding-service/service.yaml new file mode 100644 index 0000000..1dad9f1 --- /dev/null +++ b/infrastructure/k8s/embedding-service/service.yaml @@ -0,0 +1,12 @@ +apiVersion: v1 +kind: Service +metadata: + name: embedding-service +spec: + type: ClusterIP + selector: + app: embedding-service + ports: + - name: http + port: 8000 + targetPort: 8000 diff --git a/rag_sessions.db b/rag_sessions.db new file mode 100644 index 0000000..e69de29 diff --git a/run_jobs_25984820017.json b/run_jobs_25984820017.json new file mode 100644 index 0000000..e69de29 diff --git a/scripts/init-databases.sql b/scripts/init-databases.sql index 6f5155c..50ad80d 100644 --- a/scripts/init-databases.sql +++ b/scripts/init-databases.sql @@ -1,5 +1,7 @@ -- LexCam PostgreSQL initialization --- Runs automatically on first container start +-- Runs automatically on first container start. +-- Design choice: database-per-service on a single Postgres instance +-- (separate logical databases like lexcam_users, lexcam_rag_sessions). CREATE DATABASE lexcam_users; CREATE DATABASE lexcam_lawyers; diff --git a/scripts/populate_embedding_cache.ps1 b/scripts/populate_embedding_cache.ps1 new file mode 100644 index 0000000..47dc5e3 --- /dev/null +++ b/scripts/populate_embedding_cache.ps1 @@ -0,0 +1,23 @@ +Param( + [string]$Image = 'lexcam/embedding-service:from-test' +) + +Write-Output "Using image: $Image" + +$tmp = New-TemporaryFile +$tmpDir = Split-Path $tmp -Parent + +try { + $cid = (docker create $Image).Trim() + Write-Output "Created container $cid" + docker cp "$cid`:/cache/model.onnx" "$tmpDir\model.onnx" + docker rm $cid | Out-Null + + # create temporary container with volume mounted + $vcontainer = (docker create --name tmp_embedding_volume -v embedding_cache:/cache busybox).Trim() + docker cp "$tmpDir\model.onnx" "$vcontainer`:/cache/model.onnx" + docker rm $vcontainer | Out-Null + Write-Output "Model copied into volume 'embedding_cache'" +} finally { + Remove-Item "$tmpDir\model.onnx" -ErrorAction SilentlyContinue +} diff --git a/scripts/populate_embedding_cache.sh b/scripts/populate_embedding_cache.sh new file mode 100644 index 0000000..636292c --- /dev/null +++ b/scripts/populate_embedding_cache.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Populate the named Docker volume `embedding_cache` with model files +# Usage: ./scripts/populate_embedding_cache.sh [image] +# Default image: lexcam/embedding-service:from-test + +IMAGE=${1:-lexcam/embedding-service:from-test} + +echo "Using image: $IMAGE" + +TMPDIR=$(mktemp -d) +cleanup() { + rm -rf "$TMPDIR" +} +trap cleanup EXIT + +echo "Creating temporary container from image to extract model..." +CID=$(docker create "$IMAGE") +echo "Copying /cache/model.onnx from container $CID to host temp" +docker cp "$CID":/cache/model.onnx "$TMPDIR"/model.onnx +docker rm "$CID" >/dev/null + +echo "Creating temporary container with embedding_cache volume mounted..." +VCONTAINER=$(docker create --name tmp_embedding_volume -v embedding_cache:/cache busybox) +echo "Copying model into volume" +docker cp "$TMPDIR"/model.onnx tmp_embedding_volume:/cache/model.onnx +docker rm tmp_embedding_volume >/dev/null + +echo "Model copied into volume 'embedding_cache'" + +echo "Done" diff --git a/services/embedding-service/.env.example b/services/embedding-service/.env.example new file mode 100644 index 0000000..5e5284d --- /dev/null +++ b/services/embedding-service/.env.example @@ -0,0 +1,12 @@ +SERVICE_NAME=embedding-service +API_PREFIX=/api/v1 +MODEL_NAME=intfloat/multilingual-e5-small +DEVICE=cpu +MAX_BATCH_SIZE=64 +MAX_SEQ_LENGTH=512 +NORMALIZE_EMBEDDINGS=true +LOG_LEVEL=INFO +REQUIRE_API_KEY=false +API_KEY=change-me +RATE_LIMIT_ENABLED=true +RATE_LIMIT=120/minute diff --git a/services/embedding-service/Dockerfile b/services/embedding-service/Dockerfile new file mode 100644 index 0000000..90600a9 --- /dev/null +++ b/services/embedding-service/Dockerfile @@ -0,0 +1,65 @@ +FROM mcr.microsoft.com/devcontainers/python:1-3.11-bookworm + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + TRANSFORMERS_CACHE=/cache \ + HF_HOME=/cache + +ARG MODEL_NAME=intfloat/multilingual-e5-small +ENV MODEL_NAME=${MODEL_NAME} +ARG MODEL_URL +ARG MODEL_SHA256= + +WORKDIR /app + +RUN useradd --create-home appuser \ + && mkdir -p /cache \ + && chown -R appuser:appuser /cache + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +RUN python3 - <<'PY' +import hashlib +import os +import pathlib +import urllib.request + +model_url = os.environ.get("MODEL_URL", "").strip() +if not model_url: + raise SystemExit( + "MODEL_URL build arg is required. Point it at the externally stored model artifact." + ) + +expected_sha256 = os.environ.get("MODEL_SHA256", "").strip().lower() +target = pathlib.Path("/cache/model.onnx") +target.parent.mkdir(parents=True, exist_ok=True) + +hasher = hashlib.sha256() if expected_sha256 else None +with urllib.request.urlopen(model_url) as response, target.open("wb") as output_file: + while True: + chunk = response.read(1024 * 1024) + if not chunk: + break + output_file.write(chunk) + if hasher: + hasher.update(chunk) + +if hasher and hasher.hexdigest().lower() != expected_sha256: + target.unlink(missing_ok=True) + raise SystemExit( + f"MODEL_SHA256 mismatch for {target}: expected {expected_sha256}, got {hasher.hexdigest().lower()}" + ) + +print(f"Downloaded external model artifact to {target}") +PY + +COPY app ./app + +EXPOSE 8000 + +USER appuser + +ENV WEB_CONCURRENCY=1 + +CMD ["gunicorn", "-k", "uvicorn.workers.UvicornWorker", "app.main:app", "--bind", "0.0.0.0:8000", "--workers", "1", "--timeout", "120"] diff --git a/services/embedding-service/app/__init__.py b/services/embedding-service/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/embedding-service/app/api/v1/__init__.py b/services/embedding-service/app/api/v1/__init__.py new file mode 100644 index 0000000..10c442a --- /dev/null +++ b/services/embedding-service/app/api/v1/__init__.py @@ -0,0 +1,3 @@ +from app.api.v1.routes import router + +__all__ = ["router"] diff --git a/services/embedding-service/app/api/v1/routes.py b/services/embedding-service/app/api/v1/routes.py new file mode 100644 index 0000000..fc323c2 --- /dev/null +++ b/services/embedding-service/app/api/v1/routes.py @@ -0,0 +1,74 @@ +import time +from fastapi import APIRouter, Body, Depends, HTTPException, Request + +from app.config import settings +from app.limiting import rate_limit +from app.schemas import EmbeddingRequest, EmbeddingResponse, HealthResponse +from app.security import verify_api_key + +router = APIRouter() + + +async def _get_embedding_model(request: Request): + embedding_model = getattr(request.app.state, "embedding_model", None) + if embedding_model is None: + from app.services.embedding import EmbeddingModel + + embedding_model = EmbeddingModel() + await embedding_model.load() + request.app.state.embedding_model = embedding_model + return embedding_model + + +@router.get("/health", response_model=HealthResponse) +async def health(request: Request) -> HealthResponse: + embedding_model = await _get_embedding_model(request) + try: + dimension = embedding_model.dimension + ready = True + status_msg = "ok" + except Exception as exc: + # model not loaded or failed to infer dimension + dimension = 0 + ready = False + status_msg = f"model not ready: {exc}" + + return HealthResponse( + status=status_msg, + model=settings.model_name, + device=settings.device, + dimension=dimension, + ready=ready, + ) + + +@router.post( + "/embed", + response_model=EmbeddingResponse, + dependencies=[Depends(verify_api_key)], +) +@rate_limit() +async def embed(request: Request, payload: EmbeddingRequest = Body(...)) -> EmbeddingResponse: + if len(payload.texts) > settings.max_batch_size: + raise HTTPException( + status_code=413, + detail="Batch size exceeds MAX_BATCH_SIZE", + ) + + embedding_model = await _get_embedding_model(request) + start = time.perf_counter() + embeddings = await embedding_model.embed( + payload.texts, + payload.input_type, + payload.normalize, + ) + elapsed_ms = (time.perf_counter() - start) * 1000 + normalized = settings.normalize_embeddings if payload.normalize is None else payload.normalize + + return EmbeddingResponse( + model=settings.model_name, + dimension=embedding_model.dimension, + normalized=normalized, + embeddings=embeddings, + processing_ms=elapsed_ms, + ) diff --git a/services/embedding-service/app/config.py b/services/embedding-service/app/config.py new file mode 100644 index 0000000..8c4474f --- /dev/null +++ b/services/embedding-service/app/config.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_file=".env", + env_ignore_empty=True, + extra="ignore", + ) + + service_name: str = Field("embedding-service", alias="SERVICE_NAME") + api_prefix: str = Field("/api/v1", alias="API_PREFIX") + model_name: str = Field("intfloat/multilingual-e5-small", alias="MODEL_NAME") + model_path: str | None = Field(None, alias="MODEL_PATH") + device: str = Field("cpu", alias="DEVICE") + max_batch_size: int = Field(64, alias="MAX_BATCH_SIZE") + max_seq_length: int = Field(512, alias="MAX_SEQ_LENGTH") + normalize_embeddings: bool = Field(True, alias="NORMALIZE_EMBEDDINGS") + log_level: str = Field("INFO", alias="LOG_LEVEL") + require_api_key: bool = Field(False, alias="REQUIRE_API_KEY") + api_key: str | None = Field(None, alias="API_KEY") + rate_limit_enabled: bool = Field(True, alias="RATE_LIMIT_ENABLED") + rate_limit: str = Field("120/minute", alias="RATE_LIMIT") + + +settings = Settings() diff --git a/services/embedding-service/app/limiting.py b/services/embedding-service/app/limiting.py new file mode 100644 index 0000000..dc49867 --- /dev/null +++ b/services/embedding-service/app/limiting.py @@ -0,0 +1,14 @@ +from __future__ import annotations + +from slowapi import Limiter +from slowapi.util import get_remote_address + +from app.config import settings + +limiter = Limiter(key_func=get_remote_address) + + +def rate_limit(): + if settings.rate_limit_enabled: + return limiter.limit(settings.rate_limit) + return lambda func: func diff --git a/services/embedding-service/app/main.py b/services/embedding-service/app/main.py new file mode 100644 index 0000000..e62bc4b --- /dev/null +++ b/services/embedding-service/app/main.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import logging +from contextlib import asynccontextmanager + +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse +from slowapi.errors import RateLimitExceeded +from starlette.status import HTTP_429_TOO_MANY_REQUESTS + +from app.api.v1 import router as v1_router +from app.config import settings +from app.limiting import limiter +from app.services.embedding import EmbeddingModel + + +def _configure_logging() -> None: + logging.basicConfig( + level=settings.log_level, + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + _configure_logging() + model = EmbeddingModel() + await model.load() + app.state.embedding_model = model + yield + + +app = FastAPI( + title="LexCam Embedding Service", + version="0.1.0", + lifespan=lifespan, +) + + +def _rate_limit_handler( + request: Request, + exc: RateLimitExceeded, +) -> JSONResponse: + return JSONResponse( + status_code=HTTP_429_TOO_MANY_REQUESTS, + content={"detail": "Rate limit exceeded"}, + ) + + +app.state.limiter = limiter +app.add_exception_handler(RateLimitExceeded, _rate_limit_handler) + +app.include_router(v1_router, prefix=settings.api_prefix) diff --git a/services/embedding-service/app/schemas.py b/services/embedding-service/app/schemas.py new file mode 100644 index 0000000..465c612 --- /dev/null +++ b/services/embedding-service/app/schemas.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, Field + + +class EmbeddingRequest(BaseModel): + texts: list[str] = Field(..., min_length=1, description="Texts to embed") + input_type: Literal["query", "passage"] | None = Field( + default=None, + description="Optional E5 prefix to apply", + ) + normalize: bool | None = Field( + default=None, + description="Override embedding normalization", + ) + + +class EmbeddingResponse(BaseModel): + model: str + dimension: int + normalized: bool + embeddings: list[list[float]] + processing_ms: float + + +class HealthResponse(BaseModel): + status: str + model: str + device: str + dimension: int + ready: bool diff --git a/services/embedding-service/app/security.py b/services/embedding-service/app/security.py new file mode 100644 index 0000000..5e97b8f --- /dev/null +++ b/services/embedding-service/app/security.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +from fastapi import Header, HTTPException + +from app.config import settings + + +def verify_api_key( + x_api_key: str | None = Header(default=None, alias="X-API-Key"), + authorization: str | None = Header(default=None, alias="Authorization"), +) -> None: + if not settings.require_api_key: + return + + if not settings.api_key: + raise HTTPException( + status_code=500, + detail="API key auth enabled but API_KEY not set", + ) + + if x_api_key == settings.api_key: + return + + if authorization and authorization.startswith("Bearer "): + if authorization.removeprefix("Bearer ") == settings.api_key: + return + + raise HTTPException(status_code=401, detail="Unauthorized") diff --git a/services/embedding-service/app/services/embedding.py b/services/embedding-service/app/services/embedding.py new file mode 100644 index 0000000..8ef6440 --- /dev/null +++ b/services/embedding-service/app/services/embedding.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import logging +from typing import Optional + +import anyio +import numpy as np +import onnxruntime as ort +from transformers import AutoTokenizer + +from app.config import settings +from pathlib import Path + + +class EmbeddingModel: + """ONNX-based embedding model runner. + + Expects an ONNX model file at `/cache/model.onnx`. This class uses a + Hugging Face tokenizer (from `settings.model_name`) and runs the ONNX + session to obtain the last hidden state, applies mean-pooling and + optional L2-normalization to produce sentence embeddings. + """ + + def __init__(self) -> None: + self._tokenizer: Optional[AutoTokenizer] = None + self._session: Optional[ort.InferenceSession] = None + self._dimension: Optional[int] = None + + @property + def dimension(self) -> int: + if self._dimension is None: + raise RuntimeError("Embedding dimension is not available") + return self._dimension + + async def load(self) -> None: + def _load(): + tokenizer = AutoTokenizer.from_pretrained(settings.model_name, use_fast=True) + + model_path = self._resolve_model_path() + if not model_path.exists(): + raise RuntimeError( + f"ONNX model not found at {model_path}.\n" + "Provide MODEL_PATH or let the Docker build download it via MODEL_URL." + ) + session = ort.InferenceSession(str(model_path), providers=["CPUExecutionProvider"]) + + # Attempt to infer hidden dimension from the first output shape + outputs = session.get_outputs() + hidden_size: Optional[int] = None + if outputs: + out_shape = outputs[0].shape # e.g., (batch, seq_len, hidden) + if len(out_shape) >= 3 and out_shape[2] is not None: + hidden_size = int(out_shape[2]) + + return tokenizer, session, hidden_size + + self._tokenizer, self._session, self._dimension = await anyio.to_thread.run_sync(_load) + logging.getLogger(__name__).info( + "ONNX embedding model loaded: %s (dim=%s)", settings.model_name, self._dimension + ) + + def _resolve_model_path(self) -> Path: + if settings.model_path: + return Path(settings.model_path) + + repo_backup = Path(__file__).resolve().parents[4].joinpath("_safe_backups", "model.onnx") + if repo_backup.exists(): + return repo_backup + + return Path("/cache/model.onnx") + + async def embed(self, texts: list[str], input_type: str | None, normalize: bool | None) -> list[list[float]]: + if input_type: + prefix = f"{input_type}: " + texts = [f"{prefix}{t}" for t in texts] + + normalize_embeddings = settings.normalize_embeddings if normalize is None else normalize + + def _encode() -> list[list[float]]: + assert self._tokenizer is not None and self._session is not None + + enc = self._tokenizer( + texts, + padding=True, + truncation=True, + max_length=settings.max_seq_length, + return_tensors="np", + ) + + # ONNX runtime expects numpy inputs + ort_inputs = {k: v for k, v in enc.items()} + outputs = self._session.run(None, ort_inputs) + + # assume outputs[0] is last_hidden_state: (batch, seq_len, hidden) + last_hidden = outputs[0] + attention_mask = enc.get("attention_mask") + if attention_mask is None: + # fallback: average across tokens + embeddings = last_hidden.mean(axis=1) + else: + mask = attention_mask.astype(np.float32)[..., None] + summed = (last_hidden * mask).sum(axis=1) + counts = mask.sum(axis=1) + counts[counts == 0] = 1 + embeddings = summed / counts + + if normalize_embeddings: + norms = np.linalg.norm(embeddings, axis=1, keepdims=True) + norms[norms == 0] = 1 + embeddings = embeddings / norms + + return embeddings.tolist() + + return await anyio.to_thread.run_sync(_encode) diff --git a/services/embedding-service/convert.Dockerfile b/services/embedding-service/convert.Dockerfile new file mode 100644 index 0000000..1308529 --- /dev/null +++ b/services/embedding-service/convert.Dockerfile @@ -0,0 +1,16 @@ +FROM python:3.11-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /workspace + +RUN apt-get update \ + && apt-get install -y --no-install-recommends build-essential git curl ca-certificates \ + && rm -rf /var/lib/apt/lists/* \ + && pip install --no-cache-dir --upgrade pip \ + && pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu transformers onnx onnxruntime optimum sentence-transformers + +COPY services/embedding-service/scripts/convert_to_onnx.py /workspace/convert_to_onnx.py + +ENTRYPOINT ["python", "/workspace/convert_to_onnx.py"] diff --git a/services/embedding-service/model/.gitkeep b/services/embedding-service/model/.gitkeep new file mode 100644 index 0000000..0519ecb --- /dev/null +++ b/services/embedding-service/model/.gitkeep @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/services/embedding-service/requirements.txt b/services/embedding-service/requirements.txt new file mode 100644 index 0000000..c950d6b --- /dev/null +++ b/services/embedding-service/requirements.txt @@ -0,0 +1,10 @@ +fastapi>=0.115.0,<1.0.0 +uvicorn[standard]==0.30.1 +gunicorn==22.0.0 +pydantic>=2.10.0,<3.0.0 +pydantic-settings>=2.7.0,<3.0.0 +transformers>=4.47.0,<5.0.0 +numpy>=2.1.0,<3.0.0 +anyio==4.4.0 +slowapi==0.1.9 +onnxruntime==1.26.0 diff --git a/services/embedding-service/scripts/convert_to_onnx.py b/services/embedding-service/scripts/convert_to_onnx.py new file mode 100644 index 0000000..e97c77d --- /dev/null +++ b/services/embedding-service/scripts/convert_to_onnx.py @@ -0,0 +1,98 @@ +"""Helper to convert a Hugging Face / Sentence-Transformers model to ONNX. + +This script attempts to use the transformers ONNX conversion tooling. Converting +requires PyTorch and the conversion tools; run this locally and upload the +generated `model.onnx` file as an external artifact, such as a GitHub Releases +asset, before building the Docker image. + +Usage (example / recommended): + +1. Create a Python venv with PyTorch and transformers installed: + + pip install torch sentence-transformers transformers onnx onnxruntime + +2. Run this script: + + python services/embedding-service/scripts/convert_to_onnx.py --model intfloat/multilingual-e5-small --output model.onnx + +Notes: +- Conversion commands and compatibility depend on the model architecture. +- If this script cannot perform an automated conversion on your platform, + follow Hugging Face or sentence-transformers conversion guides to produce + an ONNX file and upload it as your release asset. +""" + +import argparse +import shutil +import subprocess +import sys +from pathlib import Path + + +def export_with_torch(model_id: str, output: Path) -> None: + import torch + from transformers import AutoModel, AutoTokenizer + + tokenizer = AutoTokenizer.from_pretrained(model_id, use_fast=True) + model = AutoModel.from_pretrained(model_id) + model.eval() + + sample = tokenizer( + ["This is a sample text for ONNX export."], + padding=True, + truncation=True, + max_length=128, + return_tensors="pt", + ) + + output.parent.mkdir(parents=True, exist_ok=True) + if output.exists(): + output.unlink() + data_file = Path(str(output) + ".data") + if data_file.exists(): + data_file.unlink() + + torch.onnx.export( + model, + (sample["input_ids"], sample["attention_mask"]), + str(output), + input_names=["input_ids", "attention_mask"], + output_names=["last_hidden_state"], + dynamic_axes={ + "input_ids": {0: "batch_size", 1: "sequence"}, + "attention_mask": {0: "batch_size", 1: "sequence"}, + "last_hidden_state": {0: "batch_size", 1: "sequence"}, + }, + opset_version=18, + do_constant_folding=True, + external_data=False, + ) + if data_file.exists(): + data_file.unlink() + print("Torch ONNX export complete; wrote:", output) + + +def main(): + p = argparse.ArgumentParser() + p.add_argument("--model", required=True, help="Hugging Face model id") + p.add_argument("--output", required=True, help="Path to write model.onnx") + args = p.parse_args() + + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + + # Try using the transformers.onnx CLI if available + cli = shutil.which("transformers-onnx") or shutil.which("transformers.onnx") + if cli: + print("Found transformers ONNX CLI; attempting conversion using it...") + cmd = [sys.executable, "-m", "transformers.onnx", "--model", args.model, str(output)] + subprocess.check_call(cmd) + print("Conversion complete; wrote:", output) + return + + print("transformers ONNX CLI not found; falling back to torch.onnx export.") + export_with_torch(args.model, output) + + +if __name__ == "__main__": + main() diff --git a/services/embedding-service/tests/test_embed.py b/services/embedding-service/tests/test_embed.py new file mode 100644 index 0000000..eda510d --- /dev/null +++ b/services/embedding-service/tests/test_embed.py @@ -0,0 +1,29 @@ +import os + +import pytest +from fastapi.testclient import TestClient + +from app.main import app + + +@pytest.mark.skipif( + os.getenv("LEXCAM_TEST_NO_MODEL") == "1", + reason="Model download disabled for tests", +) +def test_embed_returns_vectors(): + client = TestClient(app) + headers = {} + if os.getenv("REQUIRE_API_KEY", "false").lower() == "true": + api_key = os.getenv("API_KEY", "") + if api_key: + headers["X-API-Key"] = api_key + response = client.post( + "/api/v1/embed", + json={"texts": ["hello world"], "input_type": "query"}, + headers=headers, + ) + assert response.status_code == 200 + payload = response.json() + assert "embeddings" in payload + assert len(payload["embeddings"]) == 1 + assert len(payload["embeddings"][0]) > 0 diff --git a/services/feedback-service/.dockerignore b/services/feedback-service/.dockerignore new file mode 100644 index 0000000..ac56efb --- /dev/null +++ b/services/feedback-service/.dockerignore @@ -0,0 +1,6 @@ +__pycache__ +*.pyc +*.pyo +.pytest_cache +venv +.env diff --git a/services/feedback-service/Dockerfile b/services/feedback-service/Dockerfile new file mode 100644 index 0000000..35069f5 --- /dev/null +++ b/services/feedback-service/Dockerfile @@ -0,0 +1,13 @@ +FROM python:3.11-slim +WORKDIR /app +ENV PYTHONDONTWRITEBYTECODE=1 +ENV PYTHONUNBUFFERED=1 + +COPY requirements.txt ./ +RUN pip install --no-cache-dir -r requirements.txt + +COPY app ./app +COPY alembic.ini ./ +COPY alembic ./alembic + +CMD ["sh", "-c", "alembic -c /app/alembic.ini upgrade head && gunicorn -k uvicorn.workers.UvicornWorker app.main:app --bind 0.0.0.0:8000 --workers 2"] diff --git a/services/feedback-service/alembic.ini b/services/feedback-service/alembic.ini new file mode 100644 index 0000000..3e04d18 --- /dev/null +++ b/services/feedback-service/alembic.ini @@ -0,0 +1,24 @@ +[alembic] +script_location = alembic + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = INFO +handlers = console + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s diff --git a/services/feedback-service/alembic/env.py b/services/feedback-service/alembic/env.py new file mode 100644 index 0000000..8aa760a --- /dev/null +++ b/services/feedback-service/alembic/env.py @@ -0,0 +1,38 @@ +from alembic import context +import os +import sys + +sys.path.append(os.path.dirname(os.path.dirname(__file__))) +sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) + +from app.config import settings +from app.db import engine +from app.models import Feedback +from app.models_user import User + +# this is the Alembic Config object, which provides +# access to the values within the .ini file in use. +config = context.config + +target_metadata = [Feedback.metadata, User.metadata] + + +def run_migrations_offline(): + url = str(settings.DATABASE_URL) + context.configure(url=url, target_metadata=target_metadata, literal_binds=True) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online(): + connectable = engine + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/services/feedback-service/alembic/versions/0001_initial.py b/services/feedback-service/alembic/versions/0001_initial.py new file mode 100644 index 0000000..8d4a529 --- /dev/null +++ b/services/feedback-service/alembic/versions/0001_initial.py @@ -0,0 +1,43 @@ +"""initial + +Revision ID: 0001_initial +Revises: +Create Date: 2026-05-17 + +""" +from alembic import op + +revision = '0001_initial' +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade(): + op.execute(""" + CREATE TABLE IF NOT EXISTS "user" ( + id SERIAL PRIMARY KEY, + username VARCHAR(128) NOT NULL UNIQUE, + email VARCHAR(256), + hashed_password VARCHAR(256) NOT NULL, + is_active BOOLEAN NOT NULL DEFAULT TRUE + ) + """) + op.execute(""" + CREATE TABLE IF NOT EXISTS feedback ( + id SERIAL PRIMARY KEY, + user_id VARCHAR(128), + text TEXT NOT NULL, + rating INTEGER, + created_at TIMESTAMP WITHOUT TIME ZONE, + session_id VARCHAR(128), + message_index INTEGER, + flag_count INTEGER NOT NULL DEFAULT 0, + flagged BOOLEAN NOT NULL DEFAULT FALSE + ) + """) + + +def downgrade(): + op.execute('DROP TABLE IF EXISTS feedback') + op.execute('DROP TABLE IF EXISTS "user"') diff --git a/services/feedback-service/app/__init__.py b/services/feedback-service/app/__init__.py new file mode 100644 index 0000000..e4c7aa5 --- /dev/null +++ b/services/feedback-service/app/__init__.py @@ -0,0 +1 @@ +from .main import app diff --git a/services/feedback-service/app/api/v1/auth_routes.py b/services/feedback-service/app/api/v1/auth_routes.py new file mode 100644 index 0000000..cf4411c --- /dev/null +++ b/services/feedback-service/app/api/v1/auth_routes.py @@ -0,0 +1,32 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from fastapi.security import OAuth2PasswordRequestForm +from sqlmodel import Session +from ...db import get_session +from ...models_user import User +from ...auth import verify_password, get_password_hash, create_access_token +from datetime import timedelta +from ...config import settings + +router = APIRouter(prefix="/v1/auth") + + +@router.post("/signup", status_code=status.HTTP_201_CREATED) +def signup(username: str, password: str, session: Session = Depends(get_session)): + existing = session.exec(User.select().where(User.username == username)).first() + if existing: + raise HTTPException(status_code=400, detail="User already exists") + user = User(username=username, hashed_password=get_password_hash(password)) + session.add(user) + session.commit() + session.refresh(user) + return {"id": user.id, "username": user.username} + + +@router.post("/token") +def login_token(form_data: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)): + user = session.exec(User.select().where(User.username == form_data.username)).first() + if not user or not verify_password(form_data.password, user.hashed_password): + raise HTTPException(status_code=400, detail="Incorrect username or password") + access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) + access_token = create_access_token(data={"sub": user.username}, expires_delta=access_token_expires) + return {"access_token": access_token, "token_type": "bearer"} diff --git a/services/feedback-service/app/api/v1/routes.py b/services/feedback-service/app/api/v1/routes.py new file mode 100644 index 0000000..3a4601f --- /dev/null +++ b/services/feedback-service/app/api/v1/routes.py @@ -0,0 +1,59 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from sqlmodel import Session, select +from typing import List + +from ...db import get_session +from ...models import Feedback +from ...schemas import FeedbackCreate, FeedbackRead +from ...tasks import process_feedback_async +from ...events import publish_event + +router = APIRouter(prefix="/v1") + + +@router.get("/health", status_code=200) +def health(): + return {"status": "ok"} + + +@router.post("/feedback", response_model=FeedbackRead, status_code=status.HTTP_201_CREATED) +def create_feedback(payload: FeedbackCreate, session: Session = Depends(get_session)): + fb = Feedback.from_orm(payload) + session.add(fb) + session.commit() + session.refresh(fb) + # enqueue background processing + process_feedback_async.delay(fb.id) + return fb + + +@router.post("/feedback/{feedback_id}/flag", status_code=200) +def flag_feedback(feedback_id: int, session: Session = Depends(get_session)): + fb = session.get(Feedback, feedback_id) + if not fb: + raise HTTPException(status_code=404, detail="Feedback not found") + fb.flag_count += 1 + if fb.flag_count >= 3 and not fb.flagged: + fb.flagged = True + session.add(fb) + session.commit() + publish_event("feedback.flagged", {"feedback_id": fb.id, "session_id": fb.session_id}) + return {"flagged": True} + session.add(fb) + session.commit() + return {"flagged": fb.flagged, "flag_count": fb.flag_count} + + +@router.get("/feedback", response_model=List[FeedbackRead]) +def list_feedback(limit: int = 50, session: Session = Depends(get_session)): + statement = select(Feedback).limit(limit) + results = session.exec(statement).all() + return results + + +@router.get("/feedback/{feedback_id}", response_model=FeedbackRead) +def get_feedback(feedback_id: int, session: Session = Depends(get_session)): + fb = session.get(Feedback, feedback_id) + if not fb: + raise HTTPException(status_code=404, detail="Feedback not found") + return fb diff --git a/services/feedback-service/app/auth.py b/services/feedback-service/app/auth.py new file mode 100644 index 0000000..330289d --- /dev/null +++ b/services/feedback-service/app/auth.py @@ -0,0 +1,53 @@ +from datetime import datetime, timedelta +from typing import Optional +from passlib.context import CryptContext +from jose import JWTError, jwt +from pydantic import BaseModel +from fastapi import Depends, HTTPException, status +from fastapi.security import OAuth2PasswordBearer +from .config import settings +from .db import get_session +from sqlmodel import Session +from .models_user import User + +pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/v1/auth/token") + +class TokenData(BaseModel): + username: Optional[str] = None + + +def verify_password(plain_password, hashed_password): + return pwd_context.verify(plain_password, hashed_password) + + +def get_password_hash(password): + return pwd_context.hash(password) + + +def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): + to_encode = data.copy() + expire = datetime.utcnow() + (expires_delta or timedelta(minutes=60)) + to_encode.update({"exp": expire}) + encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) + return encoded_jwt + + +def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)): + credentials_exception = HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Could not validate credentials", + headers={"WWW-Authenticate": "Bearer"}, + ) + try: + payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) + username: str = payload.get("sub") + if username is None: + raise credentials_exception + token_data = TokenData(username=username) + except JWTError: + raise credentials_exception + user = session.exec(User.select().where(User.username == token_data.username)).first() + if user is None: + raise credentials_exception + return user diff --git a/services/feedback-service/app/config.py b/services/feedback-service/app/config.py new file mode 100644 index 0000000..ef634ab --- /dev/null +++ b/services/feedback-service/app/config.py @@ -0,0 +1,21 @@ +from pydantic import BaseSettings, AnyUrl + + +class Settings(BaseSettings): + SERVICE_NAME: str = "feedback-service" + DATABASE_URL: AnyUrl = "postgresql://postgres:postgres@db:5432/feedback_db" + REDIS_URL: str = "redis://redis:6379/0" + CELERY_BROKER_URL: str = "redis://redis:6379/1" + CELERY_RESULT_BACKEND: str = "redis://redis:6379/2" + MIGRATE_AUTO: bool = True + SECRET_KEY: str = "please-change-me" + ALGORITHM: str = "HS256" + ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 + USER_MGMT_URL: str = "http://user-management:8000" + RABBITMQ_URL: str = "amqp://lexcam:lexcam_dev@rabbitmq:5672/%2F" + + class Config: + env_file = ".env" + + +settings = Settings() diff --git a/services/feedback-service/app/db.py b/services/feedback-service/app/db.py new file mode 100644 index 0000000..58e3353 --- /dev/null +++ b/services/feedback-service/app/db.py @@ -0,0 +1,13 @@ +from sqlmodel import SQLModel, create_engine, Session +from .config import settings + +engine = create_engine(str(settings.DATABASE_URL), echo=False) + + +def init_db(): + return None + + +def get_session(): + with Session(engine) as session: + yield session diff --git a/services/feedback-service/app/events.py b/services/feedback-service/app/events.py new file mode 100644 index 0000000..8c7dd9b --- /dev/null +++ b/services/feedback-service/app/events.py @@ -0,0 +1,18 @@ +import json +import pika +from .config import settings + + +def _get_connection(): + params = pika.URLParameters(settings.RABBITMQ_URL) + return pika.BlockingConnection(params) + + +def publish_event(routing_key: str, payload: dict): + conn = _get_connection() + ch = conn.channel() + ch.exchange_declare(exchange="lexcam.events", exchange_type="topic", durable=True) + # Make published messages persistent so they survive broker restarts. + props = pika.BasicProperties(delivery_mode=2) + ch.basic_publish(exchange="lexcam.events", routing_key=routing_key, body=json.dumps(payload), properties=props) + conn.close() diff --git a/services/feedback-service/app/main.py b/services/feedback-service/app/main.py new file mode 100644 index 0000000..bdff8f5 --- /dev/null +++ b/services/feedback-service/app/main.py @@ -0,0 +1,21 @@ +from fastapi import FastAPI +from .api.v1 import routes as v1_routes +from .api.v1 import auth_routes +from .db import init_db +from .config import settings +from .monitoring import init_logging, prometheus_middleware, prometheus_endpoint + +app = FastAPI(title=settings.SERVICE_NAME) + + +@app.on_event("startup") +def on_startup(): + init_db() + init_logging() + + +app.middleware("http")(prometheus_middleware) +app.add_api_route("/metrics", prometheus_endpoint, methods=["GET"]) + +app.include_router(auth_routes.router) +app.include_router(v1_routes.router) diff --git a/services/feedback-service/app/models.py b/services/feedback-service/app/models.py new file mode 100644 index 0000000..d937a84 --- /dev/null +++ b/services/feedback-service/app/models.py @@ -0,0 +1,15 @@ +from typing import Optional +from datetime import datetime +from sqlmodel import SQLModel, Field + + +class Feedback(SQLModel, table=True): + id: Optional[int] = Field(default=None, primary_key=True) + user_id: Optional[str] = Field(index=True) + text: str + rating: Optional[int] = Field(default=None) + created_at: datetime = Field(default_factory=datetime.utcnow) + session_id: Optional[str] = None + message_index: Optional[int] = None + flag_count: int = Field(default=0) + flagged: bool = Field(default=False) diff --git a/services/feedback-service/app/models_user.py b/services/feedback-service/app/models_user.py new file mode 100644 index 0000000..37c60d6 --- /dev/null +++ b/services/feedback-service/app/models_user.py @@ -0,0 +1,13 @@ +from typing import Optional +from sqlmodel import SQLModel, Field +from datetime import datetime + + +class User(SQLModel, table=True): + id: Optional[int] = Field(default=None, primary_key=True) + username: str = Field(index=True, unique=True) + email: Optional[str] = None + hashed_password: str + is_active: bool = True + is_superuser: bool = False + created_at: datetime = Field(default_factory=datetime.utcnow) diff --git a/services/feedback-service/app/monitoring.py b/services/feedback-service/app/monitoring.py new file mode 100644 index 0000000..b21d99e --- /dev/null +++ b/services/feedback-service/app/monitoring.py @@ -0,0 +1,26 @@ +import logging +from prometheus_client import Counter, generate_latest, CONTENT_TYPE_LATEST +from fastapi import Request, Response +from fastapi.responses import PlainTextResponse +import structlog + +REQUEST_COUNT = Counter("http_requests_total", "Total HTTP requests", ["method", "endpoint", "http_status"]) + + +def init_logging(): + structlog.configure(processors=[structlog.processors.JSONRenderer()]) + logging.basicConfig(level=logging.INFO) + + +async def prometheus_endpoint(): + data = generate_latest() + return Response(content=data, media_type=CONTENT_TYPE_LATEST) + + +async def prometheus_middleware(request: Request, call_next): + response = await call_next(request) + try: + REQUEST_COUNT.labels(request.method, request.url.path, str(response.status_code)).inc() + except Exception: + pass + return response diff --git a/services/feedback-service/app/schemas.py b/services/feedback-service/app/schemas.py new file mode 100644 index 0000000..1e9108e --- /dev/null +++ b/services/feedback-service/app/schemas.py @@ -0,0 +1,20 @@ +from typing import Optional +from datetime import datetime +from pydantic import BaseModel + + +class FeedbackCreate(BaseModel): + user_id: Optional[str] + text: str + rating: Optional[int] + + +class FeedbackRead(BaseModel): + id: int + user_id: Optional[str] + text: str + rating: Optional[int] + created_at: datetime + + class Config: + orm_mode = True diff --git a/services/feedback-service/app/tasks.py b/services/feedback-service/app/tasks.py new file mode 100644 index 0000000..bc4bd20 --- /dev/null +++ b/services/feedback-service/app/tasks.py @@ -0,0 +1,22 @@ +from celery import Celery +from celery.utils.log import get_task_logger +from .config import settings + +logger = get_task_logger(__name__) + +celery_app = Celery( + settings.SERVICE_NAME, + broker=settings.CELERY_BROKER_URL, + backend=settings.CELERY_RESULT_BACKEND, +) + + +@celery_app.task(bind=True) +def placeholder_process_feedback(self, feedback_id: int): + """Placeholder: feedback processing implementation is out-of-scope for the assistant.""" + logger.warning("placeholder_process_feedback called for %s", feedback_id) + raise NotImplementedError("Feedback processing implemented by the team") + + +def get_celery_app(): + return celery_app diff --git a/services/feedback-service/requirements.txt b/services/feedback-service/requirements.txt new file mode 100644 index 0000000..5dae27b --- /dev/null +++ b/services/feedback-service/requirements.txt @@ -0,0 +1,20 @@ +fastapi==0.95.2 +uvicorn[standard]==0.22.0 +SQLModel==0.0.8 +SQLAlchemy==1.4.41 +psycopg2-binary==2.9.7 +python-dotenv==1.0.0 +celery==5.3.0 +redis==4.5.1 +gunicorn==20.1.0 +httpx==0.24.1 +pydantic==1.10.9 +pytest==7.4.0 +passlib[bcrypt]==1.7.4 +python-jose==3.3.0 +alembic==1.11.1 +prometheus_client==0.16.0 +structlog==23.3.0 +python-json-logger==2.0.7 +pika==1.3.1 +python-multipart==0.0.9 diff --git a/services/feedback-service/tests/test_health.py b/services/feedback-service/tests/test_health.py new file mode 100644 index 0000000..b5703e1 --- /dev/null +++ b/services/feedback-service/tests/test_health.py @@ -0,0 +1,9 @@ +from fastapi.testclient import TestClient +from app.main import app + + +def test_health(): + client = TestClient(app) + r = client.get("/v1/health") + assert r.status_code == 200 + assert r.json() == {"status": "ok"} diff --git a/services/knowledge-base-service/Dockerfile b/services/knowledge-base-service/Dockerfile new file mode 100644 index 0000000..5e58202 --- /dev/null +++ b/services/knowledge-base-service/Dockerfile @@ -0,0 +1,22 @@ +FROM python:3.11-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /app + +RUN useradd --create-home appuser + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY app ./app +COPY scripts ./scripts + +EXPOSE 8000 + +USER appuser + +ENV WEB_CONCURRENCY=1 + +CMD ["gunicorn", "-k", "uvicorn.workers.UvicornWorker", "app.main:app", "--bind", "0.0.0.0:8000", "--workers", "1", "--timeout", "120"] diff --git a/services/knowledge-base-service/README.md b/services/knowledge-base-service/README.md new file mode 100644 index 0000000..19ef60e --- /dev/null +++ b/services/knowledge-base-service/README.md @@ -0,0 +1,40 @@ +Knowledge Base Service + +Quick start (development with docker-compose): + +1. Ensure docker-compose.dev.yml includes the `knowledge-base-service` entry (it does by default). +2. Start dependencies: Postgres, Qdrant, Redis, Embedding Service via docker-compose: + +```bash +docker compose -f docker-compose.dev.yml up -d postgres qdrant redis embedding-service +``` + +3. Create the KB schema in the `lexcam_knowledge` database (run inside the Postgres container or via psql): + +```bash +# from repo root +cat services/knowledge-base-service/sql/init_kb_schema.sql | docker exec -i lexcam-postgres psql -U lexcam -d lexcam_knowledge +``` + +4. Build and start the Knowledge Base Service: + +```bash +docker compose -f docker-compose.dev.yml up -d knowledge-base-service +``` + +5. Seed a sample document/article into the KB (make sure services are running): + +```bash +# from repo root +python services/knowledge-base-service/scripts/seed_kb.py +``` + +6. Test the public search endpoint: + +```bash +curl -X POST http://localhost:8003/api/v1/search -H 'Content-Type: application/json' -d '{"query":"santé travailleurs"}' +``` + +Notes: +- The service uses SQLAlchemy `create_all` at startup, but you still need to run the provided SQL to create the tsvector index and trigger for full-text search. +- The Indexing Worker and ingestion workflows will use the `/internal/retrieve` endpoint and Qdrant upserts. diff --git a/services/knowledge-base-service/app/__init__.py b/services/knowledge-base-service/app/__init__.py new file mode 100644 index 0000000..0b16126 --- /dev/null +++ b/services/knowledge-base-service/app/__init__.py @@ -0,0 +1 @@ +# LexCam Knowledge Base Service package diff --git a/services/knowledge-base-service/app/api/v1/__init__.py b/services/knowledge-base-service/app/api/v1/__init__.py new file mode 100644 index 0000000..0f47bf7 --- /dev/null +++ b/services/knowledge-base-service/app/api/v1/__init__.py @@ -0,0 +1,6 @@ +from fastapi import APIRouter + +from app.api.v1.routes import router as routes_router + +router = APIRouter() +router.include_router(routes_router) diff --git a/services/knowledge-base-service/app/api/v1/routes.py b/services/knowledge-base-service/app/api/v1/routes.py new file mode 100644 index 0000000..a08b211 --- /dev/null +++ b/services/knowledge-base-service/app/api/v1/routes.py @@ -0,0 +1,301 @@ +from __future__ import annotations + +import json +from typing import Any +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException, Request +from qdrant_client.http import models as rest +from sqlalchemy import desc, func, select +from sqlalchemy.orm import Session, joinedload + +from app.config import settings +from app.db import SessionLocal +from app.models import LawArticle, LawDocument +from app.schemas import ( + ArticleResponse, + LawArticleSummary, + LawDocumentSchema, + RetrieveItem, + RetrieveRequest, + RetrieveResponse, + SearchRequest, + SearchResultItem, +) +from app.services.embedding import EmbeddingServiceClient +from redis.asyncio import Redis + +router = APIRouter() + + +def get_db() -> Session: + db = SessionLocal() + try: + yield db + finally: + db.close() + + +def get_qdrant(request: Request): + return request.app.state.qdrant + + +def get_embedding_client(request: Request) -> EmbeddingServiceClient: + return request.app.state.embedding_client + + +def get_redis(request: Request) -> Redis: + return request.app.state.redis + + +@router.get("/health") +async def health(request: Request, db: Session = Depends(get_db)) -> Any: + try: + db.execute(select(1)).scalar_one() + qdrant_client = get_qdrant(request) + qdrant_client.get_collection(settings.qdrant_collection) + # check embedding service health + embedding_client = get_embedding_client(request) + try: + emb_health = await embedding_client.health() + except Exception: + emb_health = {"status": "unreachable"} + # check redis + redis_client = get_redis(request) + try: + await redis_client.ping() + redis_ok = True + except Exception: + redis_ok = False + except Exception as exc: + raise HTTPException(status_code=503, detail=str(exc)) + + return { + "status": "ok", + "database": "ok", + "qdrant_collection": settings.qdrant_collection, + "embedding_service": emb_health, + "redis": "ok" if redis_ok else "unreachable", + } + + +def _build_limit(limit: int | None) -> int: + if limit is None or limit <= 0: + return settings.max_search_results + return min(limit, settings.max_search_results) + + +@router.post("/search") +async def search( + payload: SearchRequest, + request: Request, + db: Session = Depends(get_db), + qdrant_client=Depends(get_qdrant), + embedding_client: EmbeddingServiceClient = Depends(get_embedding_client), +) -> list[SearchResultItem]: + if not payload.query.strip(): + raise HTTPException(status_code=400, detail="query is required") + + limit = _build_limit(payload.limit) + vector_results: dict[UUID, dict[str, Any]] = {} + qdrant_filter = None + if payload.domain: + qdrant_filter = rest.Filter( + must=[ + rest.FieldCondition( + key="domain", + match=rest.MatchValue(value=payload.domain), + ) + ] + ) + + embedding = await embedding_client.embed(payload.query) + qdrant_hits = qdrant_client.search( + collection_name=settings.qdrant_collection, + query_vector=embedding, + limit=limit, + with_payload=True, + query_filter=qdrant_filter, + ) + + for rank, hit in enumerate(qdrant_hits, start=1): + payload_data = hit.payload or {} + article_id = payload_data.get("article_id") + if not article_id or article_id in vector_results: + continue + + vector_results[article_id] = { + "article_id": UUID(str(article_id)), + "law_name": payload_data.get("law_name", ""), + "article_number": payload_data.get("article_number", ""), + "title": payload_data.get("title"), + "domain": payload_data.get("domain", ""), + "language": payload_data.get("language", ""), + "text_preview": payload_data.get("text_preview", ""), + "score": 1.0 / (60 + rank), + } + + query_expression = func.plainto_tsquery("simple", payload.query) + statement = ( + select(LawArticle, func.ts_rank_cd(LawArticle.search_vector, query_expression).label("rank")) + .options(joinedload(LawArticle.document)) + .where(LawArticle.search_vector.op("@@")(query_expression)) + ) + if payload.domain: + statement = statement.where(LawArticle.domain == payload.domain) + if payload.language: + statement = statement.where(LawArticle.language == payload.language) + statement = statement.order_by(desc("rank")).limit(limit) + + keyword_rows = db.execute(statement).all() + keyword_results: dict[UUID, dict[str, Any]] = {} + for index, (article, rank) in enumerate(keyword_rows, start=1): + if article.id in keyword_results: + continue + + keyword_results[article.id] = { + "article_id": article.id, + "law_name": article.document.name if article.document else "", + "article_number": article.article_number, + "title": article.title, + "domain": article.domain, + "language": article.language, + "text_preview": article.full_text[:200], + "score": 1.0 / (60 + index), + } + + merged: dict[UUID, dict[str, Any]] = {} + for article_id, entry in vector_results.items(): + merged[article_id] = entry.copy() + + for article_id, entry in keyword_results.items(): + if article_id in merged: + merged[article_id]["score"] += entry["score"] + else: + merged[article_id] = entry.copy() + + ordered = sorted(merged.values(), key=lambda item: item["score"], reverse=True) + return [SearchResultItem(**item) for item in ordered[:limit]] + + +@router.get("/articles/{article_id}") +async def get_article( + article_id: UUID, + request: Request, + db: Session = Depends(get_db), + redis_client: Redis = Depends(get_redis), +) -> ArticleResponse: + cache_key = f"article:{article_id}" + cached = await redis_client.get(cache_key) + if cached: + return ArticleResponse.model_validate_json(cached) + + article = db.get(LawArticle, article_id) + if not article: + raise HTTPException(status_code=404, detail="Article not found") + + if not article.document: + db.refresh(article, attribute_names=["document"]) + + response = ArticleResponse( + id=article.id, + document_id=article.document_id, + document_code=article.document.code if article.document else None, + document_name=article.document.name if article.document else None, + article_number=article.article_number, + chapter=article.chapter, + title=article.title, + full_text=article.full_text, + plain_summary=article.plain_summary, + domain=article.domain, + language=article.language, + qdrant_id=article.qdrant_id, + ) + await redis_client.set( + cache_key, + response.model_dump_json(), + ex=settings.plain_summary_cache_ttl_seconds, + ) + return response + + +@router.get("/articles") +def list_articles( + law_code: str | None = None, + domain: str | None = None, + language: str | None = None, + db: Session = Depends(get_db), +) -> list[LawArticleSummary]: + statement = select(LawArticle).options(joinedload(LawArticle.document)) + if law_code: + statement = statement.join(LawArticle.document).where(LawDocument.code == law_code) + if domain: + statement = statement.where(LawArticle.domain == domain) + if language: + statement = statement.where(LawArticle.language == language) + + articles = db.scalars(statement).all() + return [ + LawArticleSummary( + id=article.id, + document_id=article.document_id, + document_code=article.document.code if article.document else None, + document_name=article.document.name if article.document else None, + article_number=article.article_number, + title=article.title, + domain=article.domain, + language=article.language, + ) + for article in articles + ] + + +@router.get("/laws") +def list_laws(db: Session = Depends(get_db)) -> list[LawDocumentSchema]: + documents = db.scalars(select(LawDocument)).all() + return [LawDocumentSchema.model_validate(document) for document in documents] + + +@router.post("/internal/retrieve") +def internal_retrieve( + payload: RetrieveRequest, + request: Request, + qdrant_client=Depends(get_qdrant), +) -> RetrieveResponse: + limit = _build_limit(payload.top_k) + qdrant_filter = None + if payload.domain_filter: + conditions = [ + rest.FieldCondition(key="domain", match=rest.MatchValue(value=value)) + for value in payload.domain_filter + ] + qdrant_filter = rest.Filter(min_should=rest.MinShould(conditions=conditions, min_count=1)) + + hits = qdrant_client.search( + collection_name=settings.qdrant_collection, + query_vector=payload.query_vector, + limit=limit, + with_payload=True, + query_filter=qdrant_filter, + ) + + results = [] + for hit in hits: + payload_data = hit.payload or {} + article_id_value = payload_data.get("article_id") + if not article_id_value: + continue + + results.append( + RetrieveItem( + article_id=UUID(str(article_id_value)), + law_name=payload_data.get("law_name", ""), + article_number=payload_data.get("article_number", ""), + domain=payload_data.get("domain", ""), + language=payload_data.get("language", ""), + text_preview=payload_data.get("text_preview", ""), + score=hit.score or 0.0, + ) + ) + + return RetrieveResponse(results=results) diff --git a/services/knowledge-base-service/app/config.py b/services/knowledge-base-service/app/config.py new file mode 100644 index 0000000..5939cea --- /dev/null +++ b/services/knowledge-base-service/app/config.py @@ -0,0 +1,32 @@ +from pydantic import Field +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_file=".env", + env_ignore_empty=True, + extra="ignore", + ) + + service_name: str = Field("knowledge-base-service", alias="SERVICE_NAME") + api_prefix: str = Field("/api/v1", alias="API_PREFIX") + database_url: str = Field( + "postgresql+psycopg://lexcam:lexcam_dev@postgres:5432/lexcam_knowledge", + alias="DATABASE_URL", + ) + qdrant_url: str = Field("http://qdrant:6333", alias="QDRANT_URL") + qdrant_api_key: str | None = Field(None, alias="QDRANT_API_KEY") + qdrant_collection: str = Field("lexcam_laws", alias="QDRANT_COLLECTION") + embedding_service_url: str = Field( + "http://embedding-service:8000", alias="EMBEDDING_SERVICE_URL" + ) + redis_url: str = Field("redis://redis:6379/0", alias="REDIS_URL") + log_level: str = Field("INFO", alias="LOG_LEVEL") + plain_summary_cache_ttl_seconds: int = Field( + 7 * 24 * 60 * 60, alias="PLAIN_SUMMARY_CACHE_TTL_SECONDS" + ) + max_search_results: int = Field(10, alias="MAX_SEARCH_RESULTS") + + +settings = Settings() diff --git a/services/knowledge-base-service/app/db.py b/services/knowledge-base-service/app/db.py new file mode 100644 index 0000000..26ed99b --- /dev/null +++ b/services/knowledge-base-service/app/db.py @@ -0,0 +1,13 @@ +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from app.config import settings + +engine = create_engine( + settings.database_url.replace("postgresql://", "postgresql+psycopg://", 1), + future=True, + echo=False, + pool_pre_ping=True, + connect_args={"check_same_thread": False} if settings.database_url.startswith("sqlite") else {}, +) +SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) diff --git a/services/knowledge-base-service/app/main.py b/services/knowledge-base-service/app/main.py new file mode 100644 index 0000000..ce0b1fc --- /dev/null +++ b/services/knowledge-base-service/app/main.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +import logging +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from app.api.v1 import router as v1_router +from app.config import settings +from app.db import engine +from app.models import Base +from app.services.cache import create_redis_client +from app.services.embedding import EmbeddingServiceClient +from app.services.qdrant import create_qdrant_client + + +def _configure_logging() -> None: + logging.basicConfig( + level=settings.log_level, + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + _configure_logging() + Base.metadata.create_all(bind=engine) + app.state.qdrant = create_qdrant_client() + app.state.embedding_client = EmbeddingServiceClient() + app.state.redis = create_redis_client() + yield + await app.state.embedding_client.close() + await app.state.redis.close() + + +app = FastAPI( + title="LexCam Knowledge Base Service", + version="0.1.0", + lifespan=lifespan, +) + +app.include_router(v1_router, prefix=settings.api_prefix) diff --git a/services/knowledge-base-service/app/models.py b/services/knowledge-base-service/app/models.py new file mode 100644 index 0000000..e0d1308 --- /dev/null +++ b/services/knowledge-base-service/app/models.py @@ -0,0 +1,41 @@ +import uuid +from sqlalchemy import Column, DateTime, ForeignKey, Index, String, Text, func +from sqlalchemy.dialects.postgresql import TSVECTOR, UUID +from sqlalchemy.orm import declarative_base, relationship + +Base = declarative_base() + + +class LawDocument(Base): + __tablename__ = "law_documents" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + code = Column(String(length=128), nullable=False) + name = Column(String(length=256), nullable=False) + jurisdiction = Column(String(length=128), nullable=False) + language = Column(String(length=16), nullable=False) + version = Column(String(length=64), nullable=True) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + + articles = relationship("LawArticle", back_populates="document") + + +class LawArticle(Base): + __tablename__ = "law_articles" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + document_id = Column(UUID(as_uuid=True), ForeignKey("law_documents.id"), nullable=False) + article_number = Column(String(length=128), nullable=False) + chapter = Column(String(length=128), nullable=True) + title = Column(String(length=512), nullable=True) + full_text = Column(Text, nullable=False) + plain_summary = Column(Text, nullable=True) + domain = Column(String(length=64), nullable=False) + language = Column(String(length=16), nullable=False) + qdrant_id = Column(String(length=128), nullable=True) + search_vector = Column(TSVECTOR, nullable=True) + + document = relationship("LawDocument", back_populates="articles") + + +Index("ix_law_articles_search_vector", LawArticle.search_vector, postgresql_using="gin") diff --git a/services/knowledge-base-service/app/schemas.py b/services/knowledge-base-service/app/schemas.py new file mode 100644 index 0000000..969fc94 --- /dev/null +++ b/services/knowledge-base-service/app/schemas.py @@ -0,0 +1,85 @@ +from datetime import datetime +from typing import List, Optional +from uuid import UUID + +from pydantic import BaseModel, ConfigDict + + +class LawDocumentSchema(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: UUID + code: str + name: str + jurisdiction: str + language: str + version: Optional[str] + created_at: datetime + + +class LawArticleSummary(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: UUID + document_id: UUID + document_code: Optional[str] + document_name: Optional[str] + article_number: str + title: Optional[str] + domain: str + language: str + + +class ArticleResponse(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: UUID + document_id: UUID + document_code: Optional[str] + document_name: Optional[str] + article_number: str + chapter: Optional[str] + title: Optional[str] + full_text: str + plain_summary: Optional[str] + domain: str + language: str + qdrant_id: Optional[str] + + +class SearchRequest(BaseModel): + query: str + language: Optional[str] = None + domain: Optional[str] = None + limit: Optional[int] = 10 + + +class SearchResultItem(BaseModel): + article_id: UUID + law_name: str + article_number: str + title: Optional[str] + domain: str + language: str + text_preview: str + score: float + + +class RetrieveRequest(BaseModel): + query_vector: List[float] + top_k: Optional[int] = 5 + domain_filter: Optional[List[str]] = None + + +class RetrieveItem(BaseModel): + article_id: UUID + law_name: str + article_number: str + domain: str + language: str + text_preview: str + score: float + + +class RetrieveResponse(BaseModel): + results: List[RetrieveItem] diff --git a/services/knowledge-base-service/app/services/cache.py b/services/knowledge-base-service/app/services/cache.py new file mode 100644 index 0000000..9daf2e2 --- /dev/null +++ b/services/knowledge-base-service/app/services/cache.py @@ -0,0 +1,7 @@ +from redis.asyncio import Redis + +from app.config import settings + + +def create_redis_client() -> Redis: + return Redis.from_url(settings.redis_url, decode_responses=True) diff --git a/services/knowledge-base-service/app/services/embedding.py b/services/knowledge-base-service/app/services/embedding.py new file mode 100644 index 0000000..f090d69 --- /dev/null +++ b/services/knowledge-base-service/app/services/embedding.py @@ -0,0 +1,31 @@ +import httpx + +from app.config import settings + + +class EmbeddingServiceClient: + def __init__(self) -> None: + self._client = httpx.AsyncClient(timeout=30.0) + + async def close(self) -> None: + await self._client.aclose() + + async def embed(self, text: str) -> list[float]: + response = await self._client.post( + f"{settings.embedding_service_url}{settings.api_prefix}/embed", + json={"texts": [text]}, + ) + response.raise_for_status() + payload = response.json() + embeddings = payload.get("embeddings") + if not embeddings or not isinstance(embeddings, list): + raise RuntimeError("Invalid embedding response from Embedding Service") + return embeddings[0] + + async def health(self) -> dict: + try: + resp = await self._client.get(f"{settings.embedding_service_url}{settings.api_prefix}/health") + resp.raise_for_status() + return resp.json() + except Exception: + return {"status": "unreachable"} diff --git a/services/knowledge-base-service/app/services/qdrant.py b/services/knowledge-base-service/app/services/qdrant.py new file mode 100644 index 0000000..653dd36 --- /dev/null +++ b/services/knowledge-base-service/app/services/qdrant.py @@ -0,0 +1,10 @@ +from qdrant_client import QdrantClient + +from app.config import settings + + +def create_qdrant_client() -> QdrantClient: + client_kwargs = {"url": settings.qdrant_url} + if settings.qdrant_api_key: + client_kwargs["api_key"] = settings.qdrant_api_key + return QdrantClient(**client_kwargs) diff --git a/services/knowledge-base-service/requirements.txt b/services/knowledge-base-service/requirements.txt new file mode 100644 index 0000000..70942c8 --- /dev/null +++ b/services/knowledge-base-service/requirements.txt @@ -0,0 +1,11 @@ +fastapi>=0.115.0,<1.0.0 +uvicorn[standard]==0.30.1 +gunicorn==22.0.0 +pydantic>=2.10.0,<3.0.0 +pydantic-settings>=2.7.0,<3.0.0 +sqlalchemy==2.0.32 +psycopg[binary]==3.3.1 +qdrant-client==1.8.1 +httpx==0.28.1 +redis==5.3.1 +pytest>=8.2.0,<10.0.0 diff --git a/services/knowledge-base-service/scripts/seed_kb.py b/services/knowledge-base-service/scripts/seed_kb.py new file mode 100644 index 0000000..bda5b39 --- /dev/null +++ b/services/knowledge-base-service/scripts/seed_kb.py @@ -0,0 +1,124 @@ +"""Seed the Knowledge Base with a sample document and article. + +Usage: + python scripts/seed_kb.py + +Requires environment variables (defaults provided for local dev with docker-compose): +- DATABASE_URL +- QDRANT_URL +- EMBEDDING_SERVICE_URL +""" +import os +import uuid +import json + +import httpx +from qdrant_client import QdrantClient +from qdrant_client.http import models as rest +import psycopg + +DATABASE_URL = os.environ.get( + "DATABASE_URL", "postgresql+psycopg://lexcam:lexcam_dev@localhost:5432/lexcam_knowledge" +) +PSYCOPG_DATABASE_URL = DATABASE_URL.replace("postgresql+psycopg://", "postgresql://", 1) +QDRANT_URL = os.environ.get("QDRANT_URL", "http://localhost:6333") +QDRANT_COLLECTION = os.environ.get("QDRANT_COLLECTION", "lexcam_laws") +EMBEDDING_SERVICE_URL = os.environ.get("EMBEDDING_SERVICE_URL", "http://localhost:8001/api/v1") + +SAMPLE_DOC = { + "code": "LABOR", + "name": "Cameroon Labour Code", + "jurisdiction": "cameroon", + "language": "fr", +} + +SAMPLE_ARTICLE = { + "article_number": "Art. 34", + "chapter": None, + "title": "Protection des travailleurs", + "full_text": "Tout employeur est tenu d'assurer la protection de la santé et de la sécurité des travailleurs.", + "plain_summary": "Employers must ensure worker health and safety.", + "domain": "labor", + "language": "fr", +} + + +def embed_text(text: str) -> list[float]: + url = f"{EMBEDDING_SERVICE_URL}/embed" + with httpx.Client(timeout=30.0) as client: + r = client.post(url, json={"texts": [text]}) + r.raise_for_status() + payload = r.json() + embeddings = payload.get("embeddings") + if not embeddings: + raise RuntimeError("No embeddings returned") + return embeddings[0] + + +def upsert_qdrant(client: QdrantClient, point_id: str, vector: list[float], payload: dict): + client.upsert( + collection_name=QDRANT_COLLECTION, + points=[rest.PointStruct(id=point_id, vector=vector, payload=payload)], + ) + + +def main(): + print("Connecting to DB...", PSYCOPG_DATABASE_URL) + with psycopg.connect(PSYCOPG_DATABASE_URL, autocommit=True) as conn: + with conn.cursor() as cur: + # Insert document + cur.execute( + "INSERT INTO law_documents (code, name, jurisdiction, language) VALUES (%s,%s,%s,%s) RETURNING id", + (SAMPLE_DOC["code"], SAMPLE_DOC["name"], SAMPLE_DOC["jurisdiction"], SAMPLE_DOC["language"]), + ) + doc_id = cur.fetchone()[0] + print("Inserted document id", doc_id) + + # Insert article + cur.execute( + "INSERT INTO law_articles (document_id, article_number, chapter, title, full_text, plain_summary, domain, language) VALUES (%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id", + ( + doc_id, + SAMPLE_ARTICLE["article_number"], + SAMPLE_ARTICLE["chapter"], + SAMPLE_ARTICLE["title"], + SAMPLE_ARTICLE["full_text"], + SAMPLE_ARTICLE["plain_summary"], + SAMPLE_ARTICLE["domain"], + SAMPLE_ARTICLE["language"], + ), + ) + article_id = cur.fetchone()[0] + print("Inserted article id", article_id) + + # Get embedding and upsert to Qdrant + print("Requesting embedding...") + vector = embed_text(SAMPLE_ARTICLE["full_text"]) # list of floats + + q = QdrantClient(url=QDRANT_URL) + point_id = str(uuid.uuid4()) + payload = { + "article_id": str(article_id), + "law_name": SAMPLE_DOC["name"], + "article_number": SAMPLE_ARTICLE["article_number"], + "domain": SAMPLE_ARTICLE["domain"], + "language": SAMPLE_ARTICLE["language"], + "text_preview": SAMPLE_ARTICLE["full_text"][:200], + } + print("Upserting to Qdrant, point id", point_id) + upsert_qdrant(q, point_id, vector, payload) + + # Update article qdrant_id in DB + with psycopg.connect(PSYCOPG_DATABASE_URL, autocommit=True) as conn: + with conn.cursor() as cur: + cur.execute( + "UPDATE law_articles SET qdrant_id = %s WHERE id = %s", + (point_id, article_id), + ) + print("Updated article qdrant_id") + + print("Seed complete") + + +if __name__ == "__main__": + main() diff --git a/services/knowledge-base-service/sql/init_kb_schema.sql b/services/knowledge-base-service/sql/init_kb_schema.sql new file mode 100644 index 0000000..4fec0c0 --- /dev/null +++ b/services/knowledge-base-service/sql/init_kb_schema.sql @@ -0,0 +1,42 @@ +-- Knowledge Base schema for lexcam_knowledge +CREATE EXTENSION IF NOT EXISTS pgcrypto; + +CREATE TABLE IF NOT EXISTS law_documents ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + code VARCHAR(128) NOT NULL, + name VARCHAR(256) NOT NULL, + jurisdiction VARCHAR(128) NOT NULL, + language VARCHAR(16) NOT NULL, + version VARCHAR(64), + created_at TIMESTAMPTZ DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS law_articles ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + document_id UUID NOT NULL REFERENCES law_documents(id) ON DELETE CASCADE, + article_number VARCHAR(128) NOT NULL, + chapter VARCHAR(128), + title VARCHAR(512), + full_text TEXT NOT NULL, + plain_summary TEXT, + domain VARCHAR(64) NOT NULL, + language VARCHAR(16) NOT NULL, + qdrant_id VARCHAR(128), + search_vector TSVECTOR +); + +CREATE INDEX IF NOT EXISTS ix_law_articles_search_vector ON law_articles USING GIN(search_vector); + +CREATE OR REPLACE FUNCTION law_articles_search_vector_trigger() RETURNS trigger AS $$ +begin + new.search_vector := to_tsvector('simple', coalesce(new.full_text, '')); + return new; +end +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS tsvectorupdate ON law_articles; +CREATE TRIGGER tsvectorupdate BEFORE INSERT OR UPDATE ON law_articles +FOR EACH ROW EXECUTE FUNCTION law_articles_search_vector_trigger(); + +-- Optional helper to reindex all existing rows +-- UPDATE law_articles SET search_vector = to_tsvector('simple', coalesce(full_text, '')); diff --git a/services/knowledge-base-service/tests/test_embedding_client.py b/services/knowledge-base-service/tests/test_embedding_client.py new file mode 100644 index 0000000..869b9e7 --- /dev/null +++ b/services/knowledge-base-service/tests/test_embedding_client.py @@ -0,0 +1,29 @@ +import sys +import respx +import pytest +import httpx + +# Ensure the knowledge-base-service `app` package is imported first +sys.path.insert(0, "services/knowledge-base-service") + +from app.services.embedding import EmbeddingServiceClient +from app.config import settings + + +@respx.mock +@pytest.mark.asyncio +async def test_embedding_client_embed_and_health(monkeypatch): + # Mock embedding service endpoints + base = settings.embedding_service_url + settings.api_prefix + embed_route = respx.post(f"{base}/embed").mock(return_value=httpx.Response(200, json={"embeddings": [[0.1, 0.2, 0.3]]})) + health_route = respx.get(f"{base}/health").mock(return_value=httpx.Response(200, json={"status": "ok"})) + + client = EmbeddingServiceClient() + emb = await client.embed("hello world") + assert isinstance(emb, list) + assert emb == [0.1, 0.2, 0.3] + + h = await client.health() + assert h["status"] == "ok" + + await client.close() diff --git a/services/knowledge-base-service/tests/test_health.py b/services/knowledge-base-service/tests/test_health.py new file mode 100644 index 0000000..85f9ded --- /dev/null +++ b/services/knowledge-base-service/tests/test_health.py @@ -0,0 +1,79 @@ +import sys +import types +import pytest +from fastapi.testclient import TestClient + + +class FakeQdrant: + def get_collection(self, name): + return {"name": name} + + +class FakeEmbeddingClient: + def __init__(self): + pass + + async def health(self): + return {"status": "ok"} + + async def close(self): + return None + + +class FakeRedis: + async def ping(self): + return True + + async def close(self): + return None + + +def test_health_endpoint(monkeypatch): + # Ensure the knowledge-base-service `app` package is imported first + import sys + kb_path = "services/knowledge-base-service" + sys.path.insert(0, kb_path) + + # Use an in-memory SQLite DB for tests to avoid attempting to connect to Postgres + monkeypatch.setenv("DATABASE_URL", "sqlite:///:memory:") + + # Ensure app config/db pick up the test DATABASE_URL + import importlib + try: + import app.config as _config + importlib.reload(_config) + except Exception: + pass + try: + import app.db as _db + importlib.reload(_db) + except Exception: + pass + + # Patch creation functions so lifespan uses our fakes + # Prevent metadata.create_all from running DB-specific DDL in tests + try: + import app.models as _models + _models.Base.metadata.create_all = lambda bind=None: None + except Exception: + pass + + import app.services.qdrant as qmod + import app.services.embedding as emod + import app.services.cache as cmod + + monkeypatch.setattr(qmod, "create_qdrant_client", lambda: FakeQdrant()) + monkeypatch.setattr(emod, "EmbeddingServiceClient", lambda: FakeEmbeddingClient()) + monkeypatch.setattr(cmod, "create_redis_client", lambda: FakeRedis()) + + # Import app after monkeypatching so lifespan uses the fakes + from app.main import app + + with TestClient(app) as client: + resp = client.get("/api/v1/health") + assert resp.status_code == 200 + body = resp.json() + assert body["status"] == "ok" + assert body["qdrant_collection"] is not None + assert body["embedding_service"]["status"] == "ok" + assert body["redis"] == "ok" diff --git a/services/notification-service/.dockerignore b/services/notification-service/.dockerignore new file mode 100644 index 0000000..ac56efb --- /dev/null +++ b/services/notification-service/.dockerignore @@ -0,0 +1,6 @@ +__pycache__ +*.pyc +*.pyo +.pytest_cache +venv +.env diff --git a/services/notification-service/Dockerfile b/services/notification-service/Dockerfile new file mode 100644 index 0000000..35069f5 --- /dev/null +++ b/services/notification-service/Dockerfile @@ -0,0 +1,13 @@ +FROM python:3.11-slim +WORKDIR /app +ENV PYTHONDONTWRITEBYTECODE=1 +ENV PYTHONUNBUFFERED=1 + +COPY requirements.txt ./ +RUN pip install --no-cache-dir -r requirements.txt + +COPY app ./app +COPY alembic.ini ./ +COPY alembic ./alembic + +CMD ["sh", "-c", "alembic -c /app/alembic.ini upgrade head && gunicorn -k uvicorn.workers.UvicornWorker app.main:app --bind 0.0.0.0:8000 --workers 2"] diff --git a/services/notification-service/alembic.ini b/services/notification-service/alembic.ini new file mode 100644 index 0000000..3e04d18 --- /dev/null +++ b/services/notification-service/alembic.ini @@ -0,0 +1,24 @@ +[alembic] +script_location = alembic + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = INFO +handlers = console + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s diff --git a/services/notification-service/alembic/env.py b/services/notification-service/alembic/env.py new file mode 100644 index 0000000..306c4d5 --- /dev/null +++ b/services/notification-service/alembic/env.py @@ -0,0 +1,36 @@ +from alembic import context +import os +import sys + +sys.path.append(os.path.dirname(os.path.dirname(__file__))) +sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) + +from app.config import settings +from app.db import engine +from app.models import Notification +from app.models_user import User + +config = context.config + +target_metadata = [Notification.metadata, User.metadata] + + +def run_migrations_offline(): + url = str(settings.DATABASE_URL) + context.configure(url=url, target_metadata=target_metadata, literal_binds=True) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online(): + connectable = engine + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/services/notification-service/alembic/versions/0001_initial.py b/services/notification-service/alembic/versions/0001_initial.py new file mode 100644 index 0000000..e4fc505 --- /dev/null +++ b/services/notification-service/alembic/versions/0001_initial.py @@ -0,0 +1,51 @@ +"""initial + +Revision ID: 0001_initial +Revises: +Create Date: 2026-05-17 + +""" +from alembic import op + +revision = '0001_initial' +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade(): + op.execute(""" + CREATE TABLE IF NOT EXISTS "user" ( + id SERIAL PRIMARY KEY, + username VARCHAR(128) NOT NULL UNIQUE, + email VARCHAR(256), + hashed_password VARCHAR(256) NOT NULL, + is_active BOOLEAN NOT NULL DEFAULT TRUE + ) + """) + op.execute(""" + CREATE TABLE IF NOT EXISTS notification ( + id SERIAL PRIMARY KEY, + user_id VARCHAR(128), + message TEXT NOT NULL, + delivered BOOLEAN NOT NULL DEFAULT FALSE, + created_at TIMESTAMP WITHOUT TIME ZONE + ) + """) + op.execute(""" + CREATE TABLE IF NOT EXISTS deliverylog ( + id SERIAL PRIMARY KEY, + notification_id INTEGER, + to_address VARCHAR(256), + subject VARCHAR(256), + status VARCHAR(64), + error TEXT, + sent_at TIMESTAMP WITHOUT TIME ZONE + ) + """) + + +def downgrade(): + op.execute('DROP TABLE IF EXISTS deliverylog') + op.execute('DROP TABLE IF EXISTS notification') + op.execute('DROP TABLE IF EXISTS "user"') diff --git a/services/notification-service/app/__init__.py b/services/notification-service/app/__init__.py new file mode 100644 index 0000000..e4c7aa5 --- /dev/null +++ b/services/notification-service/app/__init__.py @@ -0,0 +1 @@ +from .main import app diff --git a/services/notification-service/app/api/v1/auth_routes.py b/services/notification-service/app/api/v1/auth_routes.py new file mode 100644 index 0000000..cf4411c --- /dev/null +++ b/services/notification-service/app/api/v1/auth_routes.py @@ -0,0 +1,32 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from fastapi.security import OAuth2PasswordRequestForm +from sqlmodel import Session +from ...db import get_session +from ...models_user import User +from ...auth import verify_password, get_password_hash, create_access_token +from datetime import timedelta +from ...config import settings + +router = APIRouter(prefix="/v1/auth") + + +@router.post("/signup", status_code=status.HTTP_201_CREATED) +def signup(username: str, password: str, session: Session = Depends(get_session)): + existing = session.exec(User.select().where(User.username == username)).first() + if existing: + raise HTTPException(status_code=400, detail="User already exists") + user = User(username=username, hashed_password=get_password_hash(password)) + session.add(user) + session.commit() + session.refresh(user) + return {"id": user.id, "username": user.username} + + +@router.post("/token") +def login_token(form_data: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)): + user = session.exec(User.select().where(User.username == form_data.username)).first() + if not user or not verify_password(form_data.password, user.hashed_password): + raise HTTPException(status_code=400, detail="Incorrect username or password") + access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) + access_token = create_access_token(data={"sub": user.username}, expires_delta=access_token_expires) + return {"access_token": access_token, "token_type": "bearer"} diff --git a/services/notification-service/app/api/v1/routes.py b/services/notification-service/app/api/v1/routes.py new file mode 100644 index 0000000..a681b9f --- /dev/null +++ b/services/notification-service/app/api/v1/routes.py @@ -0,0 +1,32 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from sqlmodel import Session, select +from typing import List + +from ...db import get_session +from ...models import Notification +from ...schemas import NotificationCreate, NotificationRead +from ...tasks import send_notification_async + +router = APIRouter(prefix="/v1") + + +@router.get("/health", status_code=200) +def health(): + return {"status": "ok"} + + +@router.post("/notify", response_model=NotificationRead, status_code=status.HTTP_201_CREATED) +def create_notification(payload: NotificationCreate, session: Session = Depends(get_session)): + n = Notification.from_orm(payload) + session.add(n) + session.commit() + session.refresh(n) + send_notification_async.delay(n.id) + return n + + +@router.get("/notifications", response_model=List[NotificationRead]) +def list_notifications(limit: int = 50, session: Session = Depends(get_session)): + statement = select(Notification).limit(limit) + results = session.exec(statement).all() + return results diff --git a/services/notification-service/app/auth.py b/services/notification-service/app/auth.py new file mode 100644 index 0000000..330289d --- /dev/null +++ b/services/notification-service/app/auth.py @@ -0,0 +1,53 @@ +from datetime import datetime, timedelta +from typing import Optional +from passlib.context import CryptContext +from jose import JWTError, jwt +from pydantic import BaseModel +from fastapi import Depends, HTTPException, status +from fastapi.security import OAuth2PasswordBearer +from .config import settings +from .db import get_session +from sqlmodel import Session +from .models_user import User + +pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/v1/auth/token") + +class TokenData(BaseModel): + username: Optional[str] = None + + +def verify_password(plain_password, hashed_password): + return pwd_context.verify(plain_password, hashed_password) + + +def get_password_hash(password): + return pwd_context.hash(password) + + +def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): + to_encode = data.copy() + expire = datetime.utcnow() + (expires_delta or timedelta(minutes=60)) + to_encode.update({"exp": expire}) + encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) + return encoded_jwt + + +def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)): + credentials_exception = HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Could not validate credentials", + headers={"WWW-Authenticate": "Bearer"}, + ) + try: + payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) + username: str = payload.get("sub") + if username is None: + raise credentials_exception + token_data = TokenData(username=username) + except JWTError: + raise credentials_exception + user = session.exec(User.select().where(User.username == token_data.username)).first() + if user is None: + raise credentials_exception + return user diff --git a/services/notification-service/app/config.py b/services/notification-service/app/config.py new file mode 100644 index 0000000..811406c --- /dev/null +++ b/services/notification-service/app/config.py @@ -0,0 +1,25 @@ +from pydantic import BaseSettings, AnyUrl + + +class Settings(BaseSettings): + SERVICE_NAME: str = "notification-service" + DATABASE_URL: AnyUrl = "postgresql://postgres:postgres@db:5432/notification_db" + REDIS_URL: str = "redis://redis:6379/0" + CELERY_BROKER_URL: str = "redis://redis:6379/1" + CELERY_RESULT_BACKEND: str = "redis://redis:6379/2" + MIGRATE_AUTO: bool = True + SECRET_KEY: str = "please-change-me" + ALGORITHM: str = "HS256" + ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 + USER_MGMT_URL: str = "http://user-management:8000" + RABBITMQ_URL: str = "amqp://lexcam:lexcam_dev@rabbitmq:5672/%2F" + SMTP_HOST: str = "smtp.example.com" + SMTP_PORT: int = 587 + SMTP_USER: str = "" + SMTP_PASS: str = "" + + class Config: + env_file = ".env" + + +settings = Settings() diff --git a/services/notification-service/app/db.py b/services/notification-service/app/db.py new file mode 100644 index 0000000..58e3353 --- /dev/null +++ b/services/notification-service/app/db.py @@ -0,0 +1,13 @@ +from sqlmodel import SQLModel, create_engine, Session +from .config import settings + +engine = create_engine(str(settings.DATABASE_URL), echo=False) + + +def init_db(): + return None + + +def get_session(): + with Session(engine) as session: + yield session diff --git a/services/notification-service/app/events_consumer.py b/services/notification-service/app/events_consumer.py new file mode 100644 index 0000000..b74fba8 --- /dev/null +++ b/services/notification-service/app/events_consumer.py @@ -0,0 +1,64 @@ +import json +import pika +import os +from .config import settings +from .db import Session, engine +from .models import Notification + + +def _get_connection(): + params = pika.URLParameters(settings.RABBITMQ_URL) + return pika.BlockingConnection(params) + + +def on_message(ch, method, properties, body): + payload = json.loads(body) + routing_key = method.routing_key + # For simplicity, create a notification record with payload and enqueue delivery + with Session(engine) as session: + # map event to notification content + if routing_key == "feedback.flagged": + message = f"Feedback flagged: {payload.get('feedback_id')}" + user = payload.get('session_id') or 'admin@example.com' + elif routing_key == "lawyers.scraped": + message = f"New lawyers scraped batch" + user = 'admin@example.com' + else: + message = f"Event {routing_key}" + user = 'admin@example.com' + notif = Notification(user_id=user, message=message) + session.add(notif) + session.commit() + session.refresh(notif) + # Acknowledge message after successful DB write. Delivery of the + # notification (SMTP, push, etc.) is handled by dedicated worker(s) + # implemented by the team. Do NOT perform delivery here. + try: + ch.basic_ack(delivery_tag=method.delivery_tag) + except Exception: + # If ack fails, attempt to nack without requeue so DLX can capture it + try: + ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False) + except Exception: + pass + + +def start_consumer(): + conn = _get_connection() + ch = conn.channel() + # Declare main exchange and a DLX exchange for failed messages. + ch.exchange_declare(exchange="lexcam.events", exchange_type="topic", durable=True) + ch.exchange_declare(exchange="lexcam.dlx", exchange_type="fanout", durable=True) + # Declare queue with dead-letter exchange so failed messages route to DLX. + args = {"x-dead-letter-exchange": "lexcam.dlx"} + q = ch.queue_declare(queue="notification_events", durable=True, arguments=args) + ch.queue_bind(queue="notification_events", exchange="lexcam.events", routing_key="feedback.flagged") + ch.queue_bind(queue="notification_events", exchange="lexcam.events", routing_key="lawyers.scraped") + # Use explicit acknowledgements to ensure messages are not lost + ch.basic_consume(queue="notification_events", on_message_callback=on_message, auto_ack=False) + print("[notification-consumer] Waiting for events...") + ch.start_consuming() + + +if __name__ == '__main__': + start_consumer() diff --git a/services/notification-service/app/main.py b/services/notification-service/app/main.py new file mode 100644 index 0000000..bdff8f5 --- /dev/null +++ b/services/notification-service/app/main.py @@ -0,0 +1,21 @@ +from fastapi import FastAPI +from .api.v1 import routes as v1_routes +from .api.v1 import auth_routes +from .db import init_db +from .config import settings +from .monitoring import init_logging, prometheus_middleware, prometheus_endpoint + +app = FastAPI(title=settings.SERVICE_NAME) + + +@app.on_event("startup") +def on_startup(): + init_db() + init_logging() + + +app.middleware("http")(prometheus_middleware) +app.add_api_route("/metrics", prometheus_endpoint, methods=["GET"]) + +app.include_router(auth_routes.router) +app.include_router(v1_routes.router) diff --git a/services/notification-service/app/models.py b/services/notification-service/app/models.py new file mode 100644 index 0000000..143a5dc --- /dev/null +++ b/services/notification-service/app/models.py @@ -0,0 +1,11 @@ +from typing import Optional +from datetime import datetime +from sqlmodel import SQLModel, Field + + +class Notification(SQLModel, table=True): + id: Optional[int] = Field(default=None, primary_key=True) + user_id: Optional[str] = Field(index=True) + message: str + delivered: bool = Field(default=False) + created_at: datetime = Field(default_factory=datetime.utcnow) diff --git a/services/notification-service/app/models_delivery.py b/services/notification-service/app/models_delivery.py new file mode 100644 index 0000000..9b49bef --- /dev/null +++ b/services/notification-service/app/models_delivery.py @@ -0,0 +1,13 @@ +from typing import Optional +from sqlmodel import SQLModel, Field +from datetime import datetime + + +class DeliveryLog(SQLModel, table=True): + id: Optional[int] = Field(default=None, primary_key=True) + notification_id: Optional[int] = Field(index=True) + to_address: str + subject: str + status: str = Field(default="pending") + error: Optional[str] = None + sent_at: Optional[datetime] = None diff --git a/services/notification-service/app/models_user.py b/services/notification-service/app/models_user.py new file mode 100644 index 0000000..37c60d6 --- /dev/null +++ b/services/notification-service/app/models_user.py @@ -0,0 +1,13 @@ +from typing import Optional +from sqlmodel import SQLModel, Field +from datetime import datetime + + +class User(SQLModel, table=True): + id: Optional[int] = Field(default=None, primary_key=True) + username: str = Field(index=True, unique=True) + email: Optional[str] = None + hashed_password: str + is_active: bool = True + is_superuser: bool = False + created_at: datetime = Field(default_factory=datetime.utcnow) diff --git a/services/notification-service/app/monitoring.py b/services/notification-service/app/monitoring.py new file mode 100644 index 0000000..b21d99e --- /dev/null +++ b/services/notification-service/app/monitoring.py @@ -0,0 +1,26 @@ +import logging +from prometheus_client import Counter, generate_latest, CONTENT_TYPE_LATEST +from fastapi import Request, Response +from fastapi.responses import PlainTextResponse +import structlog + +REQUEST_COUNT = Counter("http_requests_total", "Total HTTP requests", ["method", "endpoint", "http_status"]) + + +def init_logging(): + structlog.configure(processors=[structlog.processors.JSONRenderer()]) + logging.basicConfig(level=logging.INFO) + + +async def prometheus_endpoint(): + data = generate_latest() + return Response(content=data, media_type=CONTENT_TYPE_LATEST) + + +async def prometheus_middleware(request: Request, call_next): + response = await call_next(request) + try: + REQUEST_COUNT.labels(request.method, request.url.path, str(response.status_code)).inc() + except Exception: + pass + return response diff --git a/services/notification-service/app/schemas.py b/services/notification-service/app/schemas.py new file mode 100644 index 0000000..95d68a0 --- /dev/null +++ b/services/notification-service/app/schemas.py @@ -0,0 +1,19 @@ +from typing import Optional +from datetime import datetime +from pydantic import BaseModel + + +class NotificationCreate(BaseModel): + user_id: Optional[str] + message: str + + +class NotificationRead(BaseModel): + id: int + user_id: Optional[str] + message: str + delivered: bool + created_at: datetime + + class Config: + orm_mode = True diff --git a/services/notification-service/app/tasks.py b/services/notification-service/app/tasks.py new file mode 100644 index 0000000..1ef51a9 --- /dev/null +++ b/services/notification-service/app/tasks.py @@ -0,0 +1,28 @@ +from celery import Celery +from celery.utils.log import get_task_logger +from .config import settings + +logger = get_task_logger(__name__) + +celery_app = Celery( + settings.SERVICE_NAME, + broker=settings.CELERY_BROKER_URL, + backend=settings.CELERY_RESULT_BACKEND, +) + + +@celery_app.task(bind=True) +def send_notification_async(self, notification_id: int): + """Placeholder Celery task for notification delivery. + + The concrete delivery implementation (SMTP, push, retry policy, + delivery logging) is intentionally left to the team per docs. This + placeholder raises `NotImplementedError` to avoid accidental sends + during development. + """ + logger.warning("send_notification_async called for %s -- placeholder", notification_id) + raise NotImplementedError("Notification delivery is implemented by the team") + + +def get_celery_app(): + return celery_app diff --git a/services/notification-service/requirements.txt b/services/notification-service/requirements.txt new file mode 100644 index 0000000..586a243 --- /dev/null +++ b/services/notification-service/requirements.txt @@ -0,0 +1,22 @@ +fastapi==0.95.2 +uvicorn[standard]==0.22.0 +SQLModel==0.0.8 +SQLAlchemy==1.4.41 +psycopg2-binary==2.9.7 +python-dotenv==1.0.0 +celery==5.3.0 +redis==4.5.1 +gunicorn==20.1.0 +httpx==0.24.1 +pydantic==1.10.9 +pytest==7.4.0 +passlib[bcrypt]==1.7.4 +python-jose==3.3.0 +alembic==1.11.1 +prometheus_client==0.16.0 +structlog==23.3.0 +python-json-logger==2.0.7 +pika==1.3.1 +jinja2==3.1.2 +aiosmtplib==5.1.0 +python-multipart==0.0.9 diff --git a/services/notification-service/tests/test_health.py b/services/notification-service/tests/test_health.py new file mode 100644 index 0000000..b5703e1 --- /dev/null +++ b/services/notification-service/tests/test_health.py @@ -0,0 +1,9 @@ +from fastapi.testclient import TestClient +from app.main import app + + +def test_health(): + client = TestClient(app) + r = client.get("/v1/health") + assert r.status_code == 200 + assert r.json() == {"status": "ok"} diff --git a/services/rag-service/.dockerignore b/services/rag-service/.dockerignore new file mode 100644 index 0000000..a7fef50 --- /dev/null +++ b/services/rag-service/.dockerignore @@ -0,0 +1,8 @@ +__pycache__/ +*.pyc +*.pyo +.venv/ +venv/ +env/ +build/ +dist/ diff --git a/services/rag-service/Dockerfile b/services/rag-service/Dockerfile new file mode 100644 index 0000000..85040c8 --- /dev/null +++ b/services/rag-service/Dockerfile @@ -0,0 +1,6 @@ +FROM python:3.11-slim +WORKDIR /app +COPY requirements.txt ./ +RUN pip install --no-cache-dir -r requirements.txt +COPY app ./app +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/services/rag-service/README.md b/services/rag-service/README.md new file mode 100644 index 0000000..2127c25 --- /dev/null +++ b/services/rag-service/README.md @@ -0,0 +1,8 @@ +Small RAG (retrieval-augmented generation) service stub. + +Endpoints: +- POST /api/v1/query — accepts {"query": str, "top_k": int} + +Behavior: +- For local dev this service calls the `knowledge-base-service` `/api/v1/search` endpoint + to retrieve top documents and returns them along with a simple synthesized answer. diff --git a/services/rag-service/alembic/env.py b/services/rag-service/alembic/env.py new file mode 100644 index 0000000..0192002 --- /dev/null +++ b/services/rag-service/alembic/env.py @@ -0,0 +1,68 @@ +import os +import sys +from logging.config import fileConfig +from sqlalchemy import engine_from_config +from sqlalchemy import pool +from alembic import context + +# ensure project path is importable +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +# this is the Alembic Config object, which provides +# access to the values within the .ini file in use. +config = context.config + +# Interpret the config file for Python logging. +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +# Import your model's MetaData object here for 'autogenerate' support +try: + from app.models import Base + target_metadata = Base.metadata +except Exception: + target_metadata = None + +# other values from the config, defined by the needs of env.py, +# can be acquired: +# my_important_option = config.get_main_option("my_important_option") +# ... etc. + + +def get_url(): + return os.getenv("DATABASE_URL", "sqlite:///./rag.db") + + +def run_migrations_offline(): + url = get_url() + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online(): + configuration = config.get_section(config.config_ini_section) or {} + configuration["sqlalchemy.url"] = get_url() + connectable = engine_from_config( + configuration, + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/services/rag-service/alembic/script.py.mako b/services/rag-service/alembic/script.py.mako new file mode 100644 index 0000000..08ca073 --- /dev/null +++ b/services/rag-service/alembic/script.py.mako @@ -0,0 +1,22 @@ +""" +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = '${up_revision}' +down_revision = ${repr(down_revision)} +branch_labels = None +dependencies = None + + +def upgrade(): +${upgrades if upgrades else " pass"} + + +def downgrade(): +${downgrades if downgrades else " pass"} diff --git a/services/rag-service/alembic/versions/0001_create_rag_sessions.py b/services/rag-service/alembic/versions/0001_create_rag_sessions.py new file mode 100644 index 0000000..2bffad3 --- /dev/null +++ b/services/rag-service/alembic/versions/0001_create_rag_sessions.py @@ -0,0 +1,30 @@ +"""create rag_sessions table + +Revision ID: 0001_create_rag_sessions +Revises: +Create Date: 2026-05-17 00:00:00.000000 +""" +from alembic import op +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision = '0001_create_rag_sessions' +down_revision = None +branch_labels = None +dependencies = None + + +def upgrade(): + op.create_table( + 'rag_sessions', + sa.Column('id', sa.Integer, primary_key=True), + sa.Column('query', sa.Text, nullable=False), + sa.Column('prompt', sa.Text, nullable=True), + sa.Column('response', sa.Text, nullable=True), + sa.Column('sources', sa.JSON(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + + +def downgrade(): + op.drop_table('rag_sessions') diff --git a/services/rag-service/app/__init__.py b/services/rag-service/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/rag-service/app/db.py b/services/rag-service/app/db.py new file mode 100644 index 0000000..62edc59 --- /dev/null +++ b/services/rag-service/app/db.py @@ -0,0 +1,8 @@ +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +import os + +DATABASE_URL = os.getenv("RAG_DATABASE_URL", "sqlite:///./rag_sessions.db") + +engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}) +SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) diff --git a/services/rag-service/app/events.py b/services/rag-service/app/events.py new file mode 100644 index 0000000..ff3cd2d --- /dev/null +++ b/services/rag-service/app/events.py @@ -0,0 +1,30 @@ +import os +import json +from typing import Optional +import asyncio + +import aio_pika + +RABBITMQ_URL = os.getenv("RABBITMQ_URL") + + +async def publish_matching_requested(domain: str, session_id: Optional[int], city: Optional[str] = None) -> None: + if not RABBITMQ_URL: + return + + payload = { + "domain": domain, + "session_id": session_id, + "city": city, + } + try: + connection = await aio_pika.connect_robust(RABBITMQ_URL) + async with connection: + channel = await connection.channel() + body = json.dumps(payload).encode() + await channel.default_exchange.publish( + aio_pika.Message(body), routing_key="matching.requested" + ) + except Exception: + # don't raise — event publishing should not block RAG responses + return diff --git a/services/rag-service/app/llm.py b/services/rag-service/app/llm.py new file mode 100644 index 0000000..7e51920 --- /dev/null +++ b/services/rag-service/app/llm.py @@ -0,0 +1,70 @@ +import os +import httpx +from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type +from typing import List, Dict + +HF_API_URL = "https://api-inference.huggingface.co/models" + + +class LLMError(Exception): + pass + + +@retry(reraise=True, stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), + retry=retry_if_exception_type((httpx.HTTPError, LLMError))) +def generate_answer_from_documents(query: str, documents: List[Dict[str, str]], max_tokens: int = 256) -> str: + hf_api_key = os.getenv("HF_API_KEY") + hf_model = os.getenv("HF_MODEL", "google/flan-t5-small") + fallback_model = os.getenv("HF_FALLBACK_MODEL", "gpt2") + + if not hf_api_key: + raise LLMError("HF_API_KEY not set") + + # Build prompt with provenance instructions + prompt_parts = [f"You are an assistant. Answer the question using only the provided documents." + , f"Question: {query}", "Documents:"] + for i, doc in enumerate(documents, start=1): + snippet = doc.get("snippet") or doc.get("content") or "" + prompt_parts.append(f"[{i}] id={doc.get('id')} score={doc.get('score', '')} text={snippet}") + + prompt_parts.append("Provide a concise answer and at the end list the document ids used as sources.") + prompt = "\n\n".join(prompt_parts) + + headers = {"Authorization": f"Bearer {hf_api_key}", "Accept": "application/json"} + url = f"{HF_API_URL}/{hf_model}" + payload = {"inputs": prompt, "parameters": {"max_new_tokens": max_tokens, "return_full_text": False}} + + with httpx.Client(timeout=30.0) as client: + resp = client.post(url, json=payload, headers=headers) + try: + resp.raise_for_status() + except httpx.HTTPStatusError as e: + # If model not found, try fallback model once + if e.response.status_code == 404: + # Model not hosted or unavailable; fall back to deterministic synthesis + # Build simple concatenation of top snippets as a safe fallback + snippets = [d.get("snippet") or d.get("content") or "" for d in documents] + synthesized = "\n\n".join(snippets[:3]) or "No relevant documents found." + return f"(fallback synthesis) {synthesized}\n\nNote: HF model {hf_model} unavailable." + if e.response.status_code == 404 and fallback_model and fallback_model != hf_model: + url_fb = f"{HF_API_URL}/{fallback_model}" + resp_fb = client.post(url_fb, json=payload, headers=headers) + try: + resp_fb.raise_for_status() + body = resp_fb.json() + except httpx.HTTPStatusError as e2: + raise LLMError(f"HF API error (fallback): {e2.response.status_code} {e2.response.text}") + else: + raise LLMError(f"HF API error: {e.response.status_code} {e.response.text}") + else: + body = resp.json() + # HF Inference API can return different shapes depending on model; handle common cases + if isinstance(body, list) and len(body) and "generated_text" in body[0]: + return body[0]["generated_text"] + if isinstance(body, dict) and "generated_text" in body: + return body["generated_text"] + # Fallback: try to pull 'text' or string + if isinstance(body, dict) and "error" in body: + raise LLMError(f"HF API returned error: {body['error']}") + # Last resort: stringify + return str(body) diff --git a/services/rag-service/app/llm_adapter.py b/services/rag-service/app/llm_adapter.py new file mode 100644 index 0000000..cca35b2 --- /dev/null +++ b/services/rag-service/app/llm_adapter.py @@ -0,0 +1,84 @@ +import os +import json +from typing import AsyncGenerator, List, Dict + +import httpx + +from .llm import generate_answer_from_documents + + +def generate(prompt: str, documents: List[Dict], max_tokens: int = 256) -> str: + # synchronous wrapper to existing llm function + return generate_answer_from_documents(prompt, documents, max_tokens=max_tokens) + + +async def stream_generate(prompt: str, documents: List[Dict], max_tokens: int = 256) -> AsyncGenerator[str, None]: + """ + Stream tokens from a configured Groq console streaming endpoint. + + Behavior: + - Reads `GROQ_API_URL` and `GROQ_API_KEY` from env. + - POSTs a JSON payload with the prompt and documents. + - Expects a text/event-stream or chunked newline-delimited stream where + each chunk is either a JSON object (containing 'token' or 'text') or + plain text. Yields text fragments as they arrive. + + Falls back to the synchronous `generate` implementation when no GROQ + configuration is available. + """ + groq_url = os.getenv("GROQ_API_URL") + groq_key = os.getenv("GROQ_API_KEY") + + if not groq_url or not groq_key: + # fallback to non-streaming generation, then chunk + answer = generate(prompt, documents, max_tokens=max_tokens) + # simple sentence splitting + import re + + parts = re.split(r'(?<=[\.\!\?])\s+', answer) + if len(parts) == 1: + chunk_size = 120 + for i in range(0, len(answer), chunk_size): + yield answer[i : i + chunk_size] + return + for p in parts: + if p.strip(): + yield p.strip() + return + + headers = {"Authorization": f"Bearer {groq_key}", "Accept": "text/event-stream", "Content-Type": "application/json"} + payload = {"prompt": prompt, "documents": documents, "max_tokens": max_tokens} + + async with httpx.AsyncClient(timeout=None) as client: + try: + async with client.stream("POST", groq_url, json=payload, headers=headers) as resp: + resp.raise_for_status() + async for chunk in resp.aiter_text(chunk_size=1024): + if not chunk: + continue + # split into lines and handle data: prefixes + for line in chunk.splitlines(): + line = line.strip() + if not line: + continue + # SSE-style 'data: ...' + if line.startswith("data:"): + content = line[len("data:"):].strip() + else: + content = line + # try JSON + try: + obj = json.loads(content) + # accept 'token' or 'text' fields + token = obj.get("token") or obj.get("text") + if token: + yield token + continue + except Exception: + # not JSON, yield raw + yield content + continue + except httpx.HTTPError: + # On stream failure, fallback to sync answer once + answer = generate(prompt, documents, max_tokens=max_tokens) + yield answer diff --git a/services/rag-service/app/main.py b/services/rag-service/app/main.py new file mode 100644 index 0000000..5377bcf --- /dev/null +++ b/services/rag-service/app/main.py @@ -0,0 +1,213 @@ +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import StreamingResponse +from pydantic import BaseModel +import os +import httpx +from typing import List, Any, Dict +from contextlib import asynccontextmanager + +from .llm import generate_answer_from_documents +from .llm_adapter import generate, stream_generate +from .translator import translate_text +from .events import publish_matching_requested +from langdetect import detect +import asyncio +from .db import engine, SessionLocal +from .models import Base, RagSession + + +class QueryRequest(BaseModel): + query: str + top_k: int = 5 + + +class ResultSource(BaseModel): + id: str + score: float | None = None + snippet: str | None = None + + +class QueryResponse(BaseModel): + query: str + results: List[ResultSource] + synthesized_answer: str + provenance: List[Dict[str, Any]] + + +@asynccontextmanager +async def lifespan(app: FastAPI): + # Ensure DB tables exist (local dev default: sqlite) + Base.metadata.create_all(bind=engine) + yield + + +app = FastAPI(title="rag-service", lifespan=lifespan) + +KB_URL = os.getenv("KNOWLEDGE_BASE_URL", "http://knowledge-base-service:8000") +EMBED_URL = os.getenv("EMBEDDING_SERVICE_URL", "http://embedding-service:8000") + + + + + +@app.get("/api/v1/health") +async def health(): + return {"status": "ok", "kb": KB_URL, "llm_configured": bool(os.getenv("HF_API_KEY"))} + + +def _normalize_kb_response(data: Any) -> List[Dict[str, Any]]: + if isinstance(data, list): + results_raw = data + else: + results_raw = data.get("results") or data.get("hits") or data + sources = [] + documents = [] + if isinstance(results_raw, list): + for r in results_raw: + if isinstance(r, dict): + sid = str(r.get("id") or r.get("qdrant_id") or r.get("doc_id") or r.get("source_id") or "") + score = r.get("score") if "score" in r else r.get("_score") + snippet = r.get("snippet") or r.get("text") or r.get("content") + language = r.get("language") or r.get("lang") + sources.append({"id": sid, "score": score, "snippet": snippet, "language": language}) + documents.append({"id": sid, "score": score, "snippet": snippet, "language": language}) + return sources, documents + + +def _save_session(query: str, prompt: str | None, response: str | None, sources: List[Dict]): + db = SessionLocal() + try: + sess = RagSession(query=query, prompt=prompt, response=response, sources=sources) + db.add(sess) + db.commit() + db.refresh(sess) + return sess.id + finally: + db.close() + + +@app.post("/api/v1/query", response_model=QueryResponse) +async def query(req: QueryRequest): + # Call KB search endpoint (KB handles embedding + vector search) + async with httpx.AsyncClient(timeout=30.0) as client: + try: + resp = await client.post(f"{KB_URL}/api/v1/search", json={"query": req.query, "k": req.top_k}) + resp.raise_for_status() + data = resp.json() + except httpx.HTTPError as e: + raise HTTPException(status_code=502, detail=f"KB search failed: {e}") + + sources, documents = _normalize_kb_response(data) + + # Detect query language and translate mismatched document snippets when possible + query_lang = None + try: + query_lang = detect(req.query) + except Exception: + query_lang = None + + # translate documents whose language differs from query_lang + async def _maybe_translate_documents(docs): + tasks = [] + for d in docs: + src = d.get("language") + if src and query_lang and not src.startswith(query_lang): + # translate snippet from src -> query_lang + tasks.append(translate_text(d.get("snippet") or "", src, query_lang)) + else: + tasks.append(None) + + results = [] + for t in tasks: + if t is None: + results.append(None) + else: + results.append(await t) + return results + + try: + translated = await _maybe_translate_documents(documents) + for idx, tr in enumerate(translated): + if tr: + documents[idx]["snippet"] = tr + sources[idx]["snippet"] = tr + except Exception: + # translation failures are non-fatal; continue with original snippets + pass + + # Build a simple prompt (could be expanded to include provenance formatting) + prompt_parts = [f"You are an assistant. Answer using only the provided documents.", f"Question: {req.query}", "Documents:"] + for i, d in enumerate(documents, start=1): + prompt_parts.append(f"[{i}] id={d.get('id')} score={d.get('score')} text={d.get('snippet')}") + prompt = "\n\n".join(prompt_parts) + + # Generate answer via adapter + try: + answer = generate(prompt, documents) + except Exception as e: + raise HTTPException(status_code=502, detail=f"LLM generation failed: {e}") + + # persist session + session_id = None + try: + session_id = _save_session(req.query, prompt, answer, sources) + except Exception: + # don't fail the request if DB save fails + pass + + # classify domain (simple keyword-based heuristic) + def classify_domain(text: str) -> str: + t = text.lower() + if any(k in t for k in ["salary", "pay", "wage", "salaire", "impay"]): + return "labor" + if any(k in t for k in ["house", "rent", "logement", "bail", "evict"]): + return "housing" + if any(k in t for k in ["family", "divorce", "marriage", "famil"]): + return "family" + if any(k in t for k in ["crime", "police", "punish", "délit"]): + return "criminal" + return "general" + + domain = classify_domain(req.query) + + # publish matching.requested asynchronously (best-effort) + try: + asyncio.create_task(publish_matching_requested(domain, session_id, None)) + except Exception: + pass + + provenance = sources + return {"query": req.query, "results": sources, "synthesized_answer": answer, "provenance": provenance} + + +@app.post("/api/v1/query/stream") +async def query_stream(req: QueryRequest, request: Request): + # Call KB search endpoint (KB handles embedding + vector search) + async with httpx.AsyncClient(timeout=30.0) as client: + try: + resp = await client.post(f"{KB_URL}/api/v1/search", json={"query": req.query, "k": req.top_k}) + resp.raise_for_status() + data = resp.json() + except httpx.HTTPError as e: + raise HTTPException(status_code=502, detail=f"KB search failed: {e}") + + sources, documents = _normalize_kb_response(data) + + prompt_parts = [f"You are an assistant. Answer using only the provided documents.", f"Question: {req.query}", "Documents:"] + for i, d in enumerate(documents, start=1): + prompt_parts.append(f"[{i}] id={d.get('id')} score={d.get('score')} text={d.get('snippet')}") + prompt = "\n\n".join(prompt_parts) + + async def event_generator(): + # stream_generate is an async generator; yield SSE 'data:' frames + try: + async for chunk in stream_generate(prompt, documents): + if not chunk: + continue + yield f"data: {chunk}\n\n" + yield "event: done\n\n" + finally: + # Optionally save final aggregated response into DB. + pass + + return StreamingResponse(event_generator(), media_type="text/event-stream") diff --git a/services/rag-service/app/models.py b/services/rag-service/app/models.py new file mode 100644 index 0000000..1c56b0a --- /dev/null +++ b/services/rag-service/app/models.py @@ -0,0 +1,17 @@ +from sqlalchemy import Column, Integer, String, Text, DateTime, JSON +from sqlalchemy.sql import func +from sqlalchemy.orm import declarative_base + + +Base = declarative_base() + + +class RagSession(Base): + __tablename__ = "rag_sessions" + + id = Column(Integer, primary_key=True, index=True) + query = Column(Text, nullable=False) + prompt = Column(Text, nullable=True) + response = Column(Text, nullable=True) + sources = Column(JSON(), nullable=True) + created_at = Column(DateTime(timezone=True), server_default=func.now()) diff --git a/services/rag-service/app/schemas.py b/services/rag-service/app/schemas.py new file mode 100644 index 0000000..957726d --- /dev/null +++ b/services/rag-service/app/schemas.py @@ -0,0 +1,20 @@ +from pydantic import BaseModel +from typing import List, Optional + + +class QueryRequest(BaseModel): + query: str + top_k: int = 5 + + +class ResultSource(BaseModel): + id: str + score: Optional[float] + snippet: Optional[str] + + +class QueryResponse(BaseModel): + query: str + results: List[ResultSource] + synthesized_answer: str + provenance: List[ResultSource] diff --git a/services/rag-service/app/translator.py b/services/rag-service/app/translator.py new file mode 100644 index 0000000..3ab1d39 --- /dev/null +++ b/services/rag-service/app/translator.py @@ -0,0 +1,48 @@ +import os +from typing import Optional +import httpx + +# Simple translator using HuggingFace translation models when HF_API_KEY is set. +# Falls back to returning the input text unchanged. + +HF_API_URL = "https://api-inference.huggingface.co/models" +HF_API_KEY = os.getenv("HF_API_KEY") + + +async def translate_text(text: str, src_lang: str, tgt_lang: str) -> str: + """Translate `text` from src_lang to tgt_lang. Uses HF translation models when + available. Returns original text on failure or when no translator configured. + """ + if not HF_API_KEY: + return text + + # choose a Helsinki model for common en<->fr translations + model = None + if src_lang.startswith("fr") and tgt_lang.startswith("en"): + model = "Helsinki-NLP/opus-mt-fr-en" + elif src_lang.startswith("en") and tgt_lang.startswith("fr"): + model = "Helsinki-NLP/opus-mt-en-fr" + else: + # for other pairs, don't attempt automatic translation + return text + + headers = {"Authorization": f"Bearer {HF_API_KEY}", "Accept": "application/json"} + payload = {"inputs": text} + + async with httpx.AsyncClient(timeout=30.0) as client: + try: + resp = await client.post(f"{HF_API_URL}/{model}", json=payload, headers=headers) + resp.raise_for_status() + body = resp.json() + # HF translation models typically return [{'translation_text': '...'}] + if isinstance(body, list) and body and "translation_text" in body[0]: + return body[0]["translation_text"] + if isinstance(body, dict) and "translation_text" in body: + return body["translation_text"] + # fallback: if model returns a string + if isinstance(body, str): + return body + except Exception: + return text + + return text diff --git a/services/rag-service/requirements.txt b/services/rag-service/requirements.txt new file mode 100644 index 0000000..50ab671 --- /dev/null +++ b/services/rag-service/requirements.txt @@ -0,0 +1,16 @@ +fastapi>=0.115.0,<1.0.0 +httpx>=0.24.0 +uvicorn[standard]>=0.22.0 +pydantic>=2.10.0,<3.0.0 +pydantic-settings>=2.7.0,<3.0.0 +tenacity>=8.2.0 + +# database +SQLAlchemy>=1.4.0 +langdetect>=1.0.9 +aio-pika>=8.1.0 + +# dev/test +pytest>=8.2.0,<10.0.0 +pytest-asyncio>=0.21.0 +respx>=0.20.0 diff --git a/services/rag-service/tests/conftest.py b/services/rag-service/tests/conftest.py new file mode 100644 index 0000000..4d3d7e5 --- /dev/null +++ b/services/rag-service/tests/conftest.py @@ -0,0 +1,8 @@ +from pathlib import Path +import sys + +# Ensure the service `app` package directory is on sys.path so tests can import `app` +ROOT = Path(__file__).resolve().parents[2] +root_str = str(ROOT) +if root_str not in sys.path: + sys.path.insert(0, root_str) diff --git a/services/rag-service/tests/test_llm_adapter.py b/services/rag-service/tests/test_llm_adapter.py new file mode 100644 index 0000000..8a29d19 --- /dev/null +++ b/services/rag-service/tests/test_llm_adapter.py @@ -0,0 +1,44 @@ +import os +import pytest +import respx +from httpx import Response +import sys +from pathlib import Path + +# Ensure the service package directory is on sys.path so `import app` works +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) +import importlib.util + +# Load the `llm.py` module directly to avoid package import issues +this_file = Path(__file__).resolve() +llm_path = None +for parent in this_file.parents: + candidate = parent / 'app' + if candidate.exists(): + llm_path = candidate / 'llm.py' + break +if llm_path is None: + raise FileNotFoundError('Could not locate app/llm.py from test location') +spec = importlib.util.spec_from_file_location('app.llm', str(llm_path)) +app_dir = str((llm_path.parent).resolve()) +import types +if 'app' not in sys.modules: + app_pkg = types.ModuleType('app') + app_pkg.__path__ = [app_dir] + sys.modules['app'] = app_pkg +llm = importlib.util.module_from_spec(spec) +spec.loader.exec_module(llm) +generate_answer_from_documents = llm.generate_answer_from_documents + + +@respx.mock +def test_generate_answer_success(monkeypatch): + # Mock HF API + model = "google/flan-t5-small" + url = f"https://api-inference.huggingface.co/models/{model}" + respx.post(url).mock(return_value=Response(200, json=[{"generated_text": "This is an answer. [sources: 1,2]"}])) + + monkeypatch.setenv("HF_API_KEY", "test-token") + docs = [{"id": "1", "snippet": "doc one"}, {"id": "2", "snippet": "doc two"}] + out = generate_answer_from_documents("What is this?", docs) + assert "This is an answer" in out diff --git a/services/rag-service/tests/test_query_flow.py b/services/rag-service/tests/test_query_flow.py new file mode 100644 index 0000000..3debd5e --- /dev/null +++ b/services/rag-service/tests/test_query_flow.py @@ -0,0 +1,54 @@ +import respx +from httpx import Response +import pytest +import os + +from fastapi.testclient import TestClient + +import sys +from pathlib import Path + +# Ensure the service package directory is on sys.path so `import app` works +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) +import importlib.util + +# Load the `main.py` module directly to avoid package import issues +this_file = Path(__file__).resolve() +main_path = None +for parent in this_file.parents: + candidate = parent / 'app' + if candidate.exists(): + main_path = candidate / 'main.py' + break +if main_path is None: + raise FileNotFoundError('Could not locate app/main.py from test location') +app_dir = str((main_path.parent).resolve()) +import types +if 'app' not in sys.modules: + app_pkg = types.ModuleType('app') + app_pkg.__path__ = [app_dir] + sys.modules['app'] = app_pkg +spec = importlib.util.spec_from_file_location('app.main', str(main_path)) +main_mod = importlib.util.module_from_spec(spec) +spec.loader.exec_module(main_mod) +app = main_mod.app + + +@respx.mock +def test_query_endpoint(monkeypatch): + # Mock KB search endpoint + kb_url = "http://knowledge-base-service:8000/api/v1/search" + respx.post(kb_url).mock(return_value=Response(200, json={"results": [{"id": "d1", "snippet": "s1", "score": 0.9}]})) + + # Mock HF inference + hf_url = "https://api-inference.huggingface.co/models/google/flan-t5-small" + respx.post(hf_url).mock(return_value=Response(200, json=[{"generated_text": "Answer from HF. [sources: d1]"}])) + + monkeypatch.setenv("HF_API_KEY", "test-token") + + client = TestClient(app) + resp = client.post("/api/v1/query", json={"query": "test", "top_k": 1}) + assert resp.status_code == 200 + body = resp.json() + assert "synthesized_answer" in body + assert body["synthesized_answer"].startswith("Answer from HF") diff --git a/services/scraper-service/.dockerignore b/services/scraper-service/.dockerignore new file mode 100644 index 0000000..ac56efb --- /dev/null +++ b/services/scraper-service/.dockerignore @@ -0,0 +1,6 @@ +__pycache__ +*.pyc +*.pyo +.pytest_cache +venv +.env diff --git a/services/scraper-service/Dockerfile b/services/scraper-service/Dockerfile new file mode 100644 index 0000000..35069f5 --- /dev/null +++ b/services/scraper-service/Dockerfile @@ -0,0 +1,13 @@ +FROM python:3.11-slim +WORKDIR /app +ENV PYTHONDONTWRITEBYTECODE=1 +ENV PYTHONUNBUFFERED=1 + +COPY requirements.txt ./ +RUN pip install --no-cache-dir -r requirements.txt + +COPY app ./app +COPY alembic.ini ./ +COPY alembic ./alembic + +CMD ["sh", "-c", "alembic -c /app/alembic.ini upgrade head && gunicorn -k uvicorn.workers.UvicornWorker app.main:app --bind 0.0.0.0:8000 --workers 2"] diff --git a/services/scraper-service/alembic.ini b/services/scraper-service/alembic.ini new file mode 100644 index 0000000..3e04d18 --- /dev/null +++ b/services/scraper-service/alembic.ini @@ -0,0 +1,24 @@ +[alembic] +script_location = alembic + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = INFO +handlers = console + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s diff --git a/services/scraper-service/alembic/env.py b/services/scraper-service/alembic/env.py new file mode 100644 index 0000000..967762e --- /dev/null +++ b/services/scraper-service/alembic/env.py @@ -0,0 +1,36 @@ +from alembic import context +import os +import sys + +sys.path.append(os.path.dirname(os.path.dirname(__file__))) +sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) + +from app.config import settings +from app.db import engine +from app.models import ScrapeJob +from app.models_user import User + +config = context.config + +target_metadata = [ScrapeJob.metadata, User.metadata] + + +def run_migrations_offline(): + url = str(settings.DATABASE_URL) + context.configure(url=url, target_metadata=target_metadata, literal_binds=True) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online(): + connectable = engine + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/services/scraper-service/alembic/versions/0001_initial.py b/services/scraper-service/alembic/versions/0001_initial.py new file mode 100644 index 0000000..a957a67 --- /dev/null +++ b/services/scraper-service/alembic/versions/0001_initial.py @@ -0,0 +1,30 @@ +"""initial + +Revision ID: 0001_initial +Revises: +Create Date: 2026-05-17 + +""" +from alembic import op + +revision = '0001_initial' +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade(): + op.execute(""" + CREATE TABLE IF NOT EXISTS scrapejob ( + id SERIAL PRIMARY KEY, + url VARCHAR(2048) NOT NULL, + status VARCHAR(64), + result TEXT, + created_at TIMESTAMP WITHOUT TIME ZONE, + updated_at TIMESTAMP WITHOUT TIME ZONE + ) + """) + + +def downgrade(): + op.execute('DROP TABLE IF EXISTS scrapejob') diff --git a/services/scraper-service/app/__init__.py b/services/scraper-service/app/__init__.py new file mode 100644 index 0000000..e4c7aa5 --- /dev/null +++ b/services/scraper-service/app/__init__.py @@ -0,0 +1 @@ +from .main import app diff --git a/services/scraper-service/app/api/v1/auth_routes.py b/services/scraper-service/app/api/v1/auth_routes.py new file mode 100644 index 0000000..cf4411c --- /dev/null +++ b/services/scraper-service/app/api/v1/auth_routes.py @@ -0,0 +1,32 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from fastapi.security import OAuth2PasswordRequestForm +from sqlmodel import Session +from ...db import get_session +from ...models_user import User +from ...auth import verify_password, get_password_hash, create_access_token +from datetime import timedelta +from ...config import settings + +router = APIRouter(prefix="/v1/auth") + + +@router.post("/signup", status_code=status.HTTP_201_CREATED) +def signup(username: str, password: str, session: Session = Depends(get_session)): + existing = session.exec(User.select().where(User.username == username)).first() + if existing: + raise HTTPException(status_code=400, detail="User already exists") + user = User(username=username, hashed_password=get_password_hash(password)) + session.add(user) + session.commit() + session.refresh(user) + return {"id": user.id, "username": user.username} + + +@router.post("/token") +def login_token(form_data: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)): + user = session.exec(User.select().where(User.username == form_data.username)).first() + if not user or not verify_password(form_data.password, user.hashed_password): + raise HTTPException(status_code=400, detail="Incorrect username or password") + access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) + access_token = create_access_token(data={"sub": user.username}, expires_delta=access_token_expires) + return {"access_token": access_token, "token_type": "bearer"} diff --git a/services/scraper-service/app/api/v1/routes.py b/services/scraper-service/app/api/v1/routes.py new file mode 100644 index 0000000..6a6f268 --- /dev/null +++ b/services/scraper-service/app/api/v1/routes.py @@ -0,0 +1,39 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from sqlmodel import Session, select +from typing import List + +from ...db import get_session +from ...models import ScrapeJob +from ...schemas import ScrapeCreate, ScrapeRead +from ...tasks import run_scrape_async + +router = APIRouter(prefix="/v1") + + +@router.get("/health", status_code=200) +def health(): + return {"status": "ok"} + + +@router.post("/scrape", response_model=ScrapeRead, status_code=status.HTTP_201_CREATED) +def create_job(payload: ScrapeCreate, session: Session = Depends(get_session)): + job = ScrapeJob(url=str(payload.url)) + session.add(job) + session.commit() + session.refresh(job) + run_scrape_async.delay(job.id) + return job + + +@router.get("/scrape/{job_id}", response_model=ScrapeRead) +def get_job(job_id: int, session: Session = Depends(get_session)): + job = session.get(ScrapeJob, job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + return job + + +@router.get("/scrape", response_model=List[ScrapeRead]) +def list_jobs(limit: int = 50, session: Session = Depends(get_session)): + statement = select(ScrapeJob).limit(limit) + return session.exec(statement).all() diff --git a/services/scraper-service/app/auth.py b/services/scraper-service/app/auth.py new file mode 100644 index 0000000..330289d --- /dev/null +++ b/services/scraper-service/app/auth.py @@ -0,0 +1,53 @@ +from datetime import datetime, timedelta +from typing import Optional +from passlib.context import CryptContext +from jose import JWTError, jwt +from pydantic import BaseModel +from fastapi import Depends, HTTPException, status +from fastapi.security import OAuth2PasswordBearer +from .config import settings +from .db import get_session +from sqlmodel import Session +from .models_user import User + +pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/v1/auth/token") + +class TokenData(BaseModel): + username: Optional[str] = None + + +def verify_password(plain_password, hashed_password): + return pwd_context.verify(plain_password, hashed_password) + + +def get_password_hash(password): + return pwd_context.hash(password) + + +def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): + to_encode = data.copy() + expire = datetime.utcnow() + (expires_delta or timedelta(minutes=60)) + to_encode.update({"exp": expire}) + encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) + return encoded_jwt + + +def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)): + credentials_exception = HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Could not validate credentials", + headers={"WWW-Authenticate": "Bearer"}, + ) + try: + payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) + username: str = payload.get("sub") + if username is None: + raise credentials_exception + token_data = TokenData(username=username) + except JWTError: + raise credentials_exception + user = session.exec(User.select().where(User.username == token_data.username)).first() + if user is None: + raise credentials_exception + return user diff --git a/services/scraper-service/app/config.py b/services/scraper-service/app/config.py new file mode 100644 index 0000000..21d5bca --- /dev/null +++ b/services/scraper-service/app/config.py @@ -0,0 +1,20 @@ +from pydantic import BaseSettings, AnyUrl + + +class Settings(BaseSettings): + SERVICE_NAME: str = "scraper-service" + DATABASE_URL: AnyUrl = "postgresql://postgres:postgres@db:5432/scraper_db" + REDIS_URL: str = "redis://redis:6379/0" + CELERY_BROKER_URL: str = "redis://redis:6379/1" + CELERY_RESULT_BACKEND: str = "redis://redis:6379/2" + MIGRATE_AUTO: bool = True + SECRET_KEY: str = "please-change-me" + ALGORITHM: str = "HS256" + ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 + RABBITMQ_URL: str = "amqp://lexcam:lexcam_dev@rabbitmq:5672/%2F" + + class Config: + env_file = ".env" + + +settings = Settings() diff --git a/services/scraper-service/app/db.py b/services/scraper-service/app/db.py new file mode 100644 index 0000000..58e3353 --- /dev/null +++ b/services/scraper-service/app/db.py @@ -0,0 +1,13 @@ +from sqlmodel import SQLModel, create_engine, Session +from .config import settings + +engine = create_engine(str(settings.DATABASE_URL), echo=False) + + +def init_db(): + return None + + +def get_session(): + with Session(engine) as session: + yield session diff --git a/services/scraper-service/app/events.py b/services/scraper-service/app/events.py new file mode 100644 index 0000000..58861d9 --- /dev/null +++ b/services/scraper-service/app/events.py @@ -0,0 +1,18 @@ +import json +import pika +from .config import settings + + +def _get_connection(): + params = pika.URLParameters(settings.RABBITMQ_URL) + return pika.BlockingConnection(params) + + +def publish_event(routing_key: str, payload: dict): + conn = _get_connection() + ch = conn.channel() + ch.exchange_declare(exchange="lexcam.events", exchange_type="topic", durable=True) + # Publish persistent messages so events survive broker restarts. + props = pika.BasicProperties(delivery_mode=2) + ch.basic_publish(exchange="lexcam.events", routing_key=routing_key, body=json.dumps(payload), properties=props) + conn.close() diff --git a/services/scraper-service/app/main.py b/services/scraper-service/app/main.py new file mode 100644 index 0000000..bdff8f5 --- /dev/null +++ b/services/scraper-service/app/main.py @@ -0,0 +1,21 @@ +from fastapi import FastAPI +from .api.v1 import routes as v1_routes +from .api.v1 import auth_routes +from .db import init_db +from .config import settings +from .monitoring import init_logging, prometheus_middleware, prometheus_endpoint + +app = FastAPI(title=settings.SERVICE_NAME) + + +@app.on_event("startup") +def on_startup(): + init_db() + init_logging() + + +app.middleware("http")(prometheus_middleware) +app.add_api_route("/metrics", prometheus_endpoint, methods=["GET"]) + +app.include_router(auth_routes.router) +app.include_router(v1_routes.router) diff --git a/services/scraper-service/app/models.py b/services/scraper-service/app/models.py new file mode 100644 index 0000000..9d84bf4 --- /dev/null +++ b/services/scraper-service/app/models.py @@ -0,0 +1,12 @@ +from typing import Optional +from datetime import datetime +from sqlmodel import SQLModel, Field + + +class ScrapeJob(SQLModel, table=True): + id: Optional[int] = Field(default=None, primary_key=True) + url: str = Field(index=True) + status: str = Field(default="pending") + result: Optional[str] = None + created_at: datetime = Field(default_factory=datetime.utcnow) + finished_at: Optional[datetime] = None diff --git a/services/scraper-service/app/models_user.py b/services/scraper-service/app/models_user.py new file mode 100644 index 0000000..37c60d6 --- /dev/null +++ b/services/scraper-service/app/models_user.py @@ -0,0 +1,13 @@ +from typing import Optional +from sqlmodel import SQLModel, Field +from datetime import datetime + + +class User(SQLModel, table=True): + id: Optional[int] = Field(default=None, primary_key=True) + username: str = Field(index=True, unique=True) + email: Optional[str] = None + hashed_password: str + is_active: bool = True + is_superuser: bool = False + created_at: datetime = Field(default_factory=datetime.utcnow) diff --git a/services/scraper-service/app/monitoring.py b/services/scraper-service/app/monitoring.py new file mode 100644 index 0000000..5acfcc1 --- /dev/null +++ b/services/scraper-service/app/monitoring.py @@ -0,0 +1,25 @@ +import logging +from prometheus_client import Counter, generate_latest, CONTENT_TYPE_LATEST +from fastapi import Request, Response +import structlog + +REQUEST_COUNT = Counter("http_requests_total", "Total HTTP requests", ["method", "endpoint", "http_status"]) + + +def init_logging(): + structlog.configure(processors=[structlog.processors.JSONRenderer()]) + logging.basicConfig(level=logging.INFO) + + +async def prometheus_endpoint(): + data = generate_latest() + return Response(content=data, media_type=CONTENT_TYPE_LATEST) + + +async def prometheus_middleware(request: Request, call_next): + response = await call_next(request) + try: + REQUEST_COUNT.labels(request.method, request.url.path, str(response.status_code)).inc() + except Exception: + pass + return response diff --git a/services/scraper-service/app/schemas.py b/services/scraper-service/app/schemas.py new file mode 100644 index 0000000..070df96 --- /dev/null +++ b/services/scraper-service/app/schemas.py @@ -0,0 +1,19 @@ +from typing import Optional +from datetime import datetime +from pydantic import BaseModel, AnyUrl + + +class ScrapeCreate(BaseModel): + url: AnyUrl + + +class ScrapeRead(BaseModel): + id: int + url: str + status: str + result: Optional[str] + created_at: datetime + finished_at: Optional[datetime] + + class Config: + orm_mode = True diff --git a/services/scraper-service/app/tasks.py b/services/scraper-service/app/tasks.py new file mode 100644 index 0000000..5bd1de2 --- /dev/null +++ b/services/scraper-service/app/tasks.py @@ -0,0 +1,25 @@ +from celery import Celery +from celery.utils.log import get_task_logger +from .config import settings + +logger = get_task_logger(__name__) + +celery_app = Celery( + settings.SERVICE_NAME, + broker=settings.CELERY_BROKER_URL, + backend=settings.CELERY_RESULT_BACKEND, +) + + +@celery_app.task(bind=True) +def placeholder_run_scrape(self, job_id: int): + """Placeholder scrape task: detailed scraping, MinIO archival and + downstream event publishing are out-of-scope for the assistant and + should be implemented by the team. + """ + logger.warning("placeholder_run_scrape called for job %s", job_id) + raise NotImplementedError("Scrape implementation is provided by the team") + + +def get_celery_app(): + return celery_app diff --git a/services/scraper-service/requirements.txt b/services/scraper-service/requirements.txt new file mode 100644 index 0000000..8e1fc8c --- /dev/null +++ b/services/scraper-service/requirements.txt @@ -0,0 +1,21 @@ +fastapi==0.95.2 +uvicorn[standard]==0.22.0 +SQLModel==0.0.8 +SQLAlchemy==1.4.41 +psycopg2-binary==2.9.7 +python-dotenv==1.0.0 +celery==5.3.0 +redis==4.5.1 +gunicorn==20.1.0 +httpx==0.24.1 +pydantic==1.10.9 +pytest==7.4.0 +passlib[bcrypt]==1.7.4 +python-jose==3.3.0 +alembic==1.11.1 +prometheus_client==0.16.0 +structlog==23.3.0 +python-json-logger==2.0.7 +pika==1.3.1 +minio==7.1.2 +python-multipart==0.0.9 diff --git a/services/scraper-service/tests/test_health.py b/services/scraper-service/tests/test_health.py new file mode 100644 index 0000000..b5703e1 --- /dev/null +++ b/services/scraper-service/tests/test_health.py @@ -0,0 +1,9 @@ +from fastapi.testclient import TestClient +from app.main import app + + +def test_health(): + client = TestClient(app) + r = client.get("/v1/health") + assert r.status_code == 200 + assert r.json() == {"status": "ok"}