Coverage for src/crawler/tasks.py: 0%

1import logging

2import os

3import subprocess

4import time

5import traceback

6from os import path

7from typing import TYPE_CHECKING

9import pypdf

10import requests

11from celery import chain, group, shared_task, states

12from django.conf import settings

13from django.contrib.auth.models import User

14from django.db.models import Q

15from history.model_data import HistoryEventStatus

16from history.utils import insert_history_event

17from opentelemetry import trace

18from ptf.display import resolver

19from ptf.models import Article, Collection, Container

20from ptf.models.classes.datastream import DataStream

21from task import TaskAborted

22from task.custom_task import PtfAbortableTask, TaskWithProgress

23from task.tasks import increment_progress

25from crawler.factory import crawler_factory

26from crawler.models import Source

27from crawler.utils import get_all_cols

29request_interval = getattr(settings, "REQUESTS_INTERVAL", 3)

30chunk_size_collections = settings.CHUNK_SIZE_COLLECTIONS

31chunk_size_sources = settings.CHUNK_SIZE_SOURCES

33if TYPE_CHECKING:

34 from history.model_data import HistoryEventDict

35 from ptf.model_data import IssueData

37tracer = trace.get_tracer(__name__)

38logger = logging.getLogger(__name__)

41def get(href, retries=0):

42 try:

43 r = requests.get(href, timeout=10.0, verify=False)

44 return r

45 except (requests.ConnectionError, requests.ConnectTimeout) as e:

46 if retries >= 3:

47 raise e

48 logger.info("Retry query %s", href)

49 time.sleep(60)

50 return get(href, retries + 1)

53def download_pdf(obj: Article, remove_first_page: bool, only_new: bool, pause_function=time.sleep):

54 collection = obj.get_collection()

56 qs = obj.extlink_set.filter(rel="article-pdf").all()

57 if not qs:

58 logger.warning(f"No PDF link for {obj.pid}")

59 return

61 extlink = qs.first()

62 href = extlink.location

64 if hasattr(obj, "my_container"):

65 container_id = obj.my_container.pid

66 obj_id = obj.pid.replace("/", "_")

67 else:

68 # Download a PDF of a book

69 container_id = obj.pid

70 obj_id = None

72 if href.find("http") == 0:

73 disk_location = resolver.get_disk_location(

74 settings.RESOURCES_ROOT,

75 collection.pid,

76 "pdf",

77 container_id=container_id,

78 article_id=obj_id,

79 do_create_folder=True,

80 )

81 logger.info("disk_location: %s", disk_location)

82 if not (os.path.isfile(disk_location) and only_new):

83 # Download unless the file is already present AND only_new = True

84 logger.info("tempo: %s", request_interval)

85 pause_function(request_interval)

86 r = get(href)

87 logger.info("http response status: %s", r.status_code)

88 if len(r.text) > 10 and r.text[0:21] == "<!DOCTYPE html PUBLIC":

89 print(f"{obj.doi} has an embargo, no PDF")

91 else:

92 # Remove front page

93 if remove_first_page:

94 temp_location = os.path.join(settings.TEMP_FOLDER, "file.pdf")

95 with open(temp_location, "wb") as f_:

96 f_.write(r.content)

98 pdf_reader = pypdf.PdfReader(temp_location)

99 pdf_writer = pypdf.PdfWriter()

100 for page in range(len(pdf_reader.pages)):

101 current_page = pdf_reader.pages[page]

102 if page > 0:

103 pdf_writer.add_page(current_page)

104

105 with open(disk_location, "wb") as f_:

106 pdf_writer.write(f_)

107 logger.info("PDF file saved on disk, %s", disk_location)

108 else:

109 with open(disk_location, "wb") as f_:

110 f_.write(r.content)

111 logger.info("PDF file saved on disk, %s", disk_location)

112

113 else:

114 disk_location = resolver.get_disk_location(

115 settings.RESOURCES_ROOT,

116 collection.pid,

117 "pdf",

118 container_id=container_id,

119 article_id=obj_id,

120 do_create_folder=False,

121 )

122

123 if not os.path.isfile(disk_location):

124 logger.debug(f"Directory does not yet exists : {disk_location}")

125 new_location = resolver.get_relative_folder(

126 collection.pid, container_id=container_id, article_id=obj_id

127 )

128 pdf_filename = os.path.join(new_location, obj.pid + ".pdf")

129

130 qs = obj.datastream_set.filter(mimetype="application/pdf")

131

132 if qs:

133 datastream = qs.first()

134 else:

