File size: 4,186 Bytes
c4331f2 a3b1498 c4331f2 782be8b c4331f2 a3b1498 c4331f2 a3b1498 c4331f2 a3b1498 8bfa348 a3b1498 c4331f2 a3b1498 c4331f2 a3b1498 0dfba83 c4331f2 a3b1498 c4331f2 a7aa9c3 a3b1498 |
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 |
from datetime import date, timedelta
import bs4
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import LocalFileStore
from langchain.storage._lc_store import create_kv_docstore
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores.chroma import Chroma
from langchain_community.document_loaders import WebBaseLoader
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
import config
DATA_URL = "https://www.sikafinance.com/marches/actualites_bourse_brvm"
embeddings_model = GoogleGenerativeAIEmbeddings(
model=config.GOOGLE_EMBEDDING_MODEL
) # type: ignore
options = webdriver.ChromeOptions()
options.add_argument("--headless")
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
driver = webdriver.Chrome(options=options)
def scrap_articles(
url="https://www.sikafinance.com/marches/actualites_bourse_brvm", num_days_past=5
):
today = date.today()
driver.get(url)
all_articles = []
for i in range(num_days_past + 1):
past_date = today - timedelta(days=i)
date_str = past_date.strftime("%Y-%m-%d")
WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.ID, "dateActu"))
)
text_box = driver.find_element(By.ID, "dateActu")
text_box.send_keys(date_str)
submit_btn = WebDriverWait(driver, 10).until(
EC.element_to_be_clickable((By.ID, "btn"))
)
submit_btn.click()
dates = driver.find_elements(By.CLASS_NAME, "sp1")
table = driver.find_element(By.ID, "tabQuotes")
titles = table.find_elements(By.TAG_NAME, "a")
articles = []
for i in range(len(titles)):
art = {
"title": titles[i].text.strip(),
"date": dates[i].text,
"link": titles[i].get_attribute("href"),
}
articles.append(art)
all_articles += articles
# driver.quit()
return all_articles
def set_metadata(documents, metadatas):
"""
#Edit a metadata of lanchain Documents object
"""
for doc in documents:
idx = documents.index(doc)
doc.metadata = metadatas[idx]
print("Metadata successfully changed")
print(documents[0].metadata)
def process_docs(
articles, persist_directory, embeddings_model, chunk_size=500, chunk_overlap=0
):
"""
#Scrap all articles urls content and save on a vector DB
"""
article_urls = [a["link"] for a in articles]
print("Starting to scrap ..")
loader = WebBaseLoader(
web_paths=article_urls,
bs_kwargs=dict(
parse_only=bs4.SoupStrainer(
class_=("inarticle txtbig", "dt_sign", "innerUp")
)
),
)
print("After scraping Loading ..")
docs = loader.load()
# Update metadata: add title,
set_metadata(documents=docs, metadatas=articles)
# print("Successfully loaded to document")
# This text splitter is used to create the child documents
child_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n"]
)
# The vectorstore to use to index the child chunks
vectorstore = Chroma(
persist_directory=persist_directory + "vectorstore",
collection_name="full_documents",
embedding_function=embeddings_model,
)
# The storage layer for the parent documents
fs = LocalFileStore(persist_directory + "docstore")
store = create_kv_docstore(fs)
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=store,
child_splitter=child_splitter,
)
retriever.add_documents(docs, ids=None)
print(len(docs), " documents added")
if __name__ == "__main__":
data = scrap_articles(DATA_URL, num_days_past=config.NUM_DAYS_PAST)
process_docs(data, config.STORAGE_PATH, embeddings_model)
|