Files
doczyai-pipelines/archive/streamlit/multipage/util.py
T
Katon Minhas afb6d5185d Merged in feature/lesser-table-caching-refactor-hybrid (pull request #847)
Feature/lesser table caching refactor hybrid

* chore: Remove unused duplicate main.py from shared pipeline

* fix: Correct crosswalk paths in aarete_derived.py

* chore: Remove unused documentation files from fieldExtraction

* docs: Add documentation files to documentation folder

* docs: Update README with uv setup, expanded project structure, and branching conventions

* docs: Add uv installation steps with Ubuntu/WSL emphasis

* Enable prompt caching for all remaining LLM calls

- Add _INSTRUCTION() functions for: EXHIBIT_HEADER, EXHIBIT_LINKAGE,
  EXHIBIT_TITLE_MATCH, DATE_FIX, DERIVED_TERM_DATE, CHECK_PROVIDER_NAME_MATCH,
  SPECIAL_CASE_ASSIGNMENT
- Update all invoke_claude() calls in saas and clover pipelines to use
  cache=True with corresponding _INSTRUCTION() functions
- Add new instructions to get_cacheable_instructions() for cache warming
- Update tests for new instruction functions

Functions now using caching:
- prompt_exhibit_level
- prompt_exhibit_lesser (EXHIBIT_LEVEL_LESSER_OF)
- prompt_fee_schedule_breakout
- prompt_grouper_breakout
- prompt_special_case_assignment
- prompt_exhibit_linkage
- prompt_exhibit_header
- prompt_smart_chunked (ONE_TO_ONE templates)
- prompt_date_fix
- prompt_derived_term_date
- prompt_exhibit_title_match
- provider_name_match_check

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Reorder

* feat: Add bcbs_promise client pipeline with OFFSET_TERM extraction

- Add new bcbs_promise client with HSC-based OFFSET_TERM field extraction
- Extract full paragraph text of offset/recoupment provisions from contracts
- Derive OFFSET_INDICATOR (Y/N) from OFFSET_TERM presence
- Fix reorder_columns to preserve extra columns not in COLUMN_ORDER
- Update QC/QA output path to outputs/qc_qa/

* fix: Update dev deps and test assertions for QC/QA output path

- Add pytest/pytest-mock to dev dependencies for mypy type checking
- Update test assertions to expect outputs/qc_qa instead of qa_qc_output

* style: Apply black formatting to prompt_templates.py

* Merge main, move scripts

* Archive some scripts

* update py version

* remove .py version file

* Remove ASCII characters

* Restore testbed code

* restore tracking

* Update testbed metrics

* Enable prompt caching for CODE_LAST_CHECK, FILL_BILL_TYPE, DUAL_LOB_CHECK, and GROUPER_BREAKOUT

- Add CODE_LAST_CHECK_INSTRUCTION() for service specificity classification
- Add FILL_BILL_TYPE_INSTRUCTION() for bill type code determination
- Add DUAL_LOB_CHECK_INSTRUCTION() for Medicare/Medicaid classification
- Update code_funcs.py to use caching for CODE_LAST_CHECK, FILL_BILL_TYPE, GROUPER_BREAKOUT
- Update postprocessing_funcs.py to use caching for DUAL_LOB_CHECK
- Add new instructions to get_cacheable_instructions() for cache warming
- Add unit tests for new instruction functions

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* Fix postprocessing_funcs to remove invalid columns

* Merge branch 'main' into feature/lesser-table-caching-refactor-hybrid

* Revert prompt caching changes from aed1b73c

* update formatting

* Update imports


Approved-by: Sha Brown
Approved-by: Praneel Panchigar
2026-01-26 16:52:55 +00:00

84 lines
2.6 KiB
Python

import streamlit as st
import security
import os
import constants
import logging
from logging.handlers import RotatingFileHandler
# Ensure the log directory exists
log_dir = "/home/ubuntu/doczy.ai/streamlit"
if not os.path.exists(log_dir):
os.makedirs(log_dir, exist_ok=True)
# Configure logging with RotatingFileHandler
log_file = f"{log_dir}/interface.log"
rotating_handler = RotatingFileHandler(
log_file,
maxBytes=10 * 1024 * 1024, # 10 MB
backupCount=5, # Keep up to 5 backup files
)
rotating_handler.setLevel(logging.INFO)
formatter = logging.Formatter("%(asctime)s %(levelname)s [%(filename)s] %(message)s")
rotating_handler.setFormatter(formatter)
logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(rotating_handler)
def setup_page(redirect_uril):
# st.set_page_config(
# page_title=page_title,
# page_icon="👋",
# )
if st.query_params.get("code"):
security.handle_redirect(redirect_uril)
access_token = st.session_state.get("access_token")
if access_token:
user_info = security.get_user_info(access_token)
st.session_state["user_info"] = user_info
return True
else:
st.write("Please sign-in to use this app.")
auth_url = security.get_auth_url(redirect_uril)
st.markdown(
f"<a href='{auth_url}' target='_self'>Sign In</a>", unsafe_allow_html=True
)
st.stop()
def load_page_details(interface):
env_var = os.environ.get("ENVIRONMENT", "DEV")
logger.info(f"env_var={env_var}")
if env_var == "UAT":
logger.info(constants.DOCZY_REDIRECT_URL_UAT + str(interface))
return (
constants.DOCZY_REDIRECT_URL_UAT + str(interface),
constants.DOCZY_CREATE_BATCH_URL_UAT,
constants.DOCZY_PIPELINE_URL_UAT,
)
elif env_var == "DEV":
logger.info(constants.DOCZY_REDIRECT_URL_DEV + str(interface))
return (
constants.DOCZY_REDIRECT_URL_DEV + str(interface),
constants.DOCZY_CREATE_BATCH_URL_DEV,
constants.DOCZY_PIPELINE_URL_DEV,
)
elif env_var == "PROD":
logger.info(constants.DOCZY_REDIRECT_URL_PROD + str(interface))
return (
constants.DOCZY_REDIRECT_URL_PROD + str(interface),
constants.DOCZY_CREATE_BATCH_URL_PROD,
constants.DOCZY_PIPELINE_URL_PROD,
)
else:
logger.info(constants.DOCZY_REDIRECT_URL_DEV + str(interface))
return (
constants.DOCZY_REDIRECT_URL_DEV + str(interface),
constants.DOCZY_CREATE_BATCH_URL_DEV,
constants.DOCZY_PIPELINE_URL_DEV,
)