wsprnet_scraper.py

#!/usr/bin/env python3
"""
WSPRNET Scraper - Simple Always-Cache Architecture
- Download thread: Always saves JSON to cache
- Insert thread: Processes cached files in order with retry on failure
- Target table: wspr.rx (supports PREWHERE for time-based queries)
- Automatic retry with exponential backoff for failed inserts
No complex gap logic, just save everything and process in order.
"""
import argparse
import json
import sys
import time

# Handle --version before heavy imports so it works without the venv packages
if '--version' in sys.argv:
    # VERSION constant is defined below, so parse it directly from this file
    import re as _re, os as _os
    _src = open(_os.path.abspath(__file__)).read()
    _m = _re.search(r'^VERSION\s*=\s*["\']([^"\']+)["\']', _src, _re.MULTILINE)
    print(f'wsprnet_scraper.py {_m.group(1) if _m else "unknown"}')
    sys.exit(0)

import requests
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import clickhouse_connect
import numpy as np
import logging
import os
from datetime import datetime
import threading
import glob
from datetime import timezone

# Version
VERSION = "2.8.0"  # Fast startup: part-metadata max(id) + bad-dir trim at startup

# Default configuration
DEFAULT_CONFIG = {
    'max_bytes_per_second': 20000,
    'request_timeout': 120,
    'clickhouse_host': 'localhost',
    'clickhouse_port': 8123,
    'clickhouse_user': '',
    'clickhouse_password': '',
    'clickhouse_database': 'wspr',
    'clickhouse_table': 'rx',
    'wsprnet_url': 'http://www.wsprnet.org/drupal/wsprnet/spots/json',
    'wsprnet_login_url': 'http://www.wsprnet.org/drupal/rest/user/login',
    'band': 'All',
    'exclude_special': 0,
    'loop_interval': 20,
    'cache_dir': '/var/lib/wsprnet/cache',
    'batch_flush_rows': 100000,   # flush accumulated rows when this many are pending
    'batch_flush_seconds': 10,    # flush if this many seconds have passed since last flush
    'bad_dir_max_files': 1000     # maximum JSON files to keep in the bad/ quarantine dir
}

# Logging configuration
LOG_FILE = 'wsprnet_scraper.log'
LOG_MAX_BYTES = 10 * 1024 * 1024
LOG_KEEP_RATIO = 0.75


class TruncatingFileHandler(logging.FileHandler):
    """File handler that truncates to newest 75% when file grows too large"""
    
    def __init__(self, filename, max_bytes, keep_ratio=0.75):
        self.max_bytes = max_bytes
        self.keep_ratio = keep_ratio
        super().__init__(filename, mode='a', encoding='utf-8')
    
    def emit(self, record):
        super().emit(record)
        self.check_truncate()
    
    def check_truncate(self):
        try:
            if os.path.exists(self.baseFilename):
                current_size = os.path.getsize(self.baseFilename)
                if current_size > self.max_bytes:
                    self.truncate_file()
        except Exception as e:
            print(f"Error checking log file size: {e}")
    
    def truncate_file(self):
        try:
            with open(self.baseFilename, 'r', encoding='utf-8') as f:
                lines = f.readlines()
            
            keep_count = int(len(lines) * self.keep_ratio)
            if keep_count < 1:
                keep_count = 1
            
            new_lines = lines[-keep_count:]
            
            with open(self.baseFilename, 'w', encoding='utf-8') as f:
                f.write(f"[Log truncated - kept newest {self.keep_ratio*100:.0f}% of {len(lines)} lines]\n")
                f.writelines(new_lines)
            
            old_size = sum(len(line.encode('utf-8')) for line in lines)
            new_size = os.path.getsize(self.baseFilename)
            logging.info(f"Log file truncated from {old_size:,} to {new_size:,} bytes")
        except Exception as e:
            print(f"Error truncating log file: {e}")


def setup_logging(log_file=None, max_bytes=LOG_MAX_BYTES, keep_ratio=LOG_KEEP_RATIO):
    """Setup logging"""
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    logger.handlers.clear()
    
    if log_file:
        file_handler = TruncatingFileHandler(log_file, max_bytes, keep_ratio)
        file_formatter = logging.Formatter('[%(asctime)s] %(levelname)s: %(message)s',
                                          datefmt='%Y-%m-%d %H:%M:%S')
        file_handler.setFormatter(file_formatter)
        logger.addHandler(file_handler)
    else:
        console_handler = logging.StreamHandler()
        console_formatter = logging.Formatter('[%(asctime)s] %(levelname)s: %(message)s',
                                             datefmt='%Y-%m-%d %H:%M:%S')
        console_handler.setFormatter(console_formatter)
        logger.addHandler(console_handler)
    
    return logger


def log(message: str, level: str = "INFO"):
    """Log a message"""
    logger = logging.getLogger()
    level_map = {
        'DEBUG': logging.DEBUG,
        'INFO': logging.INFO,
        'WARNING': logging.WARNING,
        'ERROR': logging.ERROR,
        'CRITICAL': logging.CRITICAL
    }
    logger.log(level_map.get(level, logging.INFO), message)


