Create app.py

app.py

@@ -0,0 +1,295 @@
+import streamlit as st
+import requests
+import os
+import urllib
+import base64
+from bs4 import BeautifulSoup
+import hashlib
+import json
+import uuid 
+import glob
+import zipfile
+
+EXCLUDED_FILES = ['app.py', 'requirements.txt', 'pre-requirements.txt', 'packages.txt', 'README.md','.gitattributes', "backup.py","Dockerfile"]
+URLS = {
+    "Lumiere": "https://lumiere-video.github.io/",
+    "National Library of Medicine": "https://www.nlm.nih.gov/",
+    "World Health Organization": "https://www.who.int/",
+    "UHCProvider - United Health and Optum": "https://www.uhcprovider.com/",
+    "CMS - Centers for Medicare & Medicaid Services": "https://www.cms.gov/",
+    "Mayo Clinic": "https://www.mayoclinic.org/",
+    "WebMD": "https://www.webmd.com/",
+    "MedlinePlus": "https://medlineplus.gov/",
+    "Healthline": "https://www.healthline.com/",
+    "CDC - Centers for Disease Control and Prevention": "https://www.cdc.gov/",
+    "Johns Hopkins Medicine": "https://www.hopkinsmedicine.org/"
+}
+
+if not os.path.exists("history.json"):
+    with open("history.json", "w") as f:
+        json.dump({}, f)
+
+import os
+import base64
+import zipfile
+import streamlit as st
+
+def zip_subdirs(start_dir):
+    for subdir, dirs, files in os.walk(start_dir):
+        if subdir != start_dir:  # Skip the root directory
+            zip_filename = os.path.join(start_dir, subdir.split(os.sep)[-1] + '.zip')
+            allFileSummary = ""
+            with zipfile.ZipFile(zip_filename, 'w') as zipf:
+                for file in files:
+                    file_path = os.path.join(subdir, file)
+                    zipf.write(file_path, os.path.relpath(file_path, start_dir))
+                    allFileSummary=allFileSummary+(f"Added: {file_path}")
+            st.write(allFileSummary)
+            yield zip_filename
+
+def get_zip_download_link(zip_file):
+    with open(zip_file, 'rb') as f:
+        bytes = f.read()
+    b64 = base64.b64encode(bytes).decode()
+    link_name = os.path.basename(zip_file)
+    href = f'<a href="data:file/zip;base64,{b64}" download="{link_name}">Download: {link_name}</a>'
+    return href
+
+  
+@st.cache_resource
+def create_zip_of_files(files):
+    zip_name = "all_files.zip"
+    with zipfile.ZipFile(zip_name, 'w') as zipf:
+        for file in files:
+            zipf.write(file)
+    return zip_name
+    
+@st.cache_resource
+def get_zip_download_link(zip_file):
+    with open(zip_file, 'rb') as f:
+        data = f.read()
+    b64 = base64.b64encode(data).decode()
+    href = f'<a href="data:application/zip;base64,{b64}" download="{zip_file}">Download All</a>'
+    return href
+
+
+
+def download_file(url, local_filename):
+    if url.startswith('http://') or url.startswith('https://'):
+        try:
+            with requests.get(url, stream=True) as r:
+                r.raise_for_status()
+                with open(local_filename, 'wb') as f:
+                    for chunk in r.iter_content(chunk_size=8192):
+                        f.write(chunk)
+            return local_filename
+        except requests.exceptions.HTTPError as err:
+            print(f"HTTP error occurred: {err}")
+
+def download_html_and_files(url, subdir):
+    html_content = requests.get(url).text
+    soup = BeautifulSoup(html_content, 'html.parser')
+    base_url = urllib.parse.urlunparse(urllib.parse.urlparse(url)._replace(path='', params='', query='', fragment=''))
+
+    for link in soup.find_all('a'):
+        file_url = urllib.parse.urljoin(base_url, link.get('href'))
+        local_filename = os.path.join(subdir, urllib.parse.urlparse(file_url).path.split('/')[-1])
+
+        if not local_filename.endswith('/') and local_filename != subdir:
+            link['href'] = local_filename
+            download_file(file_url, local_filename)
+    
+    with open(os.path.join(subdir, "index.html"), "w") as file:
+        file.write(str(soup))
+
+def list_files(directory_path='.'):
+    files = [f for f in os.listdir(directory_path) if os.path.isfile(os.path.join(directory_path, f))]
+    return [f for f in files if f not in EXCLUDED_FILES]
+
+def file_editor(file_path):
+    st.write(f"Editing File: {os.path.basename(file_path)}")
+    file_content = ""
+
+    with open(file_path, "r") as f:
+        file_content = f.read()
+
+    file_content = st.text_area("Edit the file content:", value=file_content, height=250)
+
+    if st.button("💾 Save"):
+        with open(file_path, "w") as f:
+            f.write(file_content)
+        st.success(f"File '{os.path.basename(file_path)}' saved!")
+
+
+def show_file_operations(file_path, sequence_number):
+    #st.write(f"File: {os.path.basename(file_path)}")
+    unique_key = hashlib.md5(file_path.encode()).hexdigest()
+    file_content = ""
+
+    col01, col02, col1, col2, col3 = st.columns(5)
+    with col01:
+        st.write(os.path.basename(file_path))
+#with col02:
+        #st.write(file_path)
+    with col1:
+        edit_key = f"edit_{unique_key}_{sequence_number}"
+        if st.button(f"✏️ Edit", key=edit_key):
+            with open(file_path, "r") as f:
+                file_content = f.read()
+            text_area_key = f"text_area_{unique_key}_{sequence_number}"
+            file_content = st.text_area("Edit the file content:", value=file_content, height=250, key=text_area_key)
+
+    with col2:
+        save_key = f"save_{unique_key}_{sequence_number}"
+        if st.button(f"💾 Save", key=save_key):
+            if file_content:  # Ensure file_content is not empty
+                with open(file_path, "w") as f:
+                    f.write(file_content)
+                st.success(f"File saved!")
+
+    with col3:
+        delete_key = f"delete_{unique_key}_{sequence_number}"
+        if st.button(f"🗑️ Delete", key=delete_key):
+            os.remove(file_path)
+            st.markdown(f"File deleted!")
+
+
+file_sequence_numbers = {}
+
+def show_download_links(subdir):
+    global file_sequence_numbers
+    for file in list_files(subdir):
+        file_path = os.path.join(subdir, file)
+        if file_path not in file_sequence_numbers:
+            file_sequence_numbers[file_path] = 1
+        else:
+            file_sequence_numbers[file_path] += 1
+        sequence_number = file_sequence_numbers[file_path]
+
+        if os.path.isfile(file_path):
+            #st.markdown(get_download_link(file_path), unsafe_allow_html=True)
+            st.markdown(file_path, unsafe_allow_html=True)  # faster than encapsulating file into base64 download link
+            show_file_operations(file_path, sequence_number)
+        else:
+            st.write(f"File not found: {file}")
+
+def get_download_link(file):
+    with open(file, "rb") as f:
+        bytes = f.read()
+        b64 = base64.b64encode(bytes).decode()
+        href = f'<a href="data:file/octet-stream;base64,{b64}" download=\'{os.path.basename(file)}\'>Download: {os.path.basename(file)}</a>'
+    return href
+    
+def main():
+    st.sidebar.title('🌐 Web Datasets Bulk Downloader')
+
+
+        
+    # Check for query parameters for file editing
+    query_params = st.experimental_get_query_params()
+    file_to_edit = query_params.get('file_to_edit', [None])[0]
+
+    if file_to_edit and os.path.exists(file_to_edit):
+        file_editor(file_to_edit)
+    else:
+        # Selecting URL input method
+        
+        # Selecting URL input method
+        url_input_method = st.sidebar.radio("Choose URL Input Method", ["Enter URL", "Select from List"], index=1)
+        url = ""
+        if url_input_method == "Enter URL":
+            url = st.sidebar.text_input('Please enter a Web URL to bulk download text and files')
+        else:
+            selected_site = st.sidebar.selectbox("Select a Website", list(URLS.keys()), index=0)
+            url = URLS[selected_site]
+
+        
+        #url_input_method = st.sidebar.radio("Choose URL Input Method", ["Enter URL", "Select from List"])
+        #url = ""
+        #if url_input_method == "Enter URL":
+        #    url = st.sidebar.text_input('Please enter a Web URL to bulk download text and files')
+        #else:
+        #    selected_site = st.sidebar.selectbox("Select a Website", list(URLS.keys()))
+        #    url = URLS[selected_site]
+
+        # Reading or creating history.json
+        if not os.path.exists("history.json"):
+            with open("history.json", "w") as f:
+                json.dump({}, f)
+
+        with open("history.json", "r") as f:
+            try:
+                history = json.load(f)
+            except:
+                print('error')
+                
+        # Handling URL submission
+        if url:
+            subdir = hashlib.md5(url.encode()).hexdigest()
+            if not os.path.exists(subdir):
+                os.makedirs(subdir)
+            if url not in history:
+                history[url] = subdir
+                with open("history.json", "w") as f:
+                    json.dump(history, f)
+
+
+        if st.sidebar.button('📥 Get All the Content', help="Download content from the selected URL"):
+            download_html_and_files(url, history[url])
+            show_download_links(history[url])
+    
+        if st.sidebar.button('📂 Show Download Links', help="Show all available download links"):
+            for subdir in history.values():
+                show_download_links(subdir)
+
+                    
+        # Button for downloading content
+        #if st.sidebar.button('📥 Get All the Content'):
+        #    download_html_and_files(url, history[url])
+        #    show_download_links(history[url])
+
+        # Button for showing download links
+        #if st.sidebar.button('📂 Show Download Links'):
+        #    for subdir in history.values():
+        #        show_download_links(subdir)
+
+        if st.sidebar.button("🗑 Delete All", help="Delete all downloaded content"):
+        #if st.sidebar.button("🗑 Delete All"):
+            # Clear history file
+            with open("history.json", "w") as f:
+                json.dump({}, f)
+    
+            # Delete all files in subdirectories
+            for subdir in glob.glob('*'):
+                if os.path.isdir(subdir) and subdir not in EXCLUDED_FILES:
+                    for file in os.listdir(subdir):
+                        file_path = os.path.join(subdir, file)
+                        os.remove(file_path)
+                        st.write(f"Deleted: {file_path}")
+                    os.rmdir(subdir)  # Remove the empty directory
+    
+            st.experimental_rerun()
+
+        if st.sidebar.button("⬇️ Download All", help="Download all files in a zip"):
+            start_directory = '.'  # Current directory
+            for zip_file in zip_subdirs(start_directory):
+                st.sidebar.markdown(zip_file, unsafe_allow_html=True)
+                st.sidebar.markdown(get_zip_download_link(zip_file), unsafe_allow_html=True)
+        #if st.sidebar.button("⬇️ Download All"):
+        #    start_directory = '.'  # Current directory
+        #    for zip_file in zip_subdirs(start_directory):
+        #        st.sidebar.markdown(get_zip_download_link(zip_file), unsafe_allow_html=True)
+            
+        # Expander for showing URL history and download links
+        with st.expander("URL History and Downloaded Files"):
+            try:
+                for url, subdir in history.items():
+                    st.markdown(f"#### {url}")
+                    # show_download_links(subdir)
+            except:
+                print('url history is empty')
+        # Update each time to show files we have
+        #for subdir in history.values():
+            # show_download_links(subdir)
+if __name__ == "__main__":
+    main()