### Set up local development environment

Source: https://github.com/inveniosoftware/invenio-stats/blob/master/CONTRIBUTING.rst

Commands to create a virtual environment and install the package in editable mode with all dependencies.

```console
$ mkvirtualenv invenio-stats
$ cd invenio-stats/
$ pip install -e .[all]
```

--------------------------------

### Install Invenio-Stats via pip

Source: https://github.com/inveniosoftware/invenio-stats/blob/master/INSTALL.rst

Use this command to install the package from the Python Package Index.

```console
$ pip install invenio-stats
```

--------------------------------

### Get Record Statistics via API

Source: https://github.com/inveniosoftware/invenio-stats/wiki/Requirements

This example shows how to retrieve statistics for a specific record using the Invenio API. The statistics, such as views and downloads, are included in the response.

```bash
$ curl -XGET /api/records/123
{
  "metadata": {...},
  "links": {...},
  "stats": {
    "views": 42,
    "downloads": 10
  }
}
```

--------------------------------

### Configure STATS_EVENTS for File Downloads

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Configure the 'file-download' event to track file downloads. This setup specifies the signal to listen for, event builders, search engine mapping templates, and the indexing processor. It also defines preprocessors for robot flagging and user anonymization, a deduplication window, and an index suffix pattern.

```python
# config.py
STATS_EVENTS = {
    "file-download": {
        # Signal that triggers event emission
        "signal": "invenio_files_rest.signals.file_downloaded",
        # Functions that build the event data from signal arguments
        "event_builders": [
            "invenio_stats.contrib.event_builders.file_download_event_builder"
        ],
        # Search engine mapping templates
        "templates": "invenio_stats.contrib.file_download",
        # Processor class for indexing events
        "cls": "invenio_stats.processors.EventsIndexer",
        "params": {
            # Preprocessors run on each event before indexing
            "preprocessors": [
                "invenio_stats.processors:flag_robots",
                "invenio_stats.processors:anonymize_user",
                "invenio_stats.contrib.event_builders:build_file_unique_id",
            ],
            # Deduplicate events within this window (seconds)
            "double_click_window": 30,
            # Index suffix pattern (creates monthly indices)
            "suffix": "%Y-%m",
        }
    },
    "record-view": {
        "signal": "invenio_records_ui.signals.record_viewed",
        "event_builders": [
            "invenio_stats.contrib.event_builders.record_view_event_builder"
        ],
        "templates": "invenio_stats.contrib.record_view",
        "cls": "invenio_stats.processors.EventsIndexer",
        "params": {
            "preprocessors": [
                "invenio_stats.processors:flag_robots",
                "invenio_stats.processors:anonymize_user",
                "invenio_stats.contrib.event_builders:build_record_unique_id",
            ],
            "double_click_window": 10,
            "suffix": "%Y-%m",
        }
    },
}

```

--------------------------------

### Configure Message Queue Exchange in Python

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Define the message queue exchange for events using `kombu.Exchange`. This example shows configuration for transient (in-memory) and persistent queues.

```python
from kombu import Exchange

# Message queue exchange configuration
STATS_MQ_EXCHANGE = Exchange(
    "events",
    type="direct",
    delivery_mode="transient",  # In-memory queue (faster but not persistent)
)

# For persistent queues (survives broker restart)
STATS_MQ_EXCHANGE = Exchange(
    "events",
    type="direct",
    delivery_mode="persistent",
)
```

--------------------------------

### CLI: Process All Configured Event Types

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Use this Flask CLI command to process all event types configured in your Invenio-Stats setup. This is useful for a full data processing run.

```bash
flask stats events process
```

--------------------------------

### Query Statistics via REST API (Date Histogram)

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Use the REST API endpoint `/stats` to query a date histogram of file downloads. This example demonstrates a POST request with a JSON payload specifying the statistic and parameters.

```bash
# Query a date histogram of file downloads
curl -X POST http://localhost:5000/stats \
  -H "Content-Type: application/json" \
  -d '{
    "downloads_over_time": {
      "stat": "bucket-file-download-histogram",
      "params": {
        "start_date": "2024-01-01",
        "end_date": "2024-01-31",
        "interval": "day",
        "bucket_id": "550e8400-e29b-41d4-a716-446655440000",
        "file_key": "document.pdf"
      }
    }
  }'
```

--------------------------------

### Custom EventsIndexer Processor

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Extend the base EventsIndexer to add custom processing logic, such as enriching events with additional fields. This example shows how to add a 'processed_by' field.

```python
from invenio_stats.processors import EventsIndexer, flag_robots, anonymize_user

# Create a custom processor with specific configuration
class CustomEventsIndexer(EventsIndexer):
    """Custom events indexer with additional preprocessing."""

    default_preprocessors = [flag_robots, anonymize_user]

    def __init__(self, queue, **kwargs):
        super().__init__(
            queue=queue,
            prefix="events",
            suffix="%Y-%m",  # Monthly indices
            double_click_window=30,  # 30-second deduplication window
            **kwargs
        )

    def actionsiter(self):
        """Override to add custom processing logic."""
        for action in super().actionsiter():
            # Add custom enrichment
            action["_source"]["processed_by"] = "custom_indexer"
            yield action

# Use in configuration
STATS_EVENTS = {
    "file-download": {
        "cls": "myapp.processors:CustomEventsIndexer",
        "params": {
            "preprocessors": [
                "invenio_stats.processors:flag_robots",
                "invenio_stats.processors:anonymize_user",
            ],
            "double_click_window": 30,
            "suffix": "%Y-%m",
        }
    },
}
```

