#!/usr/bin/env python3
"""Prepare local TrueFetch result pages or JSONL blocks for filter_jobs.py.

No network requests or model calls. Supply all files in API/download order and
the saved task response. Count/order checks detect incomplete local exports;
they do not authenticate exports or prove that files belong to the same task.
"""

import argparse
import copy
import hashlib
import json
from pathlib import Path
import sys

from filter_jobs import normalize

MAX_BYTES = 64 * 1024 * 1024


def consumer_job(row):
    """Project the Job Search record into this consumer's separate input model."""
    location = row.get("location")
    label = None
    if isinstance(location, dict):
        label = location.get("raw")
        if not label:
            parts = [location.get(key) for key in ("locality", "region", "country")]
            label = ", ".join(part for part in parts if isinstance(part, str) and part) or None
    return {
        "title": row.get("title"),
        "company": row.get("companyName"),
        "location": label,
        "description": row.get("description"),
        "source": row.get("sourceId"),
        "sourceUrl": row.get("platformUrl") or row.get("officialUrl"),
        "collectedAt": row.get("processedAt"),
    }


def load_inputs(task_path, paths, kind, allow_partial=False):
    paths = [Path(task_path), *map(Path, paths)]
    if sum(p.stat().st_size for p in paths) > MAX_BYTES:
        raise ValueError("Export exceeds the 64 MiB local import limit; no output written")
    raw = [p.read_bytes() for p in paths]
    if sum(map(len, raw)) > MAX_BYTES:
        raise ValueError("Export changed or exceeds the import limit")
    saved = json.loads(raw[0].decode("utf-8-sig"))
    task = saved.get("task") if isinstance(saved, dict) else None
    if not isinstance(task, dict) or task.get("productId") != "job-search":
        raise ValueError("Supply a Job Search task response")
    if not isinstance(task.get("taskId"), str) or not task["taskId"]:
        raise ValueError("Task identity is missing")
    status = task.get("status")
    if status not in ("succeeded", "partial", "empty"):
        raise ValueError("Task must be succeeded, partial or empty")
    if status == "partial" and not allow_partial:
        raise ValueError("Partial collection requires --allow-partial")
    expected = task.get("resultCount")
    if type(expected) is not int or expected < 0 or (status == "empty" and expected != 0):
        raise ValueError("Invalid task result count")
    rows = []
    for index, data in enumerate(raw[1:]):
        if kind == "jsonl":
            rows.extend(
                json.loads(line) for line in data.decode("utf-8-sig").splitlines() if line.strip()
            )
        else:
            page = json.loads(data.decode("utf-8-sig"))
            if not isinstance(page, dict) or not isinstance(page.get("data"), list):
                raise ValueError("Invalid result page")
            meta = page.get("meta")
            pagination = meta.get("pagination") if isinstance(meta, dict) else None
            if (
                not isinstance(pagination, dict)
                or type(meta.get("resultCount")) is not int
                or meta["resultCount"] != expected
            ):
                raise ValueError("Page result count does not match the task")
            more = index < len(raw) - 2
            cursor = pagination.get("nextCursor")
            if (
                pagination.get("hasMore") is not more
                or (more and (not isinstance(cursor, str) or not cursor))
                or (not more and cursor is not None)
            ):
                raise ValueError("Missing, extra or malformed result pages")
            rows.extend(page["data"])
    if len(rows) != expected:
        raise ValueError("Export count does not match task resultCount")
    batches, batch, mapping, seen_urls = [], [], [], {}
    previous = None
    for index, row in enumerate(rows):
        if not isinstance(row, dict):
            raise ValueError("Result record must be an object")
        record_id = row.get("recordId")
        if (
            not isinstance(record_id, str)
            or not record_id
            or (previous is not None and record_id <= previous)
        ):
            raise ValueError("Record IDs must be unique and in API order")
        previous = record_id
        # Reuse the consumer's exact field, URL and size checks. Never truncate.
        try:
            job = normalize([consumer_job(row)])[0]
        except ValueError as exc:
            raise ValueError(f"Record {index + 1} cannot be imported: {exc}") from None
        entry = {"recordId": record_id, "inputIndex": index, "sourceUrl": job["sourceUrl"]}
        if job["sourceUrl"] in seen_urls:
            entry.update(
                disposition="duplicate_source_url", duplicateOf=seen_urls[job["sourceUrl"]]
            )
        else:
            seen_urls[job["sourceUrl"]] = record_id
            entry.update(
                disposition="missing_description" if not job.get("description") else "ready",
                batch=len(batches) + 1,
                row=len(batch),
            )
            batch.append(job)
            if len(batch) == 25:
                batches.append({"status": status, "results": batch})
                batch = []
        mapping.append(entry)
    if batch:
        batches.append({"status": status, "results": batch})
    manifest = {
        "version": 1,
        "taskId": task["taskId"],
        "collectionStatus": status,
        "resultCount": expected,
        "uniqueSourceUrls": len(seen_urls),
        "sourceOutcomes": task.get("sourceOutcomes"),
        "coverageTruncated": task.get("coverageTruncated"),
        "notices": task.get("notices", []),
        "inputs": [
            {"position": i, "sha256": hashlib.sha256(data).hexdigest(), "bytes": len(data)}
            for i, data in enumerate(raw)
        ],
        "records": mapping,
        "batches": [],
        "limits": "Local export consistency only; not authenticated task provenance or model acceptance. No model calls made.",
    }
    return batches, manifest


