Metadata-Version: 2.4
Name: jobstreaming
Version: 0.0.2
Summary: Concurrent, resumable job search streams with a DataFrame batch API
License-Expression: MIT
License-File: LICENSE
Keywords: jobs-scraper,streaming,linkedin,indeed,glassdoor,ziprecruiter
Author: JobStreaming contributors
Requires-Python: >=3.10,<4.0
Classifier: Development Status :: 3 - Alpha
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Dist: beautifulsoup4 (>=4.12.2,<5.0.0)
Requires-Dist: markdownify (>=1.1.0,<2.0.0)
Requires-Dist: pandas (>=2.1.0,<3.0.0)
Requires-Dist: pydantic (>=2.3.0,<3.0.0)
Requires-Dist: requests (>=2.31.0,<3.0.0)
Requires-Dist: tls-client (>=1.0.1,<2.0.0)
Description-Content-Type: text/markdown

# JobStreaming

JobStreaming is the concurrent, restartable job-collection library for JobCtrl. Its
primary API yields each job as soon as its adapter produces it, while retaining a
convenient DataFrame batch API.

This project is a heavily modified private fork of an MIT-licensed upstream project.
It retains the original license and attribution while using a completely separate
project, distribution, and import identity.

> **Alpha:** job boards change private endpoints and markup without notice. Treat every
> adapter as a best-effort integration, respect each site's terms and rate limits, and
> persist the events you need.

## What changed

- Searches run concurrently across sites; a slow or blocked site does not hold back
  healthy sites.
- Jobs, warnings, progress, site failures, and completion are typed events.
- JSON checkpoints make searches restartable with at-least-once delivery.
- Stable job identities and checkpointed deduplication prevent acknowledged jobs from
  being emitted again after a restart.
- Adapter failures are isolated. The batch API returns healthy partial results unless
  strict failure mode is requested.
- Requests and result models are immutable and validated.
- A `scrape_jobs(...) -> pandas.DataFrame` entry point remains available.
- Adapters are registered through an extensible registry rather than a hard-coded
  dispatcher.

```mermaid
flowchart LR
    A["SearchRequest"] --> B["Concurrent coordinator"]
    B --> C1["Indeed worker"]
    B --> C2["LinkedIn worker"]
    B --> C3["Other site workers"]
    C1 --> D["Bounded event queue"]
    C2 --> D
    C3 --> D
    D --> E["Job / progress / error events"]
    E --> F["Consumer"]
    F -->|"acknowledge"| G["Atomic checkpoint"]
    G -->|"resume"| B
```

## Installation

Python 3.10 through Python 3.14 are tested.

Install the published package from PyPI:

```bash
pip install -U jobstreaming
```

Both the PyPI distribution and Python import package are `jobstreaming`. No legacy
import alias is included. The source repository remains private; PyPI releases are
performed explicitly from verified distribution artifacts.

## Stream results immediately

```python
from jobstreaming import ErrorEvent, JobEvent, SearchCompleteEvent, stream_search

with stream_search(
    site_name=["indeed", "linkedin", "zip_recruiter"],
    search_term="software engineer",
    location="Madrid",
    results_wanted=20,  # per site
    checkpoint_path=".jobstreaming/search.json",
    resume=True,
    ack_mode="explicit",
) as stream:
    for event in stream:
        if isinstance(event, JobEvent):
            print(event.site.value, event.job.title, event.job.job_url)

            # Persist the job first when durability matters, then explicitly ack it.
            save_to_database(event.job)

        elif isinstance(event, ErrorEvent):
            print(f"{event.site.value} failed: {event.message}")

        elif isinstance(event, SearchCompleteEvent):
            print("all sites completed:", event.completed)

        stream.ack(event)
```

Each site runs in its own worker. Arrival order is intentionally unspecified: faster
sites and faster pages yield first.

For a job-only iterator:

```python
from jobstreaming import stream_jobs

for job in stream_jobs(
    site_name=["indeed", "google"],
    search_term="data engineer",
    location="Barcelona",
    results_wanted=10,
):
    print(job.title)
```

Use `stream_search` when you need errors, progress, source-site metadata, or explicit
checkpoint acknowledgements.