--------------------------------

### Run project tests

Source: https://github.com/inveniosoftware/invenio-stats/blob/master/CONTRIBUTING.rst

Execute the test suite to verify changes, check code style, and build documentation.

```console
$ ./run-tests.sh
```

--------------------------------

### Clone the repository

Source: https://github.com/inveniosoftware/invenio-stats/blob/master/CONTRIBUTING.rst

Use this command to create a local copy of your forked repository.

```console
$ git clone git@github.com:your_name_here/invenio-stats.git
```

--------------------------------

### Create and Run TermsQuery

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Instantiate a TermsQuery to group statistics by field values. Define the index, metrics, and fields to aggregate on. The query can be run for a specific date range and filters.

```python
from invenio_stats.queries import TermsQuery
from invenio_search import current_search_client

# Create a terms query for file download totals
query = TermsQuery(
    name="file-downloads-by-file",
    index="stats-file-download",
    client=current_search_client,
    time_field="timestamp",
    copy_fields={
        "bucket_id": "bucket_id",
    },
    metric_fields={
        "total_downloads": ("sum", "count", {}),
        "unique_visitors": ("sum", "unique_count", {}),
    },
    required_filters={
        "bucket_id": "bucket_id",
    },
    aggregated_fields=["file_key"],  # Group by file_key
    max_bucket_size=10000,
)

# Run the query
result = query.run(
    start_date="2024-01-01",
    end_date="2024-01-31",
    bucket_id="550e8400-e29b-41d4-a716-446655440000",
)
```

--------------------------------

### Create a development branch

Source: https://github.com/inveniosoftware/invenio-stats/blob/master/CONTRIBUTING.rst

Create a new branch to isolate your changes.

```console
$ git checkout -b name-of-your-bugfix-or-feature
```

--------------------------------

### CLI: Process and Update Bookmark

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Execute aggregation processing with the `--update-bookmark` flag and `--eager` mode using the Flask CLI. This processes data and updates the bookmark for the next run.

```bash
flask stats aggregations process --update-bookmark --eager
```

--------------------------------

### Create and Run StatAggregator

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Instantiate a StatAggregator for a specific event, define aggregation fields and metrics, and run it to process events. Supports running for all events since the last bookmark or a specific date range.

```python
aggregator = StatAggregator(
    name="file-download-agg",
    event="file-download",
    client=current_search_client,
    field="unique_id",  # Field to aggregate on
    interval="day",  # Aggregation interval: hour, day, month, year
    index_interval="month",  # Index storage interval
    copy_fields={
        "file_key": "file_key",
        "bucket_id": "bucket_id",
        "file_id": "file_id",
    },
    metric_fields={
        # "destination_field": ("metric_type", "source_field", {options})
        "unique_count": ("cardinality", "unique_session_id", {}),
        "total_size": ("sum", "size", {}),
        "avg_size": ("avg", "size", {}),
    },
    query_modifiers=[filter_robots],  # Exclude robot events
    max_bucket_size=10000,
)

# Run aggregation for all events since last bookmark
results = aggregator.run(update_bookmark=True)

# Run aggregation for a specific date range
from datetime import datetime, timezone
results = aggregator.run(
    start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
    end_date=datetime(2024, 1, 31, tzinfo=timezone.utc),
    update_bookmark=False,
)
```

--------------------------------

### Commit and push changes

Source: https://github.com/inveniosoftware/invenio-stats/blob/master/CONTRIBUTING.rst

Stage, commit with descriptive messages, and push your changes to the remote repository.

```console
$ git add .
$ git commit -s
    -m "component: title without verbs"
    -m "* NEW Adds your new feature."
    -m "* FIX Fixes an existing issue."
    -m "* BETTER Improves and existing feature."
    -m "* Changes something that should not be visible in release notes."
$ git push origin name-of-your-bugfix-or-feature
```

--------------------------------

### Configure Additional InvenioStats Options in Python

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Set various advanced configuration options for InvenioStats, including signal receiver registration, index template usage, UTC datetime enablement, and permission factories.

```python
# Enable/disable signal receivers globally
STATS_REGISTER_RECEIVERS = True  # Default: True

# Use index templates instead of search templates
# (for Elasticsearch 7.8+ / OpenSearch)
STATS_REGISTER_INDEX_TEMPLATES = False  # Default: False

# Enable timezone-aware UTC datetimes
# Set to True for new installations, False for backwards compatibility
STATS_EVENTS_UTC_DATETIME_ENABLED = False  # Default: False
```