def diagnose_bad_json(cache_file: Path, je: Exception) -> None:
    """Log detailed diagnostics about a JSON file that failed to parse."""
    SNIPPET = 200  # bytes to show from head and tail

    try:
        size = cache_file.stat().st_size
    except OSError:
        size = -1

    log(f"  Bad JSON diagnostics for {cache_file.name}:", "WARNING")
    log(f"    file size  : {size} bytes", "WARNING")
    log(f"    parse error: {je}", "WARNING")

    try:
        raw = cache_file.read_bytes()
        head = raw[:SNIPPET]
        tail = raw[-SNIPPET:] if size > SNIPPET else b''

        # Try to detect format from the first non-whitespace bytes
        stripped = raw.lstrip()
        if stripped.startswith(b'{'):
            # Try to read top-level keys even from broken JSON
            try:
                import re as _re
                keys = _re.findall(rb'"([^"]{1,40})"\s*:', raw[:2048])
                unique_keys = list(dict.fromkeys(k.decode('utf-8', errors='replace') for k in keys))
                log(f"    top-level JSON keys (up to first 2048 bytes): {unique_keys}", "WARNING")
            except Exception:
                pass
        elif stripped.startswith(b'['):
            log(f"    file appears to be a JSON array (not an object)", "WARNING")
        else:
            log(f"    file does not start with '{{' or '['; may not be JSON at all", "WARNING")

        log(f"    first {min(SNIPPET, size)} bytes: {head!r}", "WARNING")
        if tail:
            log(f"    last  {min(SNIPPET, size)} bytes: {tail!r}", "WARNING")
    except Exception as read_err:
        log(f"    (could not read file for diagnostics: {read_err})", "WARNING")


def trim_bad_dir(bad_dir: Path, max_files: int) -> None:
    """Remove the oldest files from bad_dir when the count exceeds max_files."""
    if max_files <= 0:
        return
    try:
        entries = [(e.stat().st_mtime, e.path) for e in os.scandir(bad_dir) if e.is_file()]
    except OSError:
        return

    excess = len(entries) - max_files
    if excess <= 0:
        return

    # Sort oldest-first and delete the excess
    entries.sort()
    for _, path in entries[:excess]:
        try:
            os.unlink(path)
        except OSError as e:
            log(f"trim_bad_dir: could not remove {path}: {e}", "WARNING")
    log(f"trim_bad_dir: removed {excess} oldest file(s) from {bad_dir} "
        f"(limit={max_files})", "INFO")


def ensure_cache_dir(cache_dir: str) -> bool:
    """Ensure cache directory exists"""
    try:
        Path(cache_dir).mkdir(parents=True, exist_ok=True)
        test_file = Path(cache_dir) / '.test_write'
        test_file.touch()
        test_file.unlink()
        return True
    except Exception as e:
        log(f"Failed to create/access cache directory {cache_dir}: {e}", "ERROR")
        return False


def login_wsprnet(username: str, password: str, login_url: str, session_file: Path) -> Optional[Tuple[str, str]]:
    """Login to wsprnet.org"""
    log(f"Attempting to login as {username}...")
    
    login_data = {"name": username, "pass": password}
    headers = {'Content-Type': 'application/json'}
    
    try:
        response = requests.post(login_url, json=login_data, headers=headers, timeout=60)
        
        if response.status_code != 200:
            log(f"Login failed with status code {response.status_code}", "ERROR")
            return None
        
        data = response.json()
        sessid = data.get('sessid', '')
        session_name = data.get('session_name', '')
        
        if not sessid or not session_name:
            log(f"Login response missing sessid or session_name", "ERROR")
            return None
        
        session_data = {
            'sessid': sessid,
            'session_name': session_name,
            'username': username,
            'login_time': time.time()
        }
        
        session_file.parent.mkdir(parents=True, exist_ok=True)
        with open(session_file, 'w') as f:
            json.dump(session_data, f, indent=2)
        
        log(f"Login successful")
        return session_name, sessid
        
    except Exception as e:
        log(f"Login failed: {e}", "ERROR")
        return None


def read_session_file(session_file: Path) -> Optional[Tuple[str, str]]:
    """Read session from file"""
    if not session_file.exists():
        return None
    
    try:
        with open(session_file, 'r') as f:
            data = json.load(f)
        
        sessid = data.get('sessid', '')
        session_name = data.get('session_name', '')
        
        if not sessid or not session_name:
            return None
        
        login_time = data.get('login_time', 0)
        age_hours = (time.time() - login_time) / 3600
        log(f"Using session (age: {age_hours:.1f} hours)")
        
        return session_name, sessid
    except Exception as e:
        return None


def get_session_token(session_file: Path, username: str, password: str, login_url: str) -> Optional[str]:
    """Get session token"""
    session_data = read_session_file(session_file)
    
    if session_data:
        session_name, sessid = session_data
        return f"{session_name}={sessid}"
    
    if not username or not password:
        log("No session and no credentials provided", "ERROR")
        return None
    
    login_result = login_wsprnet(username, password, login_url, session_file)
    if login_result:
        session_name, sessid = login_result
        return f"{session_name}={sessid}"
    
    return None