## Restart and delivery semantics

Checkpointing is opt-in. Pass either `checkpoint_path` or a custom `CheckpointStore`.

- The default `ack_mode="implicit"` preserves the convenient behavior where requesting
  the next event acknowledges the previous event.
- Use `ack_mode="explicit"` for durable consumers. In this mode, requesting another
  event before `stream.ack(event)` raises `UnacknowledgedEventError` and does not advance
  the checkpoint.
- Call `stream.ack(event)` after a durable write when you need the checkpoint advanced
  immediately.
- Leaving the context manager early does not acknowledge the last delivered event;
  call `stream.ack(event)` first when an intentional early stop should be committed.
- If execution stops before acknowledgement, that job can be replayed on restart. This
  is at-least-once delivery: it favors avoiding data loss over pretending exactly-once
  delivery is possible.
- Acknowledged jobs are deduplicated with stable, process-independent keys.
- Page and cursor state advances only after the corresponding progress event is
  acknowledged.
- Only failures classified as `transient_network` or `rate_limited` are retried by
  default. Configure `max_retries` and `retry_backoff` on `stream_search` or
  `scrape_jobs`.
- The checkpoint is written through an `fsync` plus atomic file replacement.
- Checkpoints carry an overall schema version, a monotonically increasing revision, and
  a cursor-state schema version for every adapter. An incompatible library or adapter
  upgrade raises `CheckpointCompatibilityError` before any board worker starts.
- A checkpoint is bound to the complete request fingerprint. Changing the query,
  filters, sites, or result count raises `CheckpointMismatchError`; use a new path or
  `resume=False` for a new search.
- Board-owned cursors can expire. If a board rejects an old cursor, the stream emits an
  `ErrorEvent` with `code="cursor_expired"` and `reset_checkpoint=True`; restart that
  site from a fresh checkpoint.
- Custom stores can provide compare-and-swap ownership using `checkpoint.revision`.
  Raise `CheckpointConflictError` for a stale save; the conflict is surfaced to the
  caller immediately and the stream stops without advancing its local checkpoint.

If a process crashes while handling a job, replay is expected. Make downstream writes
idempotent using `event.job_key` or the job's stable `id`.

## Compatible batch API

```python
from jobstreaming import scrape_jobs

jobs = scrape_jobs(
    site_name=["indeed", "linkedin", "zip_recruiter", "google"],
    search_term="software engineer",
    google_search_term="software engineer jobs near Madrid since yesterday",
    location="Madrid",
    results_wanted=20,
    hours_old=72,
    country_indeed="Spain",
)

jobs.to_csv("jobs.csv", index=False)
```

`scrape_jobs` consumes the same concurrent event stream and returns a DataFrame. By
default, a failed site is logged and healthy partial results are returned. Set
`raise_on_error=True` to raise after all sites have had a chance to finish.

Checkpoints store identities and cursor state, not full job payloads. A resumed batch
call therefore contains only jobs emitted during that invocation. For a durable full
result set across restarts, use `stream_search` and upsert each `JobEvent` into your own
store before acknowledging it.

## Events

`stream_search` can yield:

| Event | Meaning |
|---|---|
| `JobEvent` | One normalized job is ready. |
| `ProgressEvent` | A restart boundary such as a page or cursor was completed. |
| `WarningEvent` | A listing was skipped or a requested filter is unsupported. |
| `ErrorEvent` | A site failed; other sites continue. |
| `SiteCompleteEvent` | One site exhausted its work or reached its result limit. |
| `SearchCompleteEvent` | Every worker stopped. `completed=False` means at least one site failed. |

`ErrorEvent.code` is a stable `ErrorCode` value. `retryable` tells an operator whether
the same board operation can be retried, while `reset_checkpoint` tells them whether
the board cursor should be discarded first.

| Error code | Retry | Reset board checkpoint |
|---|---:|---:|
| `transient_network` | yes | no |
| `rate_limited` | yes | no |
| `invalid_request` | no | no |
| `cursor_expired` | no | yes |
| `authentication_configuration` | no | no |
| `cancelled` | no | no |
| `adapter_failure` | no | no |