135 datastream = DataStream()

136 datastream.resource = obj

137

138 if datastream.location != pdf_filename:

139 datastream.location = pdf_filename

140 datastream.save()

141 return disk_location

142

143

144def crawl_sources(

145 user_name,

146 only_new=False,

147 period: tuple[int, int] = (0, 9999),

148 number: tuple[int, int] = (0, 99999),

149):

150 logger.info("Start crawling all sources")

151 sources = Source.objects.exclude(domain="NUMDAM")

152

153 # we launch the source crawlings concurrently

154 for source in sources:

155 collections = (

156 Collection.objects.filter(content__origin__source=source)

157 .distinct()

158 .order_by("pid")

159 .values("pid")

160 )

161 colids = [col["pid"] for col in collections]

162 crawl_source.delay(colids, source.domain, user_name, only_new, period, number)

163

164

165@shared_task(

166 name="crawler.tasks.download_sources",

167 bind=True,

168 queue="coordinator",

169 base=TaskWithProgress,

170)

171def download_sources(

172 self: "TaskWithProgress",

173 user_name,

174 only_new=False,

175 period: tuple[int, int] = (0, 9999),

176 number: tuple[int, int] = (0, 99999),

177):

178 logger.info("Start downloading PDF from all sources")

179

180 event_dict: "HistoryEventDict" = {

181 "type": "download-sources",

182 "pid": "all sources",

183 "col": None,

184 "status": HistoryEventStatus.PENDING,

185 }

186

187 sources = Source.objects.exclude(domain__in=["NUMDAM", "GEODESIC"])

188 # source_liste = ["SCHOLASTICA", "BMMS", "PTM", "ARSIA", "IPB"]

189 # sources = Source.objects.filter(domain__in=source_liste)

190

191 source_count = sources.count()

192 logger.info(f"Downloading article PDFs for {source_count} sources")

193 self.update_state(

194 meta={

195 "current": 1,

196 "total": source_count,

197 "progress": 1 / source_count,

198 "col": None,

199 },

200 state=states.STARTED,

201 )

202

203 # we launch the sources downloading concurrently

204 task_chains = []

205 for num, source in enumerate(sources):

206 logger.info(f"sources: {source.domain}")

207 collections = (

208 Collection.objects.filter(content__origin__source=source)

209 .distinct()

210 .order_by("pid")

211 .values("pid")

212 )

213 colids = [col["pid"] for col in collections]

214 if len(colids) > 0:

215 source_task = download_source.si(

216 colids, source.domain, user_name, only_new, period, number

217 )

218 increment_task = increment_progress.si(self.request.id)

219 task_chain = chain(source_task, increment_task)

220 task_chains.append(task_chain)

221 logger.info("Source %s : launch the PDF download.", source.domain)

222 if len(task_chains) == chunk_size_sources or num == source_count - 1:

223 try:

224 task_group = group(*task_chains).delay()

225 self.wait_child(task_group, propagate=False)

226 task_chains = []

227 except TaskAborted:

228 logger.info("Download all sources ABORTED")

229 event_dict["status"] = HistoryEventStatus.ERROR

230 event_dict["message"] = "Task aborted by user"

231 insert_history_event(event_dict)

232 logger.error(event_dict["message"])

233 raise

234 event_dict["status"] = HistoryEventStatus.OK

235 logger.info("Download all source FINISHED")

236 insert_history_event(event_dict)

237

238

239@shared_task(

240 name="crawler.tasks.download_source",

241 bind=True,

242 queue="coordinator",

243 base=TaskWithProgress,

244)

245def download_source(

246 self: "TaskWithProgress",

247 colids: list,

248 source_domain: str,

249 user_name,

250 only_new=False,

251 period: tuple[int, int] = (0, 9999),

252 number: tuple[int, int] = (0, 99999),

253):

254 # TODO comment gérer le parent_id de HistoryEvent des collections pour qu'il pointe vers celui des sources ?

255 event_dict: "HistoryEventDict" = {

256 "type": "download-source",

257 "pid": source_domain,

258 "col": None,

259 "status": HistoryEventStatus.PENDING,

260 }

261 logger.info("Start downloading the PDFs for the source: %s", source_domain)

262 logger.info("Number of collections: %s", len(colids))

263 try:

264 self.update_state(

265 meta={

266 "current": 1,

267 "total": len(colids),

268 "progress": 1 / len(colids),

269 "col": source_domain,

270 },

271 state=states.STARTED,

272 )

273 results = []

274 for col in colids:

275 logger.info("Start dowloading the pdf from the collection: %s", col)

276 promise = download_collection.delay(

277 col, source_domain, user_name, only_new, period, number

278 )

279 result = self.wait_child(promise, propagate=True)

280 logger.info(f"after wait_child: {col}")

281 # result = promise.get(disable_sync_subtasks=False, propagate=True)

282 results.append(result)

283 increment_progress.delay(self.request.id)

284 exceptions = [result for result in results if isinstance(result, Exception)]

285 if len(exceptions) > 0:

286 raise ExceptionGroup("Encountered errors while processing subtasks", exceptions)

287 event_dict["status"] = HistoryEventStatus.OK

288 # check the collection status

289

290 except TaskAborted:

291 event_dict["status"] = HistoryEventStatus.ERROR

292 event_dict["message"] = "Task aborted by user"

293 raise

294 except Exception:

295 event_dict["status"] = HistoryEventStatus.ERROR

296 event_dict["message"] = traceback.format_exc()

297 logger.error(event_dict["message"])

298 promise.abort()

299 raise

300 finally:

301 logger.info("Download one source finished")

302 # TODO fetch the article numbers

303 total_files = 0

304 total_articles = 0

305 for colid in colids:

306 file_count, article_count = articles_count(colid, source_domain)

307 total_articles += article_count

308 total_files += file_count

309 if total_files != total_articles:

310 event_dict["status"] = HistoryEventStatus.WARNING

311 event_dict["message"] = (

312 f"{total_articles} articles à télécharger \n{total_files} articles sur disque."

313 )

314 insert_history_event(event_dict)

315

316

317@shared_task(

318 name="crawler.tasks.crawl_source",

319 bind=True,

320 queue="coordinator",

321 base=TaskWithProgress,

322)

323def crawl_source(

324 self: "TaskWithProgress",

325 colids: list,

326 source_domain: str,

327 user_name,

328 only_new=False,

329 period: tuple[int, int] = (0, 9999),

330 number: tuple[int, int] = (0, 99999),

331):

332 event_dict: "HistoryEventDict" = {

333 "type": "import-source",

334 "pid": "import all",

335 "col": None,

336 "status": HistoryEventStatus.PENDING,

337 }

338 logger.info("Start crawling the source: %s", source_domain)

339 try:

340 self.update_state(

341 meta={"current": 0, "total": len(colids), "progress": 0, "col": source_domain},

342 state=states.STARTED,

343 )

344 results = []

345 for col in colids:

346 logger.info("Start crawling the collection: %s", col)

347 promise = crawl_collection.delay(

348 col, source_domain, user_name, only_new, period, number

349 )

350

351 result = self.wait_child(promise, propagate=True)

352 results.append(result)

353 increment_progress.delay(self.request.id)

354

355 event_dict["status"] = HistoryEventStatus.OK

356 exceptions = [result for result in results if isinstance(result, Exception)]

357 if len(exceptions) > 0:

358 raise ExceptionGroup("Encountered errors while processing subtasks", exceptions)

359 except Exception:

360 event_dict["status"] = HistoryEventStatus.ERROR

361 event_dict["message"] = traceback.format_exc()

362 logger.error(event_dict["message"])

363 raise

364 finally:

365 insert_history_event(event_dict)

366

367

368@shared_task(

369 name="crawler.tasks.crawl_collection",

370 bind=True,

371 queue="coordinator",

372 base=TaskWithProgress,

373)

374@tracer.start_as_current_span("CrawlCollectionTask.do")

375def crawl_collection(

376 task: "TaskWithProgress",

377 colid: str,

378 source_domain: str,

379 user_name,

380 only_new=False,

381 period: tuple[int, int] = (0, 9999),

382 number: tuple[int, int] = (0, 99999),

383):

384 event_dict: "HistoryEventDict" = {

385 "type": "import-collection",

386 "pid": f"{colid}-{source_domain}",

387 "col": None,

388 "source": source_domain,

389 "status": HistoryEventStatus.PENDING,

390 }

391

392 try:

393 logger.debug("craw_collection task")

394 user = User.objects.get(username=user_name)

395 collection = Collection.objects.filter(pid=colid).first()

396 event_dict["col"] = collection

397

398 event_dict["userid"] = user.pk

399 all_cols = get_all_cols()

400 col_data = all_cols[colid]

401 url = col_data["sources"][source_domain]

402

403 issue_list = crawl_issue_list(source_domain, colid, url, user_name)

404 if not issue_list:

405 event_dict["status"] = HistoryEventStatus.WARNING

