File size: 4,157 Bytes
c4331f2 |
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 142 143 144 145 146 147 148 149 150 |
import os
from datetime import date, timedelta
import bs4
from langchain.indexes import SQLRecordManager, index
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")
titles = driver.find_elements(By.XPATH, "//td/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=1000, chunk_overlap=100
):
"""
#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")
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n"]
)
splits = text_splitter.split_documents(docs)
# Create the storage path if it doesn't exist
if not os.path.exists(persist_directory):
os.makedirs(persist_directory)
doc_search = Chroma.from_documents(
documents=splits,
embedding=embeddings_model,
persist_directory=persist_directory,
)
# Indexing data
namespace = "chromadb/my_documents"
record_manager = SQLRecordManager(
namespace, db_url="sqlite:///record_manager_cache.sql"
)
record_manager.create_schema()
index_result = index(
docs,
record_manager,
doc_search,
cleanup="incremental",
source_id_key="link",
)
print(f"Indexing stats: {index_result}")
return doc_search
if __name__ == "__main__":
data = scrap_articles(DATA_URL, num_days_past=2)
vectordb = process_docs(data, config.STORAGE_PATH, embeddings_model)
ret = vectordb.as_retriever()
|