script.py

import json
import re
import asyncio
import logging
import time
from dataclasses import dataclass
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Literal

from playwright.async_api import async_playwright, Page, Frame, TimeoutError as PlaywrightTimeoutError

CONFIG_FILE = Path("config.json")
URLS_FILE = Path("urls.txt")
LOG_FILE = Path("log.txt")


# -------------------------- Data Models ---------------------------------------
@dataclass(frozen=True)
class Step:
    kind: Literal["sleep", "waitForShow", "waitForHide", "waitForRemove"]
    value: Any  # seconds for sleep, CSS selector for waits
    timeout_ms: Optional[int] = None
    not_if: Optional[str] = None  # selector that cancels the whole scenario


# -------------------------- Logging -------------------------------------------
def setup_logging() -> None:
    """Configure logging with rotation."""
    log_formatter = logging.Formatter(
        "[%(asctime)s] %(levelname)s: %(message)s", "%Y-%m-%d %H:%M:%S"
    )
    handler = RotatingFileHandler(
        LOG_FILE, maxBytes=1_000_000, backupCount=5, encoding="utf-8"
    )
    handler.setFormatter(log_formatter)

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(log_formatter)

    root = logging.getLogger()
    root.setLevel(logging.INFO)
    if not any(isinstance(h, RotatingFileHandler) for h in root.handlers):
        root.addHandler(handler)
        root.addHandler(console_handler)


# -------------------------- Config --------------------------------------------
def load_config() -> Dict[str, Any]:
    """Load configuration from config.json with safe defaults."""
    default = {"headless": False, "elementLocatingTimeoutInMs": 30_000}
    if not CONFIG_FILE.exists():
        logging.warning("config.json not found, using defaults.")
        return default
    try:
        with CONFIG_FILE.open("r", encoding="utf-8") as f:
            cfg = json.load(f)
        merged = {**default, **cfg}
        try:
            t = int(merged.get("elementLocatingTimeoutInMs", default["elementLocatingTimeoutInMs"]))
            if t < 0:
                raise ValueError
            merged["elementLocatingTimeoutInMs"] = t
        except Exception:
            logging.warning("Invalid elementLocatingTimeoutInMs; using default 30_000ms.")
            merged["elementLocatingTimeoutInMs"] = default["elementLocatingTimeoutInMs"]
        merged["headless"] = bool(merged.get("headless", False))
        return merged
    except Exception as e:
        logging.warning(f"Failed to load config.json: {e}")
        return default


# -------------------------- Exceptions ----------------------------------------
class _SkipScenario(Exception):
    """Internal signal to skip the entire scenario line."""
    pass


# -------------------------- Parsing -------------------------------------------
_ALLOWED_KEYS = ("sleep", "waitForShow", "waitForHide", "waitForRemove", "timeoutMs", "notIf")

_KEY_QUOTE_REGEX = re.compile(
    r'([\[{,\s])\s*(sleep|waitForShow|waitForHide|waitForRemove|timeoutMs|notIf)\s*:'
)

def _normalize_steps_text(txt: str) -> str:
    return _KEY_QUOTE_REGEX.sub(r'\1"\2":', txt)

def _find_split_index_for_array_and_url(line: str) -> Optional[int]:
    depth = 0
    in_string = False
    escape = False
    for i, ch in enumerate(line):
        if in_string:
            if escape:
                escape = False
            elif ch == '\\':
                escape = True
            elif ch == '"':
                in_string = False
            continue
        if ch == '"':
            in_string = True
        elif ch == '[':
            depth += 1
        elif ch == ']':
            depth = max(0, depth - 1)
        elif ch == ',' and depth == 0:
            return i
    return None

