Upload app.py

app.py

@@ -10,31 +10,51 @@ from pypdf import PdfReader
 import docx
 import traceback
 
-DATA_FILE = "legislation_rules.json" # Relative path for HF environment
+DATA_FILE = "legislation_rules.json"
+RULINGS_FILE = "rulings_log.json"
 
-def load_data():
-    if os.path.exists(DATA_FILE):
-        with open(DATA_FILE, 'r') as f:
+def load_data(file_path):
+    if os.path.exists(file_path):
+        with open(file_path, 'r') as f:
             try:
                 return json.load(f)
             except:
                 return []
     return []
 
-def save_data(data):
-    with open(DATA_FILE, 'w') as f:
+def save_data(data, file_path):
+    with open(file_path, 'w') as f:
         json.dump(data, f, indent=2)
 
 def get_canonical_hash(text):
     return hashlib.sha256(text.strip().encode('utf-8')).hexdigest()
 
+def extract_text_from_any(file_obj):
+    if file_obj is None: return ""
+    file_path = file_obj.name if hasattr(file_obj, 'name') else file_obj
+    if not os.path.exists(file_path): return ""
+    
+    ext = os.path.splitext(file_path)[1].lower()
+    text = ""
+    try:
+        if ext == ".pdf":
+            reader = PdfReader(file_path)
+            text = "\n".join([p.extract_text() for p in reader.pages if p.extract_text()])
+        elif ext == ".docx":
+            doc = docx.Document(file_path)
+            text = "\n".join([p.text for p in doc.paragraphs])
+        elif ext == ".txt":
+            with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
+                text = f.read()
+    except Exception as e:
+        print(f"Extraction error: {e}")
+    return text
+
 def add_rule_manually(act, title, text, source):
-    data = load_data()
+    data = load_data(DATA_FILE)
     det_id = get_canonical_hash(text)
-    
     if any(r['deterministic_id'] == det_id for r in data):
-        return f"Warning: Rule '{title}' already exists (matching hash)."
-    
+        return f"Warning: Rule '{title}' already exists."
     new_rule = {
         "act": act,
         "section_title": title,
@@ -44,152 +64,131 @@ def add_rule_manually(act, title, text, source):
         "added_at": datetime.now().isoformat()
     }
     data.append(new_rule)
-    save_data(data)
-    return f"Successfully added: {title} from {act}"
+    save_data(data, DATA_FILE)
+    return f"Successfully added: {title}"
+
+def process_rule_document(file_obj, act_name):
+    text = extract_text_from_any(file_obj)
+    if not text: return "Error: Could not extract text from rule document."
+    return add_rule_manually(act_name, os.path.basename(file_obj.name), text, f"File: {os.path.basename(file_obj.name)}")
 
-def process_document(file_obj, act_name):
+def rule_on_issue(issue_description, issue_file, llm_endpoint, llm_key, llm_model):
     """
-    Robustly extracts text from PDF, DOCX, or TXT files.
+    Framework 2.2: Regulatory Reasoning Engine with Document Intake.
     """
-    if file_obj is None:
-        return "Error: No file was uploaded."
+    # Combine text input and file input
+    file_text = extract_text_from_any(issue_file)
+    combined_issue = f"{issue_description}\n\n[EXTRACTED FROM DOCUMENT]:\n{file_text}".strip()
     
-    # Handle Gradio's file object (it can be a string path or a file-like object)
-    file_path = file_obj.name if hasattr(file_obj, 'name') else file_obj
+    if not combined_issue:
+        return "Error: No issue description or document provided.", ""
+
+    rules = load_data(DATA_FILE)
+    if not rules:
+        return "Error: Dataset is empty. Add rules (CRA, FCA, etc.) first.", ""
     
-    if not os.path.exists(file_path):
-        return f"Error: File not found at {file_path}. Please try uploading again."
+    # Context window management
+    context = "\n".join([f"[{r['act']} - {r['section_title']}]: {r['text'][:600]}..." for r in rules[:20]])
     
