add python code

app.py

@@ -1,7 +1,9 @@
 import ast
 import glob
+import time
 from itertools import islice
 from functools import partial
+from textwrap import dedent
 from typing import Optional, Type
 
 import gradio as gr
@@ -26,6 +28,7 @@ from datatrove.utils.typeshelper import Languages
 
 nltk.download('punkt_tab')
 DUMP_TO_PROCESS = "CC-MAIN-2023-50"
+TIMEOUT = 600
 
 make_gallery_image_buttons_js = """
 function load() {
@@ -281,9 +284,7 @@ with gr.Blocks(css=css, js=make_gallery_image_buttons_js) as demo:
             gopher_filtering_quality_checkbox.change(lambda visible: gr.Accordion(visible=visible), inputs=gopher_filtering_quality_checkbox, outputs=acc)
         gopher_filtering_quality_parameters_components = [language_dropdown2, min_doc_words_slider, max_doc_words_slider, min_avg_word_length_slider, max_avg_word_length_slider, max_symbol_word_ratio_slider, max_bullet_lines_ratio_slider, max_ellipsis_lines_ratio_slider, max_non_alpha_words_ratio_slider, min_stop_words_slider, stop_words_textbox]
 
-    with gr.Row():
-        view_pipeline_results_button = gr.Button("Run Pipeline & Stream Results", variant="primary", scale=4)
-        stop_button = gr.Button("Stop")
+    view_pipeline_results_button = gr.Button("Run Pipeline & Stream Results", variant="primary", scale=4)
 
     steps = [
         URLFilter,
@@ -340,7 +341,6 @@ with gr.Blocks(css=css, js=make_gallery_image_buttons_js) as demo:
         pii_removal_checkbox
     ] + sum(steps_parameters_components, [])
 
-    @view_pipeline_results_button.click(inputs=inputs, outputs=[output_tab, output_dataframe, excluded_tab] + list(excluded_dataframes.values()) + list(excluded_tabs.values()))
     def view_pipeline_results(*args):
         enable_steps, steps_parameters = args[:len(steps)], args[len(steps):]
         steps_parameters_iter = iter(steps_parameters)
@@ -358,6 +358,43 @@ with gr.Blocks(css=css, js=make_gallery_image_buttons_js) as demo:
             }
             for step_parameters_components in steps_parameters_components
         ]
+        yield {
+            python_code_markdown: dedent(
+                """
+                ```python
+                from datatrove.executor.local import LocalPipelineExecutor
+                from datatrove.pipeline.extractors import Trafilatura
+                from datatrove.pipeline.filters import (
+                    C4QualityFilter,
+                    FineWebQualityFilter,
+                    GopherQualityFilter,
+                    GopherRepetitionFilter,
+                    LanguageFilter,
+                    URLFilter,
+                )
+                from datatrove.pipeline.formatters import PIIFormatter
+                from datatrove.pipeline.readers import WarcReader
+                """
+            ).strip() + (
+                "\n\n"
+                "pipeline_executor = LocalPipelineExecutor(\n"
+                "    pipeline=[\n"
+                f'        WarcReader("s3://commoncrawl/crawl-data/{DUMP_TO_PROCESS}/segments", glob_pattern="*/warc/*"),\n'
+            ) + ",\n".join([
+                "        " + step.__name__ + "(" + ", ".join(arg + "=" + str(value) for arg, value in step_parameters.items() if value != default_step_parameters[arg] and arg != "exclusion_writer") + ")"
+                for step, step_parameters, default_step_parameters, enable_step in zip(steps, steps_parameters, default_steps_parameters, enable_steps)
+                if enable_step
+            ]) + (
+                "\n"
+                "    ]\n"
+                ")"
+            ) + dedent(
+                """
+                pipeline_executor.run()
+                ```
+                """
+            )
+        }
 
         class ExclusionWriter:
 
@@ -380,19 +417,28 @@ with gr.Blocks(css=css, js=make_gallery_image_buttons_js) as demo:
         ]
         output_docs: list[Document] = []
         num_warc_samples = 0
+        timeout_time = time.time() + TIMEOUT
 
         def increment_num_warc_samples(data, rank, world_size, num_warc_samples_per_doc=1):
             nonlocal num_warc_samples
             for x in data:
                 num_warc_samples += num_warc_samples_per_doc
                 yield x
+        
+        def check_timeout(data, rank, world_size):
+            for x in data:
+                if time.time() > timeout_time:
+                    gr.Info("Pipeline timed out")
+                    break
+                yield x
 
         if steps_parameters[:2] == default_steps_parameters[:2] and all(enable_steps[:2]):
             
             pipeline_executor = LocalPipelineExecutor(
                 pipeline=[
                     JsonlReader(data_folder=f"output_text_extraction-full/base_processing/output/{DUMP_TO_PROCESS}", glob_pattern="*.jsonl.gz"),
-                    partial(increment_num_warc_samples, num_warc_samples_per_doc=2000 / 1687)
+                    partial(increment_num_warc_samples, num_warc_samples_per_doc=2000 / 1687),
+                    check_timeout
                 ] + steps_to_run[2:] + [
                     lambda data, rank, world_size: islice(data, 100),
                     lambda data, rank, world_size: map(output_docs.append, data)
@@ -404,7 +450,8 @@ with gr.Blocks(css=css, js=make_gallery_image_buttons_js) as demo:
             pipeline_executor = LocalPipelineExecutor(
                 pipeline=[
                     WarcReader(data_folder="data", glob_pattern="*.warc.gz"),
-                    increment_num_warc_samples
+                    increment_num_warc_samples,
+                    check_timeout
                 ] + steps_to_run + [
                     lambda data, rank, world_size: islice(data, 100),
                     lambda data, rank, world_size: map(output_docs.append, data)
@@ -465,7 +512,7 @@ with gr.Blocks(css=css, js=make_gallery_image_buttons_js) as demo:
             },
         }
     
-    stop_button.click(cancels=[view_pipeline_results])
+    view_pipeline_results_button.click(view_pipeline_results, inputs=inputs, outputs=[output_tab, output_dataframe, excluded_tab, python_code_markdown] + list(excluded_dataframes.values()) + list(excluded_tabs.values()))
 
 if __name__ == "__main__":
     demo.launch()