mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-14 13:34:40 +00:00
Fixing librarian result (#6)
* Formatting and additional logging for bugfix * Formatting and additional logging for bugfix * Change of log location * fixing_librarian_result * Hopefully final fix that will be needed * Temporary * Glebokie gardlo * Fix in scrape_bot * Fix that was needed. * Scraper settings * Small fix in scraping * FIxing bad naming convention * Log message fix * Test of deep search * Fix in deep search * Refactoring * WHat and idiot coded that.... Ooops. That was me. A month ago. * Refactoring continued * Bugfix * FIx * Not needed after fix
This commit is contained in:
GitHub
@@ -1,26 +1,27 @@
|
||||
"""
|
||||
This module contains the implementation of the Librarian class and related functions for searching and refining queries.
|
||||
This module contains the implementation of the Librarian class and
|
||||
related functions for searching and refining queries.
|
||||
|
||||
Classes:
|
||||
- Librarian: Represents a librarian object that performs search and refinement operations on queries.
|
||||
- Librarian: Represents a librarian object that performs search and
|
||||
refinement operations on queries
|
||||
|
||||
Functions:
|
||||
- flask_debug: Starts a Flask application in debug mode without using the reloader.
|
||||
- waitress_run: Serves the Flask application using the Waitress WSGI server.
|
||||
- BackgroundTaskSearch: Represents a background task for running the Librarian object asynchronously.
|
||||
- flask_debug: Starts a Flask application in debug mode without using the reloader.
|
||||
- waitress_run: Serves the Flask application using the Waitress WSGI server.
|
||||
- BackgroundTaskSearch: Represents a background task for running the
|
||||
Librarian object asynchronously
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import netrc
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from json.decoder import JSONDecodeError
|
||||
from logging import handlers
|
||||
from queue import Queue
|
||||
from urllib.request import urlopen
|
||||
|
||||
import requests
|
||||
import scrape_bot
|
||||
@@ -36,7 +37,8 @@ PORT_ADDRESS = 5001
|
||||
MAIN_BOT_ADDRESS = "http://192.168.1.191:5000"
|
||||
SEND_RESULTS = "/conjurer"
|
||||
BDSM_UUID_TEST = "96b7f85a-1142-4908-8986-62a2ea25a147"
|
||||
MAX_CR_RESULTS = 150000
|
||||
MAX_CR_RESULTS = 7000
|
||||
|
||||
ENCODING = "utf-8"
|
||||
|
||||
app = Flask(__name__)
|
||||
@@ -45,8 +47,13 @@ librarian_queue = Queue()
|
||||
librarian_list = []
|
||||
|
||||
|
||||
# trunk-ignore(pylint/R0902)
|
||||
class Librarian(object):
|
||||
def __init__(self, _app, query, uuid) -> None:
|
||||
"""
|
||||
Represents a librarian object that performs search and refinement operations on queries.
|
||||
"""
|
||||
|
||||
def __init__(self, _app, query, uuid, _deep_search) -> None:
|
||||
"""
|
||||
Initializes a Librarian object.
|
||||
|
||||
@@ -71,8 +78,11 @@ class Librarian(object):
|
||||
- done: A flag indicating if the search is done.
|
||||
"""
|
||||
netrc_mod = netrc.netrc(NETRC_FILE)
|
||||
authTokens = netrc_mod.authenticators("crossref")
|
||||
self.cr = Crossref(mailto=authTokens[0])
|
||||
auth_tokens = netrc_mod.authenticators("crossref")
|
||||
self.cr = Crossref(
|
||||
mailto=auth_tokens[0],
|
||||
ua_string=f"Conjurer project. mailto:{auth_tokens[0]}"
|
||||
)
|
||||
self.query = query
|
||||
self.uuid = str(uuid)
|
||||
self.limit = MAX_CR_RESULTS
|
||||
@@ -81,12 +91,13 @@ class Librarian(object):
|
||||
self.total = 0
|
||||
self.app = _app
|
||||
self.live_results = []
|
||||
self.final_result = []
|
||||
self.not_in_db = []
|
||||
self.final_result = {}
|
||||
self.not_in_db = {}
|
||||
self.search_result_from_cr = {}
|
||||
self.done = False
|
||||
self.deep_search = _deep_search
|
||||
|
||||
async def search_crossref(self, query):
|
||||
async def search_crossref(self, query, deep_search=False):
|
||||
"""
|
||||
Performs a search on Crossref for the given query.
|
||||
|
||||
@@ -99,122 +110,124 @@ class Librarian(object):
|
||||
Raises:
|
||||
- None.
|
||||
"""
|
||||
|
||||
self.app.logger.info("STARTED SEARCH")
|
||||
result = self.cr.works(query=query, limit=1000)
|
||||
self.search_result_from_cr.update(result)
|
||||
self.total = result["message"]["total-results"]
|
||||
self.fetched += len(result["message"]["items"])
|
||||
self.app.logger.info(self.total)
|
||||
self.app.logger.info(self.fetched)
|
||||
while self.total > self.fetched and self.limit > self.fetched:
|
||||
result = self.cr.works(query=query, limit=1000)
|
||||
self.search_result_from_cr.update(result)
|
||||
self.total = result["message"]["total-results"]
|
||||
self.fetched += len(result["message"]["items"])
|
||||
|
||||
if not deep_search:
|
||||
cr_result = self.cr.works(query=query, limit=1000)
|
||||
self.search_result_from_cr.update(cr_result)
|
||||
self.total = cr_result["message"]["total-results"]
|
||||
self.fetched += len(cr_result["message"]["items"])
|
||||
self.app.logger.info(self.total)
|
||||
self.app.logger.info(self.fetched)
|
||||
time.sleep(0.1)
|
||||
self.app.logger.info("CROSSREF DONE")
|
||||
# TODO: add to sql with UUID
|
||||
# sql row - title, doi, author, uuid, last retrieved, is_available
|
||||
while self.total > self.fetched and self.limit > self.fetched:
|
||||
tmp_result = self.cr.works(query=query, limit=1000, offset=self.fetched)
|
||||
cr_result["message"]["items"].extend(tmp_result["message"]["items"])
|
||||
self.total = tmp_result["message"]["total-results"]
|
||||
self.fetched = len(cr_result["message"]["items"])
|
||||
self.app.logger.info(self.total)
|
||||
self.app.logger.info(self.fetched)
|
||||
time.sleep(0.1)
|
||||
|
||||
else:
|
||||
cr_result = self.cr.works(query=query, cursor_max=10000, cursor='*', progress_bar = True)
|
||||
result = cr_result[0]
|
||||
for item in cr_result[1:]:
|
||||
result["message"]["items"].extend(item["message"]["items"])
|
||||
self.total = item["message"]["total-results"]
|
||||
self.fetched = len(result["message"]["items"])
|
||||
self.app.logger.info(self.total)
|
||||
self.app.logger.info(self.fetched)
|
||||
cr_result = result
|
||||
self.app.logger.info("Total, fetched:")
|
||||
self.app.logger.info(self.total)
|
||||
self.app.logger.info(self.fetched)
|
||||
self.search_result_from_cr.update(cr_result)
|
||||
self.total = cr_result["message"]["total-results"]
|
||||
self.app.logger.info("CROSSREF DONE")
|
||||
|
||||
self.app.logger.info("CROSSREF DONE")
|
||||
with open("cr_results.json", "r+", encoding="utf-8") as data_file:
|
||||
# First we load existing data into a dict.
|
||||
try:
|
||||
file_data = json.load(data_file)
|
||||
except JSONDecodeError:
|
||||
file_data = {}
|
||||
data_file.truncate(0)
|
||||
data_file.seek(0)
|
||||
tmp = {self.uuid : self.search_result_from_cr}
|
||||
if file_data:
|
||||
file_data.update(tmp)
|
||||
else:
|
||||
file_data = tmp
|
||||
json.dump(file_data, data_file, indent=4)
|
||||
return cr_result
|
||||
|
||||
|
||||
async def refine_search(self, unrefined_result):
|
||||
"""
|
||||
Refines the search query based on the unrefined search result.
|
||||
|
||||
Args:
|
||||
- unrefined_result: The unrefined search result.
|
||||
|
||||
Returns:
|
||||
- refined_result: The refined search result.
|
||||
|
||||
Raises:
|
||||
- None.
|
||||
"""
|
||||
summarized_results = []
|
||||
self.app.logger.info("REFINE: Removing all derived works from the list")
|
||||
for item in unrefined_result["message"]["items"]:
|
||||
summarized_results.append(
|
||||
{
|
||||
"DOI": item["DOI"],
|
||||
"title": item["title"] if "title" in item else None,
|
||||
"type": item["type"] if "type" in item else None,
|
||||
}
|
||||
)
|
||||
|
||||
partial_result = {
|
||||
self.uuid: {
|
||||
"total_results": self.search_result_from_cr["message"][
|
||||
"total-results"
|
||||
],
|
||||
"on_page": 1,
|
||||
"summary": summarized_results,
|
||||
"results": self.search_result_from_cr["message"]["items"],
|
||||
}
|
||||
}
|
||||
self.app.logger.info("REFINE: Dumping to file")
|
||||
temp = []
|
||||
refined_result = {}
|
||||
for key in partial_result:
|
||||
self.app.logger.info("KEY:")
|
||||
self.app.logger.info(key)
|
||||
|
||||
for item in partial_result[self.uuid]["summary"]:
|
||||
if item["title"]:
|
||||
temp.append(item)
|
||||
|
||||
for item in temp:
|
||||
refined_result[item["DOI"]]= item
|
||||
with open("rr_results.json", "r+", encoding="utf-8") as data_file:
|
||||
# First we load existing data into a dict.
|
||||
try:
|
||||
file_data = json.load(data_file)
|
||||
except JSONDecodeError:
|
||||
file_data = {}
|
||||
summarized_results = []
|
||||
self.app.logger.info("REFINE: Removing all derived works from the list")
|
||||
for item in self.search_result_from_cr["message"]["items"]:
|
||||
container = "NO"
|
||||
if "container-title" in item:
|
||||
container = "YES"
|
||||
else:
|
||||
summarized_results.append(
|
||||
{
|
||||
"DOI": item["DOI"],
|
||||
"title": item["title"] if "title" in item else None,
|
||||
"type": item["type"] if "type" in item else None,
|
||||
"container": container,
|
||||
}
|
||||
)
|
||||
result = {
|
||||
self.uuid: {
|
||||
"total_results": self.search_result_from_cr["message"][
|
||||
"total-results"
|
||||
],
|
||||
"on_page": 1,
|
||||
"summary": summarized_results,
|
||||
"results": self.search_result_from_cr["message"]["items"],
|
||||
}
|
||||
}
|
||||
if file_data:
|
||||
file_data.update(result)
|
||||
else:
|
||||
file_data = result
|
||||
|
||||
self.app.logger.info("REFINE: Dumping to file")
|
||||
data_file.truncate(0)
|
||||
data_file.seek(0)
|
||||
tmp = {self.uuid: refined_result}
|
||||
if file_data:
|
||||
file_data.update(tmp)
|
||||
else:
|
||||
file_data = tmp
|
||||
json.dump(file_data, data_file, indent=4)
|
||||
return result
|
||||
return refined_result
|
||||
|
||||
async def check_dois_in_local_db(self, doi):
|
||||
"""
|
||||
Checks if the given DOI exists in the local database.
|
||||
|
||||
Args:
|
||||
- doi: The DOI to be checked.
|
||||
|
||||
Returns:
|
||||
- result: The search result from the backend app.
|
||||
|
||||
Raises:
|
||||
- None.
|
||||
"""
|
||||
self.app.logger.info("REFINE: Running search in the backend app")
|
||||
coro = asyncio.to_thread(
|
||||
search_bot.search_for_doi, doi, self.live_results, self.app.logger
|
||||
)
|
||||
result = await coro
|
||||
return result
|
||||
|
||||
async def check_if_exists_brute_force(self, page_url):
|
||||
"""
|
||||
Checks if the given page URL exists using brute force method.
|
||||
|
||||
Args:
|
||||
- page_url: The URL of the page to be checked.
|
||||
|
||||
Returns:
|
||||
- exists: A boolean indicating if the page exists.
|
||||
|
||||
Raises:
|
||||
- ValueError: If the URL does not start with 'http:' or 'https:'.
|
||||
"""
|
||||
self.app.logger.error("REFINE: Brute force search!")
|
||||
if not page_url.startswith(("http:", "https:")):
|
||||
raise ValueError("URL must start with 'http:' or 'https:'")
|
||||
# trunk-ignore(bandit/B310)
|
||||
with urlopen(page_url) as response:
|
||||
data = response.read()
|
||||
text = data.decode("utf-8")
|
||||
for line in text.splitlines():
|
||||
if re.match(
|
||||
r".*Unfortunately, Sci-Hub doesn't have the requested document.*",
|
||||
line,
|
||||
):
|
||||
return False
|
||||
if m := re.match(
|
||||
r".*<embed type=\"application.pdf\"\s*src=\"(.*\.pdf)", line
|
||||
):
|
||||
direct_download_link = "https:" + str(m.group(1))
|
||||
self.app.logger.info(direct_download_link)
|
||||
return True
|
||||
return False
|
||||
|
||||
async def check_if_exists(self, doi, brute_force=False):
|
||||
async def check_if_exists(self, refined_result):
|
||||
"""
|
||||
Checks if the given DOI exists.
|
||||
|
||||
@@ -229,56 +242,26 @@ class Librarian(object):
|
||||
- None.
|
||||
"""
|
||||
result = {}
|
||||
result = await self.check_dois_in_local_db(doi)
|
||||
if brute_force and not result:
|
||||
for item in doi:
|
||||
item_link = "https://sci-hub.se/" + item[0]
|
||||
tmp = {
|
||||
"DOI": item["DOI"],
|
||||
"exists": await self.check_if_exists_brute_force(item_link),
|
||||
"data": item,
|
||||
}
|
||||
result.update(tmp)
|
||||
return result
|
||||
|
||||
async def refine_query(self, unrefined_result):
|
||||
"""
|
||||
Refines the search query based on the unrefined search result.
|
||||
|
||||
Args:
|
||||
- unrefined_result: The unrefined search result.
|
||||
- brute_force: A flag indicating if brute force method should be used.
|
||||
|
||||
Returns:
|
||||
- final_result: The final refined search result.
|
||||
|
||||
Raises:
|
||||
- None.
|
||||
"""
|
||||
temp = []
|
||||
refined_result = []
|
||||
for key in unrefined_result:
|
||||
self.app.logger.info("KEY:")
|
||||
self.app.logger.info(key)
|
||||
|
||||
for item in unrefined_result[self.uuid]["summary"]:
|
||||
if item["container"] == "NO" and item["title"]:
|
||||
refined_result.append(item)
|
||||
|
||||
for item in refined_result:
|
||||
temp.append([item["DOI"], item])
|
||||
|
||||
refined_result = await self.check_if_exists(temp)
|
||||
|
||||
for item in refined_result:
|
||||
self.app.logger.info("REFINE: Running search in the backend app")
|
||||
dois = []
|
||||
for item, value in refined_result.items():
|
||||
dois.append([item, value])
|
||||
coro = asyncio.to_thread(
|
||||
search_bot.search_for_doi, dois, self.live_results, self.app.logger
|
||||
)
|
||||
result = await coro
|
||||
result_list = []
|
||||
result_no_db = []
|
||||
for item in result:
|
||||
if item["exists"]:
|
||||
self.final_result.append(item["data"])
|
||||
result_list.append(item)
|
||||
else:
|
||||
self.not_in_db.append(item["data"])
|
||||
self.hit = len(self.final_result)
|
||||
return self.final_result
|
||||
result_no_db.append(item)
|
||||
self.hit = len(result)
|
||||
return result_list, result_no_db
|
||||
|
||||
async def answer_query(self):
|
||||
|
||||
async def answer_query(self, deep_search=False):
|
||||
"""
|
||||
Answers the search query.
|
||||
|
||||
@@ -292,57 +275,21 @@ class Librarian(object):
|
||||
- None.
|
||||
"""
|
||||
self.app.logger.info(f"Search started {self.uuid}")
|
||||
database = None
|
||||
query = self.query
|
||||
with open("cr_results.json", "r+", encoding="utf-8") as cr_file:
|
||||
try:
|
||||
database = json.load(cr_file)
|
||||
except JSONDecodeError:
|
||||
pass
|
||||
if database:
|
||||
for item in database.keys():
|
||||
if item in self.uuid:
|
||||
self.result = await self.refine_query(database)
|
||||
temp = {}
|
||||
for item in self.result:
|
||||
temp[item["DOI"]] = {
|
||||
"Title": item["title"],
|
||||
"type": item["type"],
|
||||
}
|
||||
self.result = temp
|
||||
temp2 = {}
|
||||
for item in self.not_in_db:
|
||||
temp2[item["DOI"]] = {
|
||||
"Title": item["title"],
|
||||
"type": item["type"],
|
||||
}
|
||||
self.not_in_db = temp2
|
||||
return self.result
|
||||
answer = await self.search_crossref(query=query)
|
||||
database.update(answer)
|
||||
cr_file.truncate(0)
|
||||
cr_file.seek(0)
|
||||
json.dump(database, cr_file)
|
||||
self.result = await self.refine_query(answer)
|
||||
else:
|
||||
answer = await self.search_crossref(query=query)
|
||||
database = answer
|
||||
cr_file.truncate(0)
|
||||
cr_file.seek(0)
|
||||
json.dump(database, cr_file)
|
||||
self.result = await self.refine_query(answer)
|
||||
self.app.logger.info("Refined")
|
||||
self.done = True
|
||||
temp = {}
|
||||
for item in self.result:
|
||||
temp[item["DOI"]] = {"Title": item["title"], "type": item["type"]}
|
||||
self.result = temp
|
||||
temp2 = {}
|
||||
for item in self.not_in_db:
|
||||
temp2[item["DOI"]] = {"Title": item["title"], "type": item["type"]}
|
||||
self.not_in_db = temp2
|
||||
return self.result
|
||||
cr_result = await self.search_crossref(query=self.query, deep_search=deep_search)
|
||||
refined_result = await self.refine_search(cr_result)
|
||||
answer, negative_answer = await self.check_if_exists(refined_result)
|
||||
|
||||
self.app.logger.info("Returning result")
|
||||
self.app.logger.info(answer)
|
||||
self.app.logger.info(negative_answer)
|
||||
|
||||
for item in answer:
|
||||
self.final_result[item["DOI"]] = {"Title": item["data"]["title"], "type": item["data"]["type"]}
|
||||
for item in negative_answer:
|
||||
self.not_in_db[item["DOI"]] = {"Title": item["data"]["title"], "type": item["data"]["type"]}
|
||||
self.app.logger.info("Returning result case2")
|
||||
self.app.logger.info(self.final_result)
|
||||
return self.final_result
|
||||
|
||||
# ============================= FLASK INTERNALS===============================
|
||||
|
||||
@@ -418,7 +365,7 @@ class BackgroundTaskSearch(threading.Thread):
|
||||
ndb_database = None
|
||||
librarian = librarian_queue.get()
|
||||
self.app.logger.info("STARTED")
|
||||
result = await librarian.answer_query()
|
||||
result = await librarian.answer_query(librarian.deep_search)
|
||||
result = {librarian.uuid: result}
|
||||
self.app.logger.info("Saving to file")
|
||||
|
||||
@@ -475,8 +422,11 @@ async def query_database():
|
||||
Endpoint for querying the database.
|
||||
|
||||
This function receives a POST request containing a JSON payload with a query and a UUID.
|
||||
It creates a Librarian object with the query and UUID, and adds it to the librarian_queue and librarian_list.
|
||||
Finally, it returns a JSON response indicating the success of the operation, along with the query, UUID, and the current size of the librarian_queue.
|
||||
It creates a Librarian object with the query and UUID,
|
||||
and adds it to the librarian_queue and librarian_list.
|
||||
Finally, it returns a JSON response indicating the success
|
||||
of the operation, along with the query, UUID,
|
||||
and the current size of the librarian_queue.
|
||||
|
||||
Returns:
|
||||
tuple: A tuple containing a JSON response and a status code.
|
||||
@@ -486,7 +436,8 @@ async def query_database():
|
||||
app.logger.info(record["query"])
|
||||
app.logger.info(record["UUID"])
|
||||
uuid = record["UUID"]
|
||||
cl = Librarian(app, record["query"], uuid)
|
||||
deep_search = record["deep_search"]
|
||||
cl = Librarian(app, record["query"], uuid, deep_search)
|
||||
librarian_queue.put(cl)
|
||||
librarian_list.append(cl)
|
||||
answer_data = (record["query"], record["UUID"], librarian_queue.qsize())
|
||||
@@ -523,7 +474,7 @@ async def get_partial():
|
||||
if __name__ == "__main__":
|
||||
app.logger.setLevel(logging.DEBUG)
|
||||
h1 = handlers.RotatingFileHandler(
|
||||
filename="E:\\logs\\librarian.log",
|
||||
filename="D:\\logs\\librarian.log",
|
||||
encoding="utf-8",
|
||||
mode="a",
|
||||
maxBytes=6 * 1024 * 1024,
|
||||
|
||||
@@ -1,16 +1,33 @@
|
||||
#!/bin/bash
|
||||
# This script sets up a virtual environment for the Conjurer Librarian service and installs the required dependencies.
|
||||
|
||||
|
||||
# Create a virtual environment at /home/pi/Conjurer_librarian/env
|
||||
python3 -m venv /home/pi/Conjurer_librarian/env
|
||||
cp /home/pi/conjurer/conjurer_librarian/conjurer_librarian.py ./Conjurer_librarian/
|
||||
cp /home/pi/conjurer/conjurer_librarian/requirements_librarian.txt ./Conjurer_librarian/
|
||||
|
||||
# Copy the Conjurer Librarian Python script to the appropriate directory
|
||||
cp /home/pi/conjurer/conjurer_librarian/conjurer_librarian.py ./Conjurer_librarian/
|
||||
|
||||
# Copy the requirements file to the appropriate directory
|
||||
cp /home/pi/conjurer/conjurer_librarian/requirements_librarian.txt ./Conjurer_librarian/
|
||||
|
||||
# Change directory to /home/pi/Conjurer_librarian
|
||||
cd /home/pi/Conjurer_librarian
|
||||
|
||||
touch /home/pi/Conjurer/rr_results.json
|
||||
touch /home/pi/Conjurer/cr_results.json
|
||||
touch /home/pi/Conjurer/s_results.json
|
||||
touch /home/pi/Conjurer/rr_results.json
|
||||
touch /home/pi/Conjurer/not_in_db.json
|
||||
source /home/pi/Conjurer_librarian/env/bin/activate
|
||||
./env/bin/python3 -m pip install --upgrade pip
|
||||
./env/bin/python3 -m pip install -r requirements_librarian.txt
|
||||
deactivate
|
||||
|
||||
cp /home/pi/conjurer/conjurer_librarian/conjurer_librarian.py ./Conjurer_librarian/
|
||||
# Activate the virtual environment
|
||||
source /home/pi/Conjurer_librarian/env/bin/activate
|
||||
|
||||
# Upgrade pip
|
||||
./env/bin/python3 -m pip install --upgrade pip
|
||||
|
||||
# Install the required dependencies
|
||||
./env/bin/python3 -m pip install -r requirements_librarian.txt
|
||||
|
||||
# Deactivate the virtual environment
|
||||
deactivate
|
||||
|
||||
@@ -1,46 +1,67 @@
|
||||
"""
|
||||
This module contains the code for the scrape bot.
|
||||
"""
|
||||
|
||||
import json
|
||||
from json.decoder import JSONDecodeError
|
||||
from urllib.request import urlopen
|
||||
from requests import Timeout,ConnectTimeout, ConnectionError
|
||||
from queue import Queue
|
||||
import logging
|
||||
import random
|
||||
import re
|
||||
import time
|
||||
import random
|
||||
import logging
|
||||
from json import JSONDecodeError
|
||||
from queue import Queue
|
||||
from threading import Thread
|
||||
SCR_DATABASE_PATH = r'C:\\Database\\chunks\\'
|
||||
from urllib.request import urlopen
|
||||
|
||||
from requests import ConnectionError as RequestsConnectionError
|
||||
from requests import ConnectTimeout, Timeout
|
||||
|
||||
SCR_DATABASE_PATH = r"C:\\Database\\chunks\\"
|
||||
SCR_FILENAME = "40_chunk.txt"
|
||||
SCR_ENCODING = "utf-8"
|
||||
|
||||
WORK_Q = Queue()
|
||||
random.seed()
|
||||
|
||||
|
||||
def load_ndb_to_q(logger):
|
||||
"""
|
||||
This function loads items from the not_in_db.json file into the work queue.
|
||||
"""
|
||||
logger.info("Loader started")
|
||||
while True:
|
||||
while True:
|
||||
with open("not_in_db.json", "r+", encoding="utf-8") as ndb_file:
|
||||
try:
|
||||
ndb_database = json.load(ndb_file)
|
||||
for item in ndb_database.keys():
|
||||
logger.info(item)
|
||||
url = f"https://sci-hub.se/{item}"
|
||||
ndb_database = json.load(ndb_file)
|
||||
for _ in range (1,10):
|
||||
key = next(iter(ndb_database))
|
||||
item = ndb_database.pop(key)
|
||||
logger.info(key)
|
||||
url = f"https://sci-hub.se/{key}"
|
||||
WORK_Q.put([item, url, False])
|
||||
|
||||
ndb_file.truncate(0)
|
||||
ndb_file.seek(0)
|
||||
json.dump(ndb_database, ndb_file, indent=4)
|
||||
except JSONDecodeError:
|
||||
time.sleep(60*60*2)
|
||||
pass
|
||||
time.sleep(180)
|
||||
time.sleep(60 * 60 * 3)
|
||||
time.sleep(60*60*3)
|
||||
|
||||
|
||||
def check_if_exists_brute_force(logger):
|
||||
"""
|
||||
This function checks if a document exists using brute force search.
|
||||
"""
|
||||
# Function code here
|
||||
while True:
|
||||
logger.info("Scraper tick")
|
||||
item = WORK_Q.get()
|
||||
page_url =item[1]
|
||||
page_url = item[1]
|
||||
logger.error("REFINE: Brute force search!")
|
||||
if not page_url.startswith(("http:", "https:")):
|
||||
raise ValueError("URL must start with 'http:' or 'https:'")
|
||||
blocked = True
|
||||
try:
|
||||
# trunk-ignore(bandit/B310)
|
||||
# trunk-ignore(bandit/B310)
|
||||
with urlopen(page_url) as response:
|
||||
data = response.read()
|
||||
text = data.decode("utf-8")
|
||||
@@ -50,44 +71,64 @@ def check_if_exists_brute_force(logger):
|
||||
blocked = False
|
||||
else:
|
||||
for line in text.splitlines():
|
||||
if re.match(r".*Unfortunately, Sci-Hub doesn't have the requested document.*", line):
|
||||
if re.match(
|
||||
r".*Unfortunately, Sci-Hub doesn't have the requested document.*",
|
||||
line,
|
||||
):
|
||||
blocked = False
|
||||
logger.info("Not found")
|
||||
item[2] = False
|
||||
if m := re.match(r".*<embed type=\"application.pdf\"\s*src=\"(.*\.pdf)",line):
|
||||
if m := re.match(
|
||||
r".*<embed type=\"application.pdf\"\s*src=\"(.*\.pdf)", line
|
||||
):
|
||||
blocked = False
|
||||
direct_download_link = "https:" + str(m.group(1))
|
||||
logger.info (direct_download_link)
|
||||
with open(SCR_DATABASE_PATH + SCR_FILENAME, "a", encoding=SCR_ENCODING) as operated_file:
|
||||
logger.info(direct_download_link)
|
||||
with open(
|
||||
SCR_DATABASE_PATH + SCR_FILENAME,
|
||||
"a",
|
||||
encoding=SCR_ENCODING,
|
||||
) as operated_file:
|
||||
operated_file.write("\n")
|
||||
operated_file.write(item[0])
|
||||
item[2] = True
|
||||
except (Timeout,ConnectTimeout, ConnectionRefusedError, ConnectionError):
|
||||
except (
|
||||
Timeout,
|
||||
ConnectTimeout,
|
||||
ConnectionRefusedError,
|
||||
ConnectionError,
|
||||
RequestsConnectionError,
|
||||
):
|
||||
pass
|
||||
if blocked:
|
||||
logger.info(item)
|
||||
logger.info(text)
|
||||
logger.error("Got blocked. Fuck.")
|
||||
time.sleep(60*60*72)
|
||||
time.sleep(60 * 60 * 72)
|
||||
# trunk-ignore(bandit/B311)
|
||||
rand = random.randint(1,100)
|
||||
logger.info(f"Sleeping for {3*rand} minutes")
|
||||
time.sleep(180*rand)
|
||||
rand = random.randint(1, 60)
|
||||
logger.info(f"Sleeping for {2*rand} minutes")
|
||||
time.sleep(120 * rand)
|
||||
|
||||
def scraper(logger = None):
|
||||
|
||||
def scraper(logger=None):
|
||||
"""
|
||||
This function is responsible for scraping data.
|
||||
"""
|
||||
if not logger:
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel("DEBUG")
|
||||
h1 = logging.StreamHandler()
|
||||
logger.addHandler(h1)
|
||||
logger.info("No logger set. Test run")
|
||||
|
||||
loader = Thread(target = load_ndb_to_q, args = (logger,))
|
||||
scraper = Thread(target = check_if_exists_brute_force, args = (logger,))
|
||||
|
||||
loader = Thread(target=load_ndb_to_q, args=(logger,))
|
||||
_scraper = Thread(target=check_if_exists_brute_force, args=(logger,))
|
||||
loader.start()
|
||||
scraper.start()
|
||||
_scraper.start()
|
||||
loader.join()
|
||||
scraper.join()
|
||||
_scraper.join()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
scraper()
|
||||
scraper()
|
||||
|
||||
@@ -1,26 +1,55 @@
|
||||
#TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry
|
||||
"""
|
||||
This module contains functions for searching for DOI (Digital Object Identifier)
|
||||
in a list of live results.
|
||||
It includes a producer-consumer pattern implementation using threads and queues.
|
||||
|
||||
from queue import Queue, Empty
|
||||
Functions:
|
||||
- producer: Reads lines from a file and puts them into an output queue.
|
||||
- consumer: Consumes items from an input queue and checks if DOI exists in the live results list.
|
||||
- search_for_doi: Searches for DOI in live results using the producer-consumer pattern.
|
||||
|
||||
Global Variables:
|
||||
- MAXTHREADS: Maximum number of worker threads.
|
||||
- DATABASE_PATH: Path to the database files.
|
||||
- ENCODING: Encoding of the database files.
|
||||
- CHUNK: File name pattern for the database files.
|
||||
- _sentinel: Sentinel object used to signal termination.
|
||||
- result_list: List to store the search results.
|
||||
- WORK_Q_SIZE: Maximum size of the work queue.
|
||||
"""
|
||||
|
||||
# TODO: Wpiemdolić to wszystko w klasę z loggerem przysłanym z góry
|
||||
from queue import Empty, Queue
|
||||
from threading import Thread
|
||||
|
||||
q = Queue()
|
||||
|
||||
#TODO: DATA FOR TEST ONLY
|
||||
#MAXTHREADS = 5
|
||||
#DATABASE_PATH = r'C:\\Database\\chunks_1\\'
|
||||
# DATA FOR TEST ONLY
|
||||
# MAXTHREADS = 5
|
||||
# DATABASE_PATH = r'C:\\Database\\chunks_1\\'
|
||||
|
||||
#TODO: DEPLOYMENT DATA
|
||||
# DEPLOYMENT DATA
|
||||
MAXTHREADS = 41
|
||||
DATABASE_PATH = r'C:\\Database\\chunks\\'
|
||||
DATABASE_PATH = r"C:\\Database\\chunks\\"
|
||||
|
||||
ENCODING = "utf-8"
|
||||
CHUNk = "_chunk.txt"
|
||||
CHUNK = "_chunk.txt"
|
||||
_sentinel = object()
|
||||
result_list = []
|
||||
WORK_Q_SIZE = 15500000
|
||||
|
||||
def producer(out_q, control_q, filename, logger):
|
||||
|
||||
def producer(out_q, control_q, filename, _logger):
|
||||
"""
|
||||
Produces items from the output queue and puts them into the control queue.
|
||||
|
||||
Args:
|
||||
out_q (Queue): Output queue.
|
||||
control_q (Queue): Control queue.
|
||||
filename (str): Name of the file.
|
||||
_logger: Logger object for logging.
|
||||
"""
|
||||
with open(DATABASE_PATH + filename, "r+", encoding=ENCODING) as operated_file:
|
||||
logger.info(f"Worker {filename} ")
|
||||
_logger.info(f"Worker {filename} ")
|
||||
while True:
|
||||
line = operated_file.readline()
|
||||
if not line:
|
||||
@@ -30,77 +59,107 @@ def producer(out_q, control_q, filename, logger):
|
||||
check = control_q.get(block=False)
|
||||
except Empty:
|
||||
check = False
|
||||
pass
|
||||
if check is _sentinel:
|
||||
logger.info("TERM signal received")
|
||||
_logger.info("TERM signal received")
|
||||
control_q.put(check)
|
||||
break
|
||||
logger.info(f"Worker finished: {filename}")
|
||||
_logger.info(f"Worker finished: {filename}")
|
||||
out_q.put(_sentinel)
|
||||
|
||||
#IMPORTANT!!! ONLY ONE CONSUMER THREAD AS WE ARE NOT PUTTING SENTINELS BACK
|
||||
#TODO: Change doi to tuple list and only put term when all doi are found => (doi, True)
|
||||
def consumer(in_q, control_q, doi, live_results, logger):
|
||||
|
||||
|
||||
# IMPORTANT!!! ONLY ONE CONSUMER THREAD AS WE ARE NOT PUTTING SENTINELS BACK
|
||||
def consumer(in_q, control_q, doi, live_results, result_list, _logger, ):
|
||||
"""
|
||||
Consumes items from an input queue and checks if DOI exists in the live results list.
|
||||
|
||||
Args:
|
||||
in_q (Queue): Input queue.
|
||||
control_q (Queue): Control queue.
|
||||
doi (list): List of DOI to search for.
|
||||
live_results (list): List to store the search results.
|
||||
_logger: Logger object for logging.
|
||||
"""
|
||||
sentinels = 0
|
||||
for item in doi:
|
||||
result_list.append({"DOI" : item[0], "exists" : False, "data" : item[1]})
|
||||
result_list.append({"DOI": item[0], "exists": False, "data": item[1]})
|
||||
while True:
|
||||
done_check = True
|
||||
data = in_q.get()
|
||||
if data is _sentinel:
|
||||
logger.info("Worker finished")
|
||||
_logger.info("Worker finished")
|
||||
sentinels += 1
|
||||
else:
|
||||
for item in result_list:
|
||||
if item["DOI"] in data and not item["exists"]:
|
||||
logger.info(data)
|
||||
logger.info("HIT")
|
||||
_logger.info(data)
|
||||
_logger.info("HIT")
|
||||
item["exists"] = True
|
||||
live_results.append(item)
|
||||
done_check = done_check and item["exists"]
|
||||
if done_check:
|
||||
control_q.put(_sentinel)
|
||||
if sentinels >= MAXTHREADS:
|
||||
logger.info("All workers finished")
|
||||
_logger.info("All workers finished")
|
||||
break
|
||||
|
||||
|
||||
def search_for_doi(doi, live_results, logger):
|
||||
def search_for_doi(doi, live_results, _logger):
|
||||
"""
|
||||
Search for DOI in live_results using _logger for logging.
|
||||
|
||||
Args:
|
||||
doi (list): List of DOI to search for.
|
||||
live_results (list): List to store the search results.
|
||||
_logger: Logger object for logging.
|
||||
"""
|
||||
result_list = []
|
||||
threads = []
|
||||
work_q = Queue(maxsize=WORK_Q_SIZE)
|
||||
control_q = Queue()
|
||||
t_cons = Thread(target = consumer, args = (work_q, control_q, doi, live_results, logger))
|
||||
logger.info("Consumer thread created")
|
||||
t_cons = Thread(
|
||||
target=consumer, args=(work_q, control_q, doi, live_results, result_list, _logger)
|
||||
)
|
||||
_logger.info("Consumer thread created")
|
||||
threads.append(t_cons)
|
||||
for i in range(0,MAXTHREADS):
|
||||
#TODO: TEST DATA
|
||||
#filename = "test" + str(i) + CHUNk
|
||||
#TODO: DEPLOYMENT DATA
|
||||
filename = str(i) + CHUNk
|
||||
logger.info(f"Creating worker thread no: {i}")
|
||||
threads.append(Thread(target = producer, args = (work_q, control_q, filename,logger)))
|
||||
for i in range(0, MAXTHREADS):
|
||||
# TEST DATA
|
||||
# filename = "test" + str(i) + CHUNk
|
||||
# DEPLOYMENT DATA
|
||||
filename = str(i) + CHUNK
|
||||
_logger.info(f"Creating worker thread no: {i}")
|
||||
threads.append(
|
||||
Thread(target=producer, args=(work_q, control_q, filename, _logger))
|
||||
)
|
||||
for worker in threads:
|
||||
worker.start()
|
||||
for worker in threads:
|
||||
worker.join()
|
||||
return result_list
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel("DEBUG")
|
||||
h1 = logging.StreamHandler()
|
||||
logger.addHandler(h1)
|
||||
logger.info("TEST RUN")
|
||||
live_result = []
|
||||
logger.info(search_for_doi([
|
||||
("10.1002/9781118786352.wbieg0998.pub2","DATA"),
|
||||
("10.1002/j.2050-0416.2002.tb00563.x","DATA"),
|
||||
("10.1111/j.1365-2958.1994.tb00448.x","DATA"),
|
||||
("10.2165/00128415-200309690-00017","DATA"),
|
||||
("10.5772/48313","DATA"),
|
||||
("10.15803/ijnc.7.2_419","DATA"),
|
||||
("10.1051/0004-6361/201321596e","DATA"),
|
||||
("10.2307/40835941","DATA"),
|
||||
], live_result, logger))
|
||||
logger.info(
|
||||
search_for_doi(
|
||||
[
|
||||
("10.1002/9781118786352.wbieg0998.pub2", "DATA"),
|
||||
("10.1002/j.2050-0416.2002.tb00563.x", "DATA"),
|
||||
("10.1111/j.1365-2958.1994.tb00448.x", "DATA"),
|
||||
("10.2165/00128415-200309690-00017", "DATA"),
|
||||
("10.5772/48313", "DATA"),
|
||||
("10.15803/ijnc.7.2_419", "DATA"),
|
||||
("10.1051/0004-6361/201321596e", "DATA"),
|
||||
("10.2307/40835941", "DATA"),
|
||||
],
|
||||
live_result,
|
||||
logger,
|
||||
)
|
||||
)
|
||||
logger.info(live_result)
|
||||
|
||||
Reference in New Issue
Block a user