Extraction#

Convert the mirror into sharded JSONL corpora.

Corpus extraction#

litsync.extract.clean(s)[source]#

Parameters:: s (str | None)
Return type:: str

litsync.extract.local_tag(tag)[source]#

Parameters:: tag (str)
Return type:: str

litsync.extract.text(el)[source]#

Parameters:: el (Element | None)
Return type:: str

litsync.extract.find_local(parent, name)[source]#

Parameters:

parent (Element)
name (str)

Return type:

Element | None

litsync.extract.findall_local(parent, name)[source]#

Parameters:

parent (Element)
name (str)

Return type:

list[Element]

litsync.extract.pubmed_year(article)[source]#

Parameters:: article (Element)
Return type:: int | None

litsync.extract.parse_pubmed(raw, source_file)[source]#

Parameters:

raw (bytes)
source_file (str)

Return type:

Iterator[dict]

litsync.extract.parse_pmc_article(raw, source_file)[source]#

Parameters:

raw (bytes)
source_file (str)

Return type:

dict | None

litsync.extract.parse_pmc_tar(path, rel)[source]#

Parameters:

path (Path)
rel (str)

Return type:

Iterator[dict]

litsync.extract.parse_fda_json_file(path, rel)[source]#

Parameters:

path (Path)
rel (str)

Return type:

Iterator[dict]

litsync.extract.parse_clinicaltrials_xml_file(path, rel)[source]#

Parameters:

path (Path)
rel (str)

Return type:

dict | None

class litsync.extract.ShardWriter[source]#

Bases: object

__init__(out_dir, shard_bytes, prefix='corpus', start_idx=0)[source]#

Parameters:

out_dir (Path)
shard_bytes (int)
prefix (str)
start_idx (int)

property current_shard: int#

write(rec)[source]#

Parameters:: rec (dict)

close()[source]#

roll()[source]#: Close the current shard so the next write starts a fresh shard.

litsync.extract.iter_source_files(data_root, sources, limit)[source]#

Parameters:

data_root (Path)
sources (list[str])
limit (int | None)

litsync.extract.run_extraction(data_root, out_dir, sources, shard_size_mb, limit, ui, resume=True, reset=False, yearly=False, require_abstract=False, require_body=False, require_text=False)[source]#

Parameters:

data_root (Path)
out_dir (Path)
sources (list[str])
shard_size_mb (int)
limit (int | None)
ui (UI)
resume (bool)
reset (bool)
yearly (bool)
require_abstract (bool)
require_body (bool)
require_text (bool)

Return type:

dict

Resume state#

class litsync.extract_state.ExtractState[source]#

Bases: object

SQLite-backed resume state for litsync-extract.

Tracks which source files have already been extracted so re-runs can skip them. For yearly corpora it also tracks the last shard index written per year, so re-runs can continue without overwriting completed shards.

If the output directory, source list, or shard size changes, the stored state is reset automatically.

__init__(state_path, out_dir, sources, shard_size_mb, reset=False)[source]#

Parameters:

state_path (Path)
out_dir (Path)
sources (list[str])
shard_size_mb (int)
reset (bool)

is_done(source_file)[source]#

Parameters:: source_file (str)
Return type:: bool

mark_done(source_file, records, year_shards)[source]#

Parameters:

source_file (str)
records (int)
year_shards (dict[str, int])

Return type:

None

done_count()[source]#

Return type:: int

max_end_shard(year='')[source]#

Highest completed shard index for a given year (empty = non-yearly).

Parameters:: year (str)
Return type:: int

year_shards_for_file(source_file)[source]#

Parameters:: source_file (str)
Return type:: dict[str, int]

close()[source]#

Return type:: None