mirror of
https://github.com/bellingcat/cisticola.git
synced 2026-06-08 03:18:34 +03:00
Add option to clear registered scrapers, necessary for tests
This commit is contained in:
2
.gitignore
vendored
2
.gitignore
vendored
@@ -15,7 +15,7 @@ service_account.json
|
||||
|
||||
# Unit test / coverage reports
|
||||
reports
|
||||
.coverage
|
||||
.coverage*
|
||||
.cache
|
||||
.pytest_cache/
|
||||
cover/
|
||||
|
||||
6
app.py
6
app.py
@@ -1,12 +1,12 @@
|
||||
import argparse
|
||||
from loguru import logger
|
||||
import gspread
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy import create_engine, func
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
import os
|
||||
import time
|
||||
|
||||
from cisticola.base import Channel, mapper_registry
|
||||
from cisticola.base import Channel, RawChannelInfo, mapper_registry
|
||||
from cisticola.scraper import (
|
||||
ScraperController,
|
||||
BitchuteScraper,
|
||||
@@ -82,7 +82,7 @@ def get_scraper_controller():
|
||||
controller.connect_to_db(engine)
|
||||
|
||||
scrapers = [
|
||||
# TelegramTelethonScraper(),
|
||||
TelegramTelethonScraper(),
|
||||
TwitterScraper()]
|
||||
|
||||
controller.register_scrapers(scrapers)
|
||||
|
||||
@@ -303,6 +303,9 @@ class ScraperController:
|
||||
"""
|
||||
self.scrapers.extend(scraper)
|
||||
|
||||
def remove_all_scrapers(self):
|
||||
self.scrapers = []
|
||||
|
||||
def scrape_all_channels(self, archive_media: bool = True):
|
||||
if self.session is None:
|
||||
logger.error("No DB session")
|
||||
|
||||
@@ -37,8 +37,6 @@ class TwitterScraper(Scraper):
|
||||
if tweet.media:
|
||||
media_list += tweet.media
|
||||
|
||||
print(tweet.json())
|
||||
|
||||
if tweet.retweetedTweet and hasattr(tweet.retweetedTweet, 'media') and tweet.retweetedTweet.media:
|
||||
media_list += tweet.retweetedTweet.media
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ from cisticola.base import Channel
|
||||
from cisticola.scraper import TelegramTelethonScraper
|
||||
|
||||
def test_scrape_telegram_telethon_channel_no_media(controller, channel_kwargs):
|
||||
controller.remove_all_scrapers()
|
||||
|
||||
channels = [Channel(**channel_kwargs['telegram'])]
|
||||
controller.register_scraper(scraper = TelegramTelethonScraper())
|
||||
@@ -13,6 +14,7 @@ def test_scrape_telegram_telethon_channel_no_media(controller, channel_kwargs):
|
||||
def test_scrape_telegram_telethon_channel(controller, channel_kwargs):
|
||||
|
||||
controller.reset_db()
|
||||
controller.remove_all_scrapers()
|
||||
|
||||
channels = [Channel(**channel_kwargs['telegram'])]
|
||||
controller.register_scraper(scraper = TelegramTelethonScraper())
|
||||
|
||||
Reference in New Issue
Block a user