def get_last_spotnum_from_db(client, database: str, table: str) -> int:
    """Get highest spotnum from database.

    Wsprnet spot IDs are monotonically increasing with time, so the max(id)
    is always in the most recent data.  Bounding the query with a WHERE on
    time lets ClickHouse prune to a single partition and use the leading sort
    key, making this essentially instant even on multi-billion-row tables.

    Strategy (each step falls through only if the previous returns 0):
      1. Current month only (toStartOfMonth)
      2. Last 2 months (in case we're right at a month boundary)
      3. Full table scan — slow but always correct
    """
    table_ref = f"{database}.{table}"

    for label, where in [
        ("current month",   "time >= toStartOfMonth(now())"),
        ("last 2 months",   "time >= toStartOfMonth(now() - INTERVAL 1 MONTH)"),
        ("full table scan", "1=1"),
    ]:
        try:
            result = client.query(
                f"SELECT max(id) FROM {table_ref} WHERE {where}"
            )
            if result.result_rows and result.result_rows[0][0] not in (None, 0):
                spotnum = int(result.result_rows[0][0])
                if spotnum > 0:
                    log(f"Got last spotnum via {label}: {spotnum}")
                    return spotnum
        except Exception as e:
            log(f"Spotnum query ({label}) failed: {e}", "WARNING")

    return 0


def download_and_cache_spots(session_token: str, last_spotnum: int, config: Dict, cache_dir: str) -> Tuple[bool, bool, int]:
    """
    Download spots and save to cache file
    Returns: (success, auth_failed, highest_spotnum)
    """
    params = {
        'band': config['band'],
        'exclude_special': config['exclude_special'],
        'spotnum_start': last_spotnum  # wsprnet API returns spots with ID > spotnum_start
    }
    
    headers = {'Cookie': session_token}
    
    try:
        response = requests.get(
            config['wsprnet_url'],
            params=params,
            headers=headers,
            timeout=config['request_timeout']
        )
        
        if response.status_code == 403:
            log("Authentication expired", "WARNING")
            return False, True, last_spotnum
        
        if response.status_code != 200:
            log(f"Download failed: {response.status_code}", "ERROR")
            return False, False, last_spotnum
        
        spots = response.json()
        
        if not spots:
            log("No new spots")
            return True, False, last_spotnum
        
        # Sort spots by ID (wsprnet sometimes returns them in reverse order)
        try:
            spots.sort(key=lambda x: int(x.get('Spotnum', 0)))
        except (ValueError, TypeError) as e:
            log(f"Warning: Could not sort spots by ID: {e}", "WARNING")
        
        # Get highest spotnum from this download
        try:
            highest_spotnum = int(spots[-1].get('Spotnum', 0))
        except (ValueError, TypeError):
            highest_spotnum = last_spotnum
        
        # Check for gap between last download and this download
        if len(spots) > 0:
            try:
                first_id = int(spots[0].get('Spotnum', 0))
                if last_spotnum > 0 and first_id > 0 and first_id != last_spotnum + 1:
                    gap_size = first_id - last_spotnum - 1
                    log(f"Gap between downloads: after spot {last_spotnum}, expected {last_spotnum + 1}, got {first_id} (gap of {gap_size})", "WARNING")
            except (ValueError, TypeError):
                pass
        
        # Check for gaps within downloaded data
        if len(spots) > 1:
            gaps_found = 0
            for i in range(1, len(spots)):
                try:
                    prev_id = int(spots[i-1].get('Spotnum', 0))
                    curr_id = int(spots[i].get('Spotnum', 0))
                    
                    if prev_id > 0 and curr_id > 0 and curr_id != prev_id + 1:
                        gap_size = curr_id - prev_id - 1
                        log(f"Gap in downloaded data: after spot {prev_id}, expected {prev_id + 1}, got {curr_id} (gap of {gap_size})", "WARNING")
                        gaps_found += 1
                except (ValueError, TypeError):
                    # Skip spots with bad IDs
                    continue
            
            if gaps_found > 0:
                log(f"Total gaps in this download: {gaps_found}", "WARNING")
        
        # Always save to cache — write to a .tmp file first, then rename
        # atomically so the insert thread never sees a partially-written file.
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S_%f')
        cache_file = Path(cache_dir) / f'spots_{timestamp}.json'
        tmp_file   = cache_file.with_suffix('.tmp')

        cache_data = {
            'timestamp': timestamp,
            'download_time': time.time(),
            'spot_count': len(spots),
            'first_spotnum': spots[0].get('Spotnum', 0) if spots else 0,
            'last_spotnum': spots[-1].get('Spotnum', 0) if spots else 0,
            'spots': spots
        }

        with open(tmp_file, 'w') as f:
            json.dump(cache_data, f, indent=2)
        tmp_file.rename(cache_file)   # atomic on Linux (same filesystem)

        log(f"Downloaded and cached {len(spots)} spots to {cache_file.name}")
        return True, False, highest_spotnum
        
    except Exception as e:
        log(f"Download error: {e}", "ERROR")
        return False, False, last_spotnum


def get_cached_files(cache_dir: str) -> List[Path]:
    """Get sorted list of cached files"""
    try:
        cache_path = Path(cache_dir)
        if not cache_path.exists():
            return []
        
        files = sorted(cache_path.glob('spots_*.json'))
        return files
    except Exception as e:
        log(f"Error reading cache: {e}", "ERROR")
        return []