-    ext = os.path.splitext(file_path)[1].lower()
-    text = ""
-    title = os.path.basename(file_path)
+    prompt = f"""
+    You are a Regulatory Ruling Engine. 
+    ISSUE TO ANALYZE:
+    {combined_issue}
     
-    try:
-        if ext == ".pdf":
-            reader = PdfReader(file_path)
-            for page in reader.pages:
-                extracted = page.extract_text()
-                if extracted:
-                    text += extracted + "\n"
-        elif ext == ".docx":
-            doc = docx.Document(file_path)
-            text = "\n".join([para.text for para in doc.paragraphs])
-        elif ext == ".txt":
-            with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
-                text = f.read()
-        else:
-            return f"Error: Unsupported file type '{ext}'. Use PDF, DOCX, or TXT."
-        
-        if not text.strip():
-            return "Error: Could not extract any text. The file might be empty or an image-based PDF."
-            
-        return add_rule_manually(act_name, title, text, f"Uploaded File: {title}")
-    except Exception as e:
-        error_msg = traceback.format_exc()
-        print(error_msg) # Log to HF console
-        return f"Processing Failed: {str(e)}"
-
-def scrape_fca_prin():
-    url = "https://handbook.fca.org.uk/handbook/PRIN/2/1.html"
-    try:
-        response = requests.get(url, timeout=10)
-        soup = BeautifulSoup(response.content, 'html.parser')
-        principles_table = soup.find('table')
-        if not principles_table:
-            return "Error: Could not find the Principles table. The FCA site structure may have changed."
-        rows = principles_table.find_all('tr')
-        added_count = 0
-        for row in rows:
-            cols = row.find_all('td')
-            if len(cols) >= 2:
-                title = cols[0].get_text(strip=True)
-                text = cols[1].get_text(strip=True)
-                status = add_rule_manually("FCA Handbook: PRIN", f"Principle {title}", text, url)
-                if "Successfully" in status:
-                    added_count += 1
-        return f"Successfully ingested {added_count} FCA Principles."
-    except Exception as e:
-        return f"FCA Scraping failed: {str(e)}"
+    DETERMINISTIC REFERENCE RULES:
+    {context}
+    
+    TASK:
+    1. Identify applicable rules.
+    2. Provide a 'Formal Ruling' (Compliant/Non-Compliant/Warning).
+    3. Cite Rule Titles and Deterministic IDs.
+    4. Provide forensic justification.
+    """
+    
+    ruling_text = "LLM Inference required for automated ruling."
+    if llm_endpoint and llm_key:
+        try:
+            headers = {"Authorization": f"Bearer {llm_key}", "Content-Type": "application/json"}
+            payload = {"model": llm_model, "messages": [{"role": "user", "content": prompt}], "temperature": 0}
+            response = requests.post(f"{llm_endpoint.rstrip('/')}/v1/chat/completions", json=payload, headers=headers, timeout=45)
+            ruling_text = response.json()['choices'][0]['message']['content']
+        except Exception as e:
+            ruling_text = f"Inference Error: {str(e)}"
+    
+    # Log the ruling
+    rulings_log = load_data(RULINGS_FILE)
+    new_ruling = {
+        "timestamp": datetime.now().isoformat(),
+        "issue_summary": combined_issue[:200],
+        "ruling": ruling_text,
+        "ruling_hash": hashlib.sha256(ruling_text.encode()).hexdigest()
+    }
+    rulings_log.append(new_ruling)
+    save_data(rulings_log, RULINGS_FILE)
+    
+    return ruling_text, f"Ruling Logged (ID: {new_ruling['ruling_hash'][:8]})"
 
-def sync_to_huggingface(token, dataset_id):
-    if not token or not dataset_id:
-        return "Error: HF Token and Dataset ID are required."
+def sync_all(token, dataset_id):
+    if not token or not dataset_id: return "Error: Missing credentials."
     api = HfApi()
     try:
-        api.upload_file(
-            path_or_fileobj=DATA_FILE,
-            path_in_repo="legislation_rules.json",
-            repo_id=dataset_id,
-            repo_type="dataset",
-            token=token
-        )
-        return f"Successfully synced dataset to {dataset_id}"
-    except Exception as e:
-        return f"Sync failed: {str(e)}"
+        api.upload_file(path_or_fileobj=DATA_FILE, path_in_repo=DATA_FILE, repo_id=dataset_id, repo_type="dataset", token=token)
+        if os.path.exists(RULINGS_FILE):
+            api.upload_file(path_or_fileobj=RULINGS_FILE, path_in_repo=RULINGS_FILE, repo_id=dataset_id, repo_type="dataset", token=token)
+        return "Full Sync Successful."
+    except Exception as e: return f"Sync Failed: {str(e)}"
 
-def view_dataset_stats():
-    data = load_data()
-    if not data:
-        return "Dataset is currently empty."
-    stats = f"Total Rules: {len(data)}\n"
+def view_stats():
+    data = load_data(DATA_FILE)
+    if not data: return "Empty"
     acts = {}
-    for r in data:
-        acts[r['act']] = acts.get(r['act'], 0) + 1
-    for act, count in acts.items():
-        stats += f"- {act}: {count} rules\n"
-    return stats
+    for r in data: acts[r['act']] = acts.get(r['act'], 0) + 1
+    return "\n".join([f"{k}: {v}" for k,v in acts.items()])
 
 with gr.Blocks(theme=gr.themes.Soft()) as demo:
-    gr.Markdown("# ⚖️ Legislation & FCA Dataset Manager")
-    gr.Markdown("### Create and Expand your Deterministic Rules of Law Dataset")
+    gr.Markdown("# ⚖️ Legislation & FCA Ruling Manager")
     