```python
# Complete configuration example
app.config.update({
    "STATS_REGISTER_RECEIVERS": True,
    "STATS_REGISTER_INDEX_TEMPLATES": True,
    "STATS_EVENTS_UTC_DATETIME_ENABLED": True,
    "STATS_PERMISSION_FACTORY": "myapp.permissions:stats_permission_factory",
    "STATS_EVENTS": {...},
    "STATS_AGGREGATIONS": {...},
    "STATS_QUERIES": {...},
})
```

--------------------------------

### Configure STATS_AGGREGATIONS

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Define aggregation logic for events like file downloads and record views in the configuration file.

```python
# config.py
STATS_AGGREGATIONS = {
    "file-download-agg": {
        "templates": "invenio_stats.contrib.aggregations.aggr_file_download",
        "cls": "invenio_stats.aggregations.StatAggregator",
        "params": {
            "event": "file-download",
            "field": "unique_id",
            "interval": "day",  # Aggregate by day
            "index_interval": "month",  # Store in monthly indices
            "copy_fields": {
                "file_key": "file_key",
                "bucket_id": "bucket_id",
                "file_id": "file_id",
            },
            "metric_fields": {
                "unique_count": ("cardinality", "unique_session_id", {}),
            },
            "query_modifiers": [
                "invenio_stats.aggregations:filter_robots"
            ],
        }
    },
    "record-view-agg": {
        "templates": "invenio_stats.contrib.aggregations.aggr_record_view",
        "cls": "invenio_stats.aggregations.StatAggregator",
        "params": {
            "event": "record-view",
            "field": "unique_id",
            "interval": "day",
            "index_interval": "month",
            "copy_fields": {
                "pid_type": "pid_type",
                "pid_value": "pid_value",
                "record_id": "record_id",
            },
            "metric_fields": {
                "unique_count": ("cardinality", "unique_session_id", {}),
            },
        }
    },
}
```

--------------------------------

### StatAggregator Usage

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Demonstrates how to create and use StatAggregator instances for data aggregation.

```APIDOC
## StatAggregator

### Description
Used to create and manage statistical aggregations.

### Methods

#### `__init__`
Initializes a `StatAggregator` instance.

**Parameters**
- **name** (str) - Required - The name of the aggregator.
- **event** (str) - Required - The event type to aggregate.
- **client** (object) - Required - The search client instance.
- **field** (str) - Required - The field to aggregate on.
- **interval** (str) - Required - The aggregation interval (e.g., 'day', 'month').
- **index_interval** (str) - Required - The interval for index storage.
- **copy_fields** (dict) - Optional - Fields to copy to the aggregation results.
- **metric_fields** (dict) - Optional - Fields to calculate metrics on.
- **query_modifiers** (list) - Optional - Functions to modify queries.
- **max_bucket_size** (int) - Optional - Maximum number of buckets.

#### `run`
Runs the aggregation.

**Parameters**
- **update_bookmark** (bool) - Optional - Whether to update the bookmark.
- **start_date** (datetime) - Optional - The start date for aggregation.
- **end_date** (datetime) - Optional - The end date for aggregation.

#### `list_bookmarks`
Lists existing aggregation bookmarks.

**Parameters**
- **limit** (int) - Optional - The maximum number of bookmarks to return.

#### `delete`
Deletes aggregations for a specified date range.

**Parameters**
- **start_date** (datetime) - Required - The start date for deletion.
- **end_date** (datetime) - Required - The end date for deletion.

### Example Usage
```python
from invenio_stats.aggregators import StatAggregator
from invenio_search import current_search_client
from datetime import datetime, timezone

# Create an aggregator instance
aggregator = StatAggregator(
    name="file-download-agg",
    event="file-download",
    client=current_search_client,
    field="unique_id",  # Field to aggregate on
    interval="day",  # Aggregation interval: hour, day, month, year
    index_interval="month",  # Index storage interval
    copy_fields={
        "file_key": "file_key",
        "bucket_id": "bucket_id",
        "file_id": "file_id",
    },
    metric_fields={
        "unique_count": ("cardinality", "unique_session_id", {{}})
    },
    query_modifiers=[filter_robots],  # Exclude robot events
    max_bucket_size=10000,
)

# Run aggregation for all events since last bookmark
results = aggregator.run(update_bookmark=True)

# Run aggregation for a specific date range
results = aggregator.run(
    start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
    end_date=datetime(2024, 1, 31, tzinfo=timezone.utc),
    update_bookmark=False,
)

# List aggregation bookmarks
bookmarks = aggregator.list_bookmarks(limit=10)
for bookmark in bookmarks:
    print(f"Bookmark: {bookmark.date}")

# Delete aggregations for a date range
aggregator.delete(
    start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
    end_date=datetime(2024, 1, 15, tzinfo=timezone.utc),
)
```
```

--------------------------------

### CLI: List Aggregation Bookmarks

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

View the current aggregation bookmarks using the Flask CLI. You can list all bookmarks or filter by a specific aggregation name with a limit.

```bash
flask stats aggregations list-bookmarks
```

```bash
flask stats aggregations list-bookmarks file-download-agg --limit 10
```

--------------------------------

### List and Delete Aggregation Bookmarks

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Manage aggregation bookmarks by listing existing ones or deleting aggregations within a specified date range.