406 event_dict["message"] = f"No issue to import for the collection: {colid}"

407 return

408

409 issue_list = filter_issues(colid, issue_list, period, number, event_dict, only_new)

410 if not issue_list:

411 event_dict["message"] = (

412 f"No issue to import with the selection for the collection: {colid}"

413 )

414 event_dict["status"] = HistoryEventStatus.OK

415 logger.debug(event_dict["message"])

416 return

417

418 task.update_state(

419 meta={"current": 0, "total": len(issue_list), "progress": 0, "col": colid},

420 state=states.STARTED,

421 )

422

423 logger.info(

424 "%s issues to process for the source: %s and the collection: %s",

425 len(issue_list),

426 source_domain,

427 colid,

428 )

429 for issue in issue_list.values():

430 promise = crawl_issue.delay(issue, source_domain, colid, url, user_name)

431 task.wait_child(promise)

432 increment_progress.delay(task.request.id)

433

434 event_dict["status"] = HistoryEventStatus.OK

435

436 except BaseException:

437 event_dict["status"] = HistoryEventStatus.ERROR

438 event_dict["message"] = traceback.format_exc()

439 logger.error(event_dict["message"])

440 raise

441 finally:

442 insert_history_event(event_dict)

443 logger.info("history event inserted %s", event_dict["pid"])

444

445

446def articles_count(colid, source_domain):

447 """compute the articles number on the disk and the downloaded articles"""

448 articles = Article.objects.order_by("pid")

449 articles = Article.objects.filter(

450 Q(my_container__origin__source__domain=source_domain),

451 Q(my_container__my_collection__pid=colid)

452 | Q(my_container__my_collection__parent__pid=colid),

453 )

454 articles_count = articles.count()

455 disk_path = settings.RESOURCES_ROOT

456 disk_path = path.join(disk_path, colid)

457 folders = subprocess.run(["find", disk_path, "-type", "f"], capture_output=True)

458 file_count = len(str(folders.stdout).split("\\n")) - 1

459 return file_count, articles_count

460

461

462@shared_task(

463 name="crawler.tasks.download_collection",

464 bind=True,

465 queue="coordinator_child",

466 base=TaskWithProgress,

467)

468@tracer.start_as_current_span("DownloadCollectionTask.do")

469def download_collection(

470 self: "TaskWithProgress",

471 colid: str,

472 source_domain: str,

473 user_name,

474 only_new=False,

475 period: tuple[int, int] = (0, 9999),

476 number: tuple[int, int] = (0, 99999),

477):

478 logger.debug("download_collection task")

479 user = User.objects.get(username=user_name)

480 collection = Collection.objects.filter(pid=colid).first()

481 event_dict: "HistoryEventDict" = {

482 "type": "download-collection",

483 "pid": f"{colid}",

484 "col": collection,

485 "source": source_domain,

486 "status": HistoryEventStatus.PENDING,

487 }

488 try:

489 event_dict["userid"] = user.pk

490

491 articles = Article.objects.order_by("pid")

492 articles = Article.objects.filter(

493 Q(my_container__origin__source__domain=source_domain),

494 Q(my_container__my_collection__pid=colid)

495 | Q(my_container__my_collection__parent__pid=colid),

496 )

497 articles_count = articles.count()

498 disk_path = settings.RESOURCES_ROOT

499

500 # TODO here filtrage par période: quand start_year, end_year remplaceront year dans les conteneurs

501

502 if articles_count > 0:

503 logger.info(

504 f"Downloading PDF articles: {articles_count} articles to process for the source: {source_domain} and the collection: {colid}"

505 )

506 self.update_state(

507 meta={

508 "current": 1,

509 "total": articles_count,

510 "progress": 1 / articles_count,

511 "col": colid,

512 },

513 state=states.STARTED,

514 )

515 download_tasks = []

516 for num, article in enumerate(articles):

517 if self.is_aborted():

518 raise TaskAborted

519 promise = download_article.si(article, source_domain, colid, only_new)

520 download_tasks.append(promise)

521 download_tasks.append(increment_progress.si(self.request.id))

522 if num % chunk_size_collections == 0:

523 logger.info(f"Chunk {num}")

524 batch_tasks = chain(*download_tasks).delay()

525 self.wait_child(batch_tasks, propagate=True)

526 download_tasks = []

527 if download_tasks != []:

528 batch_tasks = chain(*download_tasks).delay()

529 self.wait_child(batch_tasks, propagate=True)

530 # Compare files on disk/file to download

