formbot/formbot.py at master · helanabi/formbot · GitHub

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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
#!/usr/bin/env python

import argparse
import csv
import sys
import time
from playwright.sync_api import expect, sync_playwright, TimeoutError

def parse_args():
    parser = argparse.ArgumentParser(prog="formbot")
    parser.add_argument("csv", help="CSV data file")
    parser.add_argument("url", help="Web page URL")
    parser.add_argument("--action", "-a", default="submit",
                        help="Action button name (default: submit)")
    parser.add_argument("--pause", "-p", action="store_true",
                        help="Run headed browser and pause at the end")
    parser.add_argument("--delay", "-d", default=0, type=float,
                        help="Add delay between submissions in seconds "
                        "(default: 0)")
    parser.add_argument("--timeout", "-t", default=30, type=float,
                        help="Time to wait for elements in seconds. "
                        "Pass 0 to disable timeout (default: 30)")
    return parser.parse_args()

def warn(msg_or_name, value=''):
    if not value:
        msg = msg_or_name
    else:
        msg = f"unexpected value for '{msg_or_name}': {value}. Skipping."
    print(msg, file=sys.stderr)

def get_control(page, field):
    """Find a form control and its type in a page.

    Return a tuple of control type and its locator.
    """
    control = page.get_by_label(field).or_(
        page.get_by_placeholder(field)).first

    tag = control.evaluate("e => e.tagName").lower()

    if tag == "input":
        return control.get_attribute("type"), control
    else:
        return tag, control

def enter_value(page, field, value, default):
    type_, ctrl = get_control(page, field)

    if type_ == "checkbox":
        true_values = ("yes", "y", "true", "1")
        false_values = ("no", "n", "false", "0")
        value = value.strip().lower()

        if value not in true_values + false_values:
            warn(field, value or "*empty*")
            value = default

        if value in true_values:
            ctrl.check()
        else:
            ctrl.uncheck()

    elif type_ == "select":
        try:
            ctrl.select_option(value)
        except TimeoutError:
            warn(field, value)
            ctrl.select_option(default)
    else:
        ctrl.fill(value)

def resolve_action_ctrl(page, text):
    def is_visible(locator):
        try:
            locator.wait_for()
        except TimeoutError:
            return False
        return True

    for control in (page.get_by_role("button", name=text),
                    page.get_by_role("link", name=text),
                    page.get_by_text(text)):
        if is_visible(control):
            return control

def main(args, page):
    page.set_default_timeout(args.timeout * 1000)
    page.goto(args.url)

    with open(args.csv, newline='') as f:
        reader = csv.reader(f)
        header = next(reader)
        defaults = []

        for field in header:
            type_, ctrl = get_control(page, field)
            if type_ == "checkbox":
                defaults.append("yes" if ctrl.is_checked() else "no")
            else:
                defaults.append(ctrl.input_value())

        for i, row in enumerate(reader):
            if len(row) < len(header):
                warn(f"short row at line: {i+2}. Some fields will be skipped.")
            elif len(row) > len(header):
                warn(f"long row at line: {i+2}. Some values will be dropped.")

            for field, value, default in zip(header, row, defaults):
                enter_value(page, field, value, default)
            if not i:
                action_control = resolve_action_ctrl(page, args.action)
                if not action_control:
                    print(f"Error: Action control '{args.action}' not found",
                          file=sys.stderr)
                    sys.exit(1)
            action_control.click()
            page.wait_for_timeout(args.delay * 1000)
            page.wait_for_load_state("networkidle")
            try:
                expect(action_control).to_be_enabled()
            except AssertionError:
                page.goto(args.url)

    if args.pause:
        page.pause()

if __name__ == "__main__":
    with sync_playwright() as p:
        args = parse_args()
        try:
            browser = p.chromium.launch(headless=not args.pause)
            main(args, browser.new_page())
        except TimeoutError as e:
            print(e, file=sys.stderr)
        finally:
            browser.close()