```python
# List aggregation bookmarks
bookmarks = aggregator.list_bookmarks(limit=10)
for bookmark in bookmarks:
    print(f"Bookmark: {bookmark.date}")

# Delete aggregations for a date range
aggregator.delete(
    start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
    end_date=datetime(2024, 1, 15, tzinfo=timezone.utc),
)
```

--------------------------------

### CLI: Process All Configured Aggregations

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Initiate the processing of all configured aggregation tasks using the Flask CLI. This command ensures all defined aggregations are updated.

```bash
flask stats aggregations process
```

--------------------------------

### Configure STATS_QUERIES

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Define predefined queries for the REST API to retrieve aggregated statistics.

```python
# config.py
STATS_QUERIES = {
    "bucket-file-download-histogram": {
        "cls": "invenio_stats.queries.DateHistogramQuery",
        "params": {
            "index": "stats-file-download",
            "copy_fields": {
                "bucket_id": "bucket_id",
                "file_key": "file_key",
            },
            "metric_fields": {
                "count": ("sum", "count", {}),
                "unique_count": ("sum", "unique_count", {}),
            },
            "required_filters": {
                "bucket_id": "bucket_id",
                "file_key": "file_key",
            }
        }
    },
    "bucket-file-download-total": {
        "cls": "invenio_stats.queries.TermsQuery",
        "params": {
            "index": "stats-file-download",
            "copy_fields": {
                "bucket_id": "bucket_id",
            },
            "metric_fields": {
                "count": ("sum", "count", {}),
                "unique_count": ("sum", "unique_count", {}),
            },
            "required_filters": {
                "bucket_id": "bucket_id",
            },
            "aggregated_fields": ["file_key"],
        }
    },
    "record-view-histogram": {
        "cls": "invenio_stats.queries.DateHistogramQuery",
        "params": {
            "index": "stats-record-view",
            "copy_fields": {
                "record_id": "record_id",
            },
            "metric_fields": {
                "count": ("sum", "count", {}),
                "unique_count": ("sum", "unique_count", {}),
            },
            "required_filters": {
                "record_id": "record_id",
            }
        }
    },
}
```

--------------------------------

### Initialize InvenioStats Extension

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Initialize the InvenioStats extension with your Flask application to enable statistics collection. Ensure other required extensions like InvenioQueues and InvenioSearch are initialized first. Register the stats REST API blueprint to make statistics accessible via API.

```python
from flask import Flask
from invenio_stats import InvenioStats
from invenio_queues import InvenioQueues
from invenio_search import InvenioSearch

app = Flask(__name__)
app.config.update({
    "BROKER_URL": "redis://localhost:6379",
    "CELERY_RESULT_BACKEND": "redis://localhost:6379",
    "SECRET_KEY": "your-secret-key",
    "SQLALCHEMY_DATABASE_URI": "postgresql://user:pass@localhost/db",
})

# Initialize required extensions
InvenioQueues(app)
InvenioSearch(app)
InvenioStats(app)

# Register the stats REST API blueprint
from invenio_stats.views import blueprint
app.register_blueprint(blueprint)
```

--------------------------------

### Configure EventEmitter for Signal Handling in Python

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Set up event emitters to automatically generate events from Flask signals. This involves defining custom signals, event builders, and configuring them within `STATS_EVENTS`.

```python
from invenio_stats.receivers import EventEmitter, build_event_emitter
from invenio_stats.proxies import current_stats
from blinker import Namespace

# Create custom signals
_signals = Namespace()
custom_action = _signals.signal("custom-action")

# Define event builder for the signal
def custom_action_event_builder(event, sender_app, action=None, **kwargs):
    """Build event from custom action signal."""
    import datetime
    event.update({
        "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "action_type": action.type,
        "action_id": str(action.id),
        "user_id": str(action.user_id),
    })
    return event

# Configure the event
STATS_EVENTS = {
    "custom-action": {
        "signal": "myapp.signals:custom_action",
        "event_builders": [
            "myapp.events:custom_action_event_builder"
        ],
        "cls": "invenio_stats.processors.EventsIndexer",
    }
}
```

```python
# Manually build and use an event emitter
with app.app_context():
    emitter = build_event_emitter("file-download")

    # Emit event directly (bypassing signal)
    if emitter:
        emitter(app, obj=file_object)

# Get cached event emitter
emitter = current_stats.get_event_emitter("file-download")
```

--------------------------------

### Create and Run DateHistogramQuery

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Instantiate a DateHistogramQuery to retrieve time-series statistics. Specify index, time field, metrics, and filters. The query can be run for a specific date range and interval.

```python
from invenio_stats.queries import DateHistogramQuery
from invenio_search import current_search_client

# Create a histogram query
query = DateHistogramQuery(
    name="file-downloads-over-time",
    index="stats-file-download",
    client=current_search_client,
    time_field="timestamp",
    copy_fields={
        "bucket_id": "bucket_id",
        "file_key": "file_key",
    },
    metric_fields={
        "downloads": ("sum", "count", {}),
        "unique_visitors": ("sum", "unique_count", {}),
    },
    required_filters={
        "bucket_id": "bucket_id",
    },
)

# Run the query
result = query.run(
    interval="day",  # year, quarter, month, week, day
    start_date="2024-01-01",
    end_date="2024-01-31",
    bucket_id="550e8400-e29b-41d4-a716-446655440000",
)
```

