-
Notifications
You must be signed in to change notification settings - Fork 0
/
streamlit_app.py
141 lines (106 loc) · 4.08 KB
/
streamlit_app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
import json
import streamlit as st
from main import main as scraper
import time
# Function to load markdown file
def get_file_content_as_string(path):
with open(path, "r") as file:
lines = file.readlines()
return "".join(lines)
# Function to load cached data from jobs.json
def load_cached_data():
try:
with open("jobs.json", "r") as file:
datas = file.read()
return json.loads(datas)["jobs"]
except Exception as e:
print(e)
return {}
# Function to save data to jobs.json
def save_data_to_cache(data):
with open("jobs.json", "w") as file:
file.write(json.dumps({"jobs": data}))
st.success("Data saved to cache!")
# Function to clear cache
def clear_cache():
with open("jobs.json", "w") as file:
file.write(json.dumps({"jobs": {}}))
st.success("Cache cleared!")
st.rerun() # Rerun the app to reflect the changes
# Function to scrape jobs
def scrape_jobs():
# Simulate scraping delay
# time.sleep(5)
# Replace the following with your actual scraping logic and return the scraped data as a dictionary
return scraper()
# Function to scrape jobs and display the result
def scrape_and_display(col2):
global data
with col2:
with st.spinner("Scraping jobs... Please wait."):
if data:
st.info("Using cached data")
scraped_data = data
st.balloons() # Add a pop-up to notify the user that the data is cached
else:
st.info(
"No cached data found... Scraping new data. This may take a while.")
scraped_data = scrape_jobs()
st.success("Scraping complete!")
st.text("Total jobs scraped: {}".format(len(scraped_data)))
return scraped_data
# Main Streamlit app
def main():
global data
st.set_page_config(page_title="Amazon Jobs Scraper",
page_icon=":briefcase:")
st.title("Amazon Jobs Scraper")
data = load_cached_data()
search_term = st.text_input(
"Search for a specific Job Category", placeholder="Enter a job category...")
col1, col2 = st.columns([1, 4])
with col1:
if st.button("Scrape Jobs"):
data = scrape_and_display(col2)
if st.button("Save to Cache"):
save_data_to_cache(data)
if st.button("Clear Cache"):
clear_cache()
# if st.button("Clear Content"):
# st.text("Content cleared!")
if st.button("Clear Content"):
with col2:
st.text("Content cleared!")
data = [] # Clear the data displayed on the right side
# st.json({}) # Clear the displayed JSON content
if st.download_button("Download Data as JSON", json.dumps(data), "jobs.json", "Click here to download data"):
st.success("Data downloaded successfully!")
if st.button("Show Source Code"):
# st.code(open(__file__).read())
with col2:
st.markdown("### Source Code")
st.code(open(__file__).read())
if st.button("Show Documentation"):
with col2:
st.markdown("### Documentation")
readme_text = st.markdown(
get_file_content_as_string("README.md"))
with col2:
if search_term:
st.text("Showing results for search term: {}".format(search_term))
filtered_data = [
job for job in data if search_term.lower() in job["job_category"].lower()]
if filtered_data:
data = filtered_data
else:
st.warning(
"No jobs found for search term: {}".format(search_term))
if data:
job_titles = [job["title"] for job in data]
selected_title = st.selectbox("Select Job Title", job_titles)
selected_job = next(
(job for job in data if job["title"] == selected_title), None)
if selected_job:
st.table(selected_job)
if __name__ == "__main__":
main()