Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python | |
| # coding=utf-8 | |
| import datetime | |
| import re | |
| from fpdf import FPDF | |
| from langdetect import detect | |
| import gradio as gr | |
| from tools.common import prepend_metadata_questions | |
| # === PDF Export Function === | |
| def export_text_to_pdf(text, metadata=None, output_path=None, language="en"): | |
| if output_path is None: | |
| timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") | |
| output_path = f"sar_record_{timestamp}.pdf" | |
| pdf = FPDF() | |
| pdf.add_page() | |
| pdf.set_auto_page_break(auto=True, margin=15) | |
| pdf.set_font("Arial", 'B', 16) | |
| pdf.set_text_color(0, 51, 102) | |
| title = "Subject Access Request (SAR) Record" | |
| pdf.cell(0, 15, title, ln=True, align='C') | |
| pdf.ln(10) | |
| # Metadata | |
| if metadata: | |
| pdf.set_font("Arial", '', 12) | |
| pdf.set_text_color(90, 90, 90) | |
| pdf.multi_cell(0, 10, f"Organization: {metadata.get('organization_name', 'N/A')}") | |
| pdf.multi_cell(0, 10, f"Completed by: {metadata.get('user_name', 'N/A')} ({metadata.get('user_role', 'N/A')})") | |
| pdf.multi_cell(0, 10, f"Timestamp: {metadata.get('timestamp', 'N/A')}") | |
| pdf.ln(5) | |
| # Content | |
| pdf.set_font("Arial", '', 12) | |
| pdf.set_text_color(0, 0, 0) | |
| for line in text.strip().split('\n'): | |
| line = line.strip() | |
| if line.startswith("## "): | |
| section = line.replace("## ", "").strip() | |
| pdf.set_font("Arial", 'B', 13) | |
| pdf.set_text_color(30, 30, 120) | |
| pdf.ln(8) | |
| pdf.cell(0, 10, section, ln=True) | |
| pdf.set_font("Arial", '', 12) | |
| pdf.set_text_color(0, 0, 0) | |
| elif line.startswith("- **"): | |
| match = re.match(r"- \*\*(.+?)\*\*: (.+)", line) | |
| if match: | |
| label, value = match.groups() | |
| pdf.set_font("Arial", 'B', 12) | |
| pdf.cell(0, 10, f"{label}:", ln=True) | |
| pdf.set_font("Arial", '', 12) | |
| pdf.multi_cell(0, 10, value) | |
| else: | |
| pdf.multi_cell(0, 10, line) | |
| pdf.output(output_path) | |
| return output_path | |
| # === Questions === | |
| QUESTIONS = prepend_metadata_questions([ | |
| ("request_date", "When was the access request received?"), | |
| ("data_subject_identity", "Who is the data subject (name or ID)?"), | |
| ("requested_info", "What information did the data subject request?"), | |
| ("verification_process", "How was the subject's identity verified?"), | |
| ("response_timeline", "What was the planned timeline for response?"), | |
| ("info_provided", "What data or response was ultimately provided?"), | |
| ("notes", "Any additional notes or observations?") | |
| ]) | |
| def get_questions(): | |
| return QUESTIONS | |
| # === Run Tool === | |
| def run_tool(): | |
| state = {"step": 0, "answers": {}} | |
| def step_by_step_agent(user_input, state): | |
| step = state["step"] | |
| answers = state["answers"] | |
| if step > 0: | |
| key, _ = QUESTIONS[step - 1] | |
| answers[key] = user_input | |
| if step < len(QUESTIONS): | |
| next_q = QUESTIONS[step][1] | |
| state["step"] += 1 | |
| return next_q, state, None | |
| # Final formatting | |
| content = "\n".join([f"- **{label}**: {answers.get(key, '')}" for key, label in QUESTIONS]) | |
| lang = detect(content) | |
| metadata = { | |
| "user_name": answers.get("user_name", "N/A"), | |
| "user_role": answers.get("user_role", "N/A"), | |
| "organization_name": answers.get("organization_name", "N/A"), | |
| "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| } | |
| pdf_path = export_text_to_pdf(content, metadata=metadata, language=lang) | |
| return "✅ SAR record created. Download your PDF below.", {"done": True}, pdf_path | |
| with gr.Blocks(title="SAR Record Tool") as demo: | |
| chatbot = gr.Chatbot(label="📥 Subject Access Request Assistant", value=[{"role": "assistant", "content": QUESTIONS[0][1]}], type="messages") | |
| msg = gr.Textbox(label="Your answer") | |
| state_var = gr.State(state) | |
| file_output = gr.File(label="Download PDF") | |
| reset_btn = gr.Button("🔁 Restart") | |
| def chat_logic(msg_in, state_in): | |
| reply, updated_state, file = step_by_step_agent(msg_in, state_in) | |
| messages = [{"role": "user", "content": msg_in}] | |
| if reply: | |
| messages.append({"role": "assistant", "content": reply}) | |
| return messages, updated_state, file | |
| def reset(): | |
| return [{"role": "assistant", "content": QUESTIONS[0][1]}], {"step": 0, "answers": {}}, None | |
| msg.submit(chat_logic, [msg, state_var], [chatbot, state_var, file_output]) | |
| reset_btn.click(reset, outputs=[chatbot, state_var, file_output]) | |
| demo.launch(show_api=False) | |