def maidenhead_to_latlon(grid: str) -> Tuple[float, float]:
    """Convert Maidenhead grid square to latitude/longitude (center of square)
    
    Returns (lat, lon) with 6 decimal places precision
    Handles 4-character (e.g., CM87) and 6-character (e.g., CM87wj) grids
    Returns (0.0, 0.0) if grid is invalid
    
    Convention: 4-character grids are centered at subsquare 'll' (index 11)
    """
    if not grid or len(grid) < 2:
        return 0.0, 0.0
    
    try:
        # Only uppercase the field letters (first 2 chars), leave subsquares lowercase
        grid = grid[:2].upper() + grid[2:]
        
        # Field (first 2 characters): 20° lon, 10° lat
        lon = (ord(grid[0]) - ord('A')) * 20 - 180
        lat = (ord(grid[1]) - ord('A')) * 10 - 90
        
        # Square (next 2 digits): 2° lon, 1° lat
        if len(grid) >= 4:
            lon += int(grid[2]) * 2
            lat += int(grid[3]) * 1
        
        # Subsquare (optional next 2 characters): 5' lon, 2.5' lat
        # Subsquares use lowercase letters 'a'-'x' (0-23)
        if len(grid) >= 6:
            # For 6-character grids, add subsquare offset and center in subsquare
            lon += (ord(grid[4].lower()) - ord('a')) * (2.0/24.0)
            lat += (ord(grid[5].lower()) - ord('a')) * (1.0/24.0)
            # Center of subsquare (half of 2/24° lon and 1/24° lat)
            lon += (1.0/24.0)
            lat += (0.5/24.0)
        else:
            # For 4-character grids, use center of 'll' subsquare (subsquare index 11)
            # This gives: 11*(2/24) + 1/24 = 23/24 for lon, 11*(1/24) + 0.5/24 = 11.5/24 for lat
            lon += 11 * (2.0/24.0) + (1.0/24.0)  # = 23/24 = 0.958
            lat += 11 * (1.0/24.0) + (0.5/24.0)  # = 11.5/24 = 0.479
        
        return round(lat, 6), round(lon, 6)
    except:
        return 0.0, 0.0


def calculate_azimuth(lat1: float, lon1: float, lat2: float, lon2: float) -> int:
    """Calculate azimuth"""
    if lat1 == 0 and lon1 == 0:
        return 0
    if lat2 == 0 and lon2 == 0:
        return 0
    
    try:
        lat1_rad = np.radians(lat1)
        lat2_rad = np.radians(lat2)
        dlon_rad = np.radians(lon2 - lon1)
        
        x = np.sin(dlon_rad) * np.cos(lat2_rad)
        y = np.cos(lat1_rad) * np.sin(lat2_rad) - np.sin(lat1_rad) * np.cos(lat2_rad) * np.cos(dlon_rad)
        
        azimuth = np.degrees(np.arctan2(x, y))
        azimuth = (azimuth + 360) % 360
        
        return int(round(azimuth))
    except:
        return 0


def process_spot(spot: Dict) -> Optional[tuple]:
    """Process a single spot"""
    try:
        # Safe field extraction
        try:
            spotnum = int(spot.get('Spotnum', 0))
        except:
            spotnum = 0
            
        if spotnum == 0:
            return None
            
        try:
            date_int = int(spot.get('Date', 0))
            date = datetime.fromtimestamp(date_int, tz=timezone.utc).replace(tzinfo=None) if date_int > 0 else datetime(1970, 1, 1)
        except:
            date = datetime(1970, 1, 1)
            
        reporter = str(spot.get('Reporter', ''))
        reporter_grid = str(spot.get('ReporterGrid', ''))
        
        try:
            db = int(spot.get('dB', 0))
        except:
            db = 0
            
        try:
            mhz = float(spot.get('MHz', 0))
        except:
            mhz = 0.0
            
        callsign = str(spot.get('CallSign', ''))
        grid = str(spot.get('Grid', ''))
        
        try:
            power = int(spot.get('Power', 0))
        except:
            power = 0
            
        try:
            drift = int(spot.get('Drift', 0))
        except:
            drift = 0
            
        try:
            distance = int(spot.get('distance', 0))
        except:
            distance = 0
            
        try:
            azimuth = int(spot.get('azimuth', 0))
        except:
            azimuth = 0
            
        try:
            band = int(spot.get('Band', 0))
        except:
            band = 0
            
        version = str(spot.get('version', ''))
        
        try:
            code = int(spot.get('code', 0))
        except:
            code = 0
        
        # Calculate positions
        rx_lat, rx_lon = maidenhead_to_latlon(reporter_grid)
        tx_lat, tx_lon = maidenhead_to_latlon(grid)
        
        # Calculate rx_azimuth
        rx_azimuth = calculate_azimuth(tx_lat, tx_lon, rx_lat, rx_lon)
        
        # Convert frequency to Hz; clamp negatives (bad wsprnet data) to 0
        try:
            frequency_hz = int(mhz * 1_000_000)
            if frequency_hz < 0:
                frequency_hz = 0
        except:
            frequency_hz = 0
        
        row = (
            spotnum, date, band, reporter, rx_lat, rx_lon, reporter_grid,
            callsign, tx_lat, tx_lon, grid, distance, azimuth, rx_azimuth,
            frequency_hz, power, db, drift, version, code
        )
        
        return row
        
    except Exception as e:
        log(f"Error processing spot {spot.get('Spotnum', '?')}: {e}", "WARNING")
        return None


