File size: 3,388 Bytes
6cc0061 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | """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()
|