def _parse_steps(array_text: str) -> Optional[List[Step]]:
    if not array_text:
        return None
    try:
        normalized = _normalize_steps_text(array_text.strip())
        raw = json.loads(normalized)
        if not isinstance(raw, list):
            return None
        steps: List[Step] = []
        for idx, obj in enumerate(raw):
            if not isinstance(obj, dict):
                return None
            for k in obj.keys():
                if k not in _ALLOWED_KEYS:
                    logging.warning(f"Unknown key '{k}' in step #{idx}")
                    return None
            timeout_ms = obj.get("timeoutMs")
            if timeout_ms is not None and (not isinstance(timeout_ms, int) or timeout_ms < 0):
                return None
            not_if_raw = obj.get("notIf")
            not_if = not_if_raw.strip() if isinstance(not_if_raw, str) and not_if_raw.strip() else None

            if "sleep" in obj:
                if not isinstance(obj["sleep"], (int, float)) or obj["sleep"] < 0 or not_if:
                    return None
                steps.append(Step("sleep", float(obj["sleep"]), timeout_ms, None))
            elif "waitForShow" in obj:
                val = obj["waitForShow"]
                if not isinstance(val, str) or not val.strip():
                    return None
                steps.append(Step("waitForShow", val.strip(), timeout_ms, not_if))
            elif "waitForHide" in obj:
                val = obj["waitForHide"]
                if not isinstance(val, str) or not val.strip():
                    return None
                steps.append(Step("waitForHide", val.strip(), timeout_ms, not_if))
            elif "waitForRemove" in obj:
                val = obj["waitForRemove"]
                if not isinstance(val, str) or not val.strip():
                    return None
                steps.append(Step("waitForRemove", val.strip(), timeout_ms, not_if))
            else:
                return None
        return steps
    except json.JSONDecodeError:
        return None

def parse_urls_file(file_path: Path) -> List[Tuple[List[Step], str]]:
    results: List[Tuple[List[Step], str]] = []
    try:
        with file_path.open("r", encoding="utf-8") as f:
            for raw_line in f:
                line = raw_line.strip()
                if not line or line.startswith("#"):
                    continue
                split_idx = _find_split_index_for_array_and_url(line)
                if split_idx is None:
                    continue
                steps_text = line[:split_idx].strip()
                url = line[split_idx + 1:].strip()
                if not steps_text or not url:
                    continue
                steps = _parse_steps(steps_text)
                if not steps:
                    continue
                results.append((steps, url))
    except OSError as e:
        logging.exception(f"Failed to read {file_path}: {e}")
    return results


# -------------------------- Frame utilities -----------------------------------
async def _element_is_visible_in_frame(frame: Frame, selector: str) -> bool:
    try:
        loc = frame.locator(selector)
        count = await loc.count()
        return count > 0 and await loc.first.is_visible()
    except Exception:
        return False

async def _element_exists_in_frame(frame: Frame, selector: str) -> bool:
    try:
        return (await frame.locator(selector).count()) > 0
    except Exception:
        return False

async def selector_exists_in_any_frame(page: Page, selector: str) -> bool:
    try:
        for fr in list(page.frames):
            if await _element_exists_in_frame(fr, selector):
                return True
        return False
    except Exception:
        return False


# -------------------------- Wait with cancel -----------------------------------
async def wait_for_selector_any_frame(
    page: Page,
    selector: str,
    state: Literal["visible", "hidden", "detached"],
    timeout_ms: int,
    poll_interval_ms: int = 150,
    cancel_if_exists_selector: Optional[str] = None,
) -> None:
    """Wait for a selector in any frame; abort if cancel selector appears."""
    if cancel_if_exists_selector and await selector_exists_in_any_frame(page, cancel_if_exists_selector):
        raise _SkipScenario(f"notIf '{cancel_if_exists_selector}' detected before wait started.")

    deadline = time.monotonic() + timeout_ms / 1000
    last_log_emit = 0.0

    while True:
        now = time.monotonic()

        # Cancel check each poll
        if cancel_if_exists_selector and await selector_exists_in_any_frame(page, cancel_if_exists_selector):
            raise _SkipScenario(f"notIf '{cancel_if_exists_selector}' detected during wait.")

        if now >= deadline:
            raise PlaywrightTimeoutError(
                f"Timeout {timeout_ms} ms waiting for '{selector}' to be {state} (any-frame)"
            )

        frames = list(page.frames)  # ensure it's a concrete iterable

        if state == "visible":
            for fr in frames:
                if await _element_is_visible_in_frame(fr, selector):
                    return

        elif state == "hidden":
            # Hidden when no frame has a visible match
            any_visible = False
            for fr in frames:
                if await _element_is_visible_in_frame(fr, selector):
                    any_visible = True
                    break
            if not any_visible:
                return

        elif state == "detached":
            # Detached when no frame has the element at all
            any_exists = False
            for fr in frames:
                if await _element_exists_in_frame(fr, selector):
                    any_exists = True
                    break
            if not any_exists:
                return

        await asyncio.sleep(poll_interval_ms / 1000)

        if now - last_log_emit > 2.0:
            msg = f"…still waiting for '{selector}' to be {state}"
            if cancel_if_exists_selector:
                msg += f" (will skip if '{cancel_if_exists_selector}' appears)"
            logging.info(msg)
            last_log_emit = now


