Page Monitoring

import os

def require_secret(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"Required secret is not configured: {name}")
    return value

def optional_secret(name: str) -> str | None:
    return os.environ.get(name) or None

import requests
from typing import List, Optional

class UptimeRobotClient:
    """UptimeRobot API client for monitoring page availability."""

    def __init__(self, api_key: str):
        self.api_key = api_key
        # v2 still works as of 2026 but is in maintenance mode; v3 is
        # the current REST API at https://api.uptimerobot.com/v3 with
        # a different request shape (Bearer auth, JSON bodies).
        self.base_url = "https://api.uptimerobot.com/v2"

    def _request(self, endpoint: str, params: dict = None) -> dict:
        data = {'api_key': self.api_key}
        if params:
            data.update(params)

        response = requests.post(
            f"{self.base_url}/{endpoint}", data=data, timeout=30
        )
        response.raise_for_status()
        return response.json()

    def get_monitors(self) -> List[dict]:
        """Get all monitors."""
        result = self._request('getMonitors')
        return result.get('monitors', [])

    def create_monitor(self, friendly_name: str, url: str,
                       monitor_type: int = 1) -> dict:
        """Create a new monitor.

        Types: 1=HTTP(s), 2=Keyword, 3=Ping, 4=Port
        """
        return self._request('newMonitor', {
            'friendly_name': friendly_name,
            'url': url,
            'type': monitor_type
        })

    def get_monitor_uptime(self, monitor_id: int,
                           custom_uptime_ratios: str = "7-30-90") -> dict:
        """Get uptime statistics for a monitor."""
        return self._request('getMonitors', {
            'monitors': monitor_id,
            'custom_uptime_ratios': custom_uptime_ratios
        })

    def pause_monitor(self, monitor_id: int) -> dict:
        """Pause a monitor."""
        return self._request('editMonitor', {
            'id': monitor_id,
            'status': 0
        })

    def resume_monitor(self, monitor_id: int) -> dict:
        """Resume a monitor."""
        return self._request('editMonitor', {
            'id': monitor_id,
            'status': 1
        })

# Usage
client = UptimeRobotClient(require_secret('UPTIMEROBOT_API_KEY'))

# Create monitors for important pages
client.create_monitor('News Homepage', 'https://example-news.com')
client.create_monitor('API Status', 'https://api.example.com/health')

# Check all monitors
for monitor in client.get_monitors():
    status = 'UP' if monitor['status'] == 2 else 'DOWN'
    print(f"{monitor['friendly_name']}: {status}")

import requests
from bs4 import BeautifulSoup
from feedgen.feed import FeedGenerator
from datetime import datetime
import hashlib

class RSSGenerator:
    """Generate RSS feeds from web pages."""

    def __init__(self, feed_id: str, title: str, link: str):
        self.fg = FeedGenerator()
        self.fg.id(feed_id)
        self.fg.title(title)
        self.fg.link(href=link)
        self.fg.description(f'Auto-generated feed for {title}')

    def add_from_page(self, url: str, item_selector: str,
                      title_selector: str, link_selector: str,
                      description_selector: Optional[str] = None):
        """Parse a page and add items to feed.

        Args:
            url: Page URL to parse
            item_selector: CSS selector for each item container
            title_selector: CSS selector for title (relative to item)
            link_selector: CSS selector for link (relative to item)
            description_selector: Optional CSS selector for description
        """

        response = requests.get(url, timeout=30)
        soup = BeautifulSoup(response.text, 'html.parser')

        items = soup.select(item_selector)

        for item in items[:20]:  # Limit to 20 items
            title_elem = item.select_one(title_selector)
            link_elem = item.select_one(link_selector)

            if not title_elem or not link_elem:
                continue

            title = title_elem.get_text(strip=True)
            link = link_elem.get('href', '')

            # Make absolute URL if relative
            if link.startswith('/'):
                from urllib.parse import urljoin
                link = urljoin(url, link)

            fe = self.fg.add_entry()
            fe.id(hashlib.md5(link.encode()).hexdigest())
            fe.title(title)
            fe.link(href=link)

            if description_selector:
                desc_elem = item.select_one(description_selector)
                if desc_elem:
                    fe.description(desc_elem.get_text(strip=True))

            fe.published(datetime.now())

    def generate_rss(self) -> str:
        """Generate RSS XML string."""
        return self.fg.rss_str(pretty=True).decode()

    def save_rss(self, filepath: str):
        """Save RSS feed to file."""
        self.fg.rss_file(filepath)

# Example: Generate feed for a news site without RSS
rss = RSSGenerator(
    'https://example.com/news',
    'Example News Feed',
    'https://example.com/news'
)

rss.add_from_page(
    'https://example.com/news',
    item_selector='.news-item',
    title_selector='h2 a',
    link_selector='h2 a',
    description_selector='.summary'
)

# Save the feed
rss.save_rss('example_feed.xml')