def insert_cached_file(client, cache_file: Path, database: str, table: str,
                       bad_dir_max_files: int = 1000) -> bool:
    """Process and insert a cached file with retry tracking"""
    try:
        with open(cache_file, 'r') as f:
            try:
                cache_data = json.load(f)
            except json.JSONDecodeError as je:
                # Corrupted/truncated download - quarantine, don't retry forever
                bad_dir = cache_file.parent / 'bad'
                bad_dir.mkdir(exist_ok=True)
                diagnose_bad_json(cache_file, je)
                cache_file.rename(bad_dir / cache_file.name)
                trim_bad_dir(bad_dir, bad_dir_max_files)
                log(f"Corrupted JSON in {cache_file.name} — quarantined to bad/", "WARNING")
                return True  # Don't trigger backoff for a parse error
        
        spots = cache_data.get('spots', [])
        retry_count = cache_data.get('retry_count', 0)
        
        if not spots:
            log(f"Cache file {cache_file.name} has no spots, deleting")
            cache_file.unlink()
            return True
        
        # Process all spots
        rows = []
        for spot in spots:
            row = process_spot(spot)
            if row:
                rows.append(row)
        
        if not rows:
            log(f"No valid spots in {cache_file.name}, deleting")
            cache_file.unlink()
            return True
        
        # Insert to ClickHouse
        column_names = [
            'id', 'time', 'band', 'rx_sign', 'rx_lat', 'rx_lon', 'rx_loc',
            'tx_sign', 'tx_lat', 'tx_lon', 'tx_loc', 'distance', 'azimuth',
            'rx_azimuth', 'frequency', 'power', 'snr', 'drift', 'version', 'code'
        ]
        
        client.insert(f"{database}.{table}", rows, column_names=column_names)
        
        # Get highest spotnum from inserted rows
        highest_spotnum = max(row[0] for row in rows)
        
        if retry_count > 0:
            log(f"Inserted {len(rows)} spots from {cache_file.name} (highest id: {highest_spotnum}) [retry {retry_count}]")
        else:
            log(f"Inserted {len(rows)} spots from {cache_file.name} (highest id: {highest_spotnum})")
        
        # Delete cache file after successful insert
        cache_file.unlink()
        
        return True
        
    except Exception as e:
        # Update retry count in cache file
        try:
            with open(cache_file, 'r') as f:
                cache_data = json.load(f)
            
            retry_count = cache_data.get('retry_count', 0) + 1
            cache_data['retry_count'] = retry_count
            cache_data['last_error'] = str(e)
            cache_data['last_retry'] = datetime.utcnow().isoformat()
            
            with open(cache_file, 'w') as f:
                json.dump(cache_data, f, indent=2)
            
            log(f"Failed to insert {cache_file.name} (retry {retry_count}): {e}", "ERROR")
            log(f"Spots cached for retry - file will be retried later", "INFO")
        except Exception as write_error:
            log(f"Failed to update retry count for {cache_file.name}: {write_error}", "ERROR")
        
        return False


def recover_spots_from_corrupt_json(cache_file: Path) -> List[Dict]:
    """Attempt to extract valid spots from a truncated or corrupted JSON cache file.

    The cache file structure is:
        {"timestamp": ..., "spots": [ {spot1}, {spot2}, ... ]}

    Truncation typically cuts off mid-array, leaving valid spot objects before
    the truncation point.  Strategy:
      1. Try normal json.load — if it works, return spots normally.
      2. Find the 'spots' array opening bracket, then walk forward collecting
         complete JSON objects one at a time using a depth counter.
      3. Return however many complete spot dicts were found.
    """
    try:
        text = cache_file.read_text(encoding='utf-8', errors='replace')
    except Exception as e:
        log(f"Recovery: cannot read {cache_file.name}: {e}", "WARNING")
        return []

    # Fast path: file is actually valid
    try:
        data = json.loads(text)
        return data.get('spots', [])
    except json.JSONDecodeError:
        pass

    # Find the start of the spots array
    marker = '"spots"'
    marker_pos = text.find(marker)
    if marker_pos == -1:
        log(f"Recovery: no 'spots' key found in {cache_file.name}", "WARNING")
        return []

    array_start = text.find('[', marker_pos + len(marker))
    if array_start == -1:
        log(f"Recovery: no '[' after 'spots' in {cache_file.name}", "WARNING")
        return []

    # Walk the array character by character, extracting complete objects
    recovered = []
    i = array_start + 1
    n = len(text)

    while i < n:
        # Skip whitespace and commas between objects
        while i < n and text[i] in ' \t\n\r,':
            i += 1
        if i >= n or text[i] != '{':
            break

        # Find the matching closing brace using depth counting
        depth = 0
        in_string = False
        escape_next = False
        obj_start = i

        for j in range(i, n):
            ch = text[j]
            if escape_next:
                escape_next = False
                continue
            if ch == '\\' and in_string:
                escape_next = True
                continue
            if ch == '"':
                in_string = not in_string
                continue
            if in_string:
                continue
            if ch == '{':
                depth += 1
            elif ch == '}':
                depth -= 1
                if depth == 0:
                    # Complete object found
                    obj_text = text[obj_start:j+1]
                    try:
                        obj = json.loads(obj_text)
                        recovered.append(obj)
                    except json.JSONDecodeError:
                        pass  # skip malformed object
                    i = j + 1
                    break
        else:
            # Reached end of text without closing brace — incomplete object
            break

    return recovered


