diff --git a/.gitignore b/.gitignore index fadf1e5..eac39a7 100644 --- a/.gitignore +++ b/.gitignore @@ -22,6 +22,9 @@ temp.jpg # Streamlit cache .streamlit/secrets.toml +# Auth credentials (generated at runtime) +code/auth.yaml + # OS files .DS_Store Thumbs.db diff --git a/code/app.py b/code/app.py index 89db3ae..2655294 100644 --- a/code/app.py +++ b/code/app.py @@ -18,6 +18,16 @@ from report_generator import generate_pdf_report from src.image_analyzer import analyze_image +import yaml +from dotenv import load_dotenv + +load_dotenv() + +try: + import streamlit_authenticator as stauth +except ImportError: + stauth = None + APP_TITLE = "ClaimVision AI" APP_SUBTITLE = "AI-Powered Damage Claim Verification" @@ -25,6 +35,64 @@ LOGO_PATH = REPO_ROOT / "assets" / "logo.png" TEMP_IMAGE_PATH = Path("temp.jpg") OUTPUT_CSV_PATH = REPO_ROOT / "dataset" / "output.csv" +MAX_FILE_SIZE = 25 * 1024 * 1024 +ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png"} +AUTH_CONFIG_PATH = Path(__file__).parent / "auth.yaml" + + +def ensure_auth_config(): + if not AUTH_CONFIG_PATH.exists(): + os.makedirs(AUTH_CONFIG_PATH.parent, exist_ok=True) + cookie_key = os.getenv("CLAIMVISION_AUTH_KEY", "claimvision_secret_key_change_me_in_production") + default = { + "credentials": { + "usernames": { + "admin": { + "email": "admin@claimvision.ai", + "name": "Admin", + "password": "$2b$12$rGtJKo3o4t5EtOPW9LTbKu34HLnk5F5EJjy/RPv/BWT6eRmGMDgAG", + } + } + }, + "cookie": { + "name": "claimvision_auth", + "key": cookie_key, + "expiry_days": 30, + }, + } + with open(AUTH_CONFIG_PATH, "w") as f: + yaml.dump(default, f, default_flow_style=False) + return AUTH_CONFIG_PATH + + +def load_auth_config(): + ensure_auth_config() + with open(AUTH_CONFIG_PATH) as f: + return yaml.safe_load(f) + + +def save_auth_config(config): + with open(AUTH_CONFIG_PATH, "w") as f: + yaml.dump(config, f, default_flow_style=False) + + +def validate_uploaded_file(uploaded_file): + errors = [] + if uploaded_file is None: + errors.append("No file was uploaded.") + return errors + + ext = Path(uploaded_file.name).suffix.lower() + if ext not in ALLOWED_EXTENSIONS: + errors.append(f"Unsupported file type '{ext}'. Only JPG, JPEG, and PNG images are allowed.") + + file_size = len(uploaded_file.getbuffer()) + if file_size == 0: + errors.append("The uploaded file is empty.") + elif file_size > MAX_FILE_SIZE: + errors.append(f"File size exceeds {MAX_FILE_SIZE // (1024 * 1024)} MB.") + + return errors def get_logo_image(): @@ -3242,6 +3310,9 @@ def create_pdf_report( def save_uploaded_image(uploaded_file): + errors = validate_uploaded_file(uploaded_file) + if errors: + raise ValueError("; ".join(errors)) TEMP_IMAGE_PATH.write_bytes(uploaded_file.getbuffer()) return str(TEMP_IMAGE_PATH) @@ -3251,43 +3322,102 @@ def main(): inject_css() render_chatbot_widget() - if "analysis_result" not in st.session_state: - st.session_state["analysis_result"] = None - if "analysis_completed" not in st.session_state: - st.session_state["analysis_completed"] = False - if "pdf_report_path" not in st.session_state: - st.session_state["pdf_report_path"] = None - if "report_error" not in st.session_state: - st.session_state["report_error"] = None - if "report_success" not in st.session_state: - st.session_state["report_success"] = False - if "confidence_score" not in st.session_state: - st.session_state["confidence_score"] = None - if "risk_score" not in st.session_state: - st.session_state["risk_score"] = None - if "risk_level" not in st.session_state: - st.session_state["risk_level"] = None - if "risk_reasons" not in st.session_state: - st.session_state["risk_reasons"] = [] - if "explanation" not in st.session_state: - st.session_state["explanation"] = None - if "cost_estimate" not in st.session_state: - st.session_state["cost_estimate"] = None - if "claim_history" not in st.session_state: - st.session_state["claim_history"] = None - if "uploaded_images" not in st.session_state: - st.session_state["uploaded_images"] = [] - if "latest_claim" not in st.session_state: - st.session_state["latest_claim"] = None - if "analysis_stage" not in st.session_state: - st.session_state["analysis_stage"] = "" - if "analysis_progress" not in st.session_state: - st.session_state["analysis_progress"] = 0 - if "analysis_status" not in st.session_state: - st.session_state["analysis_status"] = "Idle" + # --- AUTHENTICATION --- + if stauth is None: + st.error( + "**streamlit-authenticator** is required. " + "Install it with: `pip install streamlit-authenticator`" + ) + return + + if "auth_mode" not in st.session_state: + st.session_state["auth_mode"] = "login" + + config = load_auth_config() + authenticator = stauth.Authenticate( + config["credentials"], + config["cookie"]["name"], + config["cookie"]["key"], + config["cookie"]["expiry_days"], + ) + + if st.session_state["auth_mode"] == "register": + try: + reg_result = authenticator.register_user( + location="main", captcha=False, key="register_user" + ) + if reg_result: + status, username, name = reg_result + if status == "New user registered": + st.success( + f"Account created for **{username}**! You can now log in." + ) + save_auth_config(config) + st.session_state["auth_mode"] = "login" + st.rerun() + elif status: + st.info(status) + except Exception as e: + st.error(str(e)) + + if st.button("← Back to login", key="back_to_login_btn"): + st.session_state["auth_mode"] = "login" + st.rerun() + return + + auth_result = authenticator.login(location="main", key="login") + if auth_result: + name, authentication_status, username = auth_result + st.session_state["auth_name"] = name + st.session_state["auth_username"] = username + st.session_state["authentication_status"] = authentication_status + else: + authentication_status = st.session_state.get("authentication_status") + + if authentication_status is None: + st.markdown("---") + col1, col2 = st.columns([3, 1]) + with col1: + st.markdown(f"### Welcome to {APP_TITLE}") + st.markdown(f"*{APP_SUBTITLE}*") + with col2: + if st.button("Register new account", key="goto_register_btn"): + st.session_state["auth_mode"] = "register" + st.rerun() + return + elif not authentication_status: + return + + # --- SESSION STATE INIT --- + init_state = [ + ("analysis_result", None), + ("analysis_completed", False), + ("pdf_report_path", None), + ("report_error", None), + ("report_success", False), + ("confidence_score", None), + ("risk_score", None), + ("risk_level", None), + ("risk_reasons", []), + ("explanation", None), + ("cost_estimate", None), + ("claim_history", None), + ("uploaded_images", []), + ("latest_claim", None), + ("analysis_stage", ""), + ("analysis_progress", 0), + ("analysis_status", "Idle"), + ("analyzing", False), + ] + for key, default in init_state: + if key not in st.session_state: + st.session_state[key] = default init_workflow_state() active_section = render_sidebar() + + st.sidebar.markdown(f"👤 **{st.session_state['auth_name']}**") + authenticator.logout("Logout", "sidebar", key="logout_btn") if active_section in ("Overview", "Review Workspace", "AI Analysis", "Report"): render_mobile_brand_bar() render_hero() @@ -3347,9 +3477,10 @@ def main(): st.markdown("### Review Actions") action_cols = st.columns([0.35, 0.65]) with action_cols[0]: + is_analyzing = st.session_state["analyzing"] analyze_clicked = st.button( - "Run AI Analysis", - disabled=not uploaded_files, + "Analyzing image..." if is_analyzing else "Run AI Analysis", + disabled=not uploaded_files or is_analyzing, use_container_width=True, ) with action_cols[1]: @@ -3370,23 +3501,29 @@ def main(): ) if analyze_clicked and uploaded_files: - saved_image_path = save_uploaded_image(uploaded_files[0]) - start_workflow() - st.session_state["analysis_completed"] = False - st.session_state["pdf_report_path"] = None - st.session_state["report_error"] = None - st.session_state["report_success"] = False - - status_slot = st.empty() - loader_slot = st.empty() - progress_slot = st.empty() - 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) - - with st.spinner("Gemini is reviewing the uploaded evidence..."): - raw_result = analyze_image(saved_image_path, claim_object) + file_errors = validate_uploaded_file(uploaded_files[0]) + if file_errors: + for err in file_errors: + st.error(err) + else: + st.session_state["analyzing"] = True + saved_image_path = save_uploaded_image(uploaded_files[0]) + start_workflow() + st.session_state["analysis_completed"] = False + st.session_state["pdf_report_path"] = None + st.session_state["report_error"] = None + st.session_state["report_success"] = False + + status_slot = st.empty() + loader_slot = st.empty() + progress_slot = st.empty() + 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) + + with st.spinner("Analyzing image..."): + raw_result = analyze_image(saved_image_path, claim_object) claim_history = load_user_history(claim_object) st.session_state["claim_history"] = claim_history completed_result = ensure_analysis_result_contract( @@ -3401,6 +3538,7 @@ def main(): build_output_record(claim_object, user_claim, saved_image_path, completed_result) ) st.session_state["analysis_completed"] = True + st.session_state["analyzing"] = False if st.session_state["analysis_completed"]: try: @@ -3451,6 +3589,7 @@ def main(): 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.") + st.session_state["analyzing"] = False st.rerun() if st.session_state["analysis_result"]: diff --git a/code/requirements.txt b/code/requirements.txt index 5810805..bc47c95 100644 --- a/code/requirements.txt +++ b/code/requirements.txt @@ -12,3 +12,5 @@ shap opencv-python ruff black +streamlit-authenticator +pyyaml diff --git a/code/src/image_analyzer.py b/code/src/image_analyzer.py index 1f1d77f..2f7b9c7 100644 --- a/code/src/image_analyzer.py +++ b/code/src/image_analyzer.py @@ -14,6 +14,29 @@ "gemini-2.5-flash" ) +ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png"} +MAX_FILE_SIZE = 25 * 1024 * 1024 + + +def validate_image_file(image_path): + ext = os.path.splitext(str(image_path))[1].lower() + if ext not in ALLOWED_EXTENSIONS: + return False, f"Unsupported file type '{ext}'. Only JPG, JPEG, and PNG images are allowed." + + file_size = os.path.getsize(image_path) + if file_size == 0: + return False, "The uploaded file is empty." + if file_size > MAX_FILE_SIZE: + return False, f"File size exceeds {MAX_FILE_SIZE // (1024 * 1024)} MB." + + try: + with Image.open(image_path) as img: + img.verify() + except Exception: + return False, "The image appears to be corrupted or unreadable." + + return True, "" + def _estimate_repair_cost(claim_object, severity): severity = str(severity or "unknown").lower() @@ -137,6 +160,18 @@ def _apply_result_contract(result, claim_object): def analyze_image(image_path, claim_object): + is_valid, error = validate_image_file(image_path) + if not is_valid: + return _apply_result_contract({ + "object_type": claim_object, + "issue_type": "unknown", + "object_part": "unknown", + "damage_visible": False, + "severity": "unknown", + "valid_image": False, + "quality_flags": [f"validation_error: {error}"] + }, claim_object) + image = Image.open(image_path) prompt = f"""