Spaces:
Sleeping
Sleeping
File size: 2,893 Bytes
9f2df60 3c5b98e 9f2df60 | 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 | import time, os, argparse
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from src.config import config
from src.pipeline.pipeline import ImportPipeline
from src.scraping.scraper import Scraper
from src.utils.logging import init_logging, get_logger
from src.utils.tools import call_with_exponential_backoff
from src.notification.notification_center import NotificationCenter
def scraping_task(full_scrape: bool):
init_logging()
logger = get_logger('scraper.scheduler')
def scrape():
logger.info("Initiating scraping task")
try:
scraper = Scraper(scrape_all=full_scrape)
pipeline = ImportPipeline()
for target_url in config.scraping.TARGET_URLS:
manifest = scraper.scrape_target(target_url)
if not manifest:
continue
pipeline.import_from_scraper(manifest)
scraper.commit_scrape(manifest)
logger.info("Scraper task finished gracefully")
except Exception as e:
logger.error(f"Scraping task was interrupted: {e}")
raise e
result = call_with_exponential_backoff(scrape)
if result['status'] != 'OK':
center = NotificationCenter()
center.send_notification(
"⛔ ERROR: Scraping failed",
f"Scraping procedure failed after {config.scraping.MAX_RETRIES} attempts with message: {result['last_error']}",
"all",
[
os.path.join(config.paths.LOGS, 'scraping.log')
]
)
raise result['last_error']
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument('--init_sched', action='store_true')
parser.add_argument('--full_scrape', action='store_true')
return parser.parse_args()
def run_scheduler():
scheduler = BackgroundScheduler()
# Daily at 3 AM (Mon–Sat)
scheduler.add_job(
scraping_task,
trigger=CronTrigger(day_of_week='mon-sat', hour=3, minute=0),
args=[False],
id='daily_scrape',
max_instances=1,
coalesce=True,
misfire_grace_time=3600,
)
# Sunday at 2 AM (full scrape)
scheduler.add_job(
scraping_task,
trigger=CronTrigger(day_of_week='sun', hour=2, minute=0),
args=[True],
id='weekly_full_scrape',
max_instances=1,
coalesce=True,
misfire_grace_time=3600,
)
scheduler.start()
print("Scheduler started")
try:
while True:
time.sleep(3600)
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()
print("Scheduler stopped")
if __name__ == "__main__":
args = parse_args()
if args.init_sched:
run_scheduler()
else:
scraping_task(args.full_scrape)
|