extracao_diarios_regex.py

from datetime import date
import glob
import math
import pandas as pd
import re
import sys
import gc

from joblib import Parallel, delayed

# # CORONAVÍRUS
# REGULAR_EXPRESSIONS_TRUE = [
#     r"covid|coronav|corona v|pandemia|62.{1,5}CNJ",
#     r"habeas corpus",
# ]
# REGULAR_EXPRESSIONS_FALSE = []

# # TRIBUTÁRIO
# REGULAR_EXPRESSIONS_TRUE = [r"tributo|tributári[oa]"]
# REGULAR_EXPRESSIONS_FALSE = []

# # PENAL
# REGULAR_EXPRESSIONS_TRUE = [r"penal|crime|habeas corpus", r"condeno|absolvo|concedo|denego"]
# REGULAR_EXPRESSIONS_FALSE = [
#     r"mandado.{,5}segurança",
#     r"\-\s*Inquérito Policial\s*\-",
#     r"Execução fiscal",
#     r"Embargos à Execução",
#     r"\-\s*Cumprimento de sentença\s*\-",
#     r"\-\s*Procedimento comum\s*\-",
#     r"\-\s*Termo circunstanciado\s*\-",
#     r"petições para juntada",
#     r"juizado especial cível",
#     r"alienação fiduciária",
#     r"\-\s*Boletim de Ocorrência",
#     r"cláusula penal",
#     r"despejo",
#     r"sob pena de crime de desobediência",
# ]

# # MÉDIA DE MERCADO
# REGULAR_EXPRESSIONS_TRUE = [
#     r"m.dia d[eo] mercado", r"aliena..o fiduci.ria"
# ]
# REGULAR_EXPRESSIONS_FALSE = []

# # COMPENSAÇÃO CRUZADA
# REGULAR_EXPRESSIONS_TRUE = [
#     r"11\.*457\/[20]*07|13\.*670\/[20]*18|compensação cruzada|apuração anterior.{,20}(utilização|adesão).{,20}e\-*social"
# ]
# REGULAR_EXPRESSIONS_FALSE = []

# # TRANSAÇÃO TRIBUTÁRIA
# REGULAR_EXPRESSIONS_TRUE = [
#     r"transação individual|transação tributária|transação excepcional|transação extraordinária|edital de transação|transação por adesão|lei n?.? 13\.?988|mp do contribuinte legal",
# ]
# REGULAR_EXPRESSIONS_FALSE = []

# # PENAL MP
# REGULAR_EXPRESSIONS_TRUE = [
#     r"penal|crime",
#     r"habeas corpus|roubo|corrupção de menores|restritiva de direitos|execução penal|extorsão|furto qualificado|princípio da insignificância|tráfico privilegiado|tr[áa]fico de drogas"
# ]
# REGULAR_EXPRESSIONS_FALSE = [
#     r"mandado.{,5}segurança",
#     r"\-\s*Inquérito Policial\s*\-",
#     r"Execução fiscal",
#     r"Embargos à Execução",
#     r"\-\s*Cumprimento de sentença\s*\-",
#     r"\-\s*Procedimento comum\s*\-",
#     r"\-\s*Termo circunstanciado\s*\-",
#     r"petições para juntada",
#     r"juizado especial cível",
#     r"alienação fiduciária",
#     r"\-\s*Boletim de Ocorrência",
#     r"cláusula penal",
#     r"despejo",
#     r"sob pena de crime de desobediência",
# ]


# # DANO MORAL PJ
# REGULAR_EXPRESSIONS_TRUE = [
#     r"honra objetiva|dano moral.{,50}(pessoa jurídica|honra objetiva)|(pessoa jurídica|honra objetiva).{,50}dano moral|súmula 227.{,5}(STJ|Superior tribunal de justi[çc]a)",
#     r"pessoa.{,5}jur.dica|[\s\.\,]pj[\s\.\,]"
# ]
# REGULAR_EXPRESSIONS_FALSE = []

# # ANATEL
# REGULAR_EXPRESSIONS_TRUE = [
#     r"[\s\,\.]anatel[\s\,\.]"
# ]
# REGULAR_EXPRESSIONS_FALSE = []

# CADE
REGULAR_EXPRESSIONS_TRUE = [
    r"[\s\,\.]cade[\s\,\.]"
]
REGULAR_EXPRESSIONS_FALSE = []