def insert_thread_worker(config: Dict, cache_dir: str, database: str, table: str, stop_event: threading.Event):
    """Worker thread that accumulates rows across cache files and flushes in batches."""
    log("Insert thread started")

    column_names = [
        'id', 'time', 'band', 'rx_sign', 'rx_lat', 'rx_lon', 'rx_loc',
        'tx_sign', 'tx_lat', 'tx_lon', 'tx_loc', 'distance', 'azimuth',
        'rx_azimuth', 'frequency', 'power', 'snr', 'drift', 'version', 'code'
    ]

    batch_flush_rows    = config.get('batch_flush_rows', 100000)
    batch_flush_seconds = config.get('batch_flush_seconds', 10)

    # Create separate ClickHouse client for this thread
    try:
        client = clickhouse_connect.get_client(
            host=config['clickhouse_host'],
            port=config['clickhouse_port'],
            username=config['clickhouse_user'],
            password=config['clickhouse_password']
        )
        log("Insert thread: Connected to ClickHouse")
    except Exception as e:
        log(f"Insert thread: Failed to connect to ClickHouse: {e}", "ERROR")
        return

    consecutive_failures = 0
    pending_rows  = []          # accumulated rows not yet inserted
    pending_files = []          # cache files whose rows are in pending_rows
    last_flush    = time.time()

    def flush_batch():
        """Insert pending_rows and delete pending_files on success."""
        nonlocal pending_rows, pending_files, last_flush, consecutive_failures
        if not pending_rows:
            pending_files = []
            last_flush = time.time()
            return True
        try:
            client.insert(f"{database}.{table}", pending_rows, column_names=column_names)
            highest = max(r[0] for r in pending_rows)
            log(f"Flushed {len(pending_rows)} rows from {len(pending_files)} cache files "
                f"(highest id: {highest})")
            for cf in pending_files:
                try:
                    cf.unlink()
                except Exception as e:
                    log(f"Failed to delete cache file {cf.name}: {e}", "WARNING")
            pending_rows  = []
            pending_files = []
            last_flush    = time.time()
            consecutive_failures = 0
            return True
        except Exception as e:
            consecutive_failures += 1
            backoff = min(5 * (2 ** min(consecutive_failures - 1, 3)), 60)
            log(f"Batch insert failed ({len(pending_rows)} rows, "
                f"{len(pending_files)} files): {e} — backoff {backoff}s "
                f"(failure #{consecutive_failures})", "ERROR")
            # Update retry counts in all pending cache files
            for cf in pending_files:
                try:
                    with open(cf, 'r') as f:
                        cd = json.load(f)
                    cd['retry_count'] = cd.get('retry_count', 0) + 1
                    cd['last_error']  = str(e)
                    cd['last_retry']  = datetime.utcnow().isoformat()
                    with open(cf, 'w') as f:
                        json.dump(cd, f, indent=2)
                except Exception as we:
                    log(f"Failed to update retry count for {cf.name}: {we}", "WARNING")
            last_flush = time.time()
            time.sleep(backoff)
            return False

    while not stop_event.is_set():
        try:
            cached_files = get_cached_files(cache_dir)

            if cached_files:
                # Pick the oldest file not already in pending_files
                pending_set = set(pending_files)
                for cache_file in cached_files:
                    if cache_file in pending_set:
                        continue

                    # Load and parse the cache file, with partial recovery on corruption
                    try:
                        with open(cache_file, 'r') as f:
                            try:
                                cache_data = json.load(f)
                                spots = cache_data.get('spots', [])
                            except json.JSONDecodeError as je:
                                log(f"Corrupted JSON in {cache_file.name} at char {je.pos}: "
                                    f"{je.msg} — attempting partial recovery", "WARNING")
                                diagnose_bad_json(cache_file, je)
                                spots = recover_spots_from_corrupt_json(cache_file)
                                if spots:
                                    log(f"Recovered {len(spots)} spots from corrupted {cache_file.name}", "INFO")
                                else:
                                    bad_dir = cache_file.parent / 'bad'
                                    bad_dir.mkdir(exist_ok=True)
                                    cache_file.rename(bad_dir / cache_file.name)
                                    trim_bad_dir(bad_dir, config.get('bad_dir_max_files', 1000))
                                    log(f"No spots recoverable from {cache_file.name} — quarantined", "WARNING")
                                    continue
                    except Exception as e:
                        log(f"Cannot open {cache_file.name}: {e}", "WARNING")
                        continue
                    if not spots:
                        cache_file.unlink()
                        continue

                    rows = [r for r in (process_spot(s) for s in spots) if r is not None]
                    if not rows:
                        cache_file.unlink()
                        continue

                    pending_rows.extend(rows)
                    pending_files.append(cache_file)

                    # Check flush thresholds
                    elapsed = time.time() - last_flush
                    if len(pending_rows) >= batch_flush_rows or elapsed >= batch_flush_seconds:
                        log(f"Flush trigger: {len(pending_rows)} rows, "
                            f"{len(pending_files)} files, {elapsed:.1f}s elapsed")
                        flush_batch()
                        break   # restart outer loop to re-scan cache dir

            else:
                # No cache files — flush whatever's pending then idle
                if pending_rows:
                    log(f"Cache empty — flushing remaining {len(pending_rows)} rows")
                    flush_batch()
                else:
                    consecutive_failures = 0
                time.sleep(1)

            # Time-based flush even if we haven't hit the row threshold
            if pending_rows and (time.time() - last_flush) >= batch_flush_seconds:
                log(f"Timeout flush: {len(pending_rows)} rows from {len(pending_files)} files")
                flush_batch()

        except Exception as e:
            log(f"Insert thread error: {e}", "ERROR")
            consecutive_failures += 1
            time.sleep(5)

    # Final flush on shutdown
    if pending_rows:
        log(f"Shutdown flush: {len(pending_rows)} rows from {len(pending_files)} files")
        flush_batch()

    log("Insert thread stopped")


