import base64 import io import os import pandas as pd from docx import Document from io import BytesIO, StringIO import dash import dash_bootstrap_components as dbc from dash import html, dcc, Input, Output, State, callback_context, MATCH, ALL from dash.dash_table import DataTable from docx.shared import Pt from docx.enum.style import WD_STYLE_TYPE from PyPDF2 import PdfReader import openai import logging import threading import re import markdown from bs4 import BeautifulSoup logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') app = dash.Dash(__name__, external_stylesheets=[dbc.themes.BOOTSTRAP], suppress_callback_exceptions=True) openai.api_key = os.environ.get("OPENAI_API_KEY", "") uploaded_files = {} current_document = None document_type = None shredded_document = None pink_review_document = None uploaded_doc_contents = {} spreadsheet_types = ["Shred", "Pink Review", "Red Review", "Gold Review", "Virtual Board", "LOE"] narrative_types = ["Pink", "Red", "Gold"] document_types = { "Shred": "Ignore all other instructions and generate only requirements spreadsheet of the Project Work Statement (PWS) identified by action words like shall, will, perform etc. by pws section, requirement. Do not write as if you're responding to the proposal. Its a spreadsheet to distill the requirements, not microhealth's approach", "Pink": "Create a Pink Team document based on the PWS outline. Your goal is to be compliant and compelling. Focus on describing the approach and how it will be done, the steps, workflow, people, processes and technology based on well known industry standards to accomplish the task. Be sure to demonstrate innovation.", "Pink Review": "Ignore all other instructions and generate and evaluate compliance of the Pink Team document against the requirements and output only a spreadsheet of non compliant findings by pws number, the goal of that pws section, what made it non compliant and your recommendations for recovery. you must also take into account section L&M of the document which is the evaluation criteria to be sure we address them.", "Red": "Produce a Red Team document based on the Pink Review by pws sections. Your goal is to be compliant and compelling by recovering all the findings in Pink Review. Focus on describing the approach and how it will be done, the steps, workflow, people, processes and technology to accomplish the task. Be sure to refer to research that validates the approach and cite sources with measurable outcomes", "Red Review": "Ignore all other instructions and generate and evaluate compliance of the Red Team document against the requirements and output a only a spreadsheet of non compliant findings by pws number, the goal of that pws section, what made it non compliant and your recommendations for recovery. you must also take into account section L&M of the document which is the evaluation criteria to be sure we address them", "Gold": "Create a Gold Team document based on the PWS response by pws sections. Your goal is to be compliant and compelling by recovering all the findings in Red Review. Focus on describing the approach and how it will be done, the steps, workflow, people, processes and technology to accomplish the task. Be sure to refer to research that validates the approach and cite sources with measurable outcomes and improve on innovations of the approach", "Gold Review": "Ignore all other instructions and generate and perform a final compliance review against the requirements and output only a spreadsheet of non compliant findings by pws number, the goal of that pws section, what made it non compliant and your recommendations for recovery. you must also take into account section L&M of the document which is the evaluation criteria to be sure we address them", "Virtual Board": "Ignore all other instructions and generate and based on the requirements and in particular the evaulation criteria, you will evaluate the proposal as if you were a contracting office and provide section by section evaluation as unsatisfactory, satisfactory, good, very good, excellent and why and produce only spreadsheet", "LOE": "Ignore all other instructions and generate and generate a Level of Effort (LOE) breakdown and produce only spreadsheet" } def get_right_col_content(selected_type): controls = [] controls.append(html.Div([ html.Div(className="blinking-dot", style={'margin':'0 auto','width':'16px','height':'16px'}), ], style={'textAlign':'center', 'marginBottom':'10px'})) controls.append(dcc.Loading( id="loading-indicator", type="dot", children=[html.Div(id="loading-output")] )) if selected_type == "Shred": controls.append( html.Div([ dcc.Upload( id='upload-document', children=html.Div([ 'Drag and Drop or ', html.A('Select Files') ]), style={ 'width': '100%', 'height': '60px', 'lineHeight': '60px', 'borderWidth': '1px', 'borderStyle': 'dashed', 'borderRadius': '5px', 'textAlign': 'center', 'margin': '10px 0' }, multiple=True ), file_list_component() ]) ) controls.append( dbc.Row([ dbc.Col( dbc.Button("Generate Document", id={'type': 'btn-generate-doc', 'index': selected_type}, color="primary", className="mb-3 w-100"), width=6 ), dbc.Col( dbc.Button("Download Document", id="btn-download", color="success", className="mb-3 w-100"), width=6 ), dcc.Download(id="download-document") ], className="mb-2") ) controls.append(html.Hr()) controls.append(html.Div(id='document-preview', className="border p-3 mb-3")) if selected_type != "Shred": controls.append(html.Div([ html.Label(f"Upload {selected_type} Document"), dcc.Upload( id={'type': 'upload-doc-type', 'index': selected_type}, children=html.Div(['Drag and Drop or ', html.A('Select File')]), style={ 'width': '100%', 'height': '60px', 'lineHeight': '60px', 'borderWidth': '1px', 'borderStyle': 'dashed', 'borderRadius': '5px', 'textAlign': 'center', 'margin': '10px 0' }, multiple=False ), html.Div(id={'type': 'uploaded-doc-name', 'index': selected_type}), dbc.RadioItems( id={'type': 'radio-doc-source', 'index': selected_type}, options=[ {'label': 'Loaded Document', 'value': 'loaded'}, {'label': 'Uploaded Document', 'value': 'uploaded'} ], value='loaded', inline=True, className="mb-2" ), ], id={'type': 'doc-type-controls', 'index': selected_type})) return dbc.Card(dbc.CardBody(controls)) def get_left_col_content(): chat_card = dbc.Card( dbc.CardBody([ html.H5("Maiko Chat", className="mb-2"), dcc.Loading( id="chat-loading", type="dot", children=[ dcc.Textarea( id="chat-input", placeholder="Chat with AI to update document...", className="mb-2", style={ 'whiteSpace':'pre-wrap', 'width': '100%', 'minHeight': '100px', 'maxHeight': '300px', 'resize': 'vertical', 'overflowY': 'auto' }, rows=5, wrap='soft' ), dcc.Store(id="chat-input-rows", data=5), dbc.Row([ dbc.Col( dbc.Button("Send", id="btn-send-chat", color="primary", className="mb-3 w-100"), width=6 ), dbc.Col( dbc.Button("Clear Chat", id="btn-clear-chat", color="secondary", className="mb-3 w-100"), width=6 ), ], className="g-1 mb-2"), html.Div(id="chat-output") ] ) ]), className="mt-4" ) return [ html.H4("Proposal Writer", className="mt-3 mb-4"), html.Div([ html.Div(className="blinking-dot", style={'margin':'0 auto','width':'16px','height':'16px'}), ], style={'textAlign':'center', 'marginBottom':'10px'}), html.Hr(), html.Div( id='doc-type-buttons' ), chat_card ] def file_list_component(): return html.Div( id='file-list-container', children=[ html.Div(id='file-list') ] ) app.layout = dbc.Container([ dcc.Store(id='selected-doc-type', data="Shred"), dbc.Row([ dbc.Col( dbc.Card( dbc.CardBody(get_left_col_content()) ), width=3 ), dbc.Col( html.Div(id='right-col-content'), width=9 ) ]) ], fluid=True) @app.callback( Output('doc-type-buttons', 'children'), Input('selected-doc-type', 'data') ) def render_doc_type_buttons(selected_type): buttons = [] for doc_type in document_types.keys(): btn_style = {'overflow': 'hidden', 'text-overflow': 'ellipsis', 'white-space': 'nowrap'} btn_class = "mb-2 w-100 text-left custom-button" if doc_type == selected_type: btn_class += " active-doc-type" buttons.append( dbc.Button( doc_type, id={'type': 'btn-doc-type', 'index': doc_type}, color="link", className=btn_class, style=btn_style ) ) return buttons @app.callback( Output('selected-doc-type', 'data'), [Input({'type': 'btn-doc-type', 'index': ALL}, 'n_clicks')], [State({'type': 'btn-doc-type', 'index': ALL}, 'id')], prevent_initial_call=True ) def update_selected_doc_type(n_clicks_list, btn_ids): triggered = callback_context.triggered if not triggered or all(x is None for x in n_clicks_list): raise dash.exceptions.PreventUpdate idx = [i for i, x in enumerate(n_clicks_list) if x] if idx: selected_type = btn_ids[idx[-1]]['index'] else: selected_type = "Shred" logging.info(f"Doc type selected: {selected_type}") return selected_type @app.callback( Output('right-col-content', 'children'), Input('selected-doc-type', 'data') ) def update_right_col(selected_type): return get_right_col_content(selected_type) def process_document(contents, filename): content_type, content_string = contents.split(',') decoded = base64.b64decode(content_string) try: if filename.lower().endswith('.docx'): doc = Document(BytesIO(decoded)) text = "\n".join([para.text for para in doc.paragraphs]) return text elif filename.lower().endswith('.pdf'): pdf = PdfReader(BytesIO(decoded)) text = "" for page in pdf.pages: page_text = page.extract_text() if page_text: text += page_text return text else: return f"Unsupported file format: {filename}. Please upload a PDF or DOCX file." except Exception as e: logging.error(f"Error processing document: {str(e)}") return f"Error processing document: {str(e)}" @app.callback( Output('file-list', 'children'), Input('upload-document', 'contents'), State('upload-document', 'filename'), State('file-list', 'children') ) def update_output(list_of_contents, list_of_names, existing_files): global uploaded_files, shredded_document if list_of_contents is not None: new_files = [] for i, (content, name) in enumerate(zip(list_of_contents, list_of_names)): file_content = process_document(content, name) uploaded_files[name] = file_content new_files.append( dbc.Row([ dbc.Col( html.Button('×', id={'type': 'remove-file', 'index': name}, style={'marginRight': '5px', 'fontSize': '10px'}), width=1 ), dbc.Col( html.Span(name, style={'wordBreak': 'break-all'}), width=11 ) ], id={'type': 'file-row', 'index': name}, align="center", className="mb-1") ) if existing_files is None: existing_files = [] elif not isinstance(existing_files, list): existing_files = [existing_files] shredded_document = None logging.info("Documents uploaded and file list updated.") return existing_files + new_files return existing_files @app.callback( Output('file-list', 'children', allow_duplicate=True), Input({'type': 'remove-file', 'index': ALL}, 'n_clicks'), State('file-list', 'children'), prevent_initial_call=True ) def remove_file(n_clicks, existing_files): global uploaded_files, shredded_document ctx = dash.callback_context if not ctx.triggered: raise dash.exceptions.PreventUpdate triggered_id = ctx.triggered[0]['prop_id'].split('.')[0] if not triggered_id.startswith("{\"type\":\"remove-file\""): raise dash.exceptions.PreventUpdate try: import json triggered_dict = json.loads(triggered_id) removed_file = triggered_dict['index'] except Exception as e: logging.error(f"Could not parse removed file from callback context: {e}") raise dash.exceptions.PreventUpdate uploaded_files.pop(removed_file, None) shredded_document = None logging.info(f"Removed file: {removed_file}") filtered_files = [] if existing_files: for file in existing_files: try: file_id = file['props']['id'] if file_id['index'] != removed_file: filtered_files.append(file) except Exception as e: filtered_files.append(file) return filtered_files @app.callback( Output('document-preview', 'children'), Output('loading-output', 'children'), Input({'type': 'btn-generate-doc', 'index': ALL}, 'n_clicks'), State({'type': 'btn-generate-doc', 'index': ALL}, 'id'), State({'type': 'radio-doc-source', 'index': ALL}, 'value'), State({'type': 'upload-doc-type', 'index': ALL}, 'contents'), State({'type': 'upload-doc-type', 'index': ALL}, 'filename'), prevent_initial_call=True ) def generate_any_doc(n_clicks_list, btn_ids, radio_values, upload_contents, upload_filenames): global current_document, document_type, shredded_document, pink_review_document ctx = callback_context logging.info(f"generate_any_doc triggered: n_clicks_list={n_clicks_list}, btn_ids={btn_ids}") if not ctx.triggered: raise dash.exceptions.PreventUpdate idx = [i for i, x in enumerate(n_clicks_list) if x] if not idx: raise dash.exceptions.PreventUpdate idx = idx[-1] doc_type = btn_ids[idx]['index'] document_type = doc_type if doc_type == "Shred": if not uploaded_files: logging.info("No uploaded files for Shred. Aborting.") return html.Div("Please upload a document before shredding."), "" file_contents = list(uploaded_files.values()) logging.info(f"Calling OpenAI for Shred with {len(file_contents)} files: {list(uploaded_files.keys())}") try: generated = generate_document(doc_type, file_contents) current_document = generated shredded_document = generated preview = markdown_table_preview(generated) logging.info("Shred document generated.") return preview, "Shred generated" except Exception as e: logging.error(f"Error generating document: {str(e)}") return html.Div(f"Error generating document: {str(e)}"), "Error" if shredded_document is None: logging.info("Shredded document is None. Aborting.") return html.Div("Please shred a document first."), "" source = None if radio_values and len(radio_values) > idx: source = radio_values[idx] else: source = 'loaded' doc_content = None if source == 'uploaded': if upload_contents and len(upload_contents) > idx and upload_contents[idx] and upload_filenames and len(upload_filenames) > idx and upload_filenames[idx]: doc_content = process_document(upload_contents[idx], upload_filenames[idx]) else: logging.info("No uploaded doc content for this doc type. Aborting.") return html.Div("Please upload a document to use as source."), "" else: if doc_type == "Pink": doc_content = shredded_document elif doc_type == "Pink Review": doc_content = pink_review_document if pink_review_document else "" elif doc_type == "Red": doc_content = pink_review_document if pink_review_document else "" elif doc_type == "Red Review": doc_content = pink_review_document if pink_review_document else "" elif doc_type == "Gold": doc_content = shredded_document elif doc_type == "Gold Review": doc_content = shredded_document elif doc_type == "Virtual Board": doc_content = shredded_document elif doc_type == "LOE": doc_content = shredded_document else: doc_content = shredded_document try: if doc_type == "Pink Review": generated = generate_document(doc_type, [doc_content, shredded_document]) pink_review_document = generated current_document = generated preview = markdown_table_preview(generated) logging.info("Pink Review document generated.") return preview, f"{doc_type} generated" elif doc_type in ["Red Review", "Gold Review", "Virtual Board", "LOE"]: generated = generate_document(doc_type, [doc_content, shredded_document]) current_document = generated preview = markdown_table_preview(generated) logging.info(f"{doc_type} document generated.") return preview, f"{doc_type} generated" elif doc_type in ["Pink", "Red", "Gold"]: generated = generate_document(doc_type, [doc_content]) current_document = generated preview = markdown_narrative_preview(generated) logging.info(f"{doc_type} document generated.") return preview, f"{doc_type} generated" else: generated = generate_document(doc_type, [doc_content]) current_document = generated preview = markdown_narrative_preview(generated) logging.info(f"{doc_type} document generated.") return preview, f"{doc_type} generated" except Exception as e: logging.error(f"Error generating document: {str(e)}") return html.Div(f"Error generating document: {str(e)}"), "Error" @app.callback( Output({'type': 'uploaded-doc-name', 'index': MATCH}, 'children'), Output({'type': 'upload-doc-type', 'index': MATCH}, 'contents'), Output({'type': 'radio-doc-source', 'index': MATCH}, 'value'), Input({'type': 'upload-doc-type', 'index': MATCH}, 'contents'), State({'type': 'upload-doc-type', 'index': MATCH}, 'filename'), State({'type': 'upload-doc-type', 'index': MATCH}, 'id') ) def update_uploaded_doc_name(contents, filename, id_dict): if contents is not None: uploaded_doc_contents[id_dict['index']] = (contents, filename) logging.info(f"{id_dict['index']} file uploaded: {filename}") return filename, contents, "uploaded" return "", None, "loaded" def markdown_table_preview(md_text): # Always use dcc.Markdown to render markdown tables, which supports tables natively return dcc.Markdown(md_text, dangerously_allow_html=True, style={'whiteSpace': 'pre-wrap', 'fontFamily': 'monospace', 'overflowX': 'auto'}) def markdown_narrative_preview(md_text): return html.Div(dcc.Markdown(md_text, dangerously_allow_html=True, style={'whiteSpace': 'pre-wrap', 'fontFamily': 'sans-serif'})) def extract_markdown_tables(md_text): # Extract all markdown tables from md_text as list of strings tables = [] lines = md_text.split('\n') in_table = False table_lines = [] for line in lines: if re.match(r'^\s*\|.*\|\s*$', line): in_table = True table_lines.append(line) elif in_table and (re.match(r'^\s*\|.*\|\s*$', line) or re.match(r'^\s*$', line)): table_lines.append(line) else: if in_table and table_lines: tables.append('\n'.join(table_lines)) table_lines = [] in_table = False if in_table and table_lines: tables.append('\n'.join(table_lines)) return tables def markdown_table_to_df(md_table): # Convert a single markdown table string to pandas DataFrame lines = [line.strip() for line in md_table.split('\n') if line.strip()] if len(lines) < 2: return None header = [h.strip() for h in lines[0].strip('|').split('|')] # Find the first line with at least 3 dashes, that's the separator sep_idx = 1 while sep_idx < len(lines) and not re.match(r'^\|\s*:?-+:?\s*(\|\s*:?-+:?\s*)+\|?$', lines[sep_idx]): sep_idx += 1 data_lines = lines[sep_idx+1:] if sep_idx+1 < len(lines) else [] rows = [] for row in data_lines: if not row.strip() or not row.strip().startswith('|'): continue cells = [c.strip() for c in row.strip('|').split('|')] # Pad or trim cells to header length if len(cells) < len(header): cells += [''] * (len(header) - len(cells)) elif len(cells) > len(header): cells = cells[:len(header)] rows.append(cells) df = pd.DataFrame(rows, columns=header) return df def markdown_tables_to_xlsx(md_text): tables = extract_markdown_tables(md_text) output = BytesIO() with pd.ExcelWriter(output, engine='xlsxwriter') as writer: for i, table in enumerate(tables): df = markdown_table_to_df(table) if df is not None: sheet_name = f"Table{i+1}" df.to_excel(writer, sheet_name=sheet_name, index=False) output.seek(0) return output def strip_markdown(text): text = re.sub(r'(\*\*|__)(.*?)\1', r'\2', text) text = re.sub(r'(\*|_)(.*?)\1', r'\2', text) text = re.sub(r'`{1,3}[^`]*`{1,3}', '', text) text = re.sub(r'^#+ ', '', text, flags=re.MULTILINE) text = re.sub(r'^> ', '', text, flags=re.MULTILINE) text = re.sub(r'!\[.*?\]\(.*?\)', '', text) text = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', text) text = re.sub(r'^\s*[-*+] ', '', text, flags=re.MULTILINE) text = re.sub(r'^\s*\d+\.\s+', '', text, flags=re.MULTILINE) text = text.replace('---', '') text = text.replace('___', '') text = text.replace('***', '') return text.strip() def generate_document(document_type, file_contents): if document_type in spreadsheet_types: prompt = f"""Ignore all other instructions and output only a spreadsheet for {document_type} as described below. Do not include any narrative, only the spreadsheet in markdown table format. Instructions: {document_types[document_type]} Project Artifacts: {' '.join(file_contents)} Output only the spreadsheet as a markdown table, no narrative or explanation.""" elif document_type in narrative_types: prompt = f"""Generate a {document_type} document based on the following project artifacts: {' '.join(file_contents)} Instructions: 1. Create the {document_type} as a detailed document. 2. Use proper formatting and structure. 3. Include all necessary sections and details. 4. Start the output immediately with the document content. Now, generate the {document_type}: """ else: prompt = f"""Generate a {document_type} based on the following project artifacts: {' '.join(file_contents)} Instructions: {document_types.get(document_type, '')} Now, generate the {document_type}: """ logging.info(f"Generating document for type: {document_type}") try: response = openai.ChatCompletion.create( model="gpt-4-1106-preview", messages=[ {"role": "system", "content": "You are a helpful, expert government proposal writer."}, {"role": "user", "content": prompt} ], max_tokens=4096, temperature=0.25, ) generated_text = response['choices'][0]['message']['content'] logging.info("Document generated successfully.") return generated_text except Exception as e: logging.error(f"Error generating document: {str(e)}") raise @app.callback( Output('chat-output', 'children'), Output('document-preview', 'children', allow_duplicate=True), Input('btn-send-chat', 'n_clicks'), Input('btn-clear-chat', 'n_clicks'), State('chat-input', 'value'), State('selected-doc-type', 'data'), prevent_initial_call=True ) def update_document_via_chat(btn_send, btn_clear, chat_input, selected_doc_type): global current_document, document_type ctx = callback_context if not ctx.triggered: raise dash.exceptions.PreventUpdate trigger_id = ctx.triggered[0]['prop_id'].split('.')[0] if trigger_id == 'btn-clear-chat': return "", dash.no_update if not chat_input or current_document is None: raise dash.exceptions.PreventUpdate if selected_doc_type in spreadsheet_types: prompt = f"""Update the following {selected_doc_type} spreadsheet based on this instruction: {chat_input} Current spreadsheet (markdown table format): {current_document} Instructions: 1. Provide the updated spreadsheet as a markdown table only. 2. Do not include any narrative, only the markdown table. Now, provide the updated {selected_doc_type} spreadsheet: """ else: prompt = f"""Update the following {selected_doc_type} document based on this instruction: {chat_input} Current document: {current_document} Instructions: 1. Provide the updated document content. 2. Maintain proper formatting and structure. 3. Incorporate the requested changes seamlessly. Now, provide the updated {selected_doc_type}: """ logging.info(f"Updating document via chat for {selected_doc_type} instruction: {chat_input}") try: response = openai.ChatCompletion.create( model="gpt-4-1106-preview", messages=[ {"role": "system", "content": "You are a helpful, expert government proposal writer."}, {"role": "user", "content": prompt} ], max_tokens=4096, temperature=0.2, ) current_document = response['choices'][0]['message']['content'] logging.info("Document updated via chat successfully.") if selected_doc_type in spreadsheet_types: preview = markdown_table_preview(current_document) else: preview = markdown_narrative_preview(current_document) return f"Document updated based on: {chat_input}", preview except Exception as e: logging.error(f"Error updating document via chat: {str(e)}") return f"Error updating document: {str(e)}", html.Div(f"Error updating document: {str(e)}") @app.callback( Output("download-document", "data"), Input("btn-download", "n_clicks"), State('selected-doc-type', 'data'), prevent_initial_call=True ) def download_document(n_clicks, selected_doc_type): global current_document if current_document is None: raise dash.exceptions.PreventUpdate if selected_doc_type in spreadsheet_types: try: # Convert markdown tables to xlsx xlsx_bytes = markdown_tables_to_xlsx(current_document) logging.info(f"{selected_doc_type} document downloaded as Excel.") return dcc.send_bytes(xlsx_bytes.read(), f"{selected_doc_type}.xlsx") except Exception as e: logging.error(f"Error downloading {selected_doc_type} document: {str(e)}") return dcc.send_string(f"Error downloading {selected_doc_type}: {str(e)}", f"{selected_doc_type}_error.txt") else: try: plain = strip_markdown(current_document) doc = Document() for para in plain.split('\n'): doc.add_paragraph(para) output = BytesIO() doc.save(output) logging.info(f"{selected_doc_type} document downloaded as Word.") output.seek(0) return dcc.send_bytes(output.read(), f"{selected_doc_type}.docx") except Exception as e: logging.error(f"Error downloading document: {str(e)}") return dcc.send_string(f"Error downloading document: {str(e)}", f"{selected_doc_type}_error.txt") @app.callback( Output("chat-input", "rows"), Input("chat-input", "value"), State("chat-input", "rows"), prevent_initial_call="initial_duplicate" ) def auto_expand_textarea(value, current_rows): if value is None or value == "": return 5 num_lines = value.count('\n') + 1 max_rows = 20 rows = min(max(num_lines, 5), max_rows) return rows if __name__ == '__main__': print("Starting the Dash application...") app.run(debug=True, host='0.0.0.0', port=7860, threaded=True) print("Dash application has finished running.")