Mapear e documentar o fluxo dos dados

[naanadr]

Atualmente temos uma documentação mais geral sobre a arquitetura do Querido Diário, mas eu estava pensando em termos um mapeamento focado nesse repositório. Porque com esse mapeamento, talvez fique mais fácil visualizar o que é realizado aqui, quais as dependências e outputs gerados, para depois começarmos a focar no que esperamos da prova de conceito.

O entregável dessa atividade pode ser um fluxograma feito no draw.io mesmo.

[naanadr]

Fiz esse fluxograma olhando brevemente para o código

Finalmente entendi que o que é realizado pelo script 😄

@ogecece @jvanz a ideia dele é atualizar os index do elasticsearch, lendo todos os diários que ainda não foram processados procurando por termos que foram definidos no themes_config.json ?

[ogecece]

Massa @naanadr !

Daria pra complementar ainda com a parte do extract_text. O fluxo principal vai até essa tarefa. Nela os txts são coletados e salvos, as entradas no banco postgres são atualizadas como "processadas" e as novas entradas são adicionadas ao índice principal.

A partir das novas entradas no índice principal que temos o processamento temático.

Fiz um esboço aqui:

flowchart TD
    subgraph main ["(Script Python) __main__"]
        subgraph list_gazettes ["(Tarefa Python) list_gazettes_to_be_processed"]
                execution_mode_fork{"Valor de EXECUTION_MODE"}

                daily_process[["(Script Postgres)\n....\nWHERE\n‏‏‎ ‎‏‏‎ ‎‏‏‎ ‎‏‏‎ ‎scraped_at > current_timestamp - interval '1 day'"]]
                style daily_process text-align:left

                all_process[["(Script Postgres)\n....\nFROM\n‏‏‎ ‎‏‏‎ ‎‏‏‎ ‎‏‏‎ gazettes\nINNER JOIN territories ON territories.id = gazettes.territory_id"]]
                style all_process text-align:left

                unprocessed_process[["(Script Postgres)\n....\nWHERE\n‏‏‎ ‎‏‏‎ ‎‏‏‎ ‎‏‏‎ ‎processed is False"]]
                style unprocessed_process text-align:left

                list_gazettes_input(("input"))
                list_gazettes_output((("output")))

                list_gazettes_input --> execution_mode_fork
                execution_mode_fork -- DAILY --> daily_process
                execution_mode_fork -- ALL --> all_process
                execution_mode_fork -- UNPROCESSED --> unprocessed_process
                daily_process & all_process & unprocessed_process --> list_gazettes_output
        end

        subgraph extract_text ["(Tarefa Python) extract_text_from_gazettes"]
                subgraph new_gazette ["(Subrotina Python) try_process_gazette_file"]
                        gazette_id_artifact[/"(ID)\nIdentificador da publicação no banco de dados fonte"/]
                        text_file_artifact[/"(Arquivo TXT)\nPublicação em texto puro"/]    

                        upload_text_file_process[["(Subrotina Python)\nupload_gazette_raw_text\n[Sobe arquivo de texto para o sistema de arquivos]"]]
                        update_database_process[["(Subrotina Python)\nset_gazette_as_processed\n[Atualiza entrada no banco de dados fonte com 'processed=True']"]]
                        add_search_engine_record_process[["(Subrotina Python)\nindex_document\n[Adiciona entrada no índice principal do motor de busca]"]]

                        crawl_database_inner[("(Postgres)\nBanco de dados de raspagens")]
                        search_engine_main_index_database_inner[("(Elasticsearch)\nÍndice principal do motor de busca")]
                        filesystem_database_inner[("(DO Spaces)\nSistema de arquivos")]

                        new_gazette_input(("input"))
                        new_gazette_output(("output"))

                        new_gazette_input --> gazette_id_artifact & text_file_artifact & update_database_process & add_search_engine_record_process
                        text_file_artifact --> upload_text_file_process
                        upload_text_file_process --> filesystem_database_inner
                        add_search_engine_record_process --> search_engine_main_index_database_inner
                        update_database_process --> crawl_database_inner
                        gazette_id_artifact --> new_gazette_output
                end

                can_process_fork{"Foi possível processar a publicação?"}

                extract_text_input(("input"))
                extract_text_output((("output")))

                extract_text_input --> new_gazette --> can_process_fork
                can_process_fork -- Não --> discard_gazette["Descarta id da publicação"]
                can_process_fork -- Sim --> extract_text_output
        end

        subgraph process_excerpts ["(Subrotina Python) Para cada tema, para cada publicação, executar:"]
                extract_themed_excerpts_process[["(Tarefa Python)\nextract_themed_excerpts_from_gazettes\n[Busca trechos do conteúdo da publicação que tem relação com o tema e os salva no índice de busca do tema]"]]
                embedding_rerank_excerpts_process[["(Tarefa Python)\nembedding_rerank_excerpts\n[<anipula pontuação das entradas de acordo com comparação semântica com subtemas]"]]
                tag_entities_in_excerpts_process[["(Tarefa Python)\ntag_entities_in_excerpts\n[Etiqueta CNPJs e outras entidades relacionadas ao tema nos excertos]"]]
    
                process_excerpts_input(("input"))
    
                process_excerpts_input --> extract_themed_excerpts_process
                extract_themed_excerpts_process --> embedding_rerank_excerpts_process & search_engine_thematic_index_database
                embedding_rerank_excerpts_process --> tag_entities_in_excerpts_process & search_engine_thematic_index_database
                tag_entities_in_excerpts_process --> search_engine_thematic_index_database
                search_engine_thematic_index_database --> embedding_rerank_excerpts_process & tag_entities_in_excerpts_process
        end

        crawl_database[("(Postgres)\nBanco de dados de raspagens")]
        search_engine_main_index_database[("(Elasticsearch)\nÍndice principal do motor de busca")]
        search_engine_thematic_index_database[("(Elasticsearch)\nÍndices temáticos do motor de busca")]
        filesystem_database[("(DO Spaces)\nSistema de arquivos")]

        themes_artifact[/"(json)\nConfiguração de filtros temáticos"/]
        to_be_processed_artifact[/"(dict)\nPublicações a ser processadas"/]
        indexed_gazette_ids_artifact[/"(list)\nIds das publicações que foram processadas e estão no índice principal do motor de busca"/]

        crawl_database --> list_gazettes
        list_gazettes --> to_be_processed_artifact
        to_be_processed_artifact & filesystem_database --> extract_text
        extract_text --> indexed_gazette_ids_artifact
        indexed_gazette_ids_artifact & themes_artifact & search_engine_main_index_database --> process_excerpts
    end
    execution_mode{{"(Variável de ambiente)\nEXECUTION_MODE\n[DAILY | ALL | UNPROCESSED]"}}

    execution_mode --> main

Loading