# RSS-Bridge generates feeds for sites without them
# Supports Twitter, Instagram, YouTube, and many others

# Docker installation
docker pull rssbridge/rss-bridge
docker run -d -p 3000:80 rssbridge/rss-bridge

# Access at http://localhost:3000
# Select a bridge, enter parameters, get RSS feed URL

# Twarc requires X (Twitter) API credentials.
#
# IMPORTANT (2023+): X eliminated the free Twitter API tier and the
# free academic research access program. Current X API pricing has
# shifted multiple times since the 2023 changes, the Basic / Pro /
# Enterprise tier names and pay-per-use credit model have been
# revised; check the current pricing page at
# https://developer.x.com/en/products/x-api before estimating cost.
# For one-off archiving, the snscrape / nitter / web-scraping path
# may be more cost-effective, see the web-scraping skill.

# Installation
# pip install twarc

# Configure (interactive, provides API keys + bearer token)
# twarc2 configure

import subprocess
import json
from pathlib import Path

class TwitterArchiver:
    """Archive Twitter searches and timelines."""

    def __init__(self, output_dir: Path):
        self.output_dir = output_dir
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def search(self, query: str, max_results: int = 100) -> Path:
        """Search tweets and save to file."""

        output_file = self.output_dir / f"search_{query.replace(' ', '_')}.jsonl"

        subprocess.run([
            'twarc2', 'search',
            '--max-results', str(max_results),
            query,
            str(output_file)
        ], check=True)

        return output_file

    def get_timeline(self, username: str, max_results: int = 100) -> Path:
        """Get user timeline."""

        output_file = self.output_dir / f"timeline_{username}.jsonl"

        subprocess.run([
            'twarc2', 'timeline',
            '--max-results', str(max_results),
            username,
            str(output_file)
        ], check=True)

        return output_file

    def parse_archive(self, filepath: Path) -> list[dict]:
        """Parse archived tweets."""

        tweets = []
        with open(filepath) as f:
            for line in f:
                data = json.loads(line)
                if 'data' in data:
                    tweets.extend(data['data'])

        return tweets

import requests
from datetime import datetime
from typing import Optional
from urllib.parse import urlsplit, urlunsplit

def redact_url(url: str) -> str:
    """Remove credentials, query parameters, and fragments from alert text."""
    parts = urlsplit(url)
    host = parts.hostname or ''
    if ':' in host:
        host = f'[{host}]'
    if parts.port:
        host = f"{host}:{parts.port}"
    return urlunsplit((parts.scheme, host, parts.path, '', ''))

class AlertManager:
    """Send alerts when monitored pages change."""

    def __init__(self, slack_webhook: str = None,
                 discord_webhook: str = None,
                 email_config: dict = None):
        self.slack_webhook = slack_webhook
        self.discord_webhook = discord_webhook
        self.email_config = email_config

    def send_slack(self, message: str, channel: str = None):
        """Send Slack notification."""
        if not self.slack_webhook:
            return

        payload = {'text': message}
        if channel:
            payload['channel'] = channel

        try:
            response = requests.post(self.slack_webhook, json=payload, timeout=15)
            response.raise_for_status()
        except requests.RequestException:
            raise RuntimeError('Slack webhook request failed') from None

    def send_discord(self, message: str):
        """Send Discord notification."""
        if not self.discord_webhook:
            return

        try:
            response = requests.post(
                self.discord_webhook, json={'content': message}, timeout=15
            )
            response.raise_for_status()
        except requests.RequestException:
            raise RuntimeError('Discord webhook request failed') from None

    def send_email(self, subject: str, body: str, to: str):
        """Send email notification."""
        if not self.email_config:
            return

        import smtplib
        from email.mime.text import MIMEText

        msg = MIMEText(body)
        msg['Subject'] = subject
        msg['From'] = self.email_config['from']
        msg['To'] = to

        with smtplib.SMTP(self.email_config['smtp_host'],
                         self.email_config['smtp_port']) as server:
            server.starttls()
            server.login(self.email_config['username'],
                        self.email_config['password'])
            server.send_message(msg)

    def alert_change(self, page_name: str, url: str,
                     old_content: str, new_content: str):
        """Send change alert to all configured channels."""

        message = f"""
Page Changed: {page_name}
URL: {redact_url(url)}
Time: {datetime.now().isoformat()}

Previous content (preview):
{old_content[:200]}...

New content (preview):
{new_content[:200]}...
"""

        if self.slack_webhook:
            self.send_slack(message)

        if self.discord_webhook:
            self.send_discord(message)

# Edit crontab
crontab -e

# Check pages every 15 minutes
*/15 * * * * /usr/bin/python3 /path/to/monitor_script.py >> /var/log/monitor.log 2>&1

# Check critical pages every 5 minutes
*/5 * * * * /usr/bin/python3 /path/to/critical_monitor.py >> /var/log/critical.log 2>&1

# Daily summary report at 8 AM
0 8 * * * /usr/bin/python3 /path/to/daily_report.py