--------------------------------

### Celery Beat Schedule Configuration

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Configure Celery Beat to automatically run tasks for processing and aggregating statistics at defined intervals. Adjust `schedule` and `args` as needed.

```python
from datetime import timedelta

# Celery Beat schedule configuration
CELERY_BEAT_SCHEDULE = {
    # Process raw events every 30 minutes
    "stats-process-events": {
        "task": "invenio_stats.tasks.process_events",
        "schedule": timedelta(minutes=30),
        "args": [["file-download", "record-view"]],
    },
    # Aggregate events every hour
    "stats-aggregate-events": {
        "task": "invenio_stats.tasks.aggregate_events",
        "schedule": timedelta(hours=1),
        "args": [["file-download-agg", "record-view-agg"]],
    },
}
```

--------------------------------

### Custom File Download Event Builder

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Define a custom event builder to extract and format metadata for file download events. This function updates the event dictionary with details like timestamp, file metadata, request context, and user information.

```python
import datetime
from flask import request
from invenio_stats.utils import get_user, format_datetime_iso

def custom_file_download_event_builder(event, sender_app, obj=None, **kwargs):
    """Build a custom file-download event with additional metadata."""
    event.update({
        # Timestamp
        "timestamp": format_datetime_iso(
            datetime.datetime.now(datetime.timezone.utc)
        ),
        # File metadata
        "bucket_id": str(obj.bucket_id),
        "file_id": str(obj.file_id),
        "file_key": obj.key,
        "size": obj.file.size,
        "mimetype": obj.mimetype,
        # Request context
        "referrer": request.referrer,
        "via_api": request.path.startswith("/api/"),
        # User information (automatically anonymized later)
        **get_user(),
    })
    return event

def build_custom_unique_id(doc):
    """Build a unique identifier for deduplication."""
    doc["unique_id"] = f"{doc['bucket_id']}_{doc['file_id']}_{doc.get('version', 'latest')}"
    return doc

# Register in configuration
STATS_EVENTS = {
    "file-download": {
        "signal": "invenio_files_rest.signals.file_downloaded",
        "event_builders": [
            "myapp.events:custom_file_download_event_builder"
        ],
        "cls": "invenio_stats.processors.EventsIndexer",
        "params": {
            "preprocessors": [
                "invenio_stats.processors:flag_robots",
                "invenio_stats.processors:anonymize_user",
                "myapp.events:build_custom_unique_id",
            ],
        }
    },
}
```

--------------------------------

### Publish Events Programmatically

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Use the current_stats proxy to publish single or batch events directly without relying on signals.

```python
import datetime
from invenio_stats.proxies import current_stats

# Publish a single file download event
event = {
    "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    "bucket_id": "550e8400-e29b-41d4-a716-446655440000",
    "file_id": "123e4567-e89b-12d3-a456-426614174000",
    "file_key": "document.pdf",
    "size": 1024000,
    "user_id": "42",
    "ip_address": "192.168.1.1",
    "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    "session_id": "abc123",
    "referrer": "https://example.com/search",
}

current_stats.publish("file-download", [event])

# Publish multiple events in batch
events = [
    {
        "timestamp": (datetime.datetime.now(datetime.timezone.utc)
                      + datetime.timedelta(minutes=i)).isoformat(),
        "bucket_id": "550e8400-e29b-41d4-a716-446655440000",
        "file_id": "123e4567-e89b-12d3-a456-426614174000",
        "file_key": f"file_{i}.pdf",
        "user_id": str(i),
    }
    for i in range(10)
]

current_stats.publish("file-download", events)
```

--------------------------------

### Query Total Downloads Grouped by File

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Use this cURL command to query the total number of downloads for a specific file within a given date range. Requires a bucket ID.

```bash
curl -X POST http://localhost:5000/stats \
  -H "Content-Type: application/json" \
  -d '{
    "total_by_file": {
      "stat": "bucket-file-download-total",
      "params": {
        "start_date": "2024-01-01",
        "end_date": "2024-01-31",
        "bucket_id": "550e8400-e29b-41d4-a716-446655440000"
      }
    }
  }'
```

--------------------------------

### Custom Permission Factory Implementation

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Implement a custom permission factory function to control access to statistics queries based on query name, parameters, and user roles. This allows for fine-grained access control.