-    with gr.Tab("➕ Add Rule"):
+    with gr.Tab("🏛️ Rule on Issues"):
+        gr.Markdown("### 📂 Issue Document Intake")
         with gr.Row():
             with gr.Column():
-                gr.Markdown("#### 📂 Option A: Document Ingestion")
-                doc_act_name = gr.Textbox(label="Legislation/Source Name", placeholder="e.g., Consumer Rights Act 2015")
-                doc_input = gr.File(label="Upload PDF, DOCX, or TXT")
-                doc_btn = gr.Button("Process Document", variant="primary")
-                
+                issue_file = gr.File(label="Upload Issue Document (PDF/DOCX/TXT)", file_count="single")
+                issue_desc = gr.TextArea(label="Additional Context / Issue Description")
+                with gr.Accordion("⚙️ Inference Settings", open=False):
+                    end = gr.Textbox(label="Endpoint", value="http://localhost:11434/v1")
+                    key = gr.Textbox(label="Key", type="password")
+                    mod = gr.Textbox(label="Model", value="llama3")
+                rule_btn = gr.Button("Generate Formal Ruling", variant="primary")
+            with gr.Column():
+                ruling_out = gr.Markdown(label="Official Ruling")
+                log_status = gr.Textbox(label="Audit Status")
+
+    with gr.Tab("➕ Manage Rules"):
+        with gr.Row():
+            with gr.Column():
+                gr.Markdown("#### 📂 Add Legislation/Handbook via File")
+                doc_act = gr.Textbox(label="Act/Source Name")
+                doc_in = gr.File(label="Upload Rule Document")
+                doc_btn = gr.Button("Process Rule")
                 gr.Markdown("---")
-                gr.Markdown("#### ✍️ Option B: Manual Entry")
-                act_input = gr.Textbox(label="Act Name")
-                title_input = gr.Textbox(label="Section Title")
-                source_input = gr.Textbox(label="Source URL")
-                text_input = gr.TextArea(label="Rule Text")
-                add_btn = gr.Button("Add Manually")
-                
+                gr.Markdown("#### ✍️ Manual Rule Entry")
+                m_act = gr.Textbox(label="Act"); m_tit = gr.Textbox(label="Title"); m_src = gr.Textbox(label="Source"); m_txt = gr.TextArea(label="Text")
+                m_btn = gr.Button("Add Rule")
             with gr.Column():
-                status_out = gr.Textbox(label="Operation Status")
-                stats_view = gr.Textbox(label="Dataset Inventory", value=view_dataset_stats())
+                op_status = gr.Textbox(label="Status")
+                stats_out = gr.Textbox(label="Dataset Inventory", value=view_stats())
                 refresh_btn = gr.Button("Refresh Inventory")
 
-    with gr.Tab("🏦 FCA Guidelines"):
-        gr.Markdown("### 🛠️ FCA Handbook Automation")
-        fca_btn = gr.Button("Ingest FCA PRIN Principles", variant="secondary")
-        fca_status = gr.Textbox(label="FCA Ingestion Status")
-
-    with gr.Tab("☁️ Sync to Hugging Face"):
-        hf_token = gr.Textbox(label="HF Write Token", type="password")
-        hf_id = gr.Textbox(label="Dataset ID")
-        sync_btn = gr.Button("Sync to Dataset")
-        sync_status = gr.Textbox(label="Sync Status")
-
-    add_btn.click(
-        fn=add_rule_manually,
-        inputs=[act_input, title_input, text_input, source_input],
-        outputs=status_out
-    )
-    
-    doc_btn.click(
-        fn=process_document,
-        inputs=[doc_input, doc_act_name],
-        outputs=status_out
-    )
-    
-    fca_btn.click(fn=scrape_fca_prin, outputs=fca_status)
-    refresh_btn.click(fn=view_dataset_stats, outputs=stats_view)
-    sync_btn.click(fn=sync_to_huggingface, inputs=[hf_token, hf_id], outputs=sync_status)
+    with gr.Tab("☁️ Sync & Audit"):
+        hf_t = gr.Textbox(label="HF Token", type="password"); hf_d = gr.Textbox(label="Dataset ID")
+        s_btn = gr.Button("Sync Everything to Dataset")
+        s_out = gr.Textbox(label="Sync Status")
+
+    rule_btn.click(fn=rule_on_issue, inputs=[issue_desc, issue_file, end, key, mod], outputs=[ruling_out, log_status])
+    doc_btn.click(fn=process_rule_document, inputs=[doc_in, doc_act], outputs=op_status)
+    m_btn.click(fn=add_rule_manually, inputs=[m_act, m_tit, m_txt, m_src], outputs=op_status)
+    refresh_btn.click(fn=view_stats, outputs=stats_out)
+    s_btn.click(fn=sync_all, inputs=[hf_t, hf_d], outputs=s_out)
 
 if __name__ == "__main__":
     demo.launch()