# -------------------------- Runner --------------------------------------------
class PlaywrightScenarioRunner:
    def __init__(self, headless: bool, default_timeout_ms: int) -> None:
        self._headless = headless
        self._default_timeout_ms = int(default_timeout_ms)

    async def run_url(self, url: str, steps: List[Step]) -> None:
        if not url.startswith("http"):
            logging.warning(f"Invalid URL: {url}")
            return
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=self._headless)
            page = await browser.new_page()
            page.set_default_timeout(self._default_timeout_ms)
            try:
                await page.goto(url)
                logging.info(f"Opened {url} (headless={self._headless}), running {len(steps)} steps.")
                await self._execute_steps(page, steps)
                logging.info(f"✅ Done with {url}")
            except _SkipScenario as s:
                logging.info(f"⏭️ Skipped scenario for {url}: {s}")
                await asyncio.sleep(1)
            except Exception as e:
                logging.exception(f"Error running scenario for {url}: {e}")
            finally:
                await browser.close()

    async def _execute_steps(self, page: Page, steps: List[Step]) -> None:
        for i, step in enumerate(steps, 1):
            try:
                # Pre-step notIf check
                if step.not_if and await selector_exists_in_any_frame(page, step.not_if):
                    logging.info(f"Step {i}: notIf '{step.not_if}' detected early; skipping scenario.")
                    raise _SkipScenario()

                if step.kind == "sleep":
                    logging.info(f"Step {i}: sleep {step.value}s")
                    await asyncio.sleep(step.value)
                    continue

                effective_timeout = int(step.timeout_ms or self._default_timeout_ms)
                state: Literal["visible", "hidden", "detached"] = (
                    "visible" if step.kind == "waitForShow"
                    else "hidden" if step.kind == "waitForHide"
                    else "detached"
                )

                logging.info(
                    f"Step {i}: {step.kind} '{step.value}' "
                    f"(timeoutMs={effective_timeout})"
                    + (f" with notIf '{step.not_if}'" if step.not_if else "")
                )

                await wait_for_selector_any_frame(
                    page,
                    selector=step.value,
                    state=state,
                    timeout_ms=effective_timeout,
                    cancel_if_exists_selector=step.not_if,
                )
            except _SkipScenario:
                raise
            except Exception as e:
                logging.exception(f"Step {i} ({step.kind}) failed: {e}")
                return


# -------------------------- Main ---------------------------------------------
async def main() -> None:
    setup_logging()
    logging.info("🚀 Script started.")
    config = load_config()
    scenarios = parse_urls_file(URLS_FILE)
    if not scenarios:
        logging.warning("No valid lines in urls.txt.")
        return

    runner = PlaywrightScenarioRunner(
        headless=bool(config.get("headless", False)),
        default_timeout_ms=int(config.get("elementLocatingTimeoutInMs", 30_000)),
    )

    for steps, url in scenarios:
        await runner.run_url(url, steps)

    logging.info("🎉 All URLs processed successfully.")


if __name__ == "__main__":
    asyncio.run(main())