def write_batches(output, batches, manifest):
    # Validate/serialize everything before creating the output directory.
    manifest = copy.deepcopy(manifest)
    contents = []
    for i, batch in enumerate(batches, 1):
        data = (json.dumps(batch, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
        if len(data) > 1_000_000:
            raise ValueError("Batch exceeds consumer file limit")
        name = f"batch-{i:04d}.json"
        contents.append((name, data))
        manifest["batches"].append(
            {
                "file": name,
                "records": len(batch["results"]),
                "sha256": hashlib.sha256(data).hexdigest(),
            }
        )
    directory = Path(output)
    directory.mkdir()  # Existing output is never overwritten, including partial runs.
    for name, data in contents:
        with (directory / name).open("xb") as stream:
            stream.write(data)
    # Written last: absence of manifest means the import did not finish.
    with (directory / "manifest.json").open("x", encoding="utf-8") as stream:
        json.dump(manifest, stream, ensure_ascii=False, indent=2)
        stream.write("\n")


def main():
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""Examples (keep import_results.py and filter_jobs.py in the same folder):
  python import_results.py --task task.json --pages page1.json page2.json --output imported
  python import_results.py --task task.json --jsonl block1.jsonl block2.jsonl --output imported
  python filter_jobs.py prepare imported/batch-0001.json --output requests-0001.jsonl

Inspect manifest.json for every record's disposition and all batch filenames.
Repeat prepare for each listed batch with a different output filename. For a
partial task, pass --allow-partial to both import and prepare. Empty results
create a manifest with no batches. Missing descriptions remain review records.
Oversize or invalid records reject the import rather than being truncated.
Only a completed import has manifest.json; do not process an incomplete folder.
Do not rerun paid filtering of already evaluated batches: imports do not track
model execution or prevent charges across separate filtering runs.
""",
    )
    parser.add_argument("--task", required=True, help="Saved GET /v1/tasks/{id} JSON response")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--pages", nargs="+", help="All saved result pages in cursor order")
    group.add_argument("--jsonl", nargs="+", help="All downloaded JSONL blocks in manifest order")
    parser.add_argument(
        "--output", required=True, help="New directory; never overwrite an existing import"
    )
    parser.add_argument("--allow-partial", action="store_true")
    args = parser.parse_args()
    try:
        batches, manifest = load_inputs(
            args.task,
            args.pages or args.jsonl,
            "pages" if args.pages else "jsonl",
            args.allow_partial,
        )
        write_batches(args.output, batches, manifest)
        print(
            f"Imported {manifest['resultCount']} records into {len(batches)} batches; no model calls made."
        )
        return 0
    except (OSError, ValueError) as exc:
        # Do not print input content or file-system details that may contain private data.
        print(
            f"Import failed ({type(exc).__name__}); verify complete ordered exports, record limits and a new output directory.",
            file=sys.stderr,
        )
        return 1


if __name__ == "__main__":
    sys.exit(main())