diff --git a/hetzner_kube.py b/hetzner_kube.py deleted file mode 100755 index 1b0f77b..0000000 --- a/hetzner_kube.py +++ /dev/null @@ -1,237 +0,0 @@ -#!/usr/bin/env python3.9 -import base64 -import json -import logging -import pprint -import subprocess -import time -import webbrowser -from pathlib import Path - -import httpx -import typer - -hetzner_token = 'zdlkmAF9EswvOIEGvWgutULYTmUpa3c44YcKZOB0jymQfgvra7kRVJxuLiFTouM9' -client = httpx.Client(base_url='https://api.hetzner.cloud/v1', headers={'Authorization': f'Bearer {hetzner_token}'}) - - -datacenters = [ - {'id': 2, 'name': 'nbg1-dc3', 'description': 'Nuremberg 1 DC 3'}, - {'id': 3, 'name': 'hel1-dc2', 'description': 'Helsinki 1 DC 2'}, - {'id': 4, 'name': 'fsn1-dc14', 'description': 'Falkenstein 1 DC14'}, - {'id': 5, 'name': 'ash-dc1', 'description': 'Ashburn DC1'}, -] - -server_types = { - 'cx11': {'cores': 1, 'memory': 2.0, 'disk': 20}, - 'cx21': {'cores': 2, 'memory': 4.0, 'disk': 40}, - 'cx31': {'cores': 2, 'memory': 8.0, 'disk': 80}, - 'cx41': {'cores': 4, 'memory': 16.0, 'disk': 160}, - 'cx51': {'cores': 8, 'memory': 32.0, 'disk': 240}, - 'ccx11': {'cores': 2, 'memory': 8.0, 'disk': 80}, - 'ccx21': {'cores': 4, 'memory': 16.0, 'disk': 160}, - 'ccx31': {'cores': 8, 'memory': 32.0, 'disk': 240}, - 'ccx41': {'cores': 16, 'memory': 64.0, 'disk': 360}, - 'ccx51': {'cores': 32, 'memory': 128.0, 'disk': 600}, - 'cpx11': {'cores': 2, 'memory': 2.0, 'disk': 40}, - 'cpx21': {'cores': 3, 'memory': 4.0, 'disk': 80}, - 'cpx31': {'cores': 4, 'memory': 8.0, 'disk': 160}, - 'cpx41': {'cores': 8, 'memory': 16.0, 'disk': 240}, - 'cpx51': {'cores': 16, 'memory': 32.0, 'disk': 360}, - 'ccx12': {'cores': 2, 'memory': 8.0, 'disk': 80}, - 'ccx22': {'cores': 4, 'memory': 16.0, 'disk': 160}, - 'ccx32': {'cores': 8, 'memory': 32.0, 'disk': 240}, - 'ccx42': {'cores': 16, 'memory': 64.0, 'disk': 360}, - 'ccx52': {'cores': 32, 'memory': 128.0, 'disk': 600}, - 'ccx62': {'cores': 48, 'memory': 192.0, 'disk': 960}, -} - - -def create_server(name: str, datacenter: str): - payload = { - "datacenter": datacenter, - "image": "ubuntu-20.04", - "name": name, - "server_type": "cpx21", - "ssh_keys": ["entropy"], - } - - res = client.post('/servers', json=payload) - res.raise_for_status() - - data = res.json() - server = data['server'] - pprint.pprint(server) - - return { - 'id': server['id'], - 'name': server['name'], - 'ip': server['public_net']['ipv4']['ip'], - } - - -def find_server(name: str) -> dict: - res = client.get('/servers') - res.raise_for_status() - - data = res.json() - - server = [it for it in data['servers'] if it['name'] == name][0] - return { - 'id': server['id'], - 'name': server['name'], - 'ip': server['public_net']['ipv4']['ip'], - } - - -def delete_server(name: str) -> None: - server = find_server(name) - logging.info(f'deleting server={name} with id={server["id"]}') - - res = client.delete(f'/servers/{server["id"]}') - res.raise_for_status() - - logging.info(f'deleted server={name}') - - -def copy_to_clipboard(text: str) -> None: - subprocess.Popen(['pbcopy'], stdin=subprocess.PIPE).communicate(input=text.encode()) - - -class Server: - ssh_options = ['-o', 'StrictHostKeyChecking=no'] - - def __init__(self, ip_address: str, name: str = None): - self.ip = ip_address - self.ssh_host = f'root@{self.ip}' - self.name = name - - @classmethod - def from_hetzner(cls, server: dict): - return cls(server['ip'], name=server['name']) - - def run_command(self, cmd: str, return_output: bool = False) -> str: - - logging.debug(f'running command={cmd} on {self.ssh_host}') - extra_args = {} - if return_output: - extra_args['stdout'] = subprocess.PIPE - p = subprocess.run(['ssh', *self.ssh_options, self.ssh_host, '--', cmd], text=True, **extra_args) - p.check_returncode() - return p.stdout - - def install_package(self, packages: str) -> str: - return self.run_command(f'apt-get install -y {packages}') - - def install_shell(self): - self.install_package('fish') - self.run_command('chsh -s /usr/bin/fish') - - def install_microk8s(self): - self.install_package('snapd') - self.run_command('snap install microk8s --classic') - self.run_command('ufw allow in on cni0 && ufw allow out on cni0 && ufw default allow routed') - self.run_command('microk8s enable dashboard dns rbac ingress storage prometheus registry') - - def get_microk8s_token(self): - raw_json = self.run_command('microk8s kubectl -n kube-system get secret -o json', return_output=True) - secrets = json.loads(raw_json)['items'] - - default_token_secret = [it for it in secrets if it['metadata']['name'].startswith('default-token')][0] - token_base64 = default_token_secret['data']['token'] - - token = base64.decodebytes(token_base64.encode()).decode() - return token - - def show_microk8s_access_info(self): - logging.info('getting default service account token') - token = self.get_microk8s_token() - print(token) - logging.info('copying token to clipboard') - copy_to_clipboard(token) - - logging.info('getting dashboard ip') - proxy = self.get_microk8s_dashboard_proxy() - print(proxy['ssh']) - copy_to_clipboard(proxy['ssh']) - - logging.info('opening url in the browser') - print(proxy['url']) - webbrowser.open_new_tab(proxy['url']) - - def get_microk8s_dashboard_proxy(self): - cluster_ip = self.run_command( - 'microk8s kubectl get -n kube-system -o json service/kubernetes-dashboard -o jsonpath="{.spec.clusterIP}"', - return_output=True, - ).strip() - port, local_port = 443, 18433 - - return { - 'ssh': f'ssh -vNL {local_port}:{cluster_ip}:{port} {self.ssh_host}', - 'url': f'https://localhost:{local_port}', - } - - def write_kubeconfig(self, filename: str): - logging.info(f'writing kubeconfig to {filename}') - kubeconfig = self.run_command('microk8s config', return_output=True) - file_path = Path(filename).expanduser() - file_path.parent.mkdir(parents=True, exist_ok=True) - file_path.write_text(kubeconfig) - - def install_micro_editor(self): - logging.info('installing micro editor') - self.run_command('curl https://getmic.ro | bash && mv ./micro /usr/bin/') - - def setup(self): - self.run_command('apt-get update') - - self.install_shell() - self.install_micro_editor() - self.install_microk8s() - - logging.info('waiting until pods are online') - time.sleep(10) - - self.show_microk8s_access_info() - self.write_kubeconfig(f'~/Desktop/kubeconfig-{self.name}.yml') - - -cli = typer.Typer(invoke_without_command=False) - - -@cli.command() -def create( - name: str, - datacenter: str = typer.Option('nbg1-dc3'), -): - create_server(name=name, datacenter=datacenter) - - -@cli.command() -def destroy( - name: str, -): - delete_server(name=name) - - -@cli.command() -def setup( - name: str, -): - server = find_server(name) - server = Server.from_hetzner(server) - server.setup() - - -@cli.command() -def show_k8s( - name: str, -): - server = Server.from_hetzner(find_server(name)) - server.show_microk8s_access_info() - server.write_kubeconfig(f'~/Desktop/kubeconfig-{server.name}.yml') - - -if __name__ == '__main__': - logging.basicConfig(level=logging.INFO) - cli() diff --git a/imgup.py b/imgup.py deleted file mode 100755 index ccc75d8..0000000 --- a/imgup.py +++ /dev/null @@ -1,53 +0,0 @@ -#!/usr/bin/env python3.9 -import argparse -import concurrent.futures -import dataclasses -import functools -import logging -import os -import random -import re -import sys -import time -import typing -import uuid -from pathlib import Path -from typing import Optional - -import httpx - - -IMGBB_TOKEN = os.getenv('IMGBB_TOKEN', '391f0559cc209b11f37eda0084bb3281') -client = httpx.Client(base_url='https://api.imgbb.com/1/', params={'key': IMGBB_TOKEN}) - - -def upload_image(image_path: Path) -> str: - filename = f'i{image_path.suffix}' - with image_path.open('rb') as f: - res = client.post('/upload', files={'image': (filename, f)}) - res.raise_for_status() - data = res.json()['data'] - return data['url'] - - -def parse_args(argv: Optional[typing.Sequence[str]] = None) -> argparse.Namespace: - arger = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter) - arger.add_argument('image_paths', nargs='+', action='store', type=Path, help='Path to image to upload') - if not argv or len(argv) == 0: - arger.print_help() - sys.exit(1) - - return arger.parse_args(argv) - - -def main(): - args = parse_args(sys.argv[1:]) - image_paths = args.image_paths - with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool: - results = list(pool.map(upload_image, image_paths)) - for url in results: - print(url) - - -if __name__ == '__main__': - main() diff --git a/personio_attendance.py b/personio_attendance.py deleted file mode 100644 index 6356674..0000000 --- a/personio_attendance.py +++ /dev/null @@ -1,71 +0,0 @@ -import datetime -from pathlib import Path -import sys -import time - -from playwright.sync_api import BrowserContext, sync_playwright - - -class creds: - employee_id: str = "9374380" - email: str = "abduessamet.kocak@refurbed.com" - password: str = "^6LMq$LX&2#c4y" - - -def main(): - try: - day = datetime.date.today().fromisoformat(sys.argv[1]).isoformat() - except: - day = "today" - - with sync_playwright() as playwright: - data_dir = Path("/tmp/personio") - data_dir.mkdir(exist_ok=True, parents=True) - context = playwright.chromium.launch_persistent_context(headless=False, user_data_dir=data_dir) - with context: - fill_attendance( - context=context, - day=day, - work=("08:00", "17:00"), - pause=("12:00", "13:00"), - ) - - -def fill_attendance(context: BrowserContext, day: str, work: tuple[str, str], pause=[str, str]): - # Open new page - page = context.new_page() - - url = f"https://refurbed.personio.de/attendance/employee/{creds.employee_id}/" - page.goto(url) - page.wait_for_load_state() - - if "/login/index" in page.url: - page.locator('[placeholder="Email"]').fill(creds.email) - page.locator('[placeholder="Password"]').fill(creds.password) - page.locator('button:has-text("Login")').click() - - card_selector = f'[data-test-id="day_{day}"] button' - if day == "today": - card_selector = '[data-test-id="today-cell"] button' - - while True: - page.locator(card_selector).click() - if page.locator('[data-test-id="day-entry-dialog"]').is_visible(): - break - time.sleep(0.05) - - work_start, work_end = work - page.locator('[data-test-id="timerange-start"]').first.fill(work_start) - page.locator('[data-test-id="work-entry"] [data-test-id="timerange-end"]').fill(work_end) - - pause_start, pause_end = pause - page.locator('[data-test-id="break-entry"] [data-test-id="timerange-start"]').fill(pause_start) - page.locator('[data-test-id="break-entry"] [data-test-id="timerange-end"]').fill(pause_end) - - page.locator('[data-test-id="day-entry-save"]').click() - # wait until network req is sent - time.sleep(1) - - -if __name__ == "__main__": - main() diff --git a/plab_search.py b/plab_search.py deleted file mode 100644 index 9d182fb..0000000 --- a/plab_search.py +++ /dev/null @@ -1,188 +0,0 @@ -from __future__ import annotations - -import argparse -import dataclasses -import datetime -import json -import typing -from pathlib import Path -from pprint import pprint -from urllib.parse import urljoin - -import bs4 -import httpx - - -@dataclasses.dataclass() -class SearchResult: - posted_at: datetime.datetime - title: str - size_mb: float - url: str - - -class Auth(typing.Protocol): - def authenticate(self) -> dict[str, str]: - ... - - -@dataclasses.dataclass() -class CookieFileAuth: - file_path: Path - - def authenticate(self) -> dict[str, str]: - with self.file_path.open() as f: - return json.load(f) - - -@dataclasses.dataclass() -class CredentialAuth: - username: str - password: str - - def authenticate(self) -> dict[str, str]: - client = httpx.Client(base_url="https://pornolab.net/") - res = client.post( - "/forum/login.php", - data={ - "login_username": self.username, - "login_password": self.password, - "login": "Вход", - }, - follow_redirects=True, - ) - res.raise_for_status() - soup = bs4.BeautifulSoup(res.text, "html.parser") - username_el = soup.select_one(".topmenu .med") - if self.username not in username_el.text.strip(): - raise ValueError("failed to login") - return dict(client.cookies) - - -@dataclasses.dataclass() -class PersistedCookieAuth: - cookie_path: Path - auth: Auth - - def authenticate(self) -> dict[str, str]: - cookie_auth = CookieFileAuth(self.cookie_path) - try: - return cookie_auth.authenticate() - except (ValueError, FileNotFoundError): - pass - - cookies = self.auth.authenticate() - with self.cookie_path.open("w") as f: - json.dump(cookies, f) - return cookies - - -class PlabAuthMiddleware(httpx.Auth): - def auth_flow(self, request: httpx.Request) -> typing.Generator[httpx.Request, httpx.Response, None]: - # if no cookies in req - # fill cookie - # check response - # reauth - # fill cookie - - res = yield request - return super().auth_flow(request) - - -class Plab: - def __init__(self, auth: Auth): - self.auth = auth - self.client = httpx.Client(base_url="https://pornolab.net/") - - def search(self, term: str) -> list[SearchResult]: - self._ensure_session() - res = self.client.post( - "/forum/tracker.php", - data={ - "max": "1", - "to": "1", - "nm": f'{term}', - }, - follow_redirects=True, - ) - res.raise_for_status() - - if 'Введите ваше имя и пароль' in res.text: - raise - - soup = bs4.BeautifulSoup(res.text, "html.parser") - rows = soup.select("#tor-tbl tbody tr") - items = [] - for row in rows: - title_link = row.select_one(f'[href*="viewtopic.php?t="]') - if not title_link: - continue - size_mb = ( - int(row.select_one('[href*="dl.php?t="]').find_previous_sibling(name='u').text.strip()) / 1_048_576 - ) - date = datetime.datetime.fromtimestamp(int(row.select_one('[title="Добавлен"] u').text.strip())) - items.append( - SearchResult( - url=urljoin("https://pornolab.net/forum/", title_link["href"]), - title=title_link.text.strip(), - size_mb=size_mb, - posted_at=date, - ) - ) - return items - - def _ensure_session(self): - if self.client.cookies: - return - self.client.cookies = self.auth.authenticate() - - -def dump_for_alfred(items: list[SearchResult]): - alfreds = [] - for it in items: - alfreds.append( - { - "title": it.title, - "subtitle": f"{it.size_mb:.1f}MB -- {it.posted_at.isoformat()}", - "arg": it.url, - "text": { - "copy": it.url, - "largetype": f"{it.title}\nPosted at: {it.posted_at.isoformat()}\nSize: {it.size_mb:.1f}MB", - }, - "match": f"{it.title} {it.posted_at.isoformat()}", - } - ) - print(json.dumps({"items": alfreds}, indent=2)) - - -def parse_args(): - arger = argparse.ArgumentParser() - arger.add_argument('term', help='Term to search for') - arger.add_argument('--alfred', action='store_true', help='Dump for Alfred') - return arger.parse_args() - - -def main(): - args = parse_args() - - p = Plab( - auth=PersistedCookieAuth( - cookie_path=Path(__file__).parent / "plab.json", - auth=CredentialAuth( - username="zzzp", - password="3c45HzLMgSB7M5hJ", - ), - ) - ) - - results = p.search(args.term) - - if args.alfred: - dump_for_alfred(results) - else: - for it in results: - pprint(it) - - -if __name__ == "__main__": - main() diff --git a/ui_prompts.py b/ui_prompts.py deleted file mode 100644 index 5c3f575..0000000 --- a/ui_prompts.py +++ /dev/null @@ -1,22 +0,0 @@ -import webview -from webview import Window - - -def webview_file_dialog(): - def open_file_dialog(w: Window): - try: - return w.create_file_dialog(webview.FOLDER_DIALOG)[0] - except TypeError: - pass # user exited file dialog without picking - finally: - w.hide() - w.destroy() - - window = webview.create_window("", hidden=True) - webview.start(open_file_dialog, window) - # file will either be a string or None - return file - - -if __name__ == "__main__": - print(webview_file_dialog())