def consertar_ncnj(nCnj):
    if len(nCnj) >= 20:
        return nCnj
    elif len(nCnj) < 20:
        return consertar_ncnj("0" + nCnj)
    else:
        return consertar_ncnj(nCnj[:-1])


def pesquisa_diario_from_csv(
    df_path,
    rows,
    expressions_true=REGULAR_EXPRESSIONS_TRUE,
    expressions_false=REGULAR_EXPRESSIONS_FALSE,
    column_text="texto_publicacao",
    column_id="numero_processo",
):
    df = pd.read_csv(
        df_path,
        chunksize=100,
        usecols=[column_text, column_id, "tribunal", "data"],
        engine="python",
    )
    for chunk in df:
        for _, row in chunk.iterrows():
            aux = True
            for expression in expressions_true:
                if not re.search(expression, row[column_text], flags=re.I | re.DOTALL):
                    aux = False
                    break
            if aux:
                for expression in expressions_false:
                    if re.search(expression, row[column_text], flags=re.I | re.DOTALL):
                        aux = False
                        break
            if aux:
                rows.append(row)


def pesquisa_diario_from_parquet(
    df_path,
    rows,
    expressions_true=REGULAR_EXPRESSIONS_TRUE,
    expressions_false=REGULAR_EXPRESSIONS_FALSE,
    column_text="texto_publicacao",
    just_numbers=False,
    lowerBound=0,
    upperBound=1,
):
    df = pd.read_parquet(
        df_path,
        engine="pyarrow",
    )
    for row in df.to_dict("records"):
        aux = True
        for expression in expressions_true:
            if not re.search(
                expression,
                row[column_text][
                    math.ceil(lowerBound * len(row[column_text])) : math.floor(
                        upperBound * len(row[column_text])
                    )
                ],
                flags=re.I | re.DOTALL,
            ):
                aux = False
                break
        if aux:
            for expression in expressions_false:
                if re.search(
                    expression,
                    row[column_text][
                        math.ceil(lowerBound * len(row[column_text])) : math.floor(
                            upperBound * len(row[column_text])
                        )
                    ],
                    flags=re.I | re.DOTALL,
                ):
                    aux = False
                    break
        if aux:
            if not just_numbers:
                dic_aux = row.copy()
                for k in dic_aux:
                    dic_aux[k] = str(dic_aux[k])
                rows.append(dic_aux)
            else:
                rows.append(
                    {
                        "numero_processo": consertar_ncnj(
                            str(row["numero_processo"])
                            .replace(".", "")
                            .replace("/", "")
                            .replace("-", "")
                        )
                    }
                )


if __name__ == "__main__":
    # ini_path = str(sys.argv[1])
    # final_path = str(sys.argv[2])
    ini_path = "Z:/data/raw/Diarios/Diarios"
    final_path = "D:/pesquisas_textos_juridicos/finding_arguments_CADE/data/raw"
    # paths = glob.glob(ini_path + "/**/*.parquet.gzip", recursive=True)
    paths = [i for i in glob.glob(ini_path + "/**/*.parquet.gzip", recursive=True) if re.search(r"stj|stf|trf", i)]
    print("Quantidade de arquivos a serem processados ",len(paths))
    rows = []
    counter = 0
    counter_aux = len(paths)
    batch_size = 6
    rows = []
    counter_aux = len(paths)
    for index in range(0, len(paths), batch_size):
        print("Faltam {:.2f}%".format((counter_aux / (len(paths)) * 100)))
        counter_aux -= batch_size
        if index + batch_size > len(paths):
            batch = paths[index:]
        else:
            batch = [paths[i] for i in range(index, index + batch_size)]
        # Parallel(n_jobs=-1, backend="threading", require="sharedmem")(
        #     delayed(pesquisa_diario_from_parquet)(diario, rows) for diario in batch
        # )
        for p in batch:
            pesquisa_diario_from_parquet(p, rows)
            if len(rows) > 25000:
                df_new = pd.DataFrame(rows)
                df_new.to_parquet(
                    "{}/extracao_{}.parquet.gzip".format(final_path, str(counter)),
                    engine="pyarrow",
                    compression="gzip",
                    index=False,
                )
                counter += 1
                rows = []
    if len(rows):
        df_new = pd.DataFrame(rows)
        df_new.to_parquet(
            "{}/extracao_{}.parquet.gzip".format(final_path, str(counter)),
            engine="pyarrow",
            compression="gzip",
            index=False,
        )