Metadata-Version: 2.1
Name: corpus-preprocess
Version: 0.0.1
Summary: Building blocks for spacy Matcher patterns
Home-page: https://mv3.dev
Author: Marcelino G. Veloso III
Author-email: contact@mv3.dev
Requires-Python: >=3.11,<4.0
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Dist: inflect (>=7.0,<8.0)
Requires-Dist: rich (>=13.7.0,<14.0.0)
Requires-Dist: roman (>=4.1,<5.0)
Requires-Dist: spacy[apple] (>=3.7.2,<4.0.0)
Requires-Dist: sqlite-utils (>=3.36,<4.0)
Project-URL: Documentation, https://justmars.github.io/corpus-preprocess
Project-URL: Repository, https://github.com/justmars/corpus-preprocess
Description-Content-Type: text/markdown

# corpus-preprocess

![Github CI](https://github.com/justmars/corpus-preprocess/actions/workflows/main.yml/badge.svg)

Helps preprocess Philippine legal corpus.

> [!IMPORTANT]
> Relies on private [corpus-assets](ttps://github.com/justmars/corpus-assets) to be downloaded locally.

## Custom tokenizer

```py
from preprocess import set_tokenizer

nlp = spacy.blank("en")
nlp.tokenizer = set_tokenizer(nlp)
```

The tokenizer:

1. Removes dashes from infixes
2. Adds prefix/suffix rules for parenthesis/brackets
3. Adds special exceptions to treat dotted text as a single token

## SpanRuler from assets

Use in tandem with tokenizer, ensure only longest spans:

```py
from spacy.language import Language
from spacy.util import filter_spans
from preprocess import set_patterns_from_assets
import spacy

@spacy.registry.tokenizers("toktest")
def create_corpus_tokenizer():
    def create_tokenizer(nlp):
        return set_tokenizer(nlp)
    return create_tokenizer


@Language.component(name="filter_added_spans")
def filter_added_spans(doc):
    doc.spans["ruler"] = filter_spans(doc.spans["ruler"])
    return doc

nlp = spacy.blank("en", config={"nlp": {"tokenizer": {"@tokenizers": "toktest"}}})
ruler = nlp.add_pipe("span_ruler", config={"phrase_matcher_attr": "LOWER"}, validate=True) # defaults to 'ruler' key
patterns = set_patterns_from_assets(folder)
ruler.add_patterns(patterns)
nlp.add_pipe("filter_added_spans") # ensures only longest spans are included
nlp.to_disk("models/")  # will save entire directory which includes the pipeline
```

## Utils

1. `annotate_fragments()` - given an nlp object and some `*.txt` files, create a single annotation `*.jsonl` file
2. `extract_lines_from_txt_files()` - accepts an iterator of `*.txt` files and yields each line (after sorting the same and ensuring uniqueness of content).
3. `split_data()` - given a list of text strings, split the same into two groups and return a dictionary containing these groups based on the ratio provided (defaults to 0.80)

## Processes

### Asset folder

This library presumes the existence of a local `corpus-assets` folder having the following structure:

```yml
- ents:
  - casenames.txt # each line is a clean case
  - clean_statute_titles.txt # each line is a clean title
- concepts:
  - political: # main subject category
      - bill_of_rights: # sub-topic
          - patterns.json # contains matcher files
          - q.txt # contains lines which can be used to query the database
- metas:
  - artifacts:
    - axiom:
      - patterns.json # same
      - q.txt # same
```

### Generate queries

The `q.txt` lines will be used as criteria to fetch relevant segments from the database.

The db file should be have an "opinion_segments" table with fts-enabled on the "text" column. `/scripts/extract.py`
utilizes [table.search()](https://sqlite-utils.datasette.io/en/stable/python-api.html#searching-with-table-search).

See [code](./preprocess/asset_extractors.py)

```py
def extract_txt_from_db(
    source_db_file: str,
    path: Path,
    max_segments: int,
    min_char_segment: int = 100,
    max_char_segment: int = 3000,
    is_unique_txt: bool = True,
):
    """An fts expression is auto-generated by `q.txt` files found in the `path`. This
    expression is used to generate strings of text that match the aggregated query."""
    db = Database(source_db_file)
    tbl = db["opinion_segments"]
    rows = tbl.search(  # type: ignore
        q=create_fts_expr(path),
        where="category='ruling' and char_count > :min_char and char_count < :max_char ",
        where_args={"min_char": min_char_segment, "max_char": max_char_segment},
        limit=max_segments,
        columns=["text", "id"],
    )
    if is_unique_txt:
        rows = filter_unique_texts(rows)
    return rows
```

### Create matcher patterns

A [SpanRuler](https://spacy.io/api/spanruler) component will be based on `patterns.json` (with `q.txt` as phrases). These patterns are aggregated via `set_patterns_from_assets()`. See [code](./preprocess/main.py):

```py
def set_patterns_from_assets(path: Path):
    axioms = axiom.collect_patterns(path.joinpath("meta"))
    concepts = create_concept_patterns(path.joinpath("concepts"))
    ents = extract_ents(path.joinpath("ents"))
    return axioms + concepts + ents
```

### Categorize queried segments via patterns found

A [TextCategorizer](https://spacy.io/api/textcategorizer) component can be trained using the results of the span ruler: see sample code:

```py
@Language.factory(name="add_cats_from_spans")
class AddTextCatComponent:
    def __init__(self, nlp: Language, name: str, path: str):
        self.nlp = nlp
        options = list({p["id"].split("/")[0] for p in create_patterns(path)})  # type: ignore
        if len(options) == 1:
            options.append(f"not_{options[0]}")
        self.options = options

    def __call__(self, doc) -> Doc:
        default = {op: 0.0 for op in self.options}
        cats = [self.nlp.vocab.strings[s.id].split("/")[0] for s in doc.spans["sc"]]
        doc.cats = default | {k: 1.0 for k, _ in Counter(cats).items()}
        return doc
```

> [!NOTE]
> Note: if textcat is in the pipeline, if only one label is found, will error out, hence need to a _not_ option. If textcat_multilabel is used, then a single category is fine.

