diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml new file mode 100644 index 0000000..3ff38be --- /dev/null +++ b/.github/workflows/lint.yml @@ -0,0 +1,38 @@ +name: Code Quality + +on: + push: + branches: [main] + + pull_request: + branches: [main] + +jobs: + lint: + runs-on: ubuntu-latest + + defaults: + run: + working-directory: code + + steps: + - name: Checkout Repository + uses: actions/checkout@v4 + + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: "pip" + + - name: Install Development Dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements-dev.txt + + - name: Run Black + run: black --check . + + - name: Run Ruff + run: ruff check . --exclude notebooks + \ No newline at end of file diff --git a/code/app.py b/code/app.py index 89db3ae..5e6b8b6 100644 --- a/code/app.py +++ b/code/app.py @@ -1,4 +1,5 @@ ο»Ώimport csv + # -*- coding: utf-8 -*- import io import json @@ -1287,7 +1288,9 @@ def render_sidebar(): ) st.divider() st.markdown("### Review Mode") - st.caption("Single-claim visual triage powered by the existing Gemini analyzer.") + st.caption( + "Single-claim visual triage powered by the existing Gemini analyzer." + ) st.markdown( """
@@ -1605,7 +1608,9 @@ def generate_repair_guidance(claim_object, result): return list(dict.fromkeys(guidance)) -def ensure_analysis_result_contract(result, claim_object=None, claim_history=None, user_claim=""): +def ensure_analysis_result_contract( + result, claim_object=None, claim_history=None, user_claim="" +): result = dict(result or {}) result.setdefault("object_type", claim_object or "unknown") result.setdefault("issue_type", "unknown") @@ -1637,23 +1642,41 @@ def ensure_analysis_result_contract(result, claim_object=None, claim_history=Non if not fraud_risk or str(fraud_risk).strip().lower() == "unknown": fraud_risk = risk_level_from_score(result["fraud_risk_score"]) result["fraud_risk"] = fraud_risk - result["fraud_risk_reasons"] = risk_reasons or ["No historical risk factors identified"] + result["fraud_risk_reasons"] = risk_reasons or [ + "No historical risk factors identified" + ] repair_estimate = result.get("repair_estimate") - if not repair_estimate or str(repair_estimate).strip().lower() in {"none", "unknown", "--"}: + if not repair_estimate or str(repair_estimate).strip().lower() in { + "none", + "unknown", + "--", + }: repair_estimate = generate_repair_guidance(claim_object, result) if isinstance(repair_estimate, str): - repair_items = [item.strip() for item in repair_estimate.split(";") if item.strip()] + repair_items = [ + item.strip() for item in repair_estimate.split(";") if item.strip() + ] else: - repair_items = [str(item).strip() for item in repair_estimate if str(item).strip()] + repair_items = [ + str(item).strip() for item in repair_estimate if str(item).strip() + ] result["repair_estimate"] = repair_items or ["Repair inspection recommended"] estimated_cost = result.get("estimated_cost") or result.get("estimated_repair_cost") - if not estimated_cost or str(estimated_cost).strip().lower() in {"none", "unknown", "--"}: - estimated_cost = estimate_repair_cost(result.get("object_type") or claim_object, result.get("severity")) + if not estimated_cost or str(estimated_cost).strip().lower() in { + "none", + "unknown", + "--", + }: + estimated_cost = estimate_repair_cost( + result.get("object_type") or claim_object, result.get("severity") + ) result["estimated_cost"] = estimated_cost result["estimated_repair_cost"] = estimated_cost - result["ai_explanation"] = result.get("ai_explanation") or build_ai_explanation(result, claim_object, user_claim) + result["ai_explanation"] = result.get("ai_explanation") or build_ai_explanation( + result, claim_object, user_claim + ) return result @@ -1686,7 +1709,11 @@ def parse_risk_flags(raw_value): normalized = str(raw_value or "").strip() if not normalized: return "Unknown" - flags = [part.strip().lower() for part in normalized.replace(",", ";").split(";") if part.strip()] + flags = [ + part.strip().lower() + for part in normalized.replace(",", ";").split(";") + if part.strip() + ] for level in ["high", "medium", "low"]: for flag in flags: if level in flag: @@ -1701,7 +1728,10 @@ def average_severity_label(severity_counts): total = sum(severity_counts.values()) if total == 0: return "Unknown" - score = sum(lookup.get(label, 0) * count for label, count in severity_counts.items()) / total + score = ( + sum(lookup.get(label, 0) * count for label, count in severity_counts.items()) + / total + ) if score >= 2.5: return "High" if score >= 1.5: @@ -1718,7 +1748,9 @@ def load_analytics_data(): rows = [ row for row in reader - if any(value.strip() for value in row.values() if isinstance(value, str)) + if any( + value.strip() for value in row.values() if isinstance(value, str) + ) and str(row.get("user_id", "")).strip().lower() != "user_id" ] except Exception: @@ -1750,7 +1782,9 @@ def load_analytics_data(): for row in rows: stats["total"] += 1 claim_status = str(row.get("claim_status", "")).strip().lower() - normalized_status = status_map.get(claim_status, str(claim_status).replace("_", " ").title() or "Unknown") + normalized_status = status_map.get( + claim_status, str(claim_status).replace("_", " ").title() or "Unknown" + ) if normalized_status == "Approved": stats["approved"] += 1 if normalized_status == "Rejected": @@ -1761,12 +1795,16 @@ def load_analytics_data(): issue_type = str(row.get("issue_type", "Unknown")).strip().title() or "Unknown" severity = str(row.get("severity", "Unknown")).strip().title() or "Unknown" risk_flag = parse_risk_flags(row.get("risk_flags", "")) - object_type = str(row.get("claim_object", "Unknown")).strip().title() or "Unknown" + object_type = ( + str(row.get("claim_object", "Unknown")).strip().title() or "Unknown" + ) confidence = str(row.get("confidence_score", "")).strip() stats["damage_types"][issue_type] = stats["damage_types"].get(issue_type, 0) + 1 stats["severity"][severity] = stats["severity"].get(severity, 0) + 1 - stats["status"][normalized_status] = stats["status"].get(normalized_status, 0) + 1 + stats["status"][normalized_status] = ( + stats["status"].get(normalized_status, 0) + 1 + ) stats["fraud_risk"][risk_flag] = stats["fraud_risk"].get(risk_flag, 0) + 1 stats["objects"][object_type] = stats["objects"].get(object_type, 0) + 1 @@ -1798,11 +1836,15 @@ def render_analytics_dashboard(): "
ANAnalytics Summary
", unsafe_allow_html=True, ) - st.info("No analytics data found in dataset/output.csv. Upload claims or refresh the dataset to populate the dashboard.") + st.info( + "No analytics data found in dataset/output.csv. Upload claims or refresh the dataset to populate the dashboard." + ) st.markdown("
", unsafe_allow_html=True) return - high_risk_pct = round((stats["high_risk"] / stats["total"]) * 100) if stats["total"] else 0 + high_risk_pct = ( + round((stats["high_risk"] / stats["total"]) * 100) if stats["total"] else 0 + ) avg_severity = average_severity_label(stats["severity"]) top_damage = "No damage type available" if stats["damage_types"]: @@ -1859,7 +1901,10 @@ def render_analytics_dashboard(): chart_row = st.columns(2) with chart_row[0]: - st.markdown("
CSClaim Status Distribution
", unsafe_allow_html=True) + st.markdown( + "
CSClaim Status Distribution
", + unsafe_allow_html=True, + ) status_names = list(stats["status"].keys()) status_values = list(stats["status"].values()) fig = px.bar( @@ -1869,7 +1914,9 @@ def render_analytics_dashboard(): labels={"x": "Status", "y": "Claims"}, template="plotly_dark", ) - fig.update_traces(marker_color="#22c55e", hovertemplate="%{x}: %{y}") + fig.update_traces( + marker_color="#22c55e", hovertemplate="%{x}: %{y}" + ) fig.update_layout( height=360, margin=dict(l=0, r=0, t=24, b=0), @@ -1879,7 +1926,10 @@ def render_analytics_dashboard(): ) st.plotly_chart(fig, use_container_width=True) with chart_row[1]: - st.markdown("
SDSeverity Distribution
", unsafe_allow_html=True) + st.markdown( + "
SDSeverity Distribution
", + unsafe_allow_html=True, + ) severity_names = list(stats["severity"].keys()) severity_values = list(stats["severity"].values()) fig = px.pie( @@ -1899,7 +1949,10 @@ def render_analytics_dashboard(): second_row = st.columns(2) with second_row[0]: - st.markdown("
DTDamage Type Distribution
", unsafe_allow_html=True) + st.markdown( + "
DTDamage Type Distribution
", + unsafe_allow_html=True, + ) damage_names = list(stats["damage_types"].keys()) damage_values = list(stats["damage_types"].values()) fig = px.bar( @@ -1909,7 +1962,9 @@ def render_analytics_dashboard(): labels={"x": "Damage Type", "y": "Claims"}, template="plotly_dark", ) - fig.update_traces(marker_color="#38bdf8", hovertemplate="%{x}: %{y}") + fig.update_traces( + marker_color="#38bdf8", hovertemplate="%{x}: %{y}" + ) fig.update_layout( height=360, margin=dict(l=0, r=0, t=24, b=0), @@ -1919,7 +1974,10 @@ def render_analytics_dashboard(): ) st.plotly_chart(fig, use_container_width=True) with second_row[1]: - st.markdown("
RDRisk Distribution
", unsafe_allow_html=True) + st.markdown( + "
RDRisk Distribution
", + unsafe_allow_html=True, + ) fraud_names = list(stats["fraud_risk"].keys()) fraud_values = list(stats["fraud_risk"].values()) fig = px.bar( @@ -1929,7 +1987,9 @@ def render_analytics_dashboard(): labels={"x": "Risk Level", "y": "Claims"}, template="plotly_dark", ) - fig.update_traces(marker_color="#f59e0b", hovertemplate="%{x}: %{y}") + fig.update_traces( + marker_color="#f59e0b", hovertemplate="%{x}: %{y}" + ) fig.update_layout( height=360, margin=dict(l=0, r=0, t=24, b=0), @@ -1993,7 +2053,9 @@ def render_analytics_dashboard(): def load_user_history(claim_object): try: - with (REPO_ROOT / "dataset" / "user_history.csv").open(newline="", encoding="utf-8") as csvfile: + with (REPO_ROOT / "dataset" / "user_history.csv").open( + newline="", encoding="utf-8" + ) as csvfile: reader = csv.DictReader(csvfile) rows = list(reader) except Exception: @@ -2008,15 +2070,31 @@ def load_user_history(claim_object): return rows[0] if rows else None -def build_export_payload(claim_object, user_claim, result, confidence_score, risk_score, cost_estimate, explanation): - result = ensure_analysis_result_contract(result, claim_object, st.session_state.get("claim_history"), user_claim) +def build_export_payload( + claim_object, + user_claim, + result, + confidence_score, + risk_score, + cost_estimate, + explanation, +): + result = ensure_analysis_result_contract( + result, claim_object, st.session_state.get("claim_history"), user_claim + ) return { "claim_object": claim_object, "claim_description": user_claim, "analysis_timestamp": datetime.utcnow().isoformat() + "Z", "analysis_result": result, - "confidence_score": confidence_score if confidence_score is not None else result["confidence_score"], - "fraud_risk_score": risk_score if risk_score is not None else result["fraud_risk_score"], + "confidence_score": ( + confidence_score + if confidence_score is not None + else result["confidence_score"] + ), + "fraud_risk_score": ( + risk_score if risk_score is not None else result["fraud_risk_score"] + ), "risk_level": result["fraud_risk"], "repair_estimate": result["repair_estimate"], "estimated_repair_cost": cost_estimate or result["estimated_cost"], @@ -2044,12 +2122,20 @@ def build_output_record(claim_object, user_claim, image_path, result): "user_claim": user_claim or "", "claim_object": claim_object or result.get("object_type", "unknown"), "evidence_standard_met": evidence_met, - "evidence_standard_met_reason": "Sufficient visual evidence" if evidence_met else "Insufficient visual evidence", + "evidence_standard_met_reason": ( + "Sufficient visual evidence" + if evidence_met + else "Insufficient visual evidence" + ), "risk_flags": result.get("fraud_risk", "Low"), "issue_type": result.get("issue_type", "unknown"), "object_part": result.get("object_part", "unknown"), "claim_status": claim_status, - "claim_status_justification": "Damage visible in image" if result.get("damage_visible") else "Claimed damage not observed", + "claim_status_justification": ( + "Damage visible in image" + if result.get("damage_visible") + else "Claimed damage not observed" + ), "supporting_image_ids": Path(str(image_path or "")).stem, "valid_image": result.get("valid_image", False), "severity": result.get("severity", "unknown"), @@ -2198,7 +2284,9 @@ def set_workflow_step(step_name, progress, message=None): st.session_state["workflow_message"] = message or f"{step_name}..." -def complete_workflow(message="The AI evidence review workflow completed successfully."): +def complete_workflow( + message="The AI evidence review workflow completed successfully.", +): for step in st.session_state["workflow_steps"]: step["status"] = "completed" st.session_state["analysis_running"] = False @@ -2220,8 +2308,12 @@ def render_workflow_card(message=None): steps = st.session_state["workflow_steps"] progress = st.session_state["workflow_progress"] completed = st.session_state["workflow_completed"] - active_step = next((step["label"] for step in steps if step.get("status") == "running"), None) - display_message = message or st.session_state.get("workflow_message", "AI analysis workflow status.") + active_step = next( + (step["label"] for step in steps if step.get("status") == "running"), None + ) + display_message = message or st.session_state.get( + "workflow_message", "AI analysis workflow status." + ) status_items = [] for index, step in enumerate(steps, start=1): @@ -2240,7 +2332,7 @@ def render_workflow_card(message=None): status_text = "Waiting" status_items.append( - f"
" + f'
' f"
" f"
{icon}
" f"
{escape(step.get('label', 'Step'))}
" @@ -2257,9 +2349,14 @@ def render_workflow_card(message=None): "
" ) - active_copy = f"
Current step: {escape(active_step)}
" if active_step else "" + active_copy = ( + f"
Current step: {escape(active_step)}
" + if active_step + else "" + ) fill_state = "running" if st.session_state.get("analysis_running") else "" - html_content = textwrap.dedent(f""" + html_content = textwrap.dedent( + f"""
{status_banner}
@@ -2276,7 +2373,8 @@ def render_workflow_card(message=None): {''.join(status_items)}
- """) + """ + ) st.markdown(html_content, unsafe_allow_html=True) @@ -2297,7 +2395,9 @@ def render_workflow_history(): render_workflow_card() -def render_legacy_ai_progress_panel(message, current_step=None, progress=0, completed=False): +def render_legacy_ai_progress_panel( + message, current_step=None, progress=0, completed=False +): step_names = [ "Uploading Evidence", "Extracting Claim", @@ -2324,7 +2424,7 @@ def render_legacy_ai_progress_panel(message, current_step=None, progress=0, comp status_text = "Pending" status_items.append( - f"
" + f'
' f"
" f"
{icon}
" f"
{escape(name)}
" @@ -2341,7 +2441,8 @@ def render_legacy_ai_progress_panel(message, current_step=None, progress=0, comp "
" ) - html_content = textwrap.dedent(f""" + html_content = textwrap.dedent( + f"""
{status_banner}
@@ -2355,7 +2456,8 @@ def render_legacy_ai_progress_panel(message, current_step=None, progress=0, comp {''.join(status_items)}
- """) + """ + ) st.markdown(html_content, unsafe_allow_html=True) @@ -2382,7 +2484,7 @@ def render_legacy_workflow_history(): status_text = "Pending" status_items.append( - f"
" + f'
' f"
" f"
{icon}
" f"
{escape(step.get('label', 'Step'))}
" @@ -2399,7 +2501,8 @@ def render_legacy_workflow_history(): "
" ) - html_content = textwrap.dedent(f""" + html_content = textwrap.dedent( + f"""
{status_banner}
@@ -2413,7 +2516,8 @@ def render_legacy_workflow_history(): {''.join(status_items)}
- """) + """ + ) st.markdown(html_content, unsafe_allow_html=True) @@ -2868,7 +2972,9 @@ def render_metrics(result, uploaded_files, claim_object, user_claim): metric_cols = st.columns(4) with metric_cols[0]: - render_metric_card("πŸ“¦", "Object Type", str(claim_object).title(), "Selected claim category", 1) + render_metric_card( + "πŸ“¦", "Object Type", str(claim_object).title(), "Selected claim category", 1 + ) with metric_cols[1]: render_metric_card( "πŸ“·", @@ -2878,9 +2984,13 @@ def render_metrics(result, uploaded_files, claim_object, user_claim): 2, ) with metric_cols[2]: - render_metric_card("πŸ“‹", "Review Status", claim_status, "Conversation context", 3) + render_metric_card( + "πŸ“‹", "Review Status", claim_status, "Conversation context", 3 + ) with metric_cols[3]: - render_metric_card("πŸ€–", "AI Review", analyzer_status, f"Image valid: {quality_status}", 4) + render_metric_card( + "πŸ€–", "AI Review", analyzer_status, f"Image valid: {quality_status}", 4 + ) def render_claim_summary(claim_object, user_claim, result): @@ -2889,7 +2999,9 @@ def render_claim_summary(claim_object, user_claim, result): part_label = escape(str(result_value(result, "object_part", "Pending"))) severity_label = escape(str(result_value(result, "severity", "Pending"))) decision_label = escape(str(infer_claim_status(result))) - description_text = escape(user_claim if user_claim.strip() else "No claim description entered yet.") + description_text = escape( + user_claim if user_claim.strip() else "No claim description entered yet." + ) st.markdown( """ @@ -2946,7 +3058,9 @@ def render_latest_claim_summary(latest_claim): return result = latest_claim.get("result") or {} - claim_object = latest_claim.get("claim_object", result.get("object_type", "Unknown")) + claim_object = latest_claim.get( + "claim_object", result.get("object_type", "Unknown") + ) user_claim = latest_claim.get("user_claim", "") severity = result.get("severity", "unknown") confidence_score = result.get("confidence_score", 0) @@ -3012,22 +3126,32 @@ def render_dashboard_insights(stats, claim_object=None, user_claim=None, result= chart_cols = st.columns(2) with chart_cols[0]: st.markdown("### Severity distribution") - severity_names = list(stats['severity'].keys()) - severity_values = list(stats['severity'].values()) + severity_names = list(stats["severity"].keys()) + severity_values = list(stats["severity"].values()) if severity_names: fig = px.pie(values=severity_values, names=severity_names, hole=0.5) - fig.update_layout(height=320, margin=dict(l=0, r=0, t=20, b=0), paper_bgcolor='rgba(0,0,0,0)', plot_bgcolor='rgba(0,0,0,0)') + fig.update_layout( + height=320, + margin=dict(l=0, r=0, t=20, b=0), + paper_bgcolor="rgba(0,0,0,0)", + plot_bgcolor="rgba(0,0,0,0)", + ) st.plotly_chart(fig, use_container_width=True) else: st.info("No analyzed severity data available.") with chart_cols[1]: st.markdown("### Claims by status") - status_names = list(stats['status'].keys()) - status_values = list(stats['status'].values()) + status_names = list(stats["status"].keys()) + status_values = list(stats["status"].values()) if status_names: fig = px.bar(x=status_names, y=status_values, text=status_values) - fig.update_layout(height=320, margin=dict(l=0, r=0, t=20, b=0), paper_bgcolor='rgba(0,0,0,0)', plot_bgcolor='rgba(0,0,0,0)') - fig.update_traces(marker_color='#38bdf8') + fig.update_layout( + height=320, + margin=dict(l=0, r=0, t=20, b=0), + paper_bgcolor="rgba(0,0,0,0)", + plot_bgcolor="rgba(0,0,0,0)", + ) + fig.update_traces(marker_color="#38bdf8") st.plotly_chart(fig, use_container_width=True) else: st.info("No analyzed status data available.") @@ -3058,7 +3182,7 @@ def render_image_panel(uploaded_files, latest_image_path=None): thumb_html = ( '
' f'' - '
' + "
" ) st.markdown(thumb_html, unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) @@ -3116,17 +3240,26 @@ def render_analysis_panel(result, claim_object=None, final_decision=None): else status_badge("No quality flags", "ok") ) - confidence_score = result.get("confidence_score", st.session_state.get("confidence_score")) + confidence_score = result.get( + "confidence_score", st.session_state.get("confidence_score") + ) risk_score = result.get("fraud_risk_score", st.session_state.get("risk_score")) risk_level = result.get("fraud_risk", st.session_state.get("risk_level")) cost_estimate = result.get("estimated_cost", st.session_state.get("cost_estimate")) - repair_estimate = result.get("repair_estimate", st.session_state.get("repair_estimate", [])) + repair_estimate = result.get( + "repair_estimate", st.session_state.get("repair_estimate", []) + ) explanation = result.get("ai_explanation", st.session_state.get("explanation")) - risk_reasons = result.get("fraud_risk_reasons", st.session_state.get("risk_reasons", [])) + risk_reasons = result.get( + "fraud_risk_reasons", st.session_state.get("risk_reasons", []) + ) if not isinstance(repair_estimate, list): repair_estimate = [str(repair_estimate)] - repair_text = "; ".join(item for item in repair_estimate if item) or "Repair inspection recommended" + repair_text = ( + "; ".join(item for item in repair_estimate if item) + or "Repair inspection recommended" + ) if confidence_score is None: confidence_score = generate_confidence_score(result) if risk_score is None: @@ -3137,7 +3270,11 @@ def render_analysis_panel(result, claim_object=None, final_decision=None): cost_estimate = estimate_repair_cost(claim_object, severity) risk_score = risk_score or 0 - risk_class = 'risk-low' if risk_score <= 40 else 'risk-medium' if risk_score <= 70 else 'risk-high' + risk_class = ( + "risk-low" + if risk_score <= 40 + else "risk-medium" if risk_score <= 70 else "risk-high" + ) st.markdown( f""" @@ -3299,7 +3436,9 @@ def main(): clear_analysis_state() st.experimental_rerun() - claim_object = st.sidebar.selectbox("Claim object", ["car", "laptop", "package"]) + claim_object = st.sidebar.selectbox( + "Claim object", ["car", "laptop", "package"] + ) user_claim = st.sidebar.text_area( "Claim description", placeholder="Example: The front bumper was scratched after delivery.", @@ -3333,7 +3472,9 @@ def main(): with left_col: render_latest_claim_summary(latest_claim) with right_col: - latest_image_path = latest_claim.get("image_path") if latest_claim else None + latest_image_path = ( + latest_claim.get("image_path") if latest_claim else None + ) render_image_panel(uploaded_files, latest_image_path=latest_image_path) if active_section == "Review Workspace": @@ -3356,7 +3497,9 @@ def main(): if not uploaded_files: st.info("Upload an evidence image before running analysis.") elif not user_claim.strip(): - st.warning("Add a claim description for a stronger reviewer workflow.") + st.warning( + "Add a claim description for a stronger reviewer workflow." + ) else: st.success("Ready for AI evidence review.") st.markdown( @@ -3380,7 +3523,9 @@ def main(): status_slot = st.empty() loader_slot = st.empty() progress_slot = st.empty() - status_slot.markdown(verification_badge("Processing"), unsafe_allow_html=True) + status_slot.markdown( + verification_badge("Processing"), unsafe_allow_html=True + ) with loader_slot.container(): render_workflow_card() progress_bar = run_verification_progress(progress_slot, loader_slot) @@ -3396,9 +3541,13 @@ def main(): user_claim=user_claim, ) store_analysis_summary(completed_result) - remember_latest_claim(claim_object, user_claim, saved_image_path, completed_result) + remember_latest_claim( + claim_object, user_claim, saved_image_path, completed_result + ) append_analysis_to_output_csv( - build_output_record(claim_object, user_claim, saved_image_path, completed_result) + build_output_record( + claim_object, user_claim, saved_image_path, completed_result + ) ) st.session_state["analysis_completed"] = True @@ -3450,7 +3599,9 @@ def main(): st.session_state["report_error"] = str(exc) st.session_state["report_success"] = False progress_bar.progress(100, text="Analysis Complete") - complete_workflow("AI analysis completed, but the PDF report could not be generated.") + complete_workflow( + "AI analysis completed, but the PDF report could not be generated." + ) st.rerun() if st.session_state["analysis_result"]: @@ -3468,7 +3619,11 @@ def main(): if active_section == "Report": render_claim_summary(claim_object, user_claim, result) - render_analysis_panel(result, claim_object=claim_object, final_decision=infer_claim_status(result)) + render_analysis_panel( + result, + claim_object=claim_object, + final_decision=infer_claim_status(result), + ) if st.session_state["analysis_completed"]: st.markdown("### Export") @@ -3476,7 +3631,9 @@ def main(): st.success("PDF report generated successfully.") render_pdf_workflow() if st.session_state["report_error"]: - st.warning(f"PDF report could not be generated: {st.session_state['report_error']}") + st.warning( + f"PDF report could not be generated: {st.session_state['report_error']}" + ) pdf_path = st.session_state["pdf_report_path"] if pdf_path and os.path.exists(pdf_path): diff --git a/code/evaluation/main.py b/code/evaluation/main.py index 7c76736..12b38bf 100644 --- a/code/evaluation/main.py +++ b/code/evaluation/main.py @@ -1,7 +1,8 @@ # evaluation/main.py + def run_evaluation(): """ Future model evaluation module. """ - pass \ No newline at end of file + pass diff --git a/code/main.py b/code/main.py index 64b708c..e678d78 100644 --- a/code/main.py +++ b/code/main.py @@ -13,75 +13,55 @@ results = [] for _, row in claims.head(5).iterrows(): - - history_row = history_df[ - history_df["user_id"] == row["user_id"] - ] + + history_row = history_df[history_df["user_id"] == row["user_id"]] risk_flags = history_row.iloc[0]["history_flags"] - claim_info = extract_claim( - row["user_claim"] - ) + claim_info = extract_claim(row["user_claim"]) image_path = row["image_paths"].split(";")[0] - - image_id = os.path.splitext( - os.path.basename(image_path) - )[0] + + image_id = os.path.splitext(os.path.basename(image_path))[0] full_path = "../dataset/" + image_path - + if not os.path.exists(full_path): continue - image_result = analyze_image( - full_path, - row["claim_object"] - ) + image_result = analyze_image(full_path, row["claim_object"]) time.sleep(15) evidence_met, evidence_reason = check_evidence( - image_result["damage_visible"], - image_result["valid_image"] + image_result["damage_visible"], image_result["valid_image"] ) claim_status, justification = decide_claim( - evidence_met, - image_result["damage_visible"] + evidence_met, image_result["damage_visible"] ) - results.append({ - "user_id": row["user_id"], - "image_paths": row["image_paths"], - - "user_claim": row["user_claim"], - "claim_object": row["claim_object"], - - "evidence_standard_met": evidence_met, - "evidence_standard_met_reason": evidence_reason, - - "risk_flags": risk_flags, - - "issue_type": image_result["issue_type"], - "object_part": image_result["object_part"], - - "claim_status": claim_status, - "claim_status_justification": justification, - - "supporting_image_ids": image_id, - - "valid_image": image_result["valid_image"], - - "severity": image_result["severity"] - }) + results.append( + { + "user_id": row["user_id"], + "image_paths": row["image_paths"], + "user_claim": row["user_claim"], + "claim_object": row["claim_object"], + "evidence_standard_met": evidence_met, + "evidence_standard_met_reason": evidence_reason, + "risk_flags": risk_flags, + "issue_type": image_result["issue_type"], + "object_part": image_result["object_part"], + "claim_status": claim_status, + "claim_status_justification": justification, + "supporting_image_ids": image_id, + "valid_image": image_result["valid_image"], + "severity": image_result["severity"], + } + ) output_df = pd.DataFrame(results) -output_df.to_csv( - "output.csv", - index=False -) +output_df.to_csv("output.csv", index=False) print("output.csv generated successfully") -print("Rows:", len(output_df)) \ No newline at end of file +print("Rows:", len(output_df)) diff --git a/code/report_generator.py b/code/report_generator.py index 9a25cf3..64a6e7e 100644 --- a/code/report_generator.py +++ b/code/report_generator.py @@ -164,7 +164,10 @@ def generate_pdf_report( REPORTS_DIR.mkdir(parents=True, exist_ok=True) timestamp = datetime.now() - pdf_path = REPORTS_DIR / f"claim_report_{timestamp.strftime('%Y%m%d_%H%M%S')}_{uuid4().hex[:8]}.pdf" + pdf_path = ( + REPORTS_DIR + / f"claim_report_{timestamp.strftime('%Y%m%d_%H%M%S')}_{uuid4().hex[:8]}.pdf" + ) styles = _build_styles() doc = SimpleDocTemplate( @@ -203,10 +206,27 @@ def generate_pdf_report( _key_value_table( [ ("Claim Object", claim_object), - ("Claim Description", claim_description or "No claim description provided."), + ( + "Claim Description", + claim_description or "No claim description provided.", + ), ("Claim Status", claim_status or "Not assigned"), - ("AI Confidence", f"{confidence_score}%" if confidence_score is not None else "Not available"), - ("Fraud Risk Score", fraud_risk_score if fraud_risk_score is not None else "Not available"), + ( + "AI Confidence", + ( + f"{confidence_score}%" + if confidence_score is not None + else "Not available" + ), + ), + ( + "Fraud Risk Score", + ( + fraud_risk_score + if fraud_risk_score is not None + else "Not available" + ), + ), ("Risk Level", risk_level or "Not available"), ("Estimated Repair Cost", estimated_repair_cost or "Not available"), ], @@ -219,7 +239,12 @@ def generate_pdf_report( if embedded_image: story.extend([embedded_image, Spacer(1, 8)]) else: - story.append(Paragraph("Uploaded image could not be embedded in the PDF.", styles["BodyTextWrapped"])) + story.append( + Paragraph( + "Uploaded image could not be embedded in the PDF.", + styles["BodyTextWrapped"], + ) + ) story.extend( [ @@ -237,7 +262,10 @@ def generate_pdf_report( styles, ), Paragraph("AI Explanation", styles["SectionHeading"]), - Paragraph(ai_explanation or "No additional explanation provided.", styles["BodyTextWrapped"]), + Paragraph( + ai_explanation or "No additional explanation provided.", + styles["BodyTextWrapped"], + ), Paragraph("Final AI Assessment", styles["SectionHeading"]), Paragraph(final_assessment, styles["BodyTextWrapped"]), ] diff --git a/code/src/claim_extractor.py b/code/src/claim_extractor.py index b3133e3..1280760 100644 --- a/code/src/claim_extractor.py +++ b/code/src/claim_extractor.py @@ -1,31 +1,22 @@ - ISSUE_MAP = { "dent": "dent", "dented": "dent", - "scratch": "scratch", "scratched": "scratch", - "crack": "crack", "cracked": "crack", - "shatter": "glass_shatter", "shattered": "glass_shatter", - "broken": "broken_part", "damage": "broken_part", "damaged": "broken_part", "affected": "broken_part", - "missing": "missing_part", - "water": "water_damage", "wet": "water_damage", - "stain": "stain", - "torn": "torn_packaging", - "crushed": "crushed_packaging" + "crushed": "crushed_packaging", } PARTS = [ @@ -47,27 +38,19 @@ "corner", "seal", "label", - "contents" + "contents", ] + def extract_claim(claim_text): text = claim_text.lower() - - # Normalize common compound words - text = text.replace( - "frontbumper", - "front bumper" - ) - text = text.replace( - "rearbumper", - "rear bumper" - ) + # Normalize common compound words + text = text.replace("frontbumper", "front bumper") + + text = text.replace("rearbumper", "rear bumper") - text = text.replace( - "sidemirror", - "side mirror" - ) + text = text.replace("sidemirror", "side mirror") issue_type = "unknown" object_part = "unknown" @@ -82,7 +65,4 @@ def extract_claim(claim_text): object_part = part.replace(" ", "_") break - return { - "issue_type": issue_type, - "object_part": object_part - } \ No newline at end of file + return {"issue_type": issue_type, "object_part": object_part} diff --git a/code/src/decision_engine.py b/code/src/decision_engine.py index fc0f710..2bfdc9c 100644 --- a/code/src/decision_engine.py +++ b/code/src/decision_engine.py @@ -1,20 +1,8 @@ -def decide_claim( - evidence_met, - damage_visible -): +def decide_claim(evidence_met, damage_visible): if not evidence_met: - return ( - "not_enough_information", - "Insufficient visual evidence" - ) + return ("not_enough_information", "Insufficient visual evidence") if damage_visible: - return ( - "supported", - "Damage visible in image" - ) + return ("supported", "Damage visible in image") - return ( - "contradicted", - "Claimed damage not observed" - ) \ No newline at end of file + return ("contradicted", "Claimed damage not observed") diff --git a/code/src/evidence_checker.py b/code/src/evidence_checker.py index 4c76896..771b151 100644 --- a/code/src/evidence_checker.py +++ b/code/src/evidence_checker.py @@ -1,20 +1,8 @@ -def check_evidence( - damage_visible, - valid_image -): +def check_evidence(damage_visible, valid_image): if not valid_image: - return ( - False, - "Image not usable for review" - ) + return (False, "Image not usable for review") if not damage_visible: - return ( - False, - "Claimed damage not visible" - ) + return (False, "Claimed damage not visible") - return ( - True, - "Sufficient visual evidence" - ) \ No newline at end of file + return (True, "Sufficient visual evidence") diff --git a/code/src/history_checker.py b/code/src/history_checker.py index d156ffa..e4d0aef 100644 --- a/code/src/history_checker.py +++ b/code/src/history_checker.py @@ -1,12 +1,7 @@ def get_history_flags(history_row): - history_flags = str( - history_row["history_flags"] - ).strip() + history_flags = str(history_row["history_flags"]).strip() if history_flags.lower() == "none": return ["none"] - return [ - flag.strip() - for flag in history_flags.split(";") - ] \ No newline at end of file + return [flag.strip() for flag in history_flags.split(";")] diff --git a/code/src/image_analyzer.py b/code/src/image_analyzer.py index 1f1d77f..98fe955 100644 --- a/code/src/image_analyzer.py +++ b/code/src/image_analyzer.py @@ -6,13 +6,9 @@ load_dotenv() -genai.configure( - api_key=os.getenv("GEMINI_API_KEY") -) +genai.configure(api_key=os.getenv("GEMINI_API_KEY")) -model = genai.GenerativeModel( - "gemini-2.5-flash" -) +model = genai.GenerativeModel("gemini-2.5-flash") def _estimate_repair_cost(claim_object, severity): @@ -116,20 +112,32 @@ def _apply_result_contract(result, claim_object): result.setdefault("severity", "unknown") result.setdefault("valid_image", False) result.setdefault("quality_flags", []) - result["confidence_score"] = result.get("confidence_score") or _confidence_score(result) + result["confidence_score"] = result.get("confidence_score") or _confidence_score( + result + ) fraud_risk = result.get("fraud_risk") if not fraud_risk or str(fraud_risk).strip().lower() == "unknown": fraud_risk = "Low" result["fraud_risk"] = fraud_risk result["fraud_risk_score"] = result.get("fraud_risk_score") or 24 repair_estimate = result.get("repair_estimate") - if not repair_estimate or str(repair_estimate).strip().lower() in {"none", "unknown", "--"}: + if not repair_estimate or str(repair_estimate).strip().lower() in { + "none", + "unknown", + "--", + }: repair_estimate = _repair_guidance(claim_object, result) result["repair_estimate"] = repair_estimate estimated_cost = result.get("estimated_cost") - if not estimated_cost or str(estimated_cost).strip().lower() in {"none", "unknown", "--"}: - estimated_cost = _estimate_repair_cost(result.get("object_type"), result.get("severity")) + if not estimated_cost or str(estimated_cost).strip().lower() in { + "none", + "unknown", + "--", + }: + estimated_cost = _estimate_repair_cost( + result.get("object_type"), result.get("severity") + ) result["estimated_cost"] = estimated_cost result["estimated_repair_cost"] = result["estimated_cost"] return result @@ -195,16 +203,9 @@ def analyze_image(image_path, claim_object): """ try: - response = model.generate_content( - [prompt, image] - ) + response = model.generate_content([prompt, image]) - cleaned = ( - response.text - .replace("```json", "") - .replace("```", "") - .strip() - ) + cleaned = response.text.replace("```json", "").replace("```", "").strip() return _apply_result_contract(json.loads(cleaned), claim_object) @@ -212,14 +213,15 @@ def analyze_image(image_path, claim_object): print(f"Gemini Error: {e}") - return _apply_result_contract({ - "object_type": claim_object, - "issue_type": "unknown", - "object_part": "unknown", - "damage_visible": False, - "severity": "unknown", - "valid_image": False, - "quality_flags": [ - "manual_review_required" - ] - }, claim_object) + return _apply_result_contract( + { + "object_type": claim_object, + "issue_type": "unknown", + "object_part": "unknown", + "damage_visible": False, + "severity": "unknown", + "valid_image": False, + "quality_flags": ["manual_review_required"], + }, + claim_object, + ) diff --git a/code/test_gemini.py b/code/test_gemini.py index a6996f3..8a92975 100644 --- a/code/test_gemini.py +++ b/code/test_gemini.py @@ -1,8 +1,5 @@ from src.image_analyzer import analyze_image -result = analyze_image( - "../dataset/images/sample/case_001/img_1.jpg", - "car" -) +result = analyze_image("../dataset/images/sample/case_001/img_1.jpg", "car") -print(result) \ No newline at end of file +print(result) diff --git a/code/utils/data_utils.py b/code/utils/data_utils.py index ceb8010..81fd4ce 100644 --- a/code/utils/data_utils.py +++ b/code/utils/data_utils.py @@ -43,7 +43,11 @@ def column_summary(df: pd.DataFrame) -> pd.DataFrame: missing = int(series.isna().sum()) unique = int(series.nunique(dropna=True)) top = series.mode().iloc[0] if not series.mode().empty else "N/A" - freq = int(series.value_counts(dropna=True).iloc[0]) if not series.value_counts(dropna=True).empty else 0 + freq = ( + int(series.value_counts(dropna=True).iloc[0]) + if not series.value_counts(dropna=True).empty + else 0 + ) summary.append( { "column": column, @@ -58,7 +62,9 @@ def column_summary(df: pd.DataFrame) -> pd.DataFrame: return pd.DataFrame(summary) -def clean_dataset(df: pd.DataFrame, drop_duplicates: bool = True, fill_method: str | None = None) -> pd.DataFrame: +def clean_dataset( + df: pd.DataFrame, drop_duplicates: bool = True, fill_method: str | None = None +) -> pd.DataFrame: clean_df = df.copy() if drop_duplicates: clean_df = clean_df.drop_duplicates() diff --git a/code/utils/ml_utils.py b/code/utils/ml_utils.py index ebcf4ec..4682055 100644 --- a/code/utils/ml_utils.py +++ b/code/utils/ml_utils.py @@ -7,7 +7,13 @@ from joblib import dump from sklearn.ensemble import GradientBoostingClassifier, RandomForestClassifier from sklearn.linear_model import LogisticRegression -from sklearn.metrics import accuracy_score, confusion_matrix, f1_score, precision_score, recall_score +from sklearn.metrics import ( + accuracy_score, + confusion_matrix, + f1_score, + precision_score, + recall_score, +) from sklearn.model_selection import GridSearchCV, train_test_split from sklearn.neighbors import KNeighborsClassifier from sklearn.preprocessing import LabelEncoder, StandardScaler @@ -16,7 +22,9 @@ def get_default_models() -> dict[str, Any]: return { - "Logistic Regression": LogisticRegression(max_iter=500, solver="lbfgs", multi_class="auto"), + "Logistic Regression": LogisticRegression( + max_iter=500, solver="lbfgs", multi_class="auto" + ), "Decision Tree": DecisionTreeClassifier(random_state=42), "Random Forest": RandomForestClassifier(random_state=42, n_estimators=100), "Gradient Boosting": GradientBoostingClassifier(random_state=42), @@ -24,19 +32,29 @@ def get_default_models() -> dict[str, Any]: } -def encode_categorical_columns(df: pd.DataFrame, categorical_columns: list[str]) -> tuple[pd.DataFrame, dict[str, LabelEncoder]]: +def encode_categorical_columns( + df: pd.DataFrame, categorical_columns: list[str] +) -> tuple[pd.DataFrame, dict[str, LabelEncoder]]: df_copy = df.copy() encoders: dict[str, LabelEncoder] = {} for column in categorical_columns: encoder = LabelEncoder() - df_copy[column] = encoder.fit_transform(df_copy[column].astype(str).fillna("Missing")) + df_copy[column] = encoder.fit_transform( + df_copy[column].astype(str).fillna("Missing") + ) encoders[column] = encoder return df_copy, encoders -def build_feature_matrix(df: pd.DataFrame, target: str, feature_columns: list[str] | None = None) -> tuple[pd.DataFrame, pd.Series]: +def build_feature_matrix( + df: pd.DataFrame, target: str, feature_columns: list[str] | None = None +) -> tuple[pd.DataFrame, pd.Series]: if feature_columns is None: - feature_columns = [col for col in df.columns if col != target and col != "user_claim" and col != "image_paths"] + feature_columns = [ + col + for col in df.columns + if col != target and col != "user_claim" and col != "image_paths" + ] X = df[feature_columns].copy() y = df[target].copy() return X, y @@ -54,13 +72,21 @@ def standardize_numeric(X: pd.DataFrame) -> tuple[pd.DataFrame, StandardScaler | def evaluate_classification(y_true, y_pred) -> dict[str, float]: return { "accuracy": float(accuracy_score(y_true, y_pred)), - "precision": float(precision_score(y_true, y_pred, average="weighted", zero_division=0)), - "recall": float(recall_score(y_true, y_pred, average="weighted", zero_division=0)), - "f1_score": float(f1_score(y_true, y_pred, average="weighted", zero_division=0)), + "precision": float( + precision_score(y_true, y_pred, average="weighted", zero_division=0) + ), + "recall": float( + recall_score(y_true, y_pred, average="weighted", zero_division=0) + ), + "f1_score": float( + f1_score(y_true, y_pred, average="weighted", zero_division=0) + ), } -def train_models(X_train, X_test, y_train, y_test, model_map: dict[str, Any]) -> list[dict[str, Any]]: +def train_models( + X_train, X_test, y_train, y_test, model_map: dict[str, Any] +) -> list[dict[str, Any]]: results = [] for name, model in model_map.items(): try: @@ -82,7 +108,13 @@ def grid_search_model(model, param_grid: dict[str, list[Any]], X_train, y_train) return None -def save_model(model, model_name: str, target: str, metrics: dict[str, float], models_dir: Path = Path("models")) -> Path: +def save_model( + model, + model_name: str, + target: str, + metrics: dict[str, float], + models_dir: Path = Path("models"), +) -> Path: models_dir.mkdir(parents=True, exist_ok=True) model_path = models_dir / f"{model_name.lower().replace(' ', '_')}_{target}.joblib" dump(model, model_path) @@ -93,7 +125,9 @@ def save_model(model, model_name: str, target: str, metrics: dict[str, float], m leaderboard = json.loads(metadata_path.read_text(encoding="utf-8")) except Exception: leaderboard = [] - leaderboard = [entry for entry in leaderboard if entry.get("model_path") != str(model_path)] + leaderboard = [ + entry for entry in leaderboard if entry.get("model_path") != str(model_path) + ] leaderboard.append( { "model_name": model_name, @@ -119,12 +153,27 @@ def load_leaderboard(models_dir: Path = Path("models")) -> list[dict[str, Any]]: def prepare_claim_features(df: pd.DataFrame) -> pd.DataFrame: df_copy = df.copy() if "image_paths" in df_copy.columns: - df_copy["image_count"] = df_copy["image_paths"].fillna("").apply(lambda value: len(str(value).split(";")) if str(value).strip() else 0) + df_copy["image_count"] = ( + df_copy["image_paths"] + .fillna("") + .apply( + lambda value: len(str(value).split(";")) if str(value).strip() else 0 + ) + ) if "user_claim" in df_copy.columns: df_copy["claim_length"] = df_copy["user_claim"].astype(str).apply(len) - df_copy["word_count"] = df_copy["user_claim"].astype(str).apply(lambda text: len(str(text).split())) - df_copy["has_question"] = df_copy["user_claim"].astype(str).str.contains(r"\?", regex=False).astype(int) + df_copy["word_count"] = ( + df_copy["user_claim"].astype(str).apply(lambda text: len(str(text).split())) + ) + df_copy["has_question"] = ( + df_copy["user_claim"] + .astype(str) + .str.contains(r"\?", regex=False) + .astype(int) + ) if "claim_object" in df_copy.columns: encoder = LabelEncoder() - df_copy["claim_object_encoded"] = encoder.fit_transform(df_copy["claim_object"].astype(str).fillna("Missing")) + df_copy["claim_object_encoded"] = encoder.fit_transform( + df_copy["claim_object"].astype(str).fillna("Missing") + ) return df_copy