## Cancellation

Supply a `threading.Event`, a callback, or both. Queue waits, retry backoff, and blocked
adapter/network operations are observed through the same cancellation boundary.
`close()` also wakes a consumer blocked in `next()`.

```python
from threading import Event

from jobstreaming import StreamCancelledError, stream_search

cancel = Event()

try:
    with stream_search(
        site_name=["indeed", "linkedin"],
        search_term="platform engineer",
        cancel_event=cancel,
        # cancel_callback=lambda: shutdown_requested(),  # optional alternative
    ) as stream:
        for event in stream:
            process(event)
except StreamCancelledError:
    pass
```

## Supported sites and important limits

| Site | Restart boundary | Notes |
|---|---|---|
| Indeed | cursor/page | `hours_old`, `easy_apply`, and `job_type`/`is_remote` are mutually exclusive in the upstream API. |
| LinkedIn | result offset/page | Full descriptions require `linkedin_fetch_description=True` and add one request per job. Aggressive rate limiting is common. |
| ZipRecruiter | continuation token | US and Canada are the primary supported markets. |
| Glassdoor | page/cursor | A location is required unless `is_remote=True`. Availability depends on `country_indeed`. |
| Google Jobs | cursor | `google_search_term` can override the generated query. The upstream response format is opaque and fragile. |
| Bayt | page | Currently supports keyword search and international results. |
| Naukri | page | India-focused. A non-empty `search_term` is required. |
| BDJobs | page | Bangladesh-focused. Detail pages are fetched concurrently within each result page. |

Adapters declare their supported filters. If a selected adapter cannot honor a
requested filter, the stream emits a `WarningEvent` instead of silently implying that
the filter was applied.

## Validated request API

For reusable searches, construct an immutable request explicitly:

```python
from jobstreaming import Country, SearchRequest, Site, stream_search

request = SearchRequest(
    site_type=(Site.INDEED, Site.LINKEDIN),
    search_term="platform engineer",
    location="Madrid",
    country=Country.SPAIN,
    results_wanted=25,
    request_timeout=20,
    max_pages=10,
)

with stream_search(request, checkpoint_path="search.json") as stream:
    for event in stream:
        ...
```

Negative offsets/result counts, invalid timeouts, malformed compensation ranges, empty
job titles/URLs, and unsupported enum values are rejected at the boundary.

## Custom adapters

```python
from jobstreaming import (
    AdapterCapabilities,
    AdapterRegistry,
    JobResponse,
    Scraper,
    Site,
    stream_search,
)

class InternalJobs(Scraper):
    capabilities = AdapterCapabilities(
        supports_resume=True,
        resume_granularity="cursor",
        cursor_schema_version=1,
    )

    def __init__(self, **kwargs):
        super().__init__(Site.INDEED)  # this example replaces the Indeed adapter

    def scrape(self, request, context=None):
        for job in fetch_internal_jobs(request):
            context.emit_job(job, {"cursor": job.id})
        return JobResponse()

registry = AdapterRegistry()
registry.register(Site.INDEED, InternalJobs)

with stream_search(
    site_name="indeed",
    registry=registry,
    search_term="engineer",
) as stream:
    for event in stream:
        ...
```

Increment `cursor_schema_version` whenever a deployed adapter can no longer interpret
cursor state written by its previous implementation. Legacy adapters that only return
`JobResponse` are still accepted, but their results cannot be streamed until that
adapter returns.

The registry can replace any built-in adapter. Adding an entirely new site also
requires adding that board to the `Site` enum so it participates in validation,
fingerprinting, events, and checkpoints.

## Development

```bash
poetry install
poetry run pytest
poetry run ruff check jobstreaming tests
poetry run black --check jobstreaming tests
poetry build
```

The test suite is offline: it validates domain invariants, concurrency, failure
isolation, acknowledgement/replay behavior, checkpoint persistence, compatibility, and
representative adapter parsing without calling live job boards.

## License and attribution

MIT. The retained license identifies Cullen Watson as the original copyright holder.
This rebuild is independently maintained and is not affiliated with the original
creator or any supported job board.

