| """Dependency-free inference for the published FiscalCheck classifier.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import math |
| import re |
| from collections import Counter |
| from pathlib import Path |
|
|
|
|
| TOKEN_RE = re.compile(r"[A-Za-zÀ-ÿ0-9_]+", re.UNICODE) |
|
|
|
|
| def tokens(text: str) -> list[str]: |
| words = [word.casefold() for word in TOKEN_RE.findall(text)] |
| word_features = words + [f"word:{words[i]}__{words[i + 1]}" for i in range(len(words) - 1)] |
| compact = " ".join(words) |
| lower = text.casefold() |
| signals = [] |
| if re.search(r"\b(?:bigint|integer|int|numeric|decimal|number|bigint)\b", lower) and "cnpj" in lower: |
| signals.append("signal:cnpj_numeric") |
| if re.search(r"(?:\\d\{14\}|\[0-9\]\{14\}|\^\\d\{14\}|onlydigits|only_digits)", lower): |
| signals.append("signal:cnpj_digits_only") |
| if "cnpj" in lower and re.search(r"(?:re\.sub|replace|preg_replace|regexp|replaceall)", lower) and re.search(r"(?:\\d|\[\^0-9\])", lower): |
| signals.append("signal:cnpj_normalization") |
| if "ibscbs" in lower: |
| signals.append("signal:ibscbs_group") |
| if not ("cst" in lower and "cclasstrib" in lower): |
| signals.append("signal:ibscbs_incomplete") |
| if "cclasstrib" in lower and not re.search(r"cclasstrib[^0-9]{0,20}\d{6}\b", lower): |
| signals.append("signal:bad_cclasstrib") |
| if re.search(r"(?:nfe|nf-e|nfc-e|nfce|nfs-e|nfse)", lower) and "ibscbs" not in lower: |
| signals.append("signal:fiscal_context_without_rtc") |
| if not signals: |
| signals.append("signal:compatible_or_other") |
| return word_features + signals |
|
|
|
|
| def predict(model: dict, text: str) -> str: |
| labels = model["labels"] |
| vocabulary = set(model["vocabulary"]) |
| features = tokens(text) |
| signal_labels = { |
| "signal:cnpj_numeric": "cnpj_numeric_storage", |
| "signal:cnpj_digits_only": "cnpj_digit_only_validation", |
| "signal:cnpj_normalization": "cnpj_destructive_normalization", |
| "signal:ibscbs_incomplete": "rtc_incomplete", |
| "signal:bad_cclasstrib": "rtc_suspect_class_code", |
| "signal:fiscal_context_without_rtc": "rtc_context_review", |
| "signal:compatible_or_other": "compatible", |
| } |
| direct = [signal_labels[token] for token in features if token in signal_labels] |
| if len(set(direct)) == 1: |
| return direct[0] |
| counts = Counter(features) |
| total_rows = sum(model["class_counts"].values()) |
| best_label, best_score = None, float("-inf") |
| for label in labels: |
| score = math.log(model["class_counts"][label] / total_rows) |
| denominator = model["token_totals"][label] + len(vocabulary) |
| label_counts = model["token_counts"][label] |
| for token, count in counts.items(): |
| score += count * math.log((label_counts.get(token, 0) + 1) / denominator) |
| if score > best_score: |
| best_label, best_score = label, score |
| assert best_label is not None |
| return best_label |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--model", type=Path, default=Path(__file__).with_name("model.json")) |
| parser.add_argument("--text", required=True) |
| args = parser.parse_args() |
| model = json.loads(args.model.read_text(encoding="utf-8")) |
| print(json.dumps({"label": predict(model, args.text)}, ensure_ascii=False)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
