182 lines
6.4 KiB
Python
182 lines
6.4 KiB
Python
"""Entry point — run crawl scenarios or start the admin panel."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
import time
|
|
|
|
import uvicorn
|
|
|
|
from config import config
|
|
from db import flow_config_repo, init_db
|
|
from db.repositories.runs import runs_repo
|
|
from logger import get_logger
|
|
|
|
log = get_logger(__name__)
|
|
|
|
|
|
def _load_flow(cfg_id: int | None = None) -> dict[str, object]:
|
|
if cfg_id is not None:
|
|
cfg = asyncio.run(flow_config_repo.get(cfg_id))
|
|
if cfg is None:
|
|
log.error("Flow config %d not found.", cfg_id)
|
|
else:
|
|
cfg = asyncio.run(flow_config_repo.get_first())
|
|
if cfg is None:
|
|
log.warning("No flow configs found — running without dynamic flow steps.")
|
|
return cfg or {}
|
|
|
|
|
|
def _record(
|
|
scenario: str,
|
|
identifier: str,
|
|
flow_result: object | None,
|
|
exc: Exception | None,
|
|
t0: float,
|
|
) -> None:
|
|
from crawler.dynamic_flow import DynamicFlowResult
|
|
|
|
total_ms = int((time.monotonic() - t0) * 1000)
|
|
if exc is not None:
|
|
asyncio.run(runs_repo.insert(scenario, identifier, False, str(exc), [], total_ms))
|
|
log.error("%s failed: %s", identifier, exc)
|
|
return
|
|
|
|
if isinstance(flow_result, DynamicFlowResult):
|
|
steps_data = [
|
|
{"name": s.name, "success": s.success, "error": s.error, "duration_ms": s.duration_ms}
|
|
for s in flow_result.steps
|
|
]
|
|
asyncio.run(runs_repo.insert(
|
|
scenario, identifier, flow_result.success,
|
|
flow_result.failure_reason, steps_data, total_ms,
|
|
))
|
|
if flow_result.success:
|
|
log.info("%s OK — %d ms", identifier, total_ms)
|
|
else:
|
|
log.error("%s FAIL (%s) — %d ms", identifier, flow_result.failure_reason, total_ms)
|
|
else:
|
|
asyncio.run(runs_repo.insert(scenario, identifier, True, None, [], total_ms))
|
|
log.info("%s OK (no flow config) — %d ms", identifier, total_ms)
|
|
|
|
|
|
def run_otp(phone: str, cfg_id: int | None = None) -> None:
|
|
from crawler.driver import driver_session
|
|
from crawler.dynamic_flow import DynamicFlow
|
|
from crawler.scenarios.otp_login import OTPLoginScenario
|
|
|
|
def otp_resolver(p: str) -> str:
|
|
raise NotImplementedError("Implement otp_resolver to fetch OTP from your SMS gateway")
|
|
|
|
flow_cfg = _load_flow(cfg_id)
|
|
target_url: str = flow_cfg.get("target_url") or "" if flow_cfg else ""
|
|
t0 = time.monotonic()
|
|
try:
|
|
with driver_session() as driver:
|
|
OTPLoginScenario(driver, phone, otp_resolver, target_url).run()
|
|
result = DynamicFlow(driver, flow_cfg).run() if flow_cfg else None
|
|
except Exception as exc:
|
|
_record("otp", phone, None, exc, t0)
|
|
return
|
|
_record("otp", phone, result, None, t0)
|
|
|
|
|
|
def run_fresh(cfg_id: int | None = None) -> None:
|
|
from crawler.driver import driver_session
|
|
from crawler.dynamic_flow import DynamicFlow
|
|
from crawler.scenarios.fresh_session import FreshSessionScenario
|
|
|
|
flow_cfg = _load_flow(cfg_id)
|
|
target_url: str = flow_cfg.get("target_url") or "" if flow_cfg else ""
|
|
t0 = time.monotonic()
|
|
try:
|
|
with driver_session() as driver:
|
|
FreshSessionScenario(driver, target_url).run()
|
|
result = DynamicFlow(driver, flow_cfg).run() if flow_cfg else None
|
|
except Exception as exc:
|
|
_record("fresh", "fresh", None, exc, t0)
|
|
return
|
|
_record("fresh", "fresh", result, None, t0)
|
|
|
|
|
|
def run_batch(workers: int, total_runs: int, stagger_ms: int, headless: bool, cfg_id: int | None = None) -> None:
|
|
from crawler.batch import run_batch as _run_batch
|
|
|
|
flow_cfg = _load_flow(cfg_id)
|
|
if not flow_cfg:
|
|
log.error("No active flow config — activate one in the admin panel first.")
|
|
sys.exit(1)
|
|
|
|
completed_count = [0]
|
|
|
|
def _progress(completed: int, succeeded: int, failed: int) -> None:
|
|
completed_count[0] = completed
|
|
log.info("Progress: %d/%d (ok=%d fail=%d)", completed, total_runs, succeeded, failed)
|
|
|
|
result = _run_batch(
|
|
cfg=flow_cfg,
|
|
workers=workers,
|
|
total_runs=total_runs,
|
|
headless=headless,
|
|
stagger_ms=stagger_ms,
|
|
on_progress=_progress,
|
|
)
|
|
log.info(
|
|
"Batch complete — succeeded=%d/%d failed=%d",
|
|
result["succeeded"], result["total"], result["failed"],
|
|
)
|
|
|
|
|
|
def run_admin() -> None:
|
|
init_db()
|
|
from admin.main import app
|
|
uvicorn.run(app, host=config.admin_host, port=config.admin_port)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Seed crawler & admin")
|
|
sub = parser.add_subparsers(dest="cmd", required=True)
|
|
|
|
sub.add_parser("admin", help="Start the admin panel")
|
|
|
|
otp_p = sub.add_parser("otp", help="Run Scenario 1 (SMS-OTP login)")
|
|
otp_p.add_argument("--phone", help="Single phone number (default: all from config)")
|
|
otp_p.add_argument("--config-id", type=int, default=None, help="Flow config ID (default: first config)")
|
|
|
|
fresh_p = sub.add_parser("fresh", help="Run Scenario 2 (fresh session)")
|
|
fresh_p.add_argument("--config-id", type=int, default=None, help="Flow config ID (default: first config)")
|
|
|
|
batch_p = sub.add_parser("batch", help="Run many parallel fresh-session runners")
|
|
batch_p.add_argument("--config-id", type=int, default=None, help="Flow config ID (default: first config)")
|
|
batch_p.add_argument("--workers", type=int, default=3, help="Parallel Chrome instances (default: 3)")
|
|
batch_p.add_argument("--runs", type=int, default=10, help="Total runs to complete (default: 10)")
|
|
batch_p.add_argument("--stagger", type=int, default=500, help="Ms delay between worker launches (default: 500)")
|
|
batch_p.add_argument("--no-headless", dest="headless", action="store_false", help="Show browser windows")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.cmd == "admin":
|
|
run_admin()
|
|
elif args.cmd == "otp":
|
|
phones = [args.phone] if args.phone else config.phone_numbers
|
|
if not phones:
|
|
log.error("No phone numbers configured. Set PHONE_NUMBERS in .env or pass --phone")
|
|
sys.exit(1)
|
|
for phone in phones:
|
|
run_otp(phone, cfg_id=args.config_id)
|
|
elif args.cmd == "fresh":
|
|
run_fresh(cfg_id=args.config_id)
|
|
elif args.cmd == "batch":
|
|
run_batch(
|
|
workers=args.workers,
|
|
total_runs=args.runs,
|
|
stagger_ms=args.stagger,
|
|
headless=args.headless,
|
|
cfg_id=args.config_id,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|