|
import os |
|
import requests |
|
from bs4 import BeautifulSoup |
|
from urllib.parse import urljoin |
|
import pandas as pd |
|
import numpy as np |
|
import zipfile |
|
import textract |
|
import gradio as gr |
|
import shutil |
|
|
|
def browse_folder(url): |
|
if url.lower().endswith(('docs', 'docs/')): |
|
return gr.update(choices=[]) |
|
response = requests.get(url) |
|
response.raise_for_status() |
|
|
|
soup = BeautifulSoup(response.text, 'html.parser') |
|
|
|
excel_links = [a['href'] + '/' for a in soup.find_all('a', href=True) if a['href'].startswith(url)] |
|
|
|
return gr.update(choices=excel_links) |
|
|
|
|
|
|
|
def extract_statuses(url): |
|
|
|
response = requests.get(url) |
|
|
|
|
|
soup = BeautifulSoup(response.content, 'html.parser') |
|
|
|
|
|
links = soup.find_all('a') |
|
|
|
|
|
for link in links: |
|
href = link.get('href') |
|
if href and (href.endswith('.xls') or href.endswith('.xlsx')): |
|
excel_url = href if href.startswith('http') else url + href |
|
excel_response = requests.get(excel_url) |
|
file_name = 'guide_status.xlsx' |
|
|
|
|
|
with open(file_name, 'wb') as f: |
|
f.write(excel_response.content) |
|
|
|
|
|
df = pd.read_excel(file_name) |
|
|
|
|
|
if 'TDoc Status' in df.columns: |
|
unique_statuses = df['TDoc Status'].unique().tolist() |
|
print(f'Downloaded {file_name} and extracted statuses: {unique_statuses}') |
|
|
|
|
|
if 'withdrawn' in unique_statuses: |
|
unique_statuses.remove('withdrawn') |
|
return gr.update(choices=unique_statuses, value=unique_statuses) |
|
else: |
|
print(f"'TDoc Status' column not found in {file_name}") |
|
return [] |
|
|
|
|
|
def scrape(url, excel_file, folder_name, status_list, progress=gr.Progress()): |
|
filenames = [] |
|
status_filenames = [] |
|
|
|
excel_file_path = "guide_status.xlsx" |
|
|
|
if os.path.exists(excel_file_path): |
|
try: |
|
df = pd.read_excel(excel_file_path) |
|
print(f"Initial DataFrame size: {len(df)}") |
|
|
|
if 'TDoc Status' in df.columns: |
|
df = df[df['TDoc Status'].isin(status_list)] |
|
print(f"Filtered DataFrame size: {len(df)}") |
|
|
|
if df.empty: |
|
print("No files match the specified 'TDoc Status'.") |
|
else: |
|
if 'TDoc' in df.columns and not df['TDoc'].isnull().all(): |
|
status_filenames = [f"{url}{row['TDoc']}.zip" for index, row in df.iterrows()] |
|
elif 'URL' in df.columns and not df['URL'].isnull().all(): |
|
status_filenames = df['URL'].tolist() |
|
else: |
|
print("No valid 'File' or 'URL' entries found for the filtered statuses.") |
|
|
|
print(f"Filenames: {status_filenames}") |
|
else: |
|
print("'TDoc Status' column not found in the Excel file.") |
|
|
|
except Exception as e: |
|
print(f"Error reading Excel file: {e}") |
|
|
|
|
|
|
|
if excel_file and os.path.exists(excel_file): |
|
try: |
|
df = pd.read_excel(excel_file) |
|
|
|
|
|
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}") |
|
|
|
|
|
|
|
download_directory = folder_name |
|
if not os.path.exists(download_directory): |
|
os.makedirs(download_directory) |
|
|
|
pourcentss = 0.05 |
|
print(f'filenames: {status_filenames}') |
|
if not filenames and not status_filenames: |
|
print("No Excel file provided, or no valid URLs found in the file.") |
|
|
|
response = requests.get(url) |
|
|
|
|
|
soup = BeautifulSoup(response.content, "html.parser") |
|
|
|
|
|
links = soup.find_all("a", href=True) |
|
|
|
|
|
zip_links = [link['href'] for link in links if link['href'].endswith('.zip')] |
|
|
|
|
|
for zip_link in zip_links: |
|
progress(pourcentss,desc='Downloading') |
|
pourcentss+=0.4/len(df) |
|
|
|
absolute_url = urljoin(url, zip_link) |
|
|
|
|
|
filename = os.path.basename(absolute_url) |
|
|
|
|
|
save_path = os.path.join(download_directory, filename) |
|
|
|
|
|
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) |
|
|
|
elif not filenames: |
|
|
|
for file_url in status_filenames: |
|
filename = os.path.basename(file_url) |
|
save_path = os.path.join(download_directory, filename) |
|
progress(pourcentss,desc='Downloading') |
|
pourcentss+=0.4/len(df) |
|
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"skipped file: {file_url}: {e}") |
|
|
|
else: |
|
|
|
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, len(df) |
|
|
|
|
|
|
|
def extractZip(url): |
|
|
|
nom_extract = url.split("/")[-3] + "_extraction" |
|
if os.path.exists(nom_extract): |
|
shutil.rmtree(nom_extract) |
|
extract_directory = nom_extract |
|
|
|
download_directory = url.split("/")[-3] + "_downloads" |
|
|
|
|
|
|
|
for zip_file in os.listdir(download_directory): |
|
zip_path = os.path.join(download_directory, zip_file) |
|
|
|
if zip_file.endswith(".zip"): |
|
extract_dir = os.path.join(extract_directory, os.path.splitext(zip_file)[0]) |
|
|
|
|
|
if os.path.exists(zip_path): |
|
|
|
if not os.path.exists(extract_dir): |
|
os.makedirs(extract_dir) |
|
|
|
|
|
|
|
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() |
|
|
|
|
|
soup = BeautifulSoup(response.text, 'html.parser') |
|
|
|
|
|
excel_links = [a['href'] for a in soup.find_all('a', href=True) if a['href'].endswith(('.xlsx', '.xls'))] |
|
|
|
|
|
if excel_links: |
|
excel_url = excel_links[0] |
|
if not excel_url.startswith('http'): |
|
excel_url = os.path.join(url, excel_url) |
|
|
|
|
|
excel_response = requests.get(excel_url) |
|
excel_response.raise_for_status() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
nom_guide = 'guide.xlsx' |
|
if os.path.exists(nom_guide): |
|
os.remove(nom_guide) |
|
filepath = nom_guide |
|
|
|
|
|
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 update_excel(data, excel_file, url): |
|
new_df_columns = ["URL", "File", "Type", "Title", "Source", "Status", "Content"] |
|
temp_df = pd.DataFrame(data, columns=new_df_columns) |
|
|
|
try: |
|
|
|
if os.path.exists(excel_file): |
|
old_df = pd.read_excel(excel_file) |
|
df = pd.concat([old_df, temp_df], axis=0, ignore_index=True) |
|
else: |
|
df = temp_df |
|
|
|
|
|
df.to_excel(excel_file, index=False) |
|
except Exception as e: |
|
print(f"Error updating Excel file: {e}") |
|
|
|
def extractionPrincipale(url, excel_file=None, status_list=None, progress=gr.Progress()): |
|
nom_download = url.split("/")[-3] + "_downloads" |
|
if os.path.exists(nom_download): |
|
shutil.rmtree(nom_download) |
|
folder_name = nom_download |
|
|
|
nom_status = url.split("/")[-3] + "_status.xlsx" |
|
if os.path.exists(nom_status): |
|
os.remove(nom_status) |
|
temp_excel = nom_status |
|
|
|
progress(0.0,desc='Downloading') |
|
|
|
result, count = scrape(url, excel_file, folder_name, status_list) |
|
if result: |
|
print("Success") |
|
else: |
|
return(None) |
|
|
|
progress(0.4,desc='Extraction') |
|
extractZip(url) |
|
progress(0.5,desc='Extraction 2') |
|
excel3gpp(url) |
|
progress(0.6,desc='Creating Excel File') |
|
|
|
|
|
extract_directory = url.split("/")[-3] + "_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"] |
|
} |
|
|
|
pourcents2=0.6 |
|
data = [] |
|
errors_count = 0 |
|
processed_count = 0 |
|
|
|
pre_title_section = None |
|
|
|
try: |
|
df = pd.read_excel(temp_excel) |
|
except Exception as e: |
|
print(f"Initializing a new DataFrame because: {e}") |
|
df = pd.DataFrame(columns=["URL", "File", "Type", "Title", "Source", "Status", "Content"]) |
|
|
|
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): |
|
progress(min(pourcents2,0.99),desc='Creating Excel File') |
|
pourcents2+=0.4/count |
|
|
|
|
|
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" |
|
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 = "" |
|
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: |
|
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) |
|
else: |
|
result_intro = "Introduction section not found." |
|
|
|
|
|
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: |
|
start_index_proposal = extracted_text.find("Proposal", end_index_intro) |
|
if start_index_proposal != -1: |
|
end_index_proposal = len(extracted_text) |
|
proposal_text = extracted_text[start_index_proposal + len("Proposal"):end_index_proposal].strip() |
|
result_conclusion = remod_text(proposal_text) |
|
else: |
|
result_conclusion = "Conclusion/Proposal section not found." |
|
else: |
|
|
|
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." |
|
|
|
|
|
discussion_details = Discussion |
|
extracted_content.append(discussion_details) |
|
|
|
contenu = "\n".join(extracted_content) |
|
|
|
|
|
|
|
source = "" |
|
status = "" |
|
data.append([url+ "/" + folder + '.zip', folder , category, title, source,status, contenu]) |
|
|
|
|
|
|
|
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()} |
|
|
|
for item in data: |
|
nom_du_fichier = item[1] |
|
if nom_du_fichier in tdoc_source_map: |
|
item[4] = tdoc_source_map[nom_du_fichier] |
|
item[5] = tdoc_status_map[nom_du_fichier] |
|
|
|
|
|
processed_count += 1 |
|
|
|
|
|
if processed_count % 20 == 0: |
|
update_excel(data, temp_excel, url) |
|
print(f"Updated after processing {processed_count} files.") |
|
data = [] |
|
|
|
if data: |
|
|
|
update_excel(data, temp_excel, url) |
|
print(f"Final update after processing all files.") |
|
|
|
file_name = temp_excel |
|
|
|
return file_name |
|
|
|
|