Module pd3f.parsr_wrapper

Interaction with Parsr's REST API

Functions

def run_parsr(file_path,
out_dir=None,
config=None,
adjust_cleaner_config=(),
text=False,
markdown=False,
check_tables=False,
fast=False,
parsr_location='localhost:3001',
**kwargs)

Expand source code

def run_parsr(
    file_path,
    out_dir=None,
    config=None,
    adjust_cleaner_config=(),
    text=False,
    markdown=False,
    check_tables=False,
    fast=False,
    parsr_location="localhost:3001",
    **kwargs,
):
    """Send a PDF to Parsr and wait for the result.

    Returns Parsr's JSON and, if `check_tables`, the tables as CSV strings.
    `out_dir`: also write the results (and `text` / `markdown` if set) to this directory.
    """
    api = f"http://{parsr_location}/api/v1"
    parsr_config = setup_config(config or {}, adjust_cleaner_config, check_tables, fast)
    parsr_config["output"]["formats"]["text"] |= text
    parsr_config["output"]["formats"]["markdown"] |= markdown

    logger.info("sending PDF to Parsr")
    logger.debug(parsr_config)
    doc_id = _send_document(api, file_path, parsr_config)

    # 200 while processing, 201 when done
    while _request(f"{api}/queue/{doc_id}")[0] != 201:
        time.sleep(2)
    logger.info("got response from Parsr")

    data = json.loads(_request(f"{api}/json/{doc_id}")[1])
    tables = _get_tables(api, doc_id) if check_tables else None

    if out_dir is not None:
        out_dir = Path(out_dir) / Path(file_path).stem
        out_dir.mkdir(exist_ok=True, parents=True)

        if text:
            (out_dir / "text.txt").write_text(_request(f"{api}/text/{doc_id}")[1])

        if markdown:
            (out_dir / "text.md").write_text(_request(f"{api}/markdown/{doc_id}")[1])

        for idx, t in enumerate(tables or []):
            (out_dir / f"table_{idx}.csv").write_text(t)

        write_dict(data, out_dir / "data.json")

    return data, tables

Send a PDF to Parsr and wait for the result.

Returns Parsr's JSON and, if check_tables, the tables as CSV strings. out_dir: also write the results (and text / markdown if set) to this directory.

def setup_config(config, adjust_cleaner_config, check_tables, fast)

Expand source code

def setup_config(config, adjust_cleaner_config, check_tables, fast):
    """Merge pd3f's base config for Parsr with the user's adjustments"""
    # update base config of parsr
    cfg = importlib.resources.files("pd3f").joinpath("pd3fConfig.json")
    jdata = update_dict(json.loads(cfg.read_text()), config)

    # Update parsr cleaner config since it's more complicated.
    # The cleaner consists of a pipeline, so we first have to find the matching module.
    # Then update its configuration.
    for new_cl in adjust_cleaner_config:
        for idx, cl in enumerate(jdata["cleaner"]):
            if not isinstance(cl, list):
                continue
            if cl[0] != new_cl[0]:
                continue
            jdata["cleaner"][idx] = [cl[0], {**cl[1], **new_cl[1]}]

    if not check_tables:
        jdata["cleaner"] = [
            x
            for x in jdata["cleaner"]
            if isinstance(x, str) or "table-detection" not in x[0]
        ]

    if fast:
        jdata["cleaner"] = [
            x
            for x in jdata["cleaner"]
            if _module_name(x) not in ("drawing-detection", "image-detection")
        ]
    return jdata

Merge pd3f's base config for Parsr with the user's adjustments