refactor: Centralize scanner report loading with universal utility function
This commit is contained in:
parent
28713d9fcd
commit
c1382ba694
@ -3,30 +3,8 @@ import pandas as pd
|
||||
from datetime import datetime
|
||||
from screener.canslim_controller import run_canslim_screener
|
||||
from db.db_connection import create_client
|
||||
from utils.report_utils import load_scanner_reports
|
||||
|
||||
def load_scanner_reports():
|
||||
"""Load and return available scanner reports"""
|
||||
import os
|
||||
import pandas as pd
|
||||
from datetime import datetime
|
||||
|
||||
reports = []
|
||||
reports_dir = "reports"
|
||||
|
||||
if os.path.exists(reports_dir):
|
||||
for file in os.listdir(reports_dir):
|
||||
if file.endswith(".csv"):
|
||||
file_path = os.path.join(reports_dir, file)
|
||||
# Get file creation time
|
||||
created = datetime.fromtimestamp(os.path.getctime(file_path))
|
||||
reports.append({
|
||||
'name': file,
|
||||
'path': file_path,
|
||||
'created': created
|
||||
})
|
||||
|
||||
# Sort by creation time, newest first
|
||||
return sorted(reports, key=lambda x: x['created'], reverse=True)
|
||||
|
||||
def canslim_screener_page():
|
||||
st.header("CANSLIM Screener")
|
||||
@ -106,7 +84,7 @@ def canslim_screener_page():
|
||||
with reports_tab:
|
||||
st.subheader("CANSLIM Reports")
|
||||
|
||||
reports = load_scanner_reports()
|
||||
reports = load_scanner_reports(scanner_type="canslim")
|
||||
if reports:
|
||||
# Create a selectbox to choose the report
|
||||
selected_report = st.selectbox(
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
import streamlit as st
|
||||
from screener.scanner_controller import run_technical_scanner
|
||||
from pages.screener.canslim_screener_page import load_scanner_reports
|
||||
from utils.report_utils import load_scanner_reports
|
||||
import pandas as pd
|
||||
|
||||
def technical_scanner_page():
|
||||
@ -83,7 +83,7 @@ def technical_scanner_page():
|
||||
with reports_tab:
|
||||
st.subheader("Scanner Reports")
|
||||
|
||||
reports = load_scanner_reports()
|
||||
reports = load_scanner_reports(scanner_type="technical")
|
||||
if reports:
|
||||
# Create a selectbox to choose the report
|
||||
selected_report = st.selectbox(
|
||||
|
||||
38
src/utils/report_utils.py
Normal file
38
src/utils/report_utils.py
Normal file
@ -0,0 +1,38 @@
|
||||
import os
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
def load_scanner_reports(scanner_type: str = None):
|
||||
"""
|
||||
Load and return available scanner reports
|
||||
|
||||
Args:
|
||||
scanner_type (str, optional): Filter reports by scanner type (e.g., 'technical', 'canslim')
|
||||
|
||||
Returns:
|
||||
list: List of dictionaries containing report information
|
||||
"""
|
||||
reports = []
|
||||
reports_dir = Path("src/reports")
|
||||
|
||||
# Create reports directory if it doesn't exist
|
||||
reports_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if reports_dir.exists():
|
||||
for file in reports_dir.glob("*.csv"):
|
||||
# Get file creation time
|
||||
created = datetime.fromtimestamp(file.stat().st_ctime)
|
||||
|
||||
# If scanner_type is specified, only include matching reports
|
||||
if scanner_type:
|
||||
if not file.name.lower().startswith(scanner_type.lower()):
|
||||
continue
|
||||
|
||||
reports.append({
|
||||
'name': file.name,
|
||||
'path': str(file),
|
||||
'created': created
|
||||
})
|
||||
|
||||
# Sort by creation time, newest first
|
||||
return sorted(reports, key=lambda x: x['created'], reverse=True)
|
||||
Loading…
Reference in New Issue
Block a user