from __future__ import annotations
import html
import json
import uuid
from pathlib import Path
import streamlit as st
ROOT = Path(__file__).resolve().parent
from core.analyzer import build_packet
from core.llm import LocalGemmaClient
from core.models import EvidenceItem
from core.storage import load_evidence, save_evidence
SAMPLE_CASE = ROOT / "data" / "sample_unpaid_wages_case.json"
RIGHTS_GUIDE = ROOT / "data" / "worker_rights_guide.md"
st.set_page_config(
page_title="ProofPrint",
page_icon="PP",
layout="wide",
initial_sidebar_state="expanded",
)
def apply_theme() -> None:
st.markdown(
"""
""",
unsafe_allow_html=True,
)
def read_text_file(uploaded_file) -> str:
data = uploaded_file.getvalue()
try:
return data.decode("utf-8")
except UnicodeDecodeError:
try:
return data.decode("latin-1")
except UnicodeDecodeError:
return ""
def load_rights_guide() -> str:
return RIGHTS_GUIDE.read_text(encoding="utf-8") if RIGHTS_GUIDE.exists() else ""
def load_sample_items() -> list[EvidenceItem]:
payload = json.loads(SAMPLE_CASE.read_text(encoding="utf-8"))
return [
EvidenceItem(
id=item["id"],
title=item["title"],
source_type=item["source_type"],
raw_text=item["raw_text"],
filename=item.get("filename", ""),
created_on=item.get("created_on", "2026-05-11"),
)
for item in payload["evidence"]
]
def init_state() -> None:
if "evidence" not in st.session_state:
st.session_state.evidence = load_evidence()
if "packet" not in st.session_state:
st.session_state.packet = None
if "worker_name" not in st.session_state:
st.session_state.worker_name = "Ravi Kumar"
def add_evidence(title: str, source_type: str, raw_text: str, filename: str = "") -> None:
if not raw_text.strip():
return
st.session_state.evidence.append(
EvidenceItem(
id=f"E{len(st.session_state.evidence) + 1:03d}",
title=title.strip() or filename or "Untitled evidence",
source_type=source_type,
raw_text=raw_text.strip(),
filename=filename,
)
)
save_evidence(st.session_state.evidence)
def packet_to_markdown(packet) -> str:
timeline = "\n".join(
f"- {event.date}: {event.event} Evidence: {', '.join(event.evidence_ids)} Amount: {event.amount or 'N/A'}"
for event in packet.timeline
)
evidence_rows = "\n".join(
f"| {row['id']} | {row['title']} | {row['type']} | {row['key_dates']} | {row['amounts']} | {row['sensitivity']} |"
for row in packet.evidence_table
)
return f"""# ProofPrint Case Packet
Worker: {packet.worker_name}
Case type: {packet.case_type}
Model status: {packet.model_status}
## Summary
{packet.summary}
## Timeline
{timeline or "No timeline events extracted."}
## Evidence Table
| ID | Title | Type | Key dates | Amounts | Sensitivity |
|---|---|---|---|---|---|
{evidence_rows}
## Missing Evidence
{chr(10).join(f"- {item}" for item in packet.missing_evidence)}
## Risk Flags
{chr(10).join(f"- {item}" for item in packet.risk_flags)}
## Privacy Notes
{chr(10).join(f"- {item}" for item in packet.privacy_notes)}
## Advocate Packet
{packet.advocate_packet}
## Complaint Draft
{packet.complaint_draft}
"""
def render_header() -> None:
st.markdown(
"""
PP
ProofPrint
Private evidence packets for worker advocates
Digital Equity
Safety & Trust
Ollama Track
Local-first justice access
ProofPrint turns scattered worker evidence into an advocate-ready packet.
A private Gemma 4 workflow for wage theft, unsafe work, retaliation, and denied benefits.
It converts chats, payment notes, shift logs, and voice transcripts into a timeline,
risk review, and human advocate handoff.
Gemma 4 local inference
Ollama on device
Evidence-grounded output
No fallback generation
Demo case
Ravi worked five shifts. Only Rs 1,500 arrived.
Proof is split across hiring messages, shift notes, a UPI transcript, follow-up messages,
and a voice note. ProofPrint turns it into one reviewable packet.
unpaid wages
retaliation risk
privacy redaction
""",
unsafe_allow_html=True,
)
def render_sidebar() -> None:
with st.sidebar:
st.markdown("## ProofPrint")
st.caption("Local Gemma evidence kit")
st.divider()
st.subheader("Case setup")
st.session_state.worker_name = st.text_input("Worker name", st.session_state.worker_name)
model = st.text_input("Local Gemma model", "gemma4:e2b")
st.caption("Use a local Ollama Gemma model name. Packet generation is disabled unless Gemma is reachable.")
if st.button("Load sample case", use_container_width=True):
st.session_state.evidence = load_sample_items()
st.session_state.packet = None
save_evidence(st.session_state.evidence)
st.rerun()
if st.button("Clear evidence", use_container_width=True):
st.session_state.evidence = []
st.session_state.packet = None
save_evidence([])
st.rerun()
st.divider()
st.subheader("Engine")
st.markdown(
"""
Gemma extraction
timeline reasoning
risk checks
packet drafting
""",
unsafe_allow_html=True,
)
st.divider()
st.caption("For public demos, use synthetic or anonymized evidence only.")
return model
def render_intake() -> None:
st.markdown("### Evidence Intake")
st.caption("Add the fragments a worker actually has: chats, payment notes, shift logs, letters, or transcripts.")
left, right = st.columns([1.08, 0.92], gap="large")
with left:
st.markdown('Paste evidence manually
Use this for chats, OCR text, voice transcripts, or notes copied from an image.
', unsafe_allow_html=True)
with st.form("manual_evidence", clear_on_submit=True):
title = st.text_input("Evidence title", placeholder="UPI payment screenshot text, shift log, employer chat...")
source_type = st.selectbox("Source type", ["chat", "payment", "shift log", "receipt", "letter", "photo note", "other"])
raw_text = st.text_area("Paste text or transcript", height=210, placeholder="Paste OCR text, chat transcript, voice note transcript, or notes from an image.")
submitted = st.form_submit_button("Add evidence", use_container_width=True)
if submitted:
add_evidence(title, source_type, raw_text)
st.rerun()
with right:
st.markdown(
"""
1
Collect proof
Bring in the messy material exactly as it exists.
2
Run Gemma locally
Extract facts without sending worker data to a hosted model.
3
Export packet
Give advocates a timeline, risk review, missing-proof list, and draft.
""",
unsafe_allow_html=True,
)
st.markdown('Upload text evidence
Import .txt, .md, or .csv evidence files.
', unsafe_allow_html=True)
uploaded_files = st.file_uploader(
"Choose evidence files",
accept_multiple_files=True,
type=["txt", "md", "csv"],
)
if uploaded_files:
st.caption(f"{len(uploaded_files)} file(s) ready to import.")
if st.button("Import uploaded files", use_container_width=True):
for uploaded_file in uploaded_files:
add_evidence(uploaded_file.name, uploaded_file.type or "uploaded file", read_text_file(uploaded_file), uploaded_file.name)
st.rerun()
def render_evidence_list() -> None:
st.markdown("### Evidence Board")
if not st.session_state.evidence:
st.markdown('No evidence yet.
Load the sample case from the sidebar or add evidence manually.
', unsafe_allow_html=True)
return
for item in st.session_state.evidence:
preview = html.escape(item.raw_text[:180] + ("..." if len(item.raw_text) > 180 else ""))
title = html.escape(item.title)
source_type = html.escape(item.source_type)
st.markdown(
f"""
{item.id} / {source_type}
{title}
stored locally
{preview}
""",
unsafe_allow_html=True,
)
with st.expander(f"{item.id} - {item.title}", expanded=False):
st.write(f"Type: `{item.source_type}`")
st.text_area("Raw evidence", item.raw_text, height=130, key=f"raw_{item.id}", disabled=True)
def render_packet(packet) -> None:
st.markdown("### Generated Case Packet")
top = st.columns(4)
counters = [
("Case type", packet.case_type),
("Evidence items", str(len(packet.evidence_table))),
("Timeline events", str(len(packet.timeline))),
("Missing checks", str(len(packet.missing_evidence))),
]
for column, (label, value) in zip(top, counters):
column.markdown(
f"""
{html.escape(label)}
{html.escape(value)}
""",
unsafe_allow_html=True,
)
st.markdown(f'{packet.model_status}', unsafe_allow_html=True)
st.markdown(f'Case summary
{html.escape(packet.summary)}
', unsafe_allow_html=True)
timeline_tab, evidence_tab, risks_tab, packet_tab, draft_tab = st.tabs(
["Timeline", "Evidence", "Risk checks", "Advocate packet", "Complaint draft"]
)
with timeline_tab:
for event in packet.timeline:
event_text = html.escape(event.event)
event_date = html.escape(str(event.date))
event_amount = html.escape(event.amount or "N/A")
evidence_ids = html.escape(", ".join(event.evidence_ids))
st.markdown(
f"""
{event_date}
{event_text}
Evidence: {evidence_ids}
Amount: {event_amount}
""",
unsafe_allow_html=True,
)
with evidence_tab:
st.dataframe(packet.evidence_table, use_container_width=True, hide_index=True)
with risks_tab:
st.markdown("##### Missing evidence")
for item in packet.missing_evidence:
st.markdown(f'{item}
', unsafe_allow_html=True)
st.markdown("##### Risk and privacy flags")
for item in packet.risk_flags:
st.markdown(f'{item}
', unsafe_allow_html=True)
st.markdown("##### Privacy notes")
for item in packet.privacy_notes:
st.write(f"- {item}")
with packet_tab:
st.markdown(packet.advocate_packet)
with draft_tab:
st.text_area("Draft", packet.complaint_draft, height=340)
markdown = packet_to_markdown(packet)
st.download_button(
"Download advocate packet",
data=markdown,
file_name=f"proofprint_packet_{uuid.uuid4().hex[:8]}.md",
mime="text/markdown",
use_container_width=True,
)
def main() -> None:
apply_theme()
init_state()
model_name = render_sidebar()
render_header()
stats = st.columns(3)
stats[0].markdown('Problem
Scattered proof blocks justice.Workers often have evidence, but not a usable case packet.Gemma 4
Private local reasoning.Facts are extracted on device through Ollama and Gemma 4 E2B.Output
Advocate-ready handoff.Timeline, evidence index, missing proof, risk flags, and draft.