#!/usr/bin/python3 import argparse import base64 import random import string import sys from datetime import datetime, timedelta from typing import List, Union, Callable import requests from tqdm import tqdm MACHINE = "devmachine" UA = 'wakatime/13.0.7 (Linux-4.15.0-91-generic-x86_64-with-glibc2.4) Python3.8.0.final.0 generator/1.42.1 generator-wakatime/4.0.0' LANGUAGES = { 'Go': 'go', 'Java': 'java', 'JavaScript': 'js', 'Python': 'py' } class Heartbeat: def __init__( self, entity: str, project: str, language: str, time: float, is_write: bool = True, branch: str = 'master', type: str = 'file' ): self.entity: str = entity self.project: str = project self.language: str = language self.time: float = time self.is_write: bool = is_write self.branch: str = branch self.type: str = type self.category: Union[str, None] = None class ConfigParams: def __init__(self): self.api_url = '' self.api_key = '' self.n = 0 self.n_projects = 0 self.offset = 0 self.seed = 0 def generate_data(n: int, n_projects: int = 5, n_past_hours: int = 24) -> List[Heartbeat]: data: List[Heartbeat] = [] now: datetime = datetime.today() projects: List[str] = [randomword(random.randint(5, 10)) for _ in range(n_projects)] languages: List[str] = list(LANGUAGES.keys()) for _ in range(n): p: str = random.choice(projects) l: str = random.choice(languages) f: str = randomword(random.randint(2, 8)) delta: timedelta = timedelta( hours=random.randint(0, n_past_hours - 1), minutes=random.randint(0, 59), seconds=random.randint(0, 59), milliseconds=random.randint(0, 999), microseconds=random.randint(0, 999) ) data.append(Heartbeat( entity=f'/home/me/dev/{p}/{f}.{LANGUAGES[l]}', project=p, language=l, time=(now - delta).timestamp() )) return data def post_data_sync(data: List[Heartbeat], url: str, api_key: str): encoded_key: str = str(base64.b64encode(api_key.encode('utf-8')), 'utf-8') for h in data: r = requests.post(url, json=[h.__dict__], headers={ 'User-Agent': UA, 'Authorization': f'Basic {encoded_key}', 'X-Machine-Name': MACHINE, }) if r.status_code != 201: print(r.text) sys.exit(1) def make_gui(callback: Callable[[ConfigParams, Callable[[int], None]], None]) -> ('QApplication', 'QWidget'): from PyQt5.QtCore import Qt from PyQt5.QtWidgets import QApplication, QWidget, QFormLayout, QHBoxLayout, QVBoxLayout, QGroupBox, QLabel, \ QLineEdit, QSpinBox, QProgressBar, QPushButton # Main app app = QApplication([]) window = QWidget() window.setWindowTitle('Wakapi Sample Data Generator') window.setFixedSize(window.sizeHint()) window.setMinimumWidth(350) container_layout = QVBoxLayout() # Top Controls form_layout_1 = QFormLayout() url_input_label = QLabel('URL:') url_input = QLineEdit() url_input.setPlaceholderText('Wakatime API Url') url_input.setText('http://localhost:3000/api') api_key_input_label = QLabel('API Key:') api_key_input = QLineEdit() api_key_input.setPlaceholderText(f'{"x"*8}-{"x"*4}-{"x"*4}-{"x"*4}-{"x"*12}') form_layout_1.addRow(url_input_label, url_input) form_layout_1.addRow(api_key_input_label, api_key_input) # Middle controls form_layout_2 = QFormLayout() params_container = QGroupBox('Parameters') params_container.setLayout(form_layout_2) heartbeats_input_label = QLabel('# Heartbeats') heartbeats_input = QSpinBox() heartbeats_input.setMaximum(2147483647) heartbeats_input.setValue(100) projects_input_label = QLabel('# Projects:') projects_input = QSpinBox() projects_input.setMinimum(1) projects_input.setValue(5) offset_input_label = QLabel('Time Offset (hrs):') offset_input = QSpinBox() offset_input.setMinimum(-2147483647) offset_input.setMaximum(0) offset_input.setValue(-12) seed_input_label = QLabel('Random Seed:') seed_input = QSpinBox() seed_input.setMaximum(2147483647) seed_input.setValue(1337) form_layout_2.addRow(heartbeats_input_label, heartbeats_input) form_layout_2.addRow(projects_input_label, projects_input) form_layout_2.addRow(offset_input_label, offset_input) form_layout_2.addRow(seed_input_label, seed_input) # Bottom controls bottom_layout = QHBoxLayout() start_button = QPushButton('Generate') progress_bar = QProgressBar() bottom_layout.addWidget(progress_bar) bottom_layout.addWidget(start_button) # Wiring up container_layout.addLayout(form_layout_1) container_layout.addWidget(params_container) container_layout.addLayout(bottom_layout) container_layout.setStretch(1, 1) window.setLayout(container_layout) # Done dialog done_dialog = QWidget() done_dialog.setWindowTitle('Done') done_ok_button = QPushButton('Ok') done_layout = QVBoxLayout() done_layout.addWidget(QLabel('Done!'), alignment=Qt.AlignCenter) done_layout.addWidget(done_ok_button, alignment=Qt.AlignCenter) done_dialog.setFixedSize(done_dialog.sizeHint()) done_ok_button.clicked.connect(done_dialog.close) done_dialog.setLayout(done_layout) # Logic def parse_params() -> ConfigParams: params = ConfigParams() params.api_url = url_input.text() params.api_key = api_key_input.text() params.n = heartbeats_input.value() params.n_projects = projects_input.value() params.offset = offset_input.value() params.seed = seed_input.value() return params def update_progress(inc=1): current = progress_bar.value() updated = current + inc if updated == progress_bar.maximum() - 1: progress_bar.setValue(0) start_button.setEnabled(True) done_dialog.show() return progress_bar.setValue(updated) def call_back(): params = parse_params() progress_bar.setMaximum(params.n) start_button.setEnabled(False) callback(params, update_progress) start_button.clicked.connect(call_back) return app, window def parse_arguments(): parser = argparse.ArgumentParser(description='Wakapi test data insertion script.') parser.add_argument('--headless', default=False, help='do not show a gui', action='store_true') parser.add_argument('-n', type=int, default=20, help='total number of random heartbeats to generate and insert') parser.add_argument('-u', '--url', type=str, default='http://localhost:3000/api', help='url of your api\'s heartbeats endpoint') parser.add_argument('-k', '--apikey', type=str, help='your api key (to get one, go to the web interface, create a new user, log in and copy the key)') parser.add_argument('-p', '--projects', type=int, default=5, help='number of different fake projects to generate') parser.add_argument('-o', '--offset', type=int, default=24, help='negative time offset in hours from now for to be used as an interval within which to generate heartbeats for') parser.add_argument('-s', '--seed', type=int, default=2020, help='a seed for initializing the pseudo-random number generator') return parser.parse_args() def args_to_params(parsed_args: argparse.Namespace) -> (ConfigParams, bool): params = ConfigParams() params.n = parsed_args.n params.n_projects = parsed_args.projects params.offset = parsed_args.offset params.seed = parsed_args.seed params.api_url = parsed_args.url params.api_key = parsed_args.apikey return params, not parsed_args.headless def randomword(length: int) -> str: letters = string.ascii_lowercase + 'äöü💩' # test utf8 and utf8mb4 characters as well return ''.join(random.choice(letters) for _ in range(length)) def run(params: ConfigParams, update_progress: Callable[[int], None]): random.seed(params.seed) data: List[Heartbeat] = generate_data( params.n, params.n_projects, params.offset * -1 if params.offset < 0 else params.offset ) for d in data: post_data_sync([d], f'{params.api_url}/heartbeats', params.api_key) update_progress(1) if __name__ == '__main__': params, show_gui = args_to_params(parse_arguments()) if show_gui: app, window = make_gui(callback=run) window.show() app.exec() else: pbar = tqdm(total=params.n) run(params, pbar.update)