Metadata-Version: 2.4
Name: airflow-provider-avito
Version: 0.4.0
Summary: Apache Airflow provider for Avito CPA — collect call statistics
Author-email: Michael Kozhin <michael@kozhin.cc>
License: MIT
Project-URL: Homepage, https://github.com/mkozhin/airflow-provider-avito
Project-URL: Documentation, https://github.com/mkozhin/airflow-provider-avito
Project-URL: Repository, https://github.com/mkozhin/airflow-provider-avito
Project-URL: Changelog, https://github.com/mkozhin/airflow-provider-avito/blob/main/CHANGELOG.md
Keywords: airflow,avito,provider,cpa,calls
Classifier: Framework :: Apache Airflow
Classifier: Framework :: Apache Airflow :: Provider
Classifier: Development Status :: 4 - Beta
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: apache-airflow<3.0,>=2.9.1
Requires-Dist: requests>=2.28
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Dynamic: license-file

# airflow-provider-avito

Apache Airflow provider for [Avito CPA](https://developers.avito.ru/api-catalog/cpa/documentation) — collect call statistics from the Avito advertising platform.

---

*Powered by [Claude Code](https://claude.ai/code)*

---

## Installation

```bash
pip install airflow-provider-avito
```

Requires Python 3.10+ and `apache-airflow>=2.9.1`.

## Connection

Create an Airflow connection of type **HTTP** with `conn_id = avito_default` (or any name you pass to the operator).

Authentication uses only the **Extra** field; `password` is ignored. In the single-account form `login` is read too — but only to populate the `account_id` column of output records, never for authentication.

### Single account

```json
{
  "client_id": "your_client_id",
  "client_secret": "your_client_secret"
}
```

### Multiple accounts

```json
{
  "accounts": [
    {"id": "main",    "client_id": "id1", "client_secret": "secret1"},
    {"id": "agency",  "client_id": "id2", "client_secret": "secret2"}
  ]
}
```

Use `account_id` parameter on the operator to select which account to use.

> **Note:** the `account_id` column stamped onto every output record is the record's provenance. In the multi-account form it **matches** the `account_id` selector you pass to the operator — the same value used to name file/GCS/S3 paths and BigQuery table suffixes (the example DAG uses `{BASE_DIR}/{account_id}/...` and `{BQ_TABLE}_{account_id}`). In the single-account form it is the connection's `login` (or `null` when `login` is empty).

## Quick start

```python
from airflow.decorators import dag
from airflow.models.param import Param
from airflow_provider_avito.operators.calls import AvitoCallsOperator

@dag(schedule=None, params={"date_from": Param("2026-06-01"), "date_to": Param("2026-06-07")})
def avito_calls_example():
    AvitoCallsOperator(
        task_id="collect_calls",
        avito_conn_id="avito_default",
        date_from="{{ params.date_from }}",
        date_to="{{ params.date_to }}",
        base_dir="/tmp/avito",
        output_format="json",   # or "csv"
        add_snapshot_ts=True,   # optional, see "Snapshot versioning" below
    )

avito_calls_example()
```

The operator writes one JSONL (or CSV) file per date to `{base_dir}/{safe_run_id}/{date}.json` and returns a `list[dict]` with `{"date": ..., "path": ..., "snapshot_ts": ...}` entries (`snapshot_ts` is `None` unless `add_snapshot_ts=True`).

### Snapshot versioning (`add_snapshot_ts`)

By default, each DAG run writes to the same per-date path, so re-running the DAG overwrites previous output and any history of call-status changes is lost.

Set `add_snapshot_ts=True` to inject `snapshot_ts` — the DAG run's `start_date` (actual wall-clock UTC start time of the run), formatted as `YYYY-MM-DDTHH:MM:SS` — into every JSON record and into the operator's returned `snapshot_ts` key. This lets a downstream task build a unique, non-overwriting path per run (e.g. an S3 key suffixed with the snapshot timestamp) and lets ClickHouse/Spark queries pick the latest snapshot or trace status history over time:

```sql
-- ClickHouse: latest snapshot only
SELECT * FROM s3('s3://bucket/prefix/**/*.json', 'JSONEachRow')
WHERE toDateTime(snapshot_ts) = (
    SELECT MAX(toDateTime(snapshot_ts)) FROM s3('s3://bucket/prefix/**/*.json', 'JSONEachRow')
)
```

`add_snapshot_ts` only applies to `output_format="json"`; it is ignored when `output_format="csv"` (the CSV column schema is fixed).

## Output record schema

Each record contains 18 fields. The canonical ordered list of field names is also available as `CALL_FIELDS` (a `tuple[str, ...]` exported from `airflow_provider_avito.hooks.avito`) for use in downstream schema definitions or validation.

| Field | Type | Description |
|---|---|---|
| `account_id` | str \| null | The cabinet's business identifier — record provenance: the `account_id` selector (multi-account form) or the connection's `login` (single-account form); `null` when unavailable |
| `id` | int | Call ID |
| `buyer_phone` | str | Buyer phone |
| `seller_phone` | str | Seller phone |
| `virtual_phone` | str | Virtual (masked) phone |
| `create_time` | str | Creation time (RFC3339) |
| `start_time` | str | Call start time (RFC3339) |
| `date` | str | Date (YYYY-MM-DD) derived from `start_time` |
| `duration` | int | Call duration, seconds |
| `waiting_duration` | float | Wait time before answer, seconds |
| `price` | int | Price in kopecks |
| `price_rub` | float | Price in rubles (`price / 100`) |
| `status_id` | int | Status code |
| `status` | str | Status label (e.g. "Целевой") |
| `item_id` | int | Ad ID |
| `group_title` | str | Campaign name |
| `is_arbitrage_available` | bool | Whether arbitrage is available |
| `record_url` | str | Call recording URL |

When `add_snapshot_ts=True` and `output_format="json"`, a 19th field is added to every record:

| Field | Type | Description |
|---|---|---|
| `snapshot_ts` | str | DAG run's `start_date`, ISO 8601 (`YYYY-MM-DDTHH:MM:SS`). Only present when `add_snapshot_ts=True` and `output_format="json"`. |

### Call statuses

| `status_id` | `status` |
|---|---|
| 0 | Целевой |
| 1 | На модерации |
| 2 | Целевой после модерации |
| 3 | Нецелевой после модерации |

## Request diagnostics in Loki (`loki_conn_id`)

Optional, off by default. With `loki_conn_id` set, the operator emits one compact diagnostic event per HTTP attempt against the Avito `callsByTime` endpoint — retries and the request repeated after a 401 token refresh each count as an attempt — to a [Loki](https://grafana.com/docs/loki/latest/) instance. The events describe how each attempt went (outcome, timing, request parameters, HTTP status, the shape of the raw page), so a past run can be explained afterwards in Grafana. **No call data is sent.**

Turning diagnostics on does not change the export: the same files, the same operator return value, the same exceptions with the same types and messages. A Loki outage cannot fail the task — the first push failure logs one WARNING and disables diagnostics for the rest of the run. The one cost is wall-clock: the push is synchronous, with a 2 s connect timeout and a 3 s read timeout, so an unresponsive Loki holds an attempt for about 5 s — once, before diagnostics switch themselves off. The read half bounds the quiet between received bytes rather than the whole exchange, so a Loki answering in a slow dribble can hold an attempt longer than that; only the response status is used, and the body is never downloaded. What diagnostics never absorb is the task being stopped: an `execution_timeout` firing or a SIGTERM arriving while a push is in flight interrupts the task there and then, exactly as it does with diagnostics off. A stop that arrives earlier — during the Avito request itself or the backoff pause — cancels the push instead of being held for it, so the interrupted attempt goes unreported and the task ends as promptly as it would with diagnostics off.

Only the `callsByTime` requests are instrumented. A run that fails before the first page — a broken connection `extra`, a failing OAuth2 token request, an unreadable connection during `account_id` resolution — sends nothing, so the absence of events for a `dag_run` is not evidence about it: it reads the same as diagnostics being off or Loki being unreachable.

```python
AvitoCallsOperator(
    task_id="collect_calls",
    avito_conn_id="avito_default",
    loki_conn_id="loki_default",   # optional; without it nothing is sent
    date_from="{{ params.date_from }}",
    date_to="{{ params.date_to }}",
    base_dir="/tmp/avito",
)
```

### Loki connection

Create an Airflow connection with `conn_type = http`:

| Airflow UI field | Meaning |
|---|---|
| **Host** | Loki base URL, either with an explicit scheme (`https://loki.example.ru`, port allowed: `https://loki.example.ru:3100`) or a bare host (`loki.example.ru`) paired with **Schema**. An IPv6 address goes in brackets: `[::1]`, `http://[::1]:3100` |
| **Schema** | `https` or `http`. Required when **Host** carries no scheme |
| **Port** | Optional (e.g. `3100`), used only when **Host** carries neither a scheme nor a port of its own |
| **Login** / **Password** | Optional Basic Auth. Set both or neither |

The push path `/loki/api/v1/push` is appended automatically; a trailing slash on **Host** is fine, and a **Host** that already ends in the push path is taken as is.

Two configurations are equivalent: `Host = https://loki.example.ru` alone, or `Host = loki.example.ru` plus `Schema = https`.

Credentials belong in **Login**/**Password**, never in the URL: a **Host** carrying userinfo (`https://user:token@loki.example.ru`, the form Grafana Cloud publishes) is rejected with a WARNING, as are a query string and a fragment.

The scheme is never guessed. A bare **Host** with an empty **Schema** is a broken connection: diagnostics are disabled with a WARNING naming the fix, rather than silently defaulting to `http`. The same happens for an empty **Host** and for any scheme other than http/https.

Basic Auth requires HTTPS: with **Login** set and a non-HTTPS URL, nothing is sent. Half-filled credentials (**Login** without **Password**, or the reverse) count as a misconfiguration and disable diagnostics too.

Multi-tenant Loki is not supported — no `X-Scope-OrgID` header is sent. The target must be single-tenant or sit behind a gateway that stamps the tenant itself.

A push counts as delivered only on HTTP 204, the status Loki answers with. Anything else — a `200` from a reverse proxy, a redirect (redirects are not followed) — is a failure: one WARNING, and diagnostics are off for the rest of the run.

Each entry carries a single stream label, `service="airflow-provider-avito"`, so label cardinality stays constant. Everything else lives in the JSON log line and is queried with LogQL over the parsed body:

```logql
{service="airflow-provider-avito"} | json | outcome != "success"
```

Because that label is the same for every task, all tasks write into one stream. On a Loki that rejects out-of-order writes, concurrent tasks can therefore have a push refused with a 4xx, which disables diagnostics for that task.

Outside an operator, the same client can be handed to the hook directly: `AvitoHook(avito_conn_id=..., loki=LokiClient(conn_id="loki_default", context={...}))`.

### Event fields

| Field | Description |
|---|---|
| `schema_version` | Event format version, currently `1` |
| `dag_id`, `task_id`, `dag_run_id`, `try_number`, `map_index` | Correlation with the Airflow task instance (`map_index` is `-1` when not mapped) |
| `outcome` | How the attempt ended — see the table below |
| `account_id` | The cabinet whose calls are being collected, as stamped onto output records |
| `offset`, `date_time_from` | Request parameters of the paginated page |
| `attempt`, `max_attempts` | Retry counters for one page request: `attempt` counts from 1 up to `max_attempts` as 429/5xx responses are retried. The request repeated after a 401 token refresh starts its own count, so a refresh shows up as two events with `attempt = 1`, told apart by `sent_at` |
| `sent_at` | UTC ISO 8601 timestamp taken just before the request is sent |
| `duration_ms` | Wall-clock duration of the HTTP attempt |
| `http_status` | Response status, `null` when the request never got one |
| `calls_count` | Number of entries in the raw page, `null` when no `calls` list was recognised |
| `calls_shape_ok` | Whether the page held a `calls` list of dicts |
| `payload_kind` | Which shape the body turned out to have: `dict` (the export reads a page out of it), `calls_non_list` (`calls` holds a non-empty value that is not a list), `result_non_dict` (`result` holds something other than a dict), `non_dict` (the body itself is not a dict) |
| `error_code`, `error_message` | `code` and truncated `message` from the API's `error` object |
| `exception_type`, `exception_message` | Type of the exception that ended the attempt; the message is filled only for a JSON parse error reported by the standard decoder, from a fixed vocabulary |
| `rate_limit_limit`, `rate_limit_remaining` | `X-RateLimit-*` headers, collected on HTTP 429 |

`calls_count`, `calls_shape_ok` and `payload_kind` all stay `null` for any attempt that never produced a parsed HTTP-200 body.

### `outcome` values

| Value | Meaning |
|---|---|
| `success` | HTTP 200 with a well-formed page, including an empty one — "no calls in this period" is a valid answer |
| `empty_shape` | HTTP 200 in which no `calls` list of dicts was recognised — see below, the outcome covers both a quiet empty page and a run that fails |
| `api_error` | HTTP 200 carrying an `error` object |
| `auth_error` | HTTP 401 — the token is refreshed and the request repeated once |
| `retryable_error` | HTTP 429, 500, 502, 503 or 504 — retried with backoff, or, on the last attempt, raised. Any other 5xx (`501`, `505`, …) is an `http_error` |
| `http_error` | Any other non-200 status |
| `network_error` | The request never completed (timeout, DNS, TLS, proxy) |
| `invalid_json` | HTTP 200 whose body could not be parsed |
| `unexpected_error` | Safety net: an attempt that ended some other way, e.g. a body that is valid JSON but not an object |

`empty_shape` splits in two, and the other fields say which half an event belongs to:

- `payload_kind = "dict"` with `calls_count = null` — a body with no `calls` key, `calls: null`, or another empty `calls` value (`""`, `{}`). The export treats it as an empty page: green task, no file.
- `payload_kind = "calls_non_list"` or `"result_non_dict"`, or `calls_shape_ok = false` with a non-`null` `calls_count` — a non-empty `calls` that is not a list (`{"calls": "abc"}`, `{"calls": {"1": {…}}}`), a non-dict `result` (`{"result": null}`, `{"result": []}`), or a `calls` list holding non-dicts (`{"calls": [null]}`). The export fails downstream with `AttributeError` or `TypeError`: the task ends red.

The quiet half — `empty_shape` with `payload_kind = "dict"` and `calls_count = null` — and `success` with `calls_count = 0` are the pair worth watching: both end with a green task and no file, and only the event tells them apart.

### Content policy

The event is assembled from the fixed allowlist above. The raw response body, response headers other than the two `X-RateLimit-*`, call records, and the `error` object as a whole never leave the process.

- From `error` only `code` (an `int`) and `message` (a value whose type is exactly `str`, truncated to 300 characters) are taken. A value of an unexpected type is described by its type — `<non-dict error: list>`, `<non-str message: dict>` — rather than serialised, so nested keys such as `details` or `trace` cannot ride along.
- `exception_message` is filled only for `invalid_json`, and only when the standard JSON decoder reported the failure. It is rebuilt from the exception's own attributes rather than from its rendered text, and the wording is chosen from a fixed vocabulary of the decoder's own literals — `Expecting value`, `Expecting ',' delimiter`, `Expecting ':' delimiter`, `Expecting property name enclosed in double quotes`, `Extra data`, `Unterminated string starting at`, `Invalid control character at`, `Invalid \escape`, `Invalid \uXXXX escape` — followed by the position counted in the document: `Expecting value: line 1 column 1 (char 0)`. Anything the decoder words differently is reported as `<other decoder message>` with the same position, because some decoder messages are formatted around a character taken from the document (the pure-Python scanner writes `Invalid \escape: 'q'`), and a single character of the body is still the body. A parse failure of any other origin — a third-party decoder, a response object of unknown provenance — may quote what it was parsing and records `exception_type` alone, as does every other outcome: those exception texts embed the response body or, for network failures, the environment's proxy URL — credentials included.
- Of the response headers, only the two `X-RateLimit-*` are copied, and only when their type is exactly `str`, truncated to 32 characters. A value of any other type is described by its type (`<non-str header: int>`), so no unknown object is ever rendered into the event.
- Truncation bounds length, not content. `error_message` is free text written by the API, so it is size-limited but not redacted — that is the honest edge of the guarantee.

## Examples

Full production examples with BigQuery + S3 upload are in [`examples/`](examples/):

- [`bq_and_s3_multi_account_dag.py`](examples/bq_and_s3_multi_account_dag.py) — multiple accounts in parallel

## License

MIT
