File size: 7,881 Bytes
58bde27
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
import gradio as gr

# import spaces

from typing import Dict, List, Tuple
from pydantic import HttpUrl

from task_management import TaskManager
from config import LANGUAGES


# Gradio interface
# @spaces.GPU
def process_rss(
    rss_url: HttpUrl,
    source_lang: str,
    target_lang: str,
    entries_limit: int = None,
) -> List[Dict]:
    """The wrapper to the respective task management function to retrieve the
    summarized and translated entries from the feed

    Args:
        rss_url (HttpUrl): the url
        src_lang (str): _description_
        tgt_lang (str): _description_
        entries_limit (int, optional): _description_. Defaults to None.

    Raises:
        gr.Error: _description_

    Returns:
        List[Dict]: _description_
    """
    try:
        tm = TaskManager()
        processed_entries = tm.parse_and_process_feed(
            rss_url, source_lang, target_lang, entries_limit
        )
    except Exception as e:
        raise gr.Error(e)

    return processed_entries, len(processed_entries)


# Custom css
custom_css = """
#messOut textarea {
    font-weight: bold;
}

#entriesTab {
    background-color: white;
}
"""

# Create a scrollable Markdown component
with gr.Blocks(
    theme=gr.themes.Soft(),
    css=custom_css,
) as demo:
    # Add a title using Markdown
    gr.Markdown("# RSS Feed Summarizer and Translator")

    # Add a description using Markdown
    gr.Markdown(
        "Input an RSS feed URL and specify the source and target languages to get summarized and translated content."
    )

    rss_entries = gr.State([])

    with gr.Row():
        # Step for starting points and options' steps for entries' dropdowns (retrieve and view)
        step = 5

        with gr.Column():
            rss_url = gr.Textbox(label="RSS Feed URL")

            languages_lst = LANGUAGES.keys()

            source_lang = gr.Dropdown(
                choices=languages_lst,
                value="",
                label="Source Language",
            )
            target_lang = gr.Dropdown(
                choices=languages_lst,
                value="",
                label="Target Language",
            )

            options_lst = list(range(5, 205, 5))
            entries_to_retrieve = gr.Dropdown(
                choices=options_lst,
                value=options_lst[0],
                label="Max Entries To Retrieve",
            )

            with gr.Row():
                clear_btn = gr.ClearButton(value="Clear")  # Clear button
                submit_btn = gr.Button("Submit", variant="primary")

        with gr.Column():
            # Message for feed entries retrieved and spinner purposes
            message_output = gr.Textbox(
                label="Entries Retrieved: ",
                interactive=False,
                elem_id="messOut",
            )

            def submit_request(
                feed_url: HttpUrl,
                src_lang: str,
                tgt_lang: str,
                entries_limit: int,
                latest_entries_num: int,
            ) -> Tuple[List[Dict], int, str]:
                """Calls format_processed_entries and format_processed_entries,
                everytime submit button is pressed in order to retrieve feed entries,
                format them and show them in the respective output component

                Args:
                    feed_url (HttpUrl): the feed url
                    src_lang (str): source language
                    tgt_lang (str): target_language
                    entries_limit (int): the entries' limit (to retrieve)
                    latest_entries_num (int): the number of the latest entries retrieved (if submission button has been pressed before)

                Returns:
                    Tuple[List[Dict], int, str]: the feed entries retrieved, the number of those entries, the entries properly formatted
                """

                proc_entries, entries_num = process_rss(
                    feed_url, src_lang, tgt_lang, entries_limit
                )
                # entries_updated = update_entries(latest_entries_num)
                formatted_updated_entries = format_processed_entries(proc_entries)
                return proc_entries, entries_num, formatted_updated_entries

            with gr.Tab("Feed Summaries:", visible=True, elem_id="entriesTab"):
                # Create a scrollable Markdown component
                markdown_output = gr.Markdown(height="400px")

                entries_to_view = gr.Dropdown(
                    choices=[options_lst[0]],
                    value=options_lst[0],
                    label="Max Entries To View",
                )

                @gr.on(
                    [entries_to_view.change],
                    inputs=[
                        rss_entries,
                        entries_to_view,
                    ],
                    outputs=[markdown_output],
                )
                def format_processed_entries(
                    processed_entries: List[Dict], entries_limit: int = None
                ) -> str:
                    """Format the output entries

                    Args:
                        processed_entries (List[Dict]): the entries retrieved from the feed that have been processed
                        entries_limit (int): a limit for the entries to view

                    Returns:
                        str: the formatted output containing the entries
                    """
                    entries_limit = entries_limit or len(processed_entries) or None

                    # Format the output for Gradio
                    output = ""
                    for entry in processed_entries[:entries_limit]:
                        output += f"### {entry.get('title', '---')}\n\n"
                        output += f"**Author:** {entry.get('author', '-')}\n\n"
                        output += f"{entry.get('content', '')}\n\n"
                        link = entry.get("link", "")
                        if link:
                            output += f"[Read more]({link})\n\n"
                        output += "---\n\n"

                    return output

                # Function to handle dropdown options for viewing entries
                @gr.on(
                    [rss_entries.change],
                    inputs=[rss_entries],
                    outputs=[entries_to_view],
                )
                def update_view_dropdown(view_entries: List[Dict]) -> gr.Dropdown:
                    """Update the options for view dropdown

                    Args:
                        view_entries (List[Dict]): the view entries list

                    Returns:
                        gr.Dropdown: a dropdown component with the updated options regarding view entries
                    """
                    max_entries_shown = len(view_entries) or None

                    # Update the dropdown options with the new length
                    dropdown_options = list(range(step, max_entries_shown + step, step))

                    # Return outputs to update components
                    return gr.Dropdown(
                        choices=dropdown_options,
                        value=entries_to_view.value,
                        label="Entries to view",
                    )

    # Link the function to the button
    submit_btn.click(
        submit_request,
        inputs=[rss_url, source_lang, target_lang, entries_to_retrieve, message_output],
        outputs=[rss_entries, message_output, markdown_output],
    )

    # Link the Clear button to reset inputs and outputs
    clear_btn.add(
        components=[
            rss_url,
            source_lang,
            target_lang,
            markdown_output,
            entries_to_view,
            entries_to_retrieve,
        ]
    )

# Launch the interface
demo.launch()