```python
from flask_principal import Permission, RoleNeed
from invenio_stats.utils import AllowAllPermission

def custom_permission_factory(query_name, params):
    """Custom permission factory for statistics access control."""
    # Public queries - allow everyone
    public_queries = ["bucket-file-download-histogram"]
    if query_name in public_queries:
        return AllowAllPermission

    # Admin-only queries
    admin_queries = ["detailed-user-stats", "system-metrics"]
    if query_name in admin_queries:
        return Permission(RoleNeed("admin"))

    # Record-specific queries - check record access
    if query_name.startswith("record-"):
        record_id = params.get("record_id")
        if record_id:
            # Return a permission based on record access
            return RecordStatsPermission(record_id)

    # Default: require authentication
    from flask_login import current_user
    class AuthenticatedPermission:
        def can(self):
            return current_user.is_authenticated
    return AuthenticatedPermission()

# Configure in app
STATS_PERMISSION_FACTORY = custom_permission_factory

# Or per-query permission factory
STATS_QUERIES = {
    "admin-stats": {
        "cls": "invenio_stats.queries.TermsQuery",
        "permission_factory": "myapp.permissions:admin_only_permission",
        "params": {...}
    },
}
```

--------------------------------

### Task: process_events

Source: https://github.com/inveniosoftware/invenio-stats/blob/master/docs/api.rst

Background task to process raw events within the Invenio Stats system.

```APIDOC
## Task: process_events

### Description
Processes raw events collected by the system.

### Task Name
invenio_stats.tasks.process_events
```

--------------------------------

### Query Multiple Statistics in One Request

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

This cURL command demonstrates how to query multiple statistics, such as daily downloads and record views, simultaneously. Specify the desired date range, interval, and relevant IDs.

```bash
curl -X POST http://localhost:5000/stats \
  -H "Content-Type: application/json" \
  -d '{
    "daily_downloads": {
      "stat": "bucket-file-download-histogram",
      "params": {
        "start_date": "2024-01-01",
        "end_date": "2024-01-31",
        "interval": "day",
        "bucket_id": "550e8400-e29b-41d4-a716-446655440000",
        "file_key": "document.pdf"
      }
    },
    "record_views": {
      "stat": "record-view-histogram",
      "params": {
        "start_date": "2024-01-01",
        "end_date": "2024-01-31",
        "interval": "week",
        "record_id": "12345"
      }
    }
  }'
```

--------------------------------

### CLI: Process Events Synchronously

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Run the Flask CLI command to process events in eager mode, meaning tasks are executed synchronously. This is helpful for debugging or when immediate results are needed.

```bash
flask stats events process --eager
```

--------------------------------

### CLI: Process Specific Aggregations

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Use this Flask CLI command to process only specific aggregation tasks, such as 'file-download-agg' or 'record-view-agg'. This provides granular control over aggregation jobs.

```bash
flask stats aggregations process file-download-agg record-view-agg
```

--------------------------------

### CLI: Process Specific Event Types

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Execute the Flask CLI command to process only specific event types, such as 'file-download' and 'record-view'. This allows for targeted data processing.

```bash
flask stats events process file-download record-view
```

--------------------------------

### Configure Celery Beat for Event Processing

Source: https://github.com/inveniosoftware/invenio-stats/blob/master/docs/configuration.rst

Use this Celery Beat schedule to trigger the `process_events` task every 3 hours. Ensure Celery is properly configured in your project.

```python
from datetime import timedelta
CELERY_BEAT_SCHEDULE = {
    'indexer': {
        'task': 'invenio_stats.tasks.process_events',
        'schedule': timedelta(hours=3),
    },
}
```

--------------------------------

### Preprocessor Functions for Event Filtering and Flagging

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Utilize built-in preprocessor functions to flag or filter events based on robot or machine traffic, and to anonymize user data. `flag_robots` and `flag_machines` add boolean fields, while `filter_robots` removes events.

```python
from invenio_stats.processors import (
    flag_robots,
    filter_robots,
    flag_machines,
    filter_machines,
    anonymize_user,
)

# flag_robots: Adds is_robot field but keeps the event
doc = {
    "user_agent": "Googlebot/2.1 (+http://www.google.com/bot.html)",
    "timestamp": "2024-01-15T10:30:00",
}
result = flag_robots(doc)
# result = {"user_agent": "...", "timestamp": "...", "is_robot": True}

# filter_robots: Returns None for robot events (filters them out)
doc = {
    "user_agent": "Googlebot/2.1 (+http://www.google.com/bot.html)",
    "timestamp": "2024-01-15T10:30:00",
}
result = filter_robots(doc)  # Returns None

# flag_machines: Adds is_machine field for automated tools
doc = {
    "user_agent": "python-requests/2.28.0",
    "timestamp": "2024-01-15T10:30:00",
}
result = flag_machines(doc)
# result = {"user_agent": "...", "timestamp": "...", "is_machine": True}

# anonymize_user: Removes PII and generates anonymous IDs
doc = {
    "timestamp": "2024-01-15T10:30:00",
    "ip_address": "192.168.1.100",
    "user_id": "42",
    "session_id": "abc123xyz",
    "user_agent": "Mozilla/5.0...",
}
result = anonymize_user(doc)
# result = {
#     "timestamp": "2024-01-15T10:30:00",
#     "country": "US",  # Derived from IP
#     "visitor_id": "a1b2c3...",  # Hashed, daily-unique identifier
#     "unique_session_id": "x9y8z7...",  # Hashed, hourly-unique identifier
# }

```

--------------------------------