def setup_clickhouse_tables(admin_user: str, admin_password: str, 
                            readonly_user: str, readonly_password: str,
                            config: Dict) -> bool:
    """Setup ClickHouse database and tables"""
    try:
        admin_client = clickhouse_connect.get_client(
            host=config['clickhouse_host'],
            port=config['clickhouse_port'],
            username=admin_user,
            password=admin_password
        )
        
        log("Connected to ClickHouse with admin privileges")
        
        # Create readonly user if needed
        try:
            users = admin_client.query("SELECT name FROM system.users").result_rows
            user_names = [user[0] for user in users]
            
            if readonly_user not in user_names:
                log(f"Creating read-only user {readonly_user}...")
                admin_client.command(f"CREATE USER IF NOT EXISTS `{readonly_user}` IDENTIFIED BY '{readonly_password}'")
                admin_client.command(f"GRANT SELECT ON {config['clickhouse_database']}.* TO `{readonly_user}`")
                log(f"Read-only user {readonly_user} created")
        except Exception as e:
            log(f"Could not check/create readonly user: {e}", "WARNING")
        
        # Create database
        result = admin_client.query(
            f"SELECT 1 FROM system.databases WHERE name = '{config['clickhouse_database']}'"
        )
        
        if not result.result_rows:
            log(f"Creating database {config['clickhouse_database']}...")
            admin_client.command(f"CREATE DATABASE {config['clickhouse_database']}")
            log(f"Database {config['clickhouse_database']} created")
        else:
            log(f"Database {config['clickhouse_database']} exists")
        
        # Create table
        create_table_sql = f"""
        CREATE TABLE IF NOT EXISTS {config['clickhouse_database']}.{config['clickhouse_table']}
        (
            id UInt64 CODEC(Delta(8), ZSTD(1)),
            time DateTime CODEC(Delta(4), ZSTD(1)),
            band Int16 CODEC(T64, ZSTD(1)),
            rx_sign LowCardinality(String),
            rx_lat Float32 CODEC(ZSTD(1)),
            rx_lon Float32 CODEC(ZSTD(1)),
            rx_loc LowCardinality(String),
            tx_sign LowCardinality(String),
            tx_lat Float32 CODEC(ZSTD(1)),
            tx_lon Float32 CODEC(ZSTD(1)),
            tx_loc LowCardinality(String),
            distance UInt16 CODEC(T64, ZSTD(1)),
            azimuth UInt16 CODEC(T64, ZSTD(1)),
            rx_azimuth UInt16 CODEC(T64, ZSTD(1)),
            frequency UInt64 CODEC(T64, ZSTD(1)),
            power Int8 CODEC(T64, ZSTD(1)),
            snr Int8 CODEC(ZSTD(1)),
            drift Int8 CODEC(ZSTD(1)),
            version LowCardinality(String),
            code Int8,
            
            Spotnum UInt64 ALIAS id,
            Date UInt32 ALIAS toUnixTimestamp(time),
            Reporter String ALIAS rx_sign,
            ReporterGrid String ALIAS rx_loc,
            dB Int8 ALIAS snr,
            MHz Float32 ALIAS frequency / 1000000.0,
            CallSign String ALIAS tx_sign,
            Grid String ALIAS tx_loc,
            Power Int8 ALIAS power,
            Drift Int8 ALIAS drift,
            Band Int16 ALIAS band,
            rx_az UInt16 ALIAS rx_azimuth
        )
        ENGINE = MergeTree
        PARTITION BY toYYYYMM(time)
        ORDER BY (time, band, id)
        SETTINGS index_granularity = 8192
        """
        
        admin_client.command(create_table_sql)
        log(f"Table {config['clickhouse_database']}.{config['clickhouse_table']} created/verified")
        
        return True
    
    except Exception as e:
        log(f"Setup failed: {e}", "ERROR")
        return False


