Create scrape_3gpp.py

scrape_3gpp.py

@@ -0,0 +1,375 @@
+import os
+import requests
+from bs4 import BeautifulSoup
+from urllib.parse import urljoin
+import pandas as pd
+import numpy as np
+import zipfile
+
+def scrape(url, excel_file, folder_name):
+    filenames = []
+    # Check if the excel_file argument is provided and if the file exists.
+    if excel_file and os.path.exists(excel_file):
+        try:
+            df = pd.read_excel(excel_file)
+
+            # If 'Actions' in df.columns and filter based on it, and construct URLs from 'TDoc' or 'URL' columns
+            if 'Actions' in df.columns:
+                df = df[df['Actions'] == 'x']
+
+            elif 'File' in df.columns:
+                filenames = [f"{url}/{row['File']}.zip" for index, row in df.iterrows()]
+            elif 'URL' in df.columns:
+                filenames = df['URL'].tolist()
+        except Exception as e:
+            print(f"Error reading Excel file: {e}")
+            # Optionally, handle the error or return a message if needed
+
+    # If no Excel file is provided or found, or if it lacks 'TDoc'/'URL', the function can still continue with predefined URLs or other logic
+    download_directory = folder_name
+    if not os.path.exists(download_directory):
+        os.makedirs(download_directory)
+
+
+    if not filenames:
+        print("No Excel file provided, or no valid URLs found in the file.")
+        # You can either return here or continue with other predefined logic
+        response = requests.get(url)
+
+        # Analyser le contenu HTML de la page
+        soup = BeautifulSoup(response.content, "html.parser")
+
+        # Trouver tous les balises <a> avec des attributs href (liens)
+        links = soup.find_all("a", href=True)
+
+        # Filtrer les liens se terminant par ".zip"
+        zip_links = [link['href'] for link in links if link['href'].endswith('.zip')]
+
+        # Télécharger chaque fichier zip
+        for zip_link in zip_links:
+            # Construire l'URL absolue du fichier zip
+            absolute_url = urljoin(url, zip_link)
+
+            # Extraire le nom de fichier de l'URL
+            filename = os.path.basename(absolute_url)
+
+            # Chemin où le fichier sera enregistré
+            save_path = os.path.join(download_directory, filename)
+
+            # Envoyer une requête GET pour télécharger le fichier
+            with requests.get(absolute_url, stream=True) as r:
+                r.raise_for_status()
+                with open(save_path, 'wb') as f:
+                    for chunk in r.iter_content(chunk_size=8192):
+                        f.write(chunk)
+
+
+
+    else:
+    # Proceed with downloading files using the filenames list
+      for file_url in filenames:
+          filename = os.path.basename(file_url)
+          save_path = os.path.join(download_directory, filename)
+
+          try:
+              with requests.get(file_url, stream=True) as r:
+                  r.raise_for_status()
+                  with open(save_path, 'wb') as f:
+                      for chunk in r.iter_content(chunk_size=8192):
+                          f.write(chunk)
+          except requests.exceptions.HTTPError as e:
+              print(f"HTTP error occurred: {file_url}: {e}")
+              return False, "Il n'y a pas de colonne action ou alors celle ci n'est pas bien écrite, format attendu: 'Actions'"
+
+    return True, "Téléchargement terminé !"
+
+
+def extractZip(folder_name):
+    # Répertoire où les fichiers zip sont déjà téléchargés
+    download_directory = folder_name
+    extract_directory = folder_name + " extraction"  # Répertoire où le contenu des fichiers zip sera extrait
+
+    # Extraire le contenu de tous les fichiers zip dans le répertoire de téléchargement
+    for zip_file in os.listdir(download_directory):
+        zip_path = os.path.join(download_directory, zip_file)
+        # Vérifier si le fichier est un fichier zip
+        if zip_file.endswith(".zip"):
+            extract_dir = os.path.join(extract_directory, os.path.splitext(zip_file)[0])  # Supprimer l'extension .zip
+
+            # Vérifier si le fichier zip existe
+            if os.path.exists(zip_path):
+                # Créer un répertoire pour extraire le contenu s'il n'existe pas
+                if not os.path.exists(extract_dir):
+                    os.makedirs(extract_dir)
+
+                # Extraire le contenu du fichier zip
+                print(f"Extraction en cours pour {zip_file}")
+                with zipfile.ZipFile(zip_path, 'r') as zip_ref:
+                    zip_ref.extractall(extract_dir)
+
+                print(f"Extraction terminée pour {zip_file}")
+            else:
+                print(f"Fichier zip {zip_file} introuvable")
+
+    print("Toutes les extractions sont terminées !")
+
+def excel3gpp(url):
+    response = requests.get(url)
+    response.raise_for_status()  # This will raise an exception if there's an error
+
+    # Use BeautifulSoup to parse the HTML content
+    soup = BeautifulSoup(response.text, 'html.parser')
+
+    # Look for Excel file links; assuming they have .xlsx or .xls extensions
+    excel_links = [a['href'] for a in soup.find_all('a', href=True) if a['href'].endswith(('.xlsx', '.xls'))]
+
+    # Download the first Excel file found (if any)
+    if excel_links:
+        excel_url = excel_links[0]  # Assuming you want the first Excel file
+        if not excel_url.startswith('http'):
+            excel_url = os.path.join(url, excel_url)  # Handle relative URLs
+
+        # Download the Excel file
+        excel_response = requests.get(excel_url)
+        excel_response.raise_for_status()
+
+        # Define the path where you want to save the file
+        filename = excel_url.split('/')[-1]
+        filepath = os.path.join('path_to_save_directory', filename)  # Replace 'path_to_save_directory' with your desired path
+
+        # Write the content of the Excel file to a local file
+        # Write the content of the Excel file to a local file named 'guide.xlsx'
+        filepath = 'guide.xlsx'  # Directly specify the filename
+
+        with open(filepath, 'wb') as f:
+            f.write(excel_response.content)
+        print(f'Excel file downloaded and saved as: {filepath}')
+
+
+def replace_line_breaks(text):
+    return text.replace("\n", "/n")
+
+def remod_text(text):
+    return text.replace("/n", "\n")
+
+def extractionPrincipale(url, excel_file=None):
+    folder_name = url.split("/")[-2]
+
+    result, message = scrape(url, excel_file, folder_name)
+    if result:
+        print("Success:", message)
+    else:
+        return(None, message)
+
+
+    extractZip(folder_name)
+    excel3gpp(url)
+
+
+    extract_directory = folder_name +" extraction"
+    categories = {
+        "Other": ["URL", "File", "Type", "Title", "Source", "Content"],
+        "CR": ["URL", "File", "Type", "Title", "Source", "Content"],
+        "pCR":["URL", "File", "Type", "Title", "Source", "Content"],
+        "LS": ["URL", "File", "Type", "Title", "Source", "Content"],
+        "WID": ["URL", "File", "Type", "Title", "Source", "Content"],
+        "SID": ["URL", "File", "Type", "Title", "Source", "Content"],
+        "DISCUSSION": ["URL", "File", "Type", "Title", "Source", "Content"],
+        "pdf": ["URL", "File", "Type", "Title", "Source", "Content"],
+        "ppt": ["URL", "File", "Type", "Title", "Source", "Content"],
+        "pptx": ["URL", "File", "Type", "Title", "Source", "Content"]
+    }
+
+    data = []
+    errors_count = 0
+    pre_title_section = None
+    for folder in os.listdir(extract_directory):
+        folder_path = os.path.join(extract_directory, folder)
+        if os.path.isdir(folder_path):
+            for file in os.listdir(folder_path):
+                if file == "__MACOSX":
+                    continue
+                file_path = os.path.join(folder_path, file)
+                if file.endswith((".pptx", ".ppt", ".pdf", ".docx", ".doc", ".DOCX")):
+                    try:
+                        text = textract.process(file_path).decode('utf-8')
+                    except Exception as e:
+                        print(f"Error processing {file_path}: {e}")
+                        errors_count += 1
+                        continue
+
+                    cleaned_text_lines = text.split('\n')
+                    cleaned_text = '\n'.join([line.strip('|').strip() for line in cleaned_text_lines if line.strip()])
+
+                    title = ""
+                    debut = ""
+                    sections = cleaned_text.split("Title:")
+                    if len(sections) > 1:
+                        pre_title_section = sections[0].strip().split()
+                        title = sections[1].strip().split("\n")[0].strip()
+                        debut = sections[0].strip()
+
+                    category = "Other"
+                    if file.endswith(".pdf"):
+                        category = "pdf"
+                    elif file.endswith((".ppt", ".pptx")):
+                        category = "ppt"  # assuming all ppt and pptx files go into the same category
+                    elif "CHANGE REQUEST" in debut:
+                        category = "CR"
+                    elif "Discussion" in title:
+                        category = "DISCUSSION"
+                    elif "WID" in title:
+                        category = "WID"
+                    elif "SID" in title:
+                        category = "SID"
+                    elif "LS" in title:
+                        category = "LS"
+                    elif pre_title_section and pre_title_section[-1] == 'pCR':
+                        category = "pCR"
+                    elif "Pseudo-CR" in title:
+                        category = "pCR"
+
+
+                    contenu = ""  # This will hold the concatenated content for 'Contenu' column
+                    if category in categories:
+                        columns = categories[category]
+                        extracted_content = []
+                        if category == "CR":
+                            reason_for_change = ""
+                            summary_of_change = ""
+                            if len(sections) > 1:
+                                reason_for_change = sections[1].split("Reason for change", 1)[-1].split("Summary of change")[0].strip()
+                                summary_of_change = sections[1].split("Summary of change", 1)[-1].split("Consequences if not")[0].strip()
+                            extracted_content.append(f"Reason for change: {reason_for_change}")
+                            extracted_content.append(f"Summary of change: {summary_of_change}")
+                        elif category == "pCR":
+                            if len(sections) > 1:# Handle 'pCR' category-specific content extraction
+                                pcr_specific_content = sections[1].split("Introduction", 1)[-1].split("First Change")[0].strip()
+                            extracted_content.append(f"Introduction: {pcr_specific_content}")
+                        elif category == "LS":
+                            overall_review = ""
+                            if len(sections) > 1:
+                                overall_review = sections[1].split("Overall description", 1)[-1].strip()
+                            extracted_content.append(f"Overall review: {overall_review}")
+                        elif category in ["WID", "SID"]:
+                            objective = ""
+                            start_index = cleaned_text.find("Objective")
+                            end_index = cleaned_text.find("Expected Output and Time scale")
+                            if start_index != -1 and end_index != -1:
+                                objective = cleaned_text[start_index + len("Objective"):end_index].strip()
+                            extracted_content.append(f"Objective: {objective}")
+                        elif category == "DISCUSSION":
+                            Discussion = ""
+                            extracted_text = replace_line_breaks(cleaned_text)
+                            start_index_doc_for = extracted_text.find("Document for:")
+                            if start_index_doc_for != -1:
+                                start_index_word_after_doc_for = start_index_doc_for + len("Document for:")
+                                end_index_word_after_doc_for = start_index_word_after_doc_for + extracted_text[start_index_word_after_doc_for:].find("/n")
+                                word_after_doc_for = extracted_text[start_index_word_after_doc_for:end_index_word_after_doc_for].strip()
+                                result_intro = ''
+                                result_conclusion = ''
+                                result_info = ''
+                                if word_after_doc_for.lower() == "discussion":
+                                    start_index_intro = extracted_text.find("Introduction")
+                                    end_index_intro = extracted_text.find("Discussion", start_index_intro)
+
+                                    intro_text = ""
+                                    if start_index_intro != -1 and end_index_intro != -1:
+                                        intro_text = extracted_text[start_index_intro + len("Introduction"):end_index_intro].strip()
+                                        result_intro = remod_text(intro_text)  # Convert back line breaks
+                                    else:
+                                        result_intro = "Introduction section not found."
+
+                                    # Attempt to find "Conclusion"
+                                    start_index_conclusion = extracted_text.find("Conclusion", end_index_intro)
+                                    end_index_conclusion = extracted_text.find("Proposal", start_index_conclusion if start_index_conclusion != -1 else end_index_intro)
+
+                                    conclusion_text = ""
+                                    if start_index_conclusion != -1 and end_index_conclusion != -1:
+                                        conclusion_text = extracted_text[start_index_conclusion + len("Conclusion"):end_index_conclusion].strip()
+                                        result_conclusion = remod_text(conclusion_text)
+                                    elif start_index_conclusion == -1:  # Conclusion not found, look for Proposal directly
+                                        start_index_proposal = extracted_text.find("Proposal", end_index_intro)
+                                        if start_index_proposal != -1:
+                                            end_index_proposal = len(extracted_text)  # Assuming "Proposal" section goes till the end if present
+                                            proposal_text = extracted_text[start_index_proposal + len("Proposal"):end_index_proposal].strip()
+                                            result_conclusion = remod_text(proposal_text)  # Using "Proposal" content as "Conclusion"
+                                        else:
+                                            result_conclusion = "Conclusion/Proposal section not found."
+                                    else:
+                                        # Handle case where "Conclusion" exists but no "Proposal" to mark its end
+                                        conclusion_text = extracted_text[start_index_conclusion + len("Conclusion"):].strip()
+                                        result_conclusion = remod_text(conclusion_text)
+                                    Discussion=f"Introduction: {result_intro}\nConclusion/Proposal: {result_conclusion}"
+                                elif word_after_doc_for.lower() == "information":
+                                    start_index_info = extracted_text.find(word_after_doc_for)
+                                    if start_index_info != -1:
+                                        info_to_end = extracted_text[start_index_info + len("Information"):].strip()
+                                        result_info = remod_text(info_to_end)
+                                    Discussion = f"Discussion:{result_info}"
+                                else:
+                                    Discussion = "The word after 'Document for:' is not 'Discussion', 'DISCUSSION', 'Information', or 'INFORMATION'."
+                            else:
+                                Discussion = "The phrase 'Document for:' was not found."
+                            # Since DISCUSSION category handling requires more specific processing, adapt as necessary
+                            # Here's a simplified example
+                            discussion_details = Discussion
+                            extracted_content.append(discussion_details)
+                        # Add more categories as needed
+                        contenu = "\n".join(extracted_content)
+
+                    # Assuming 'source' needs to be filled from the guide.xlsx mapping
+                    # Placeholder for source value calculation
+                    source = ""  # Update this with actual source determination logic
+                    status = ""
+                    data.append([url+ "/" + folder + '.zip', folder , category, title, source,status, contenu])
+
+                    # After processing all files and directories
+                    # Read the guide.xlsx file into a DataFrame to map 'TDoc' to 'Source'
+                    guide_df = pd.read_excel('guide.xlsx', usecols=['Source', 'TDoc','TDoc Status'])
+                    tdoc_source_map = {row['TDoc']: row['Source'] for index, row in guide_df.iterrows()}
+                    tdoc_status_map = {row['TDoc']: row['TDoc Status'] for index, row in guide_df.iterrows()}
+                    # Update the 'Source' in your data based on matching 'Nom du fichier' with 'TDoc'
+                    for item in data:
+                        nom_du_fichier = item[1]  # Assuming 'Nom du fichier' is the first item in your data list
+                        if nom_du_fichier in tdoc_source_map:
+                            item[4] = tdoc_source_map[nom_du_fichier]  # Update the 'Source' field, assuming it's the fourth item
+                            item[5] = tdoc_status_map[nom_du_fichier]
+
+
+
+
+    new_df_columns = ["URL", "File", "Type", "title", "Source", "Status", "Content"]  # Create a DataFrame with the updated data
+    new_df = pd.DataFrame(data, columns=new_df_columns)
+    try:
+        old_df = pd.read_excel(excel_file)
+
+        # Check if 'Actions' column exists in the old DataFrame
+        if 'Actions' in old_df.columns:
+            # Assuming you want to update 'Content' in old_df for matching 'TDoc' values in 'File'
+            for index, new_row in new_df.iterrows():
+                # Find matching rows in old_df where 'TDoc' matches 'File' from new_df
+                match_indices = old_df[old_df['TDoc'] == new_row['File']].index
+                # Update 'Content' in old_df for matching rows
+                for i in match_indices:
+                    old_df.at[i, 'Content'] = new_row['Content']
+                    old_df.at[i, 'URL'] = new_row['URL']
+
+            df = old_df
+            ###placer la colonne content en 4eme position
+            # current_columns = df.columns.tolist()
+            # current_columns.remove('URL')
+            # # Insert 'Content' at the desired position
+            # new_columns_order = current_columns[:1] + ['URL'] + current_columns[3:]
+            # df = df[new_columns_order]
+        else:
+            # If 'Actions' column doesn't exist, simply concatenate the DataFrames
+            df = pd.concat([old_df, new_df], axis=0, ignore_index=True)
+    except Exception as e:
+        print("The provided excel file seems invalid:", e)
+        df = new_df
+
+    file_name = url.split("/")[-2] + ".xlsx"
+    # Save the updated DataFrame to Excel
+    df.to_excel(file_name, index=False)
+    return file_name, "Téléchargement réussi"