### CLI: Process Aggregations for a Date Range

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Process aggregations within a specified date range using the Flask CLI. This is useful for recalculating or updating historical aggregation data.

```bash
flask stats aggregations process \
  --start-date "2024-01-01" \
  --end-date "2024-01-31"
```

--------------------------------

### REST API - Query Statistics

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Query statistics through the REST API endpoint at `/stats`.

```APIDOC
## REST API - Query Statistics

### Endpoint
`POST /stats`

### Description
Query aggregated statistics through the REST API.

### Request Body
- **\[query_name\]** (object) - Required - An object defining the query to run.
  - **stat** (string) - Required - The name of the statistic to query (e.g., `bucket-file-download-histogram`).
  - **params** (object) - Required - Parameters for the statistic query.
    - **start_date** (string) - Required - The start date for the query.
    - **end_date** (string) - Required - The end date for the query.
    - **interval** (string) - Optional - The aggregation interval for date histograms.
    - **\[filter_field\]** (string) - Optional - Additional filter fields.

### Request Example
```bash
# Query a date histogram of file downloads
curl -X POST http://localhost:5000/stats \
  -H "Content-Type: application/json" \
  -d '{
    "downloads_over_time": {
      "stat": "bucket-file-download-histogram",
      "params": {
        "start_date": "2024-01-01",
        "end_date": "2024-01-31",
        "interval": "day",
        "bucket_id": "550e8400-e29b-41d4-a716-446655440000",
        "file_key": "document.pdf"
      }
    }
  }'
```

### Response Example
```json
{
  "downloads_over_time": {
    "interval": "day",
    "key_type": "date",
    "start_date": "2024-01-01T00:00:00",
    "end_date": "2024-01-31T00:00:00",
    "buckets": [
      // ... bucket data ...
    ]
  }
}
```
```

--------------------------------

### POST /stats

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Query statistics by providing a JSON object containing the statistic name and parameters. Supports querying multiple statistics in a single request.

```APIDOC
## POST /stats

### Description
Query statistics by providing a JSON object containing the statistic name and parameters. Supports querying multiple statistics in a single request.

### Method
POST

### Endpoint
http://localhost:5000/stats

### Request Body
- **[key]** (object) - Required - A dictionary where the key is the identifier for the result and the value contains the 'stat' name and 'params' object.

### Request Example
{
  "total_by_file": {
    "stat": "bucket-file-download-total",
    "params": {
      "start_date": "2024-01-01",
      "end_date": "2024-01-31",
      "bucket_id": "550e8400-e29b-41d4-a716-446655440000"
    }
  }
}
```

--------------------------------

### StatAggregator for Event Aggregation

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

The StatAggregator class is used to compress raw events into statistics documents for efficient querying. It works in conjunction with preprocessors like `filter_robots`.

```python
from invenio_stats.aggregations import StatAggregator, filter_robots
from invenio_search import current_search_client

```

--------------------------------

### TermsQuery

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Query aggregated statistics grouped by field values (terms).

```APIDOC
## TermsQuery

### Description
Query aggregated statistics grouped by field values (terms).

### Parameters
- **name** (str) - Required - The name of the query.
- **index** (str) - Required - The search index to query.
- **client** (object) - Required - The search client instance.
- **time_field** (str) - Required - The field representing time.
- **copy_fields** (dict) - Optional - Fields to copy to the result buckets.
- **metric_fields** (dict) - Optional - Fields to calculate metrics on.
- **required_filters** (dict) - Optional - Filters that must be applied.
- **aggregated_fields** (list) - Required - Fields to group the terms by.
- **max_bucket_size** (int) - Optional - Maximum number of buckets.

### Method
#### `run`
Executes the terms query.

**Parameters**
- **start_date** (str) - Required - The start date for the query.
- **end_date** (str) - Required - The end date for the query.
- **\[filter_field\]** (str) - Optional - Any additional filter fields.

### Request Example
```python
from invenio_stats.queries import TermsQuery
from invenio_search import current_search_client

# Create a terms query for file download totals
query = TermsQuery(
    name="file-downloads-by-file",
    index="stats-file-download",
    client=current_search_client,
    time_field="timestamp",
    copy_fields={
        "bucket_id": "bucket_id",
    },
    metric_fields={
        "total_downloads": ("sum", "count", {{}}),
        "unique_visitors": ("sum", "unique_count", {{}})
    },
    required_filters={
        "bucket_id": "bucket_id",
    },
    aggregated_fields=["file_key"],  # Group by file_key
    max_bucket_size=10000,
)

# Run the query
result = query.run(
    start_date="2024-01-01",
    end_date="2024-01-31",
    bucket_id="550e8400-e29b-41d4-a716-446655440000",
)
```

### Response Example
```json
{
    "start_date": "2024-01-01T00:00:00",
    "end_date": "2024-01-31T00:00:00",
    "total_downloads": 500,
    "unique_visitors": 120,
    "type": "bucket",
    "field": "file_key",
    "key_type": "terms",
    "buckets": [
        {"key": "document.pdf", "total_downloads": 200, "unique_visitors": 50},
        {"key": "data.csv", "total_downloads": 180, "unique_visitors": 45},
        {"key": "image.png", "total_downloads": 120, "unique_visitors": 25},
    ]
}
```
```