#!/usr/bin/env python3
"""Page monitoring script for cron execution."""

import sys
import os
from pathlib import Path
from datetime import datetime

# Add project to path
sys.path.insert(0, str(Path(__file__).parent))

from monitor import PageMonitor
from alerts import AlertManager
from secure_config import optional_secret, require_secret

def main():
    # Initialize
    monitor = PageMonitor(Path('./data'))

    email_config = None
    if optional_secret('SMTP_HOST'):
        email_config = {
            'from': require_secret('ALERT_FROM_EMAIL'),
            'smtp_host': require_secret('SMTP_HOST'),
            'smtp_port': int(os.environ.get('SMTP_PORT', '587')),
            'username': require_secret('SMTP_USERNAME'),
            'password': require_secret('SMTP_APP_PASSWORD')
        }

    alerts = AlertManager(
        slack_webhook=optional_secret('SLACK_WEBHOOK_URL'),
        discord_webhook=optional_secret('DISCORD_WEBHOOK_URL'),
        email_config=email_config
    )

    # Check all pages
    results = monitor.check_all()

    # Process results
    changes = [r for r in results if r['status'] == 'changed']
    errors = [r for r in results if r['status'] == 'error']

    # Alert on changes
    for change in changes:
        alerts.alert_change(
            change['name'],
            change['url'],
            change['previous_content'],
            change['new_content']
        )
        print(f"[{datetime.now()}] CHANGE: {change['name']}")

    # Alert on errors
    for error in errors:
        alerts.send_slack(f"Monitor error for {error['name']}: {error['error']}")
        print(f"[{datetime.now()}] ERROR: {error['name']} - {error['error']}")

    # Summary
    print(f"[{datetime.now()}] Checked {len(results)} pages, "
          f"{len(changes)} changes, {len(errors)} errors")

if __name__ == '__main__':
    main()

# `MultiArchiver` is the cascade-archive helper from the sibling
# web-archiving skill (see research-toolkit/skills/web-archiving/).
# Replace this import with your own multi-service archiver, or
# port the MultiArchiver class from that skill.
from multiarchiver import MultiArchiver

class ArchivingMonitor(PageMonitor):
    """Page monitor that archives content when changes detected."""

    def __init__(self, storage_dir: Path):
        super().__init__(storage_dir)
        self.archiver = MultiArchiver()

    def check_page(self, url: str) -> dict:
        """Check page and archive if changed."""

        result = super().check_page(url)

        if result and result['status'] == 'changed':
            # Archive to multiple services
            archive_results = self.archiver.archive_url(url)

            successful_archives = [
                r.archived_url for r in archive_results
                if r.success
            ]

            result['archives'] = successful_archives

            # Log archive URLs
            print(f"Archived {result['name']} to:")
            for archive_url in successful_archives:
                print(f"  - {redact_url(archive_url)}")

        return result

## News/Current Events Monitoring

### Pages to monitor:
- Breaking news sections
- Press release pages
- Government announcement pages
- Company newsrooms

### Monitoring frequency:
- Breaking news: Every 5 minutes
- Press releases: Every 15-30 minutes
- General news: Every hour

### Archive strategy:
- Archive immediately on detection
- Use both Wayback Machine and Archive.today
- Save local copy with timestamp

## Academic/Research Monitoring

### Pages to monitor:
- Preprint servers (arXiv, SSRN)
- Journal table of contents
- Conference proceedings
- Researcher profiles

### Monitoring frequency:
- Daily for active topics
- Weekly for general monitoring

### Tools recommended:
- Google Scholar alerts (free, built-in)
- Semantic Scholar alerts
- RSS feeds where available
- Custom monitors for specific pages

## Competitor Monitoring

### Pages to monitor:
- Pricing pages
- Product pages
- Job postings
- Press releases
- Executive bios

### Monitoring frequency:
- Pricing: Daily
- Products: Daily
- Jobs: Weekly
- Press: Daily

### Legal considerations:
- Don't violate terms of service
- Don't circumvent access controls
- Public pages only
- Don't scrape at high frequency

## Before monitoring a page:

- [ ] Is the page publicly accessible?
- [ ] Are you respecting robots.txt?
- [ ] Is monitoring frequency reasonable?
- [ ] Do you have a legitimate purpose?
- [ ] Are you storing data securely?
- [ ] Do you have alerts configured?
- [ ] Is archiving set up for important pages?

## Maintenance:

- [ ] Review monitors monthly
- [ ] Remove stale monitors
- [ ] Update selectors if pages change
- [ ] Check alert delivery
- [ ] Verify archives are working

import time
from functools import wraps

def rate_limit(min_interval: float = 1.0):
    """Decorator to rate limit function calls."""
    last_call = [0.0]

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_call[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            last_call[0] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

# Usage
@rate_limit(min_interval=2.0)  # Max once per 2 seconds
def check_page(url: str):
    return requests.get(url)