531 disk_path = path.join(disk_path, colid)

532 folders = subprocess.run(["find", disk_path, "-type", "f"], capture_output=True)

533 file_count = len(str(folders.stdout).split("\\n")) - 1

534 event_dict["status"] = HistoryEventStatus.OK

535 event_dict["message"] = (

536 f"{articles_count} articles à télécharger \n{file_count} articles sur disque."

537 )

538 if file_count != articles_count:

539 event_dict["status"] = HistoryEventStatus.WARNING

540

541 else:

542 event_dict["status"] = HistoryEventStatus.WARNING

543 event_dict["message"] = f"No article to download for the collection: {colid}"

544

545 except TaskAborted:

546 event_dict["status"] = HistoryEventStatus.ERROR

547 event_dict["message"] = "Task aborted by user"

548 # insert_history_event(event_dict)

549 raise

550 except BaseException:

551 event_dict["status"] = HistoryEventStatus.ERROR

552 event_dict["message"] = traceback.format_exc()

553 # insert_history_event(event_dict)

554 raise

555 finally:

556 insert_history_event(event_dict)

557

558

559@shared_task(

560 name="crawler.tasks.crawl_issue_list", bind=True, base=TaskWithProgress, queue="executor"

561)

562def crawl_issue_list(

563 self: "TaskWithProgress", source_domain: str, colid: str, url: str, username: str

564):

565 crawler = crawler_factory(source_domain, colid, username, pause_function=self.wait)

566

567 return crawler.crawl_collection()

568

569

570@shared_task(name="crawler.tasks.crawl_issue", queue="executor", bind=True, base=TaskWithProgress)

571def crawl_issue(

572 self: "TaskWithProgress",

573 issue: "IssueData",

574 source_domain: str,

575 colid: str,

576 url: str,

577 username: str,

578):

579 crawler = crawler_factory(source_domain, colid, username, pause_function=self.wait)

580 logger.info("crawl issue: %s, %s", source_domain, colid)

581 crawler.crawl_issue(issue)

582

583

584@shared_task(

585 name="crawler.tasks.download_article", queue="executor", bind=True, base=PtfAbortableTask

586)

587def download_article(self, article: Article, source_domain: str, colid: str, only_new: bool):

588 logger.info("Download article: %s, %s", source_domain, colid)

589 download_pdf(article, False, only_new, pause_function=self.wait)

590

591

592def filter_issues(

593 colid: str,

594 issues: "dict[str, IssueData]",

595 period: tuple[int, int] = (0, 9999),

596 number: tuple[int, int] = (0, 99999),

597 event_dict=None,

598 only_new=False,

599):

600 if event_dict is None:

601 event_dict = {}

602

603 def is_year_in_range(year):

604 try:

605 return period[0] <= int(year) <= period[1]

606 except ValueError:

607 event_dict["status"] = HistoryEventStatus.ERROR

608 insert_history_event(event_dict)

609 logger.error("Missing the year property for issues in the collection %s", colid)

610 return False

611

612 def is_number_in_range(n):

613 try:

614 # n can be "1-2"

615 if "-" in n or "–" in n:

616 n = n.split("-" if "-" in n else "–")

617 return number[0] <= int(n[0]) <= int(n[1]) <= number[1]

618 # n can be "3"

619 return number[0] <= int(n) <= number[1]

620 except ValueError:

621 # issue.number is not an integer, nor an integer range

622 event_dict["status"] = HistoryEventStatus.ERROR

623 insert_history_event(event_dict)

624 logger.warning(

625 "The number property for issues in the collection %s is not defined or is not a number",

626 colid,

627 )

628 return False

629

630 if period == (0, 9999) and number != (0, 99999):

631 # no filter expected for the period, let's filter only on the volume number

632 # we select the issue if the issue.number is "" (not always defined)

633 issues = {

634 pid: issue

635 for pid, issue in issues.items()

636 if is_number_in_range(issue.number) or issue.number == ""

637 }

638

639 if number == (0, 99999) and period != (0, 9999):

640 # no filter expected for the volume numbers, let's filter only on the period

641 issues = {pid: issue for pid, issue in issues.items() if is_year_in_range(issue.fyear)}

642

643 if only_new:

644 all_containers = Container.objects.filter(my_collection__pid=colid).all()

645 issues = {

646 pid: issue

647 for pid, issue in issues.items()

648 if not any(container.pid == pid for container in all_containers)

649 }

650 logger.info("issues filtered: %s", issues)

651 return issues

Coverage for src / crawler / tasks.py: 0%

337 statements