--------------------------------

### Manual Celery Task Invocation

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Manually invoke Celery tasks for processing and aggregating statistics. Use `.delay()` for asynchronous execution or `.apply()` for synchronous execution (testing).

```python
from invenio_stats.tasks import process_events, aggregate_events

# Process specific event types
result = process_events.delay(["file-download", "record-view"])
print(f"Task ID: {result.id}")

# Aggregate with date range
result = aggregate_events.delay(
    ["file-download-agg"],
    start_date="2024-01-01T00:00:00",
    end_date="2024-01-31T23:59:59",
    update_bookmark=True,
)

# Synchronous execution (for testing)
result = process_events.apply(
    args=[["file-download"]],
    throw=True,
)
print(f"Processed: {result}")
```

--------------------------------

### Define Custom Query Modifiers in Python

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Create Python functions to filter or transform search queries. These modifiers can be included in query configurations to apply specific filtering logic.

```python
from invenio_search.engine import dsl

def filter_by_community(query, community_id=None, **kwargs):
    """Filter statistics by community."""
    if community_id:
        return query.filter("term", community_id=community_id)
    return query

def exclude_internal_users(query, **kwargs):
    """Exclude internal/staff user activity."""
    return query.filter("bool", must_not=[
        dsl.Q("term", is_internal=True),
        dsl.Q("prefix", visitor_id="internal-"),
    ])

def time_of_day_filter(query, hour_start=None, hour_end=None, **kwargs):
    """Filter by hour of day (useful for business hours analysis)."""
    if hour_start is not None and hour_end is not None:
        return query.filter("script", script={
            "source": "doc['timestamp'].value.getHour() >= params.start && doc['timestamp'].value.getHour() < params.end",
            "params": {"start": hour_start, "end": hour_end}
        })
    return query
```

```python
# Use in query configuration
STATS_QUERIES = {
    "community-downloads": {
        "cls": "invenio_stats.queries.DateHistogramQuery",
        "params": {
            "index": "stats-file-download",
            "query_modifiers": [
                "myapp.modifiers:filter_by_community",
                "myapp.modifiers:exclude_internal_users",
            ],
            "metric_fields": {
                "count": ("sum", "count", {}),
            },
        }
    },
}
```

--------------------------------

### Task: aggregate_events

Source: https://github.com/inveniosoftware/invenio-stats/blob/master/docs/api.rst

Background task to perform aggregation on processed events.

```APIDOC
## Task: aggregate_events

### Description
Aggregates processed events based on defined aggregation rules.

### Task Name
invenio_stats.tasks.aggregate_events
```

--------------------------------

### DateHistogramQuery

Source: https://context7.com/inveniosoftware/invenio-stats/llms.txt

Query aggregated statistics as a time-series histogram.

```APIDOC
## DateHistogramQuery

### Description
Query aggregated statistics as a time-series histogram.

### Parameters
- **name** (str) - Required - The name of the query.
- **index** (str) - Required - The search index to query.
- **client** (object) - Required - The search client instance.
- **time_field** (str) - Required - The field representing time.
- **copy_fields** (dict) - Optional - Fields to copy to the result buckets.
- **metric_fields** (dict) - Optional - Fields to calculate metrics on.
- **required_filters** (dict) - Optional - Filters that must be applied.
- **max_bucket_size** (int) - Optional - Maximum number of buckets.

### Method
#### `run`
Executes the date histogram query.

**Parameters**
- **interval** (str) - Required - The aggregation interval (e.g., 'day', 'month').
- **start_date** (str) - Required - The start date for the query.
- **end_date** (str) - Required - The end date for the query.
- **\[filter_field\]** (str) - Optional - Any additional filter fields.

### Request Example
```python
from invenio_stats.queries import DateHistogramQuery
from invenio_search import current_search_client

# Create a histogram query
query = DateHistogramQuery(
    name="file-downloads-over-time",
    index="stats-file-download",
    client=current_search_client,
    time_field="timestamp",
    copy_fields={
        "bucket_id": "bucket_id",
        "file_key": "file_key",
    },
    metric_fields={
        "downloads": ("sum", "count", {{}}),
        "unique_visitors": ("sum", "unique_count", {{}})
    },
    required_filters={
        "bucket_id": "bucket_id",
    },
)

# Run the query
result = query.run(
    interval="day",  # year, quarter, month, week, day
    start_date="2024-01-01",
    end_date="2024-01-31",
    bucket_id="550e8400-e29b-41d4-a716-446655440000",
)
```

### Response Example
```json
{
    "interval": "day",
    "key_type": "date",
    "start_date": "2024-01-01T00:00:00",
    "end_date": "2024-01-31T00:00:00",
    "buckets": [
        {
            "key": 1704067200000,
            "date": "2024-01-01T00:00:00.000Z",
            "downloads": 150,
            "unique_visitors": 45,
            "bucket_id": "550e8400-e29b-41d4-a716-446655440000",
        },
        ...
    ]
}
```
```