def main():
    parser = argparse.ArgumentParser(description='WSPRNET Scraper - Simple Always-Cache')
    
    parser.add_argument('--session-file', required=True, help='Session file path')
    parser.add_argument('--username', help='WSPRNET username')
    parser.add_argument('--password', help='WSPRNET password')
    
    parser.add_argument('--clickhouse-user', required=True, help='ClickHouse admin user')
    parser.add_argument('--clickhouse-password', required=True, help='ClickHouse admin password')
    parser.add_argument('--setup-readonly-user', required=True, help='Read-only user')
    parser.add_argument('--setup-readonly-password', required=True, help='Read-only password')
    
    parser.add_argument('--config', help='Config file (JSON)')
    parser.add_argument('--cache-dir', help='Cache directory')
    parser.add_argument('--loop', type=int, metavar='SECONDS', help='Loop interval')
    parser.add_argument('--log-file', default=LOG_FILE, help='Log file path')
    parser.add_argument('--log-max-mb', type=int, default=10, help='Max log size MB')
    parser.add_argument('--verbose', type=int, default=1, choices=[0, 1, 2], 
                        help='Verbosity level (ignored, for compatibility)')
    parser.add_argument('--version', action='version',
                        version=f'wsprnet_scraper.py {VERSION}')

    args = parser.parse_args()
    
    # Validate credentials are not placeholder values
    def check_changeme(value, name):
        if value and value.upper() in ('CHANGEME', 'CHANGE_ME', 'CHANGE-ME', 'CHANGME'):
            return name
        return None
    
    placeholder_errors = []
    if err := check_changeme(args.username, 'WSPRNET username (--username)'):
        placeholder_errors.append(err)
    if err := check_changeme(args.password, 'WSPRNET password (--password)'):
        placeholder_errors.append(err)
    if err := check_changeme(args.clickhouse_user, 'ClickHouse user (--clickhouse-user)'):
        placeholder_errors.append(err)
    if err := check_changeme(args.clickhouse_password, 'ClickHouse password (--clickhouse-password)'):
        placeholder_errors.append(err)
    
    if placeholder_errors:
        print("ERROR: The following credentials still have placeholder values:", file=sys.stderr)
        for err in placeholder_errors:
            print(f"  - {err}", file=sys.stderr)
        print("\nPlease edit /etc/wsprdaemon/wsprnet.conf and /etc/wsprdaemon/clickhouse.conf", file=sys.stderr)
        print("and replace 'CHANGEME' with actual credentials.", file=sys.stderr)
        sys.exit(1)
    
    if args.log_file:
        setup_logging(args.log_file, args.log_max_mb * 1024 * 1024)
    else:
        setup_logging()
    
    # Log startup banner with version
    log("")  # Blank line for separation
    log("=" * 70)
    log(f"WSPRNET Scraper version {VERSION} starting...")
    log("Architecture: Always-cache with separate insert thread")
    log("=" * 70)
    
    # Load config
    config = DEFAULT_CONFIG.copy()
    if args.config:
        with open(args.config) as f:
            config.update(json.load(f))
    
    if args.cache_dir:
        config['cache_dir'] = args.cache_dir
    
    if args.loop:
        config['loop_interval'] = args.loop
    
    # Ensure cache dir
    if not ensure_cache_dir(config['cache_dir']):
        log("Cannot create cache directory!", "ERROR")
        sys.exit(1)

    log(f"Cache directory: {config['cache_dir']}")

    # Trim the bad/ quarantine dir at startup so it doesn't grow unbounded
    # across restarts (trim_bad_dir is normally called per-quarantine, but
    # that doesn't catch files that accumulated while the scraper was down).
    bad_dir_startup = Path(config['cache_dir']) / 'bad'
    if bad_dir_startup.exists():
        trim_bad_dir(bad_dir_startup, config.get('bad_dir_max_files', 1000))
    
    config['clickhouse_user'] = args.clickhouse_user
    config['clickhouse_password'] = args.clickhouse_password
    
    # Setup ClickHouse
    log("Setting up ClickHouse...")
    success = setup_clickhouse_tables(
        admin_user=args.clickhouse_user,
        admin_password=args.clickhouse_password,
        readonly_user=args.setup_readonly_user,
        readonly_password=args.setup_readonly_password,
        config=config
    )
    
    if not success:
        log("Setup failed", "ERROR")
        sys.exit(1)
    
    session_file = Path(args.session_file)
    
    # Get session
    session_token = get_session_token(
        session_file,
        args.username or '',
        args.password or '',
        config['wsprnet_login_url']
    )
    
    if not session_token:
        log("Failed to get session", "ERROR")
        sys.exit(1)
    
    # Connect to ClickHouse
    try:
        client = clickhouse_connect.get_client(
            host=config['clickhouse_host'],
            port=config['clickhouse_port'],
            username=config['clickhouse_user'],
            password=config['clickhouse_password']
        )
        log("Connected to ClickHouse")
    except Exception as e:
        log(f"Failed to connect: {e}", "ERROR")
        sys.exit(1)
    
    # Get starting point
    last_spotnum = get_last_spotnum_from_db(client, config['clickhouse_database'], config['clickhouse_table'])
    log(f"Starting from spotnum: {last_spotnum}")
    
    # Start insert thread
    stop_event = threading.Event()
    insert_thread = threading.Thread(
        target=insert_thread_worker,
        args=(config, config['cache_dir'], config['clickhouse_database'], config['clickhouse_table'], stop_event),
        daemon=True
    )
    insert_thread.start()
    
    # Main download loop
    loop_count = 0
    
    try:
        while True:
            loop_count += 1
            log(f"Download cycle {loop_count}")
            
            success, auth_failed, new_spotnum = download_and_cache_spots(
                session_token,
                last_spotnum,
                config,
                config['cache_dir']
            )
            
            if auth_failed:
                log("Re-authenticating...")
                if session_file.exists():
                    session_file.unlink()
                
                session_token = get_session_token(
                    session_file,
                    args.username or '',
                    args.password or '',
                    config['wsprnet_login_url']
                )
                
                if not session_token:
                    log("Re-auth failed", "ERROR")
                    time.sleep(60)
                    continue
                
                # Retry
                success, auth_failed, new_spotnum = download_and_cache_spots(
                    session_token,
                    last_spotnum,
                    config,
                    config['cache_dir']
                )
            
            # Update last_spotnum from download
            if new_spotnum > last_spotnum:
                last_spotnum = new_spotnum
            
            # Show cache status
            cached_files = get_cached_files(config['cache_dir'])
            if cached_files:
                log(f"Cache status: {len(cached_files)} files pending")
            
            if not args.loop:
                break
            
            log(f"Sleeping {config['loop_interval']} seconds...")
            time.sleep(config['loop_interval'])
    
    except KeyboardInterrupt:
        log("Shutting down...")
        stop_event.set()
        insert_thread.join(timeout=5)
    
    log("Stopped")


if __name__ == '__main__':
    main()