diff --git a/backup-site.service b/backup-site.service
deleted file mode 100644
index e69de29..0000000
diff --git a/backup-site.timer b/backup-site.timer
deleted file mode 100644
index e69de29..0000000
diff --git a/berlin.py b/berlin.py
deleted file mode 100644
index 3cb5ecb..0000000
--- a/berlin.py
+++ /dev/null
@@ -1,48 +0,0 @@
-import contextlib
-
-from playwright.sync_api import sync_playwright, Page
-from time import sleep
-
-
-@contextlib.contextmanager
-def launch_browser() -> Page:
- with sync_playwright() as playwright:
- with playwright.chromium.launch(headless=False) as browser:
- with browser.new_context() as context:
- with context.new_page() as page:
- yield page
-
-
-def book_slot():
- with launch_browser() as page:
- page.goto("https://service.berlin.de/dienstleistung/120686/")
- page.get_by_role("complementary").get_by_role("link", name="Termin berlinweit suchen").click()
-
- with page.expect_navigation(url='https://service.berlin.de/terminvereinbarung/termin/time/*', timeout=0):
- pass
-
- with page.expect_navigation(url='https://service.berlin.de/terminvereinbarung/termin/register/*', timeout=0):
- pass
-
- page.get_by_label("Ihr Vor- und Nachname *").fill("Paulina Gutkowska")
- page.get_by_label("Ihre E-Mail Adresse *").fill("gutpaula11@gmail.com")
- if page.is_visible("Ihre Telefonnummer oder Mobilfunknummer *"):
- page.get_by_label("Ihre Telefonnummer oder Mobilfunknummer *").fill("+48501099723")
- page.get_by_role("combobox").select_option("0")
- page.get_by_label(
- "Ich erkläre mich mit den Nutzungsbedingungen einverstanden und akzeptiere diese. (Pflichtangabe. Termineintrag nur bei Einverständnis möglich) *"
- ).check()
-
- with page.expect_navigation():
- page.get_by_role("button", name="Termin eintragen").click()
-
- sleep(10)
- input()
-
-
-def main():
- book_slot()
-
-
-if __name__ == '__main__':
- main()
diff --git a/berlin_anmeldung.py b/berlin_anmeldung.py
deleted file mode 100644
index 5fa4307..0000000
--- a/berlin_anmeldung.py
+++ /dev/null
@@ -1,75 +0,0 @@
-import contextlib
-import datetime
-import locale
-
-import httpx
-from bs4 import BeautifulSoup
-
-import berlin
-
-http = httpx.Client(
- headers={
- 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36'
- },
- follow_redirects=True,
-)
-
-
-@contextlib.contextmanager
-def override_locale(category: int, val: str) -> None:
- prev = locale.getlocale(category)
- locale.setlocale(category, val)
- yield
- locale.setlocale(category, prev)
-
-
-def get_available_slots():
- res = http.get('https://service.berlin.de/dienstleistung/120686/')
- res.raise_for_status()
-
- soup = BeautifulSoup(res.text, 'html.parser')
- link = soup.select_one('[role="complementary"] .zmstermin-multi a')
- calendar_url = link.attrs['href']
-
- try:
- # redirect fails, we just need the cookies
- _ = http.get(calendar_url)
- except:
- pass
- res = http.get('https://service.berlin.de/terminvereinbarung/termin/day/')
- res.raise_for_status()
-
- return extract_days(res.text)
-
-
-def extract_days(html: str) -> list[datetime.date]:
- soup = BeautifulSoup(html, 'html.parser')
-
- found = []
- for it in soup.select('.calendar-month-table .buchbar'):
- month_year = it.find_parent(attrs={'class': 'calendar-month-table'}).select_one('thead .month').text.strip()
- day = it.text.strip()
- found.append(f'{day} {month_year}')
-
- with override_locale(locale.LC_TIME, 'de_DE'):
- days = [datetime.datetime.strptime(it, '%d %B %Y').date() for it in found]
-
- return days
-
-
-def main():
- days = get_available_slots()
- good_days = [it for it in days if it < datetime.date(2023, 1, 27)]
-
- if not good_days:
- print('no slots')
- return
-
- print('found slots')
- berlin.book_slot()
- for it in days:
- print(it)
-
-
-if __name__ == '__main__':
- main()
diff --git a/berlin_immigration.py b/berlin_immigration.py
deleted file mode 100644
index 84ecb0e..0000000
--- a/berlin_immigration.py
+++ /dev/null
@@ -1,156 +0,0 @@
-import argparse
-import asyncio
-import contextlib
-import dataclasses
-import datetime
-import logging
-import random
-import re
-import subprocess
-import urllib.parse
-from pathlib import Path
-
-from playwright.async_api import async_playwright, Browser, BrowserContext, Page
-
-
-@contextlib.asynccontextmanager
-async def launch_browser(headless: bool = False) -> Page:
- async with async_playwright() as playwright:
- browser = await playwright.chromium.launch(headless=headless)
- browser: Browser
- async with browser:
- ctx = await browser.new_context()
- ctx: BrowserContext
- async with ctx:
- page = await ctx.new_page()
- async with page:
- yield page
-
-
-class Error(Exception):
- pass
-
-
-def alert(text: str):
- subprocess.run(['open', f'alfred://runtrigger/dev.abdus.integrations/remind/?argument={urllib.parse.quote(text)}'])
-
-
-dump_dir = Path('~/Desktop').expanduser() / 'immigration'
-dump_dir.mkdir(parents=True, exist_ok=True)
-
-
-@dataclasses.dataclass
-class ResidencyApplication:
- citizenship_country: str
- num_applicants: int
- is_living_with_family: bool
- residency_category: str
- residency_purpose: str
-
-
-async def check_slots(page: Page, values: ResidencyApplication, retry_count: int = 1) -> list[datetime.datetime]:
- await page.add_init_script('''Object.defineProperty(navigator, 'webdriver', { get: () => false })''')
- window_id = random.randint(1000, 9999)
- req_id = random.randint(0, 999)
- await page.goto(
- f'https://otv.verwalt-berlin.de/ams/TerminBuchen/wizardng?dswid={window_id}&dsrid={req_id}',
- wait_until='networkidle',
- )
-
- async with page.expect_navigation():
- await page.click('.langBlock [name="txtEn"] a')
-
- await page.check('[name="gelesen"]')
-
- async with page.expect_navigation(url=re.compile('st=2'), wait_until='networkidle'):
- await page.click('[name="applicationForm:managedForm:proceed"]')
-
- await page.get_by_role("combobox", name="Citizenship *").select_option(label=values.citizenship_country)
- await asyncio.sleep(0.3)
-
- # how many applicants
- await page.select_option('[name="personenAnzahl_normal"]', value=str(values.num_applicants))
- await asyncio.sleep(0.3)
-
- # living with family
- await page.select_option('[name="lebnBrMitFmly"]', value='1' if values.is_living_with_family else '2')
- await asyncio.sleep(0.3)
-
- await page.get_by_text("Apply for a residence title").click()
- await page.locator("label").filter(has_text=values.residency_category).click()
- await page.get_by_text(values.residency_purpose).click()
-
- stop_at = datetime.datetime.now() + datetime.timedelta(minutes=28)
-
- while 'st=2' in page.url:
- retry_count -= 1
-
- async with page.expect_navigation(url=re.compile(r'st='), wait_until='networkidle', timeout=60_000):
- await page.get_by_role("button", name="Next").click()
-
- if datetime.datetime.now() > stop_at:
- raise TimeoutError('could not find a slot')
-
- if not retry_count:
- return []
-
- await asyncio.sleep(random.randint(4, 10))
- continue
-
- if 'st=3' in page.url:
- dates = []
- for el in await page.query_selector_all('.ui-datepicker-calendar td:not(.ui-datepicker-unselectable)'):
- dates.append(
- datetime.date(
- year=int(await el.get_attribute('data-year')),
- month=int(await el.get_attribute('data-month')),
- day=int(await el.inner_text()),
- )
- )
-
- alert("\n".join(['empty slots found', *[it.isoformat() for it in dates]]))
-
- now = int(datetime.datetime.now().timestamp())
- html_path = dump_dir / f'immigration_{now}.html'
- html_path.write_text(await page.content())
- await page.screenshot(path=dump_dir / f'immigration_{now}.png')
-
- input()
-
- input()
-
-
-def parse_args():
- arger = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
- arger.add_argument(
- '--headless', '--silent', '--quiet', action='store_true', dest='headless', help='Run in headless mode'
- )
- arger.add_argument('--retry-count', '--retry', type=int, default=1, help='Run in headless mode')
-
- args, _ = arger.parse_known_args()
- return args
-
-
-async def main():
- args = parse_args()
- async with launch_browser(headless=args.headless) as page:
- try:
- values = ResidencyApplication(
- citizenship_country='India',
- num_applicants=1,
- is_living_with_family=False,
- residency_category='Educational purposes',
- residency_purpose='Residence permit for the purpose of studying (sect. 16b)',
- )
- slots = await check_slots(page, values=values, retry_count=args.retry_count)
- if not slots:
- print('no slots yet')
- return
- except Exception as e:
- logging.exception('got an error')
- input()
-
-
-if __name__ == '__main__':
- asyncio.run(main())
- logging.basicConfig(level=logging.INFO)
diff --git a/csv_to_html.py b/csv_to_html.py
deleted file mode 100644
index 843e20a..0000000
--- a/csv_to_html.py
+++ /dev/null
@@ -1,167 +0,0 @@
-import csv
-import html
-import sys
-
-# language=css
-page_css = """
-body {
- margin: 0;
- padding: 0.5rem;
- font-family: menlo, consolas, monospace;
- font-size: 14px;
-}
-input {
- font-size: inherit;
- font-family: inherit;
- background-color: transparent;
- padding: 0.25em 0.5em;
-}
-table {
- border-collapse: collapse;
- width: 100%;
-}
-th, td {
- border: 1px solid black;
- padding: 8px;
- text-align: left;
-}
-th {
- background-color: #f2f2f2;
-}
-tr:nth-child(even) {
- background-color: #fdfdfd;
-}
-"""
-
-# language=js
-page_script = """
-function sortableTable(table) {
- const sortStates = new Map();
-
- table.querySelectorAll('th').forEach((header, colIndex) => {
- header.style.cursor = 'pointer';
- header.onclick = () => {
- const isAsc = !sortStates.get(header);
- const tbody = table.querySelector('tbody');
- const rows = [...tbody.rows];
-
- rows.sort((a, b) => {
- const [valA, valB] = [a, b].map(row =>
- row.cells[colIndex].textContent.trim());
-
- return /^\d+$/.test(valA)
- ? (valA - valB) * (isAsc ? 1 : -1)
- : valA.localeCompare(valB) * (isAsc ? 1 : -1);
- });
-
- sortStates.set(header, isAsc);
- tbody.replaceChildren(...rows); // Better than innerHTML = ''
- };
- });
-}
-
-function debounce(func, delay) {
- let timeoutId;
-
- return function(...args) {
- clearTimeout(timeoutId);
- timeoutId = setTimeout(() => {
- func.apply(this, args);
- }, delay);
- };
-}
-
-function addFilterInput(tableElement) {
- const filterInput = document.createElement('input');
- filterInput.type = 'text';
- filterInput.placeholder = "Search...";
-
- tableElement.parentNode.insertBefore(filterInput, tableElement);
-
- const debouncedFilter = debounce(filter, 100);
-
- filterInput.addEventListener('input', () => {
- debouncedFilter();
- });
-
- function filter() {
- const filterValue = filterInput.value.trim();
-
- const rows = tableElement.querySelectorAll('tbody tr');
-
- rows.forEach(row => {
- const rowData = row.textContent.toLowerCase();
-
- if (filterValue.startsWith('/') && filterValue.endsWith('/')) {
- const regex = new RegExp(filterValue.slice(1, -1), 'ig');
- row.style.display = regex.test(rowData) ? '' : 'none';
- } else {
- row.style.display = rowData.includes(filterValue.toLowerCase()) ? '' : 'none';
- }
- });
- }
-}
-
-sortableTable(document.querySelector('table'));
-addFilterInput(document.querySelector('table'));
-"""
-
-
-def csv_to_html(csv_input: str) -> str:
- try:
- lines = csv_input.splitlines()
- delimiter = "\t" if "\t" in lines[0] else ","
- reader = csv.reader(lines, delimiter=delimiter)
- header = next(reader) # Get the header row
-
- header_html = f'{"".join(f"
{html.escape(col)} | " for col in header)}'
- rows = [
- f"{''.join(f'| {html.escape(cell)} | ' for cell in row)}
"
- for row in reader
- ]
- rows_html = "\n".join(rows)
- except csv.Error as e:
- raise ValueError(f"Error parsing CSV: {e}")
- except StopIteration:
- return "No data found in CSV input.
"
-
- rendered = f"""
-
-
-
- Query Results
-
-
-
-
-
- {header_html}
-
-
- {rows_html}
-
-
-
-
-
- """
-
- return rendered
-
-
-def main():
- try:
- csv_data = sys.stdin.read()
- html_output = csv_to_html(csv_data)
- if html_output:
- print(html_output)
- except UnicodeDecodeError as e:
- raise ValueError(
- f"Error decoding input: {e}. Ensure your CSV data is encoded correctly (e.g., UTF-8)."
- )
- except Exception as e:
- raise Exception(f"An unexpected error occurred: {e}")
-
-
-if __name__ == "__main__":
- main()
diff --git a/cull_edits.py b/cull_edits.py
deleted file mode 100755
index 3b6569e..0000000
--- a/cull_edits.py
+++ /dev/null
@@ -1,939 +0,0 @@
-#!/usr/bin/env -S uv run
-# /// script
-# requires-python = ">=3.14"
-# dependencies = [
-# "bottle>=0.13.0",
-# ]
-# ///
-
-import argparse
-import functools
-import os
-import re
-import socket
-import sys
-import threading
-import time
-import webbrowser
-from pathlib import Path
-from datetime import datetime
-from typing import TypedDict
-import bottle
-
-IMAGE_EXTENSIONS: set[str] = {".jpg", ".jpeg", ".png", ".webp", ".webm"}
-
-
-class ImageGroup(TypedDict):
- """Type definition for grouped images."""
-
- base: str
- original: Path
- edits: list[Path]
-
-
-def find_available_port() -> int:
- """Find an available random port."""
- with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
- s.bind(("127.0.0.1", 0))
- s.listen(1)
- port = s.getsockname()[1]
- return port
-
-
-def is_image(path: str | Path) -> bool:
- """Check if file is a supported image."""
- return Path(path).suffix.lower() in IMAGE_EXTENSIONS
-
-
-def discover_images(paths: list[str]) -> list[Path]:
- """Discover all images from mixed list of files and folders."""
- images = []
- for path_str in paths:
- path = Path(path_str)
- if path.is_file() and is_image(path):
- images.append(path.resolve())
- elif path.is_dir():
- for item in path.iterdir():
- if item.is_file() and is_image(item):
- images.append(item.resolve())
- return sorted(set(images))
-
-
-def group_images(images: list[Path]) -> list[ImageGroup]:
- """
- Group images by original + edits.
- Original: image.jpg
- Edits: image-edit1.jpg, image-edit2.jpg, etc.
- Returns list of dicts: {original: Path, edits: [Path, ...]}
- """
- groups = {}
-
- for img in images:
- stem = img.stem
-
- # Check if this is an edit
- match = re.match(r"^(.+?)-edit", stem)
- if match:
- base = match.group(1)
- if base not in groups:
- groups[base] = {"original": None, "edits": []}
- groups[base]["edits"].append(img)
- else:
- # This is an original
- if stem not in groups:
- groups[stem] = {"original": None, "edits": []}
- groups[stem]["original"] = img
-
- # Filter out groups without originals, convert to list
- result = []
- for base, group in groups.items():
- if group["original"]:
- result.append(ImageGroup(base=base, original=group["original"], edits=sorted(group["edits"])))
-
- return sorted(result, key=lambda g: g["original"].name)
-
-
-def safe_image_path(filepath: str) -> Path:
- """Verify image path is safe (prevent directory traversal)."""
- # Search for the image in all_images by filename
- for img in all_images:
- if img.name == filepath:
- return img
- raise ValueError(f"Unauthorized path: {filepath}")
-
-
-def hardlink_images(src_paths: list[Path | str], dest_dir: str | Path) -> None:
- """Hard link multiple images to destination directory."""
- dest_path = Path(dest_dir)
- make_dirs(dest_path)
-
- for src in src_paths:
- src = Path(src)
- dest = dest_path / src.name
- try:
- # Remove existing file if present
- if dest.exists():
- dest.unlink()
- os.link(src, dest)
- except Exception as e:
- print(f"Error hard linking {src} to {dest}: {e}", file=sys.stderr)
- raise
-
-
-# Global state
-app = bottle.Bottle()
-all_images = set()
-image_groups = []
-current_group_idx = 0
-last_heartbeat = datetime.now()
-should_exit = False
-picks_dir = Path()
-flags = []
-flag_dirs = {}
-
-HTML_TEMPLATE = """
-
-
-
-
-
- Image Culler
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- No images
-
-
-
-
-
-
- Position:
-
- /
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
![]()
-
-
-
-
-
- No variations
-
-
-
-
-
-
-
-
-
-
-
-"""
-
-
-@app.get("/")
-def index() -> str:
- return HTML_TEMPLATE
-
-
-@app.get("/api/images")
-def api_images():
- """Return grouped images and current flag assignments."""
- global image_groups, flags, flag_dirs
- result = []
- for group in image_groups:
- result.append(
- {
- "base": group["base"],
- "original": {"name": group["original"].name, "path": str(group["original"])},
- "edits": [{"name": e.name, "path": str(e)} for e in group["edits"]],
- }
- )
-
- # Build flag assignments by checking which flag directories contain which original images
- flag_assignments = {}
- for flag, flag_dir in flag_dirs.items():
- if flag_dir.exists():
- for item in flag_dir.iterdir():
- if item.is_file():
- # Map the file name to its flag
- flag_assignments[item.name] = flag
-
- return {"groups": result, "flags_list": flags, "flag_assignments": flag_assignments}
-
-
-@app.get("/image/")
-def serve_image(filename: str):
- """Serve an image file safely."""
- try:
- target = safe_image_path(filename)
- return bottle.static_file(target.name, root=str(target.parent), mimetype="image/jpeg")
- except ValueError:
- bottle.response.status = 403
- return {"error": "Unauthorized path"}
-
-
-@app.post("/api/pick")
-def api_pick():
- """Hard link picked images to _picks directory."""
- global image_groups, current_group_idx, picks_dir
-
- # Create directory before writing
- make_dirs(picks_dir)
-
- data = bottle.request.json
- original_name = data.get("original")
- edits_names = data.get("edits", [])
- include_original = data.get("includeOriginal", False)
-
- files_to_pick = []
-
- # Add original if requested
- if include_original:
- for img in all_images:
- if img.name == original_name:
- files_to_pick.append(img)
- break
-
- # Add edits by finding them in all_images
- for edit_name in edits_names:
- for img in all_images:
- if img.name == edit_name:
- files_to_pick.append(img)
- break
-
- try:
- hardlink_images(files_to_pick, picks_dir)
- return {"status": "ok"}
- except Exception as e:
- bottle.response.status = 500
- return {"error": str(e)}
-
-
-@app.post("/api/unpick")
-def api_unpick():
- """Remove picked images from _picks directory."""
- global picks_dir
-
- data = bottle.request.json
- original_name = data.get("original")
-
- try:
- # Find and remove the original and any edits from picks directory
- if picks_dir.exists():
- for item in picks_dir.iterdir():
- if item.is_file() and (item.name == original_name or item.stem.startswith(original_name.rsplit(".", 1)[0] + "-edit")):
- item.unlink()
- return {"status": "ok"}
- except Exception as e:
- bottle.response.status = 500
- return {"error": str(e)}
-
-
-@app.post("/api/assign-flag")
-def api_assign_flag():
- """Assign an image to a flag."""
- global flag_dirs, all_images
-
- data = bottle.request.json
- original_name = data.get("original")
- flag_name = data.get("flag")
-
- # Find original file
- original_path = None
- for img in all_images:
- if img.name == original_name:
- original_path = img
- break
-
- if not original_path or flag_name not in flag_dirs:
- bottle.response.status = 400
- return {"error": "Invalid image or flag"}
-
- # Create directory before writing
- make_dirs(flag_dirs[flag_name])
-
- try:
- hardlink_images([original_path], flag_dirs[flag_name])
- return {"status": "ok"}
- except Exception as e:
- bottle.response.status = 500
- return {"error": str(e)}
-
-
-@app.post("/api/unassign-flag")
-def api_unassign_flag():
- """Remove an image from a flag."""
- global flag_dirs
-
- data = bottle.request.json
- original_name = data.get("original")
- flag_name = data.get("flag")
-
- if flag_name not in flag_dirs:
- bottle.response.status = 400
- return {"error": "Invalid flag"}
-
- try:
- target_file = flag_dirs[flag_name] / original_name
- if target_file.exists():
- target_file.unlink()
- return {"status": "ok"}
- except Exception as e:
- bottle.response.status = 500
- return {"error": str(e)}
-
-
-@app.post("/api/heartbeat")
-def api_heartbeat():
- """Update last heartbeat timestamp."""
- global last_heartbeat
- last_heartbeat = datetime.now()
- return {"status": "ok"}
-
-
-@app.post("/api/shutdown")
-def api_shutdown():
- """Shutdown the server."""
- global should_exit
- should_exit = True
- return {"status": "ok"}
-
-
-@functools.cache
-def make_dirs(path: Path) -> None:
- """Create directory if it doesn't exist."""
- path.mkdir(parents=True, exist_ok=True)
-
-
-def heartbeat_monitor() -> None:
- """Monitor for shutdown signal."""
- global should_exit
-
- while not should_exit:
- time.sleep(0.5)
-
- print("\nShutting down server.", file=sys.stderr)
- os._exit(0)
-
-
-def main() -> None:
- global all_images, image_groups, last_heartbeat, picks_dir, flags, flag_dirs
-
- parser = argparse.ArgumentParser(description="Image culling app for AI image2image transforms")
- parser.add_argument("paths", nargs="+", help="Image files or folders to cull")
- parser.add_argument("--picks-dir", default=None, help="Directory for picked images (default: $cwd/_picks)")
- parser.add_argument("--rejects-dir", default=None, help="Directory for rejected images (default: $cwd/_rejects)")
- parser.add_argument("--flags", default=None, help="Comma-separated flag names (e.g., 'a,b,c,d')")
- parser.add_argument("--flags-dir", default=None, help="Directory for flag subdirectories (default: $cwd)")
-
- args = parser.parse_args()
-
- # Set directories
- cwd = Path.cwd()
- picks_dir = Path(args.picks_dir or (cwd / "_picks"))
- rejects_dir = Path(args.rejects_dir or (cwd / "_rejects"))
- flags_dir = Path(args.flags_dir or cwd)
-
- # Set up flags
- if args.flags:
- flags = [f.strip() for f in args.flags.split(",")]
- for flag in flags:
- flag_dir = flags_dir / f"_picks_flag_{flag}"
- flag_dirs[flag] = flag_dir
-
- # Always add reject as a flag at the end
- flags.append("reject")
- flag_dirs["reject"] = rejects_dir
-
- # Discover images
- images = discover_images(args.paths)
- all_images = set(images)
-
- if not images:
- print("No images found in provided paths.", file=sys.stderr)
- sys.exit(1)
-
- # Group images
- image_groups = group_images(images)
-
- if not image_groups:
- print("No original images found (looking for files matching naming convention).", file=sys.stderr)
- sys.exit(1)
-
- print(f"Found {len(image_groups)} original image(s) with variations", file=sys.stderr)
-
- # Find available port
- port = find_available_port()
- url = f"http://127.0.0.1:{port}"
-
- # Start heartbeat monitor
- monitor_thread = threading.Thread(target=heartbeat_monitor, daemon=True)
- monitor_thread.start()
-
- # Open browser
- print(f"Opening browser at {url}", file=sys.stderr)
- webbrowser.open(url)
-
- # Start Flask server
- last_heartbeat = datetime.now()
- bottle.run(app, host="127.0.0.1", port=port, quiet=True)
-
-
-if __name__ == "__main__":
- main()
diff --git a/download_js_sourcemaps.py b/download_js_sourcemaps.py
deleted file mode 100755
index c7aec85..0000000
--- a/download_js_sourcemaps.py
+++ /dev/null
@@ -1,194 +0,0 @@
-#!/usr/bin/env -S uv run --script
-# /// script
-# dependencies = ["playwright", "httpx"]
-# ///
-
-import argparse
-import json
-import logging
-import re
-import sys
-from pathlib import Path
-from typing import Optional
-from urllib.parse import urljoin, urlparse
-
-import httpx
-from playwright.sync_api import sync_playwright
-
-logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
-logger = logging.getLogger(__name__)
-
-
-def str_to_bool(value: str) -> bool:
- """Convert string to boolean."""
- if isinstance(value, bool):
- return value
- return value.lower() in ("true", "1", "yes", "on")
-
-
-def parse_args() -> argparse.Namespace:
- parser = argparse.ArgumentParser(
- description="Download JavaScript files and reconstruct sources from source maps",
- formatter_class=argparse.ArgumentDefaultsHelpFormatter,
- )
- parser.add_argument("url", help="URL of the page to scrape")
- parser.add_argument(
- "--output-dir",
- "-o",
- type=Path,
- default=Path("output"),
- help="Output directory for reconstructed sources",
- )
- parser.add_argument(
- "--headless",
- type=str_to_bool,
- default=True,
- help="Run browser in headless mode",
- )
- return parser.parse_args()
-
-
-def get_script_urls(page_url: str, headless: bool = True) -> list[str]:
- """Extract all script URLs from a page using Playwright."""
- script_urls = []
-
- with sync_playwright() as playwright:
- browser = playwright.chromium.launch(headless=headless)
- page = browser.new_page()
-
- def handle_response(response):
- if response.request.resource_type == "script":
- script_urls.append(response.url)
- logger.debug(f"Detected script: {response.url}")
-
- page.on("response", handle_response)
- logger.info("Navigating to page...")
- page.goto(url=page_url)
- logger.info("Waiting for DOM ready...")
- page.wait_for_load_state(state="domcontentloaded")
- logger.info("Waiting for network idle (30s timeout)...")
- try:
- page.wait_for_load_state(state="networkidle", timeout=30000)
- except Exception as e:
- logger.warning(f"Network idle timeout: {e}")
-
- browser.close()
-
- return script_urls
-
-
-def download_file(url: str, timeout: int = 30) -> bytes:
- """Download a file from a URL."""
- try:
- with httpx.Client(timeout=timeout, follow_redirects=True) as client:
- response = client.get(url=url)
- response.raise_for_status()
- return response.content
- except httpx.HTTPError as e:
- logger.error(f"Failed to download {url}: {e}")
- raise
-
-
-def get_sourcemap_url(script_content: str, script_url: str) -> Optional[str]:
- """Extract source map URL from script content."""
- match = re.search(r"//# sourceMappingURL=(.+?)(?:\n|$)", script_content)
- if not match:
- return None
-
- sourcemap_ref = match.group(1).strip()
-
- if sourcemap_ref.startswith("http"):
- return sourcemap_ref
-
- return urljoin(base=script_url, url=sourcemap_ref)
-
-
-def decode_sourcemap(sourcemap_data: dict) -> dict:
- """Extract sources and content from source map."""
- sources = sourcemap_data.get("sources", [])
- sources_content = sourcemap_data.get("sourcesContent", [])
-
- return {
- "sources": sources,
- "sources_content": sources_content,
- }
-
-
-def reconstruct_sources(
- script_content: str,
- sourcemap_data: dict,
- output_dir: Path,
- script_name: str,
-) -> None:
- """Reconstruct original sources from source map."""
- sources = sourcemap_data.get("sources", [])
- sources_content = sourcemap_data.get("sourcesContent", [])
-
- logger.info(f"Source map has {len(sources)} source files")
- if sources:
- logger.debug(f"First few sources: {sources[:3]}")
-
- if not sources_content or all(c is None for c in sources_content):
- logger.error(f"No source content found in source map for {script_name}")
- sys.exit(1)
-
- for i, source_file in enumerate(sources):
- if i < len(sources_content) and sources_content[i]:
- source_path = output_dir / source_file
- source_path.parent.mkdir(parents=True, exist_ok=True)
-
- with open(source_path, mode="w", encoding="utf-8") as f:
- f.write(sources_content[i])
-
- logger.info(f"Reconstructed {source_file}")
-
-
-def main() -> None:
- args = parse_args()
- args.output_dir.mkdir(parents=True, exist_ok=True)
-
- logger.info(f"Fetching page: {args.url}")
- script_urls = get_script_urls(page_url=args.url, headless=args.headless)
- logger.info(f"Found {len(script_urls)} script(s)")
-
- if not script_urls:
- logger.error("No scripts found on page")
- sys.exit(1)
-
- for script_url in script_urls:
- logger.info(f"Processing {script_url}")
-
- try:
- script_content = download_file(url=script_url).decode(encoding="utf-8")
- except Exception as e:
- logger.error(f"Failed to get script {script_url}: {e}")
- continue
-
- sourcemap_url = get_sourcemap_url(script_content=script_content, script_url=script_url)
-
- if not sourcemap_url:
- logger.error(f"No source map found for {script_url}")
- sys.exit(1)
-
- logger.info(f"Downloading source map: {sourcemap_url}")
-
- try:
- sourcemap_content = download_file(url=sourcemap_url).decode(encoding="utf-8")
- sourcemap_data = json.loads(sourcemap_content)
- except Exception as e:
- logger.error(f"Failed to download/parse source map: {e}")
- sys.exit(1)
-
- script_name = urlparse(script_url).path.split("/")[-1]
- reconstruct_sources(
- script_content=script_content,
- sourcemap_data=sourcemap_data,
- output_dir=args.output_dir,
- script_name=script_name,
- )
-
- logger.info(f"All sources reconstructed to {args.output_dir}")
-
-
-if __name__ == "__main__":
- main()
diff --git a/duration_parser.py b/duration_parser.py
deleted file mode 100644
index 5a8a08a..0000000
--- a/duration_parser.py
+++ /dev/null
@@ -1,40 +0,0 @@
-import sys
-import re
-import math
-import datetime
-
-
-def parse_duration(text: str) -> datetime.timedelta:
- re_duration = re.compile(r'(\d+(?:\.\d+)?(?:sec|min|hour|hr|h|m|s))')
- re_digits = re.compile(r'([\d.]+)')
- time_scales = {
- 's': 1,
- 'sec': 1,
- 'm': 60,
- 'min': 60,
- 'h': 3600,
- 'hr': 3600,
- }
-
- parts = re_duration.findall(text)
- if parts:
- total_secs = 0
- for it in parts:
- if m := re_digits.search(it):
- _, end = m.span()
- scale = it[end:]
- total_secs += math.ceil(float(m.group(1)) * time_scales[scale])
- return datetime.timedelta(seconds=total_secs)
-
- parts = [float(it) for it in text.split(':')]
- if len(parts) == 1:
- return datetime.timedelta(seconds=parts[0])
- if len(parts) == 2:
- return datetime.timedelta(minutes=parts[0], seconds=parts[1])
- if len(parts) == 3:
- return datetime.timedelta(hours=parts[0], minutes=parts[1], seconds=parts[2])
-
-
-if __name__ == '__main__':
- parsed = parse_duration(sys.argv[1])
- seconds = int(parsed.total_seconds())
diff --git a/extract_archives.py b/extract_archives.py
deleted file mode 100755
index 29717e7..0000000
--- a/extract_archives.py
+++ /dev/null
@@ -1,230 +0,0 @@
-#!/usr/bin/env -S uv run
-# /// script
-# requires-python = ">=3.13"
-# ///
-import argparse
-import logging
-import re
-import subprocess
-import sys
-from pathlib import Path
-
-logger = logging.getLogger(__name__)
-
-KNOWN_PASSWORDS = """
-torgo
-Yuzuki
-oron.com
-MonikGonPs
-koth
-drdoom-psuzy
-psuzy
-suzy
-idle
-dan4260
-""".strip().splitlines(keepends=False)
-
-
-class CorruptArchiveError(Exception):
- pass
-
-
-def find_archive_files(root_dir: Path) -> list[Path]:
- extractables = []
- globs = ["*.rar", "*.zip*", "*.7z*"]
- for glob in globs:
- for f in root_dir.rglob(glob):
- extractables.append(f)
-
- extractables = [f for f in extractables if is_extractable(f)]
-
- return sorted(extractables, key=str)
-
-
-def is_extractable(archive: Path) -> bool:
- if ".corrupt" in archive.name:
- return False
- if re.search(r"\.zip|\.7z", archive.name):
- zip_part = re.search(r"(\d+)$", archive.suffix)
- if zip_part and int(zip_part.group(1)) == 1:
- return True
- if not zip_part:
- return True
-
- if archive.name.endswith(".rar"):
- rar_part = re.search(r"\.part(\d+)", archive.name)
- if not rar_part:
- return True
- if rar_part and int(rar_part.group(1)) == 1:
- if archive.with_name(archive.name.replace("part1", "part2")).is_file():
- return True
-
- return False
-
-
-def extract_archive_rar(archive: Path, output_dir: Path, password: str | None = None) -> None:
- args = [
- "rar",
- "e",
- *([f"-p{password}"] if password else []),
- f"{archive.name}",
- f"{output_dir}",
- ]
- try:
- subprocess.run(args, cwd=archive.parent, capture_output=True, text=True, check=True)
- except subprocess.CalledProcessError as e:
- output: str = e.stderr or e.stdout
- if "corrupt file or wrong password" in output.lower():
- raise LookupError("corrupt or invalid password")
- elif e.returncode == 3:
- raise FileNotFoundError("missing parts")
- elif e.returncode == 10:
- raise ValueError("not a rar file")
- elif e.returncode == 11:
- raise LookupError("invalid password")
-
- raise
-
-
-def extract_archive_7z(archive: Path, output_dir: Path, password: str | None = None):
- args = [
- "7z",
- "e",
- *(["-p", password] if password else []),
- f"-o{output_dir}",
- "-y",
- f"{archive}",
- ]
-
- try:
- subprocess.run(args, capture_output=True, text=True, check=True)
- except subprocess.CalledProcessError as e:
- output: str = e.stderr or e.stdout
- if "missing volume" in output.lower():
- raise CorruptArchiveError("missing parts")
- if "wrong password" in output.lower():
- raise LookupError("invalid password")
- raise
-
-
-def extract_archive_zip(archive: Path, output_dir: Path, password: str | None = None):
- args = [
- "unzip",
- "-o", # overwrite files without prompting
- *(["-P", password] if password else []),
- f"{archive}",
- "-d",
- f"{output_dir}",
- ]
-
- try:
- subprocess.run(args, capture_output=True, text=True, check=True)
- except subprocess.CalledProcessError as e:
- output: str = e.stderr or e.stdout
- if "zipfile corrupt" in output.lower():
- raise CorruptArchiveError("corrupt zip file")
- if "cannot find zipfile" in output.lower():
- raise FileNotFoundError("zip file not found")
- if "incorrect password" in output.lower() or "password required" in output.lower():
- raise LookupError("invalid password")
- raise
-
-
-def extract_archive(archive_path: Path, output_dir: Path, passwords: list[str]) -> None:
- for p in ["", *passwords]:
- try:
- fn: callable
- if ".rar" in archive_path.name:
- fn = extract_archive_rar
- elif ".zip" == archive_path.suffix:
- fn = extract_archive_zip
- else:
- fn = extract_archive_7z
- fn(archive_path, output_dir, p)
- except subprocess.CalledProcessError as e:
- logger.error(f"error extracting {archive_path}: {e.stderr}")
- raise
- except LookupError as e:
- logger.info(f"invalid password: {p!r}. {e}")
- continue
- except Exception as e:
- logger.error(f"failed to extract: {e}")
- raise
-
-
-def delete_archive(archive: Path):
- logger.info(f"deleting: {archive.name}")
- archive.unlink()
- stem = archive.stem.rsplit(".", maxsplit=1)[0]
- for f in archive.parent.rglob(f"*{archive.suffix}"):
- if archive.stem.startswith(stem):
- logger.info(f"deleting: {f.name}")
- f.unlink()
-
-
-def unlink_if_empty(dir_path: Path) -> None:
- """Unlink the directory if it is empty."""
- try:
- if not any(dir_path.iterdir()):
- dir_path.rmdir()
- logger.info(f"deleted empty directory: {dir_path}")
- except OSError as e:
- logger.error(f"error deleting directory: {e}")
- raise
-
-
-def parse_args():
- arger = argparse.ArgumentParser()
- arger.add_argument("archive_paths", type=lambda v: Path(v.strip()), nargs="+", help="List of archive paths")
- arger.add_argument("--cwd", type=Path, default=Path.cwd(), help="working directory")
- arger.add_argument("--find", default=False, help="find archives and print to stdout", action="store_true")
- arger.add_argument("--clean", default=False, help="delete file(s) after successful extraction", action="store_true")
-
- if len(sys.argv) < 2:
- arger.print_help()
- exit(1)
-
- return arger.parse_args()
-
-
-def main():
- logging.basicConfig(level=logging.INFO)
-
- args = parse_args()
- cwd: Path = args.cwd
-
- if args.find:
- archives = find_archive_files(cwd)
- for f in archives:
- print(str(f.resolve()))
- return
-
- archives = args.archive_paths
-
- failed = []
- for i, f in enumerate(archives, start=1):
- f: Path
- progress = f"{i:02}/{len(archives):02}"
- logger.info(f"extracting {progress}: {f.relative_to(cwd)}")
- output_dir = f.parent / f.stem.strip()
- try:
- output_dir.mkdir(exist_ok=True, parents=True)
- extract_archive(f, output_dir=output_dir, passwords=KNOWN_PASSWORDS)
- if args.clean:
- delete_archive(f)
- except CorruptArchiveError as e:
- if ".corrupt" not in f.name:
- f.rename(f.with_stem(f"{f.stem}.corrupt"))
- except (ValueError, FileNotFoundError) as e:
- logger.error(str(e))
- failed.append(f)
- finally:
- unlink_if_empty(output_dir)
- if failed:
- logger.info(f"failed to extract {len(failed)} archives")
- for f in failed:
- logger.info(f"{f.relative_to(cwd)}")
-
-
-if __name__ == "__main__":
- main()
diff --git a/ffmpeg_strip.py b/ffmpeg_strip.py
deleted file mode 100755
index 2801e9d..0000000
--- a/ffmpeg_strip.py
+++ /dev/null
@@ -1,81 +0,0 @@
-#!/usr/bin/env python3.9
-import argparse
-import json
-import logging
-import subprocess
-import sys
-from pathlib import Path
-import typing
-
-
-def parse_args(argv: list[str]):
- arger = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
- arger.add_argument("video_path", type=Path, help="Video path")
- arger.add_argument("save_path", nargs="?", type=Path, help="Output path")
- if len(argv) == 0:
- arger.print_help()
- exit(1)
-
- return arger.parse_args(argv)
-
-
-def ffprobe(video_path: Path) -> dict:
- proc = subprocess.run(
- [
- "ffprobe",
- "-v",
- "quiet",
- "-print_format",
- "json",
- "-show_format",
- "-show_streams",
- str(video_path),
- ],
- check=True,
- capture_output=True,
- )
- return json.loads(proc.stdout)
-
-
-def clean_video(video_path: Path, save_path: Path) -> Path:
- media_info = ffprobe(video_path)
- # check if the first video stream has hevc codec
- if media_info["streams"][0]["codec_name"] == "hevc":
- logging.debug(f"Video codec is hevc, adding hvc1 tag")
- extra_args = ["-tag:v", "hvc1"]
- else:
- extra_args = []
-
- # fmt: off
- args = [
- 'ffmpeg',
- '-i',
- str(video_path),
- '-c:v', 'copy',
- # '-movflags', '+faststart',
- '-map_metadata', '-1',
- *extra_args,
- '-c:a', 'copy',
- '-y',
- str(save_path),
- ]
- # fmt: on
- logging.info(f"calling ffmpeg with {args=}")
- subprocess.run(args, check=True)
- return save_path
-
-
-def main():
- logging.basicConfig(level=logging.INFO)
- args = parse_args(sys.argv[1:])
- save_path: typing.Optional[Path] = args.save_path
- if save_path and save_path.resolve().is_dir():
- save_path = save_path / args.video_path.name
- if save_path is None:
- save_path = args.video_path.with_suffix(".clean.mp4")
-
- clean_video(video_path=args.video_path, save_path=save_path)
-
-
-if __name__ == "__main__":
- main()
diff --git a/flux2_private_server.py b/flux2_private_server.py
deleted file mode 100644
index c93aa10..0000000
--- a/flux2_private_server.py
+++ /dev/null
@@ -1,461 +0,0 @@
-#!/usr/bin/env -S uv run --script
-# /// script
-# requires-python = ">=3.10,<3.11"
-# dependencies = [
-# "torch==2.9.1+cu128",
-# "torchvision==0.24.1+cu128",
-# "torchaudio==2.9.1+cu128",
-# "diffusers @ git+https://github.com/huggingface/diffusers.git",
-# "transformers>=4.57.0",
-# "accelerate>=1.0.0",
-# "safetensors>=0.4.0",
-# "huggingface_hub>=0.25.0",
-# "numpy>=1.26.0",
-# "pillow>=10.0.0",
-# "fastapi>=0.115.0",
-# "uvicorn>=0.29.0",
-# ]
-# [tool.uv]
-# extra-index-url = ["https://download.pytorch.org/whl/cu128"]
-# ///
-
-import argparse
-import base64
-import binascii
-import io
-import os
-import random
-import threading
-import time
-from typing import Any
-
-os.environ.setdefault("HF_XET_HIGH_PERFORMANCE", "1")
-
-import torch
-from diffusers import Flux2KleinPipeline
-from fastapi import FastAPI, HTTPException
-from fastapi.responses import HTMLResponse, Response
-from pydantic import BaseModel, Field
-from PIL import Image
-
-
-def _resolve_model_size() -> str:
- parser = argparse.ArgumentParser(add_help=False)
- parser.add_argument("--model", choices=["4b", "9b"], default="9b")
- args, _ = parser.parse_known_args()
- return args.model
-
-
-def _checkpoint_for(model_size: str) -> str:
- if model_size == "4b":
- return "black-forest-labs/FLUX.2-klein-4B"
- return "black-forest-labs/FLUX.2-klein-9B"
-
-
-MODEL_SIZE = _resolve_model_size()
-CHECKPOINT = _checkpoint_for(MODEL_SIZE)
-USE_COMPILE = os.environ.get("FLUX2_COMPILE", "1") != "0"
-WARMUP_RUNS = int(os.environ.get("FLUX2_WARMUP_RUNS", "2"))
-WARMUP_STEPS = int(os.environ.get("FLUX2_WARMUP_STEPS", "4"))
-WARMUP_PROMPT = "A high-resolution photo of a red fox in natural daylight"
-MAX_SEED = (2**31) - 1
-MAX_PIXELS = 4 * 1024 * 1024
-
-DEFAULT_STEPS = 4
-DEFAULT_GUIDANCE = 1.0
-
-app = FastAPI(title="FLUX2 Klein Private Server")
-PIPE: Flux2KleinPipeline | None = None
-PIPE_LOCK = threading.Lock()
-
-
-class GenerateRequest(BaseModel):
- prompt: str = Field(..., min_length=1)
- width: int = 1024
- height: int = 1024
- num_inference_steps: int = DEFAULT_STEPS
- guidance_scale: float = DEFAULT_GUIDANCE
- seed: int = 0
- randomize_seed: bool = False
- image_data_url: str | None = None
-
-
-class CompatRequest(BaseModel):
- model: str | None = None
- prompt: str = Field(..., min_length=1)
- imageDataUrl: str | None = None
- size: str | None = None
- width: int | None = None
- height: int | None = None
- num_inference_steps: int | None = None
- guidance_scale: float | None = None
- seed: int | None = None
- randomize_seed: bool = False
- response_format: str = "b64_json"
- n: int = 1
- negative_prompt: str | None = None
- negativePrompt: str | None = None
-
-
-def _ensure_cuda() -> None:
- if not torch.cuda.is_available():
- raise RuntimeError("CUDA is required for this server")
-
-
-def _compile_pipe(pipe: Flux2KleinPipeline) -> bool:
- if not USE_COMPILE:
- return False
- print("Compiling transformer and VAE decoder...")
- start = time.perf_counter()
- pipe.transformer = torch.compile(
- pipe.transformer,
- mode="max-autotune",
- dynamic=False,
- fullgraph=True,
- )
- pipe.vae.decode = torch.compile(
- pipe.vae.decode,
- mode="max-autotune",
- dynamic=False,
- fullgraph=True,
- )
- print(f"Compile finished in {time.perf_counter() - start:.2f}s")
- return True
-
-
-def _warmup_pipe(pipe: Flux2KleinPipeline) -> None:
- if WARMUP_RUNS <= 0:
- return
- print(f"Warmup started ({WARMUP_RUNS} runs)...")
- start = time.perf_counter()
- for run_idx in range(WARMUP_RUNS):
- generator = torch.Generator(device="cuda").manual_seed(run_idx)
- pipe(
- prompt=WARMUP_PROMPT,
- height=1024,
- width=1024,
- num_inference_steps=WARMUP_STEPS,
- guidance_scale=1.0,
- generator=generator,
- ).images[0]
- torch.cuda.synchronize()
- print(f"Warmup finished in {time.perf_counter() - start:.2f}s")
-
-
-def _load_pipe(ckpt: str) -> Flux2KleinPipeline:
- print(f"Loading checkpoint: {ckpt}")
- pipe = Flux2KleinPipeline.from_pretrained(ckpt, torch_dtype=torch.bfloat16)
- pipe = pipe.to("cuda")
- pipe.transformer.fuse_qkv_projections()
- pipe.vae.fuse_qkv_projections()
- pipe.vae.to(memory_format=torch.channels_last)
- pipe.transformer.set_attention_backend("_native_flash")
- if _compile_pipe(pipe):
- _warmup_pipe(pipe)
- return pipe
-
-
-def _normalize_dims(width: int, height: int) -> tuple[int, int]:
- if width <= 0 or height <= 0:
- raise HTTPException(status_code=422, detail="width and height must be > 0")
- if width % 16 != 0 or height % 16 != 0:
- raise HTTPException(
- status_code=422, detail="width and height must be multiples of 16"
- )
- if width * height > MAX_PIXELS:
- raise HTTPException(status_code=422, detail="max resolution is 4 megapixels")
- return width, height
-
-
-def _parse_size(
- size: str | None, width: int | None, height: int | None
-) -> tuple[int, int]:
- if width and height:
- return _normalize_dims(width, height)
- if not size:
- return _normalize_dims(1024, 1024)
- parts = size.lower().split("x")
- if len(parts) != 2:
- raise HTTPException(status_code=422, detail="size must look like 1024x1024")
- try:
- parsed_w = int(parts[0])
- parsed_h = int(parts[1])
- except ValueError as exc:
- raise HTTPException(
- status_code=422, detail="size values must be integers"
- ) from exc
- return _normalize_dims(parsed_w, parsed_h)
-
-
-def _decode_image_data_url(image_data_url: str | None) -> Image.Image | None:
- if not image_data_url:
- return None
- if "," not in image_data_url:
- raise HTTPException(status_code=422, detail="imageDataUrl must be a data URL")
- _, payload = image_data_url.split(",", 1)
- try:
- raw = base64.b64decode(payload)
- except (ValueError, binascii.Error) as exc:
- raise HTTPException(
- status_code=422, detail="invalid base64 in imageDataUrl"
- ) from exc
- with Image.open(io.BytesIO(raw)) as image:
- return image.convert("RGB")
-
-
-def _generate_png_bytes(
- prompt: str,
- width: int,
- height: int,
- num_inference_steps: int,
- guidance_scale: float,
- seed: int,
- image: Image.Image | None,
-) -> tuple[bytes, int]:
- if PIPE is None:
- raise HTTPException(status_code=503, detail="model not loaded yet")
-
- generator = torch.Generator(device="cuda").manual_seed(seed)
- kwargs: dict[str, Any] = {
- "prompt": prompt,
- "height": height,
- "width": width,
- "num_inference_steps": num_inference_steps,
- "guidance_scale": guidance_scale,
- "generator": generator,
- }
-
- with PIPE_LOCK:
- try:
- if image is not None:
- try:
- output = PIPE(image=image, **kwargs)
- except TypeError:
- output = PIPE(images=[image], **kwargs)
- else:
- output = PIPE(**kwargs)
- except Exception as exc:
- raise HTTPException(
- status_code=500, detail=f"generation failed: {exc}"
- ) from exc
-
- result_image = output.images[0]
- buf = io.BytesIO()
- result_image.save(buf, format="PNG")
- return buf.getvalue(), seed
-
-
-@app.on_event("startup")
-def _startup() -> None:
- global PIPE
- _ensure_cuda()
- PIPE = _load_pipe(CHECKPOINT)
- print("Server ready")
-
-
-@app.get("/", response_class=HTMLResponse)
-def index() -> str:
- return """
-
-
-
-
- FLUX.2 Klein Playground
-
-
-
-
-
-
FLUX.2 Klein Playground
-
Private single-user UI at /. Distilled defaults: 4 steps, guidance 1.0.
-
-
-
-
-
Output
-
![\"Generated]()
-
Tip: right-click image to save.
-
-
-
-
-
-
-"""
-
-
-@app.get("/health")
-def health() -> dict[str, str]:
- return {"status": "ok", "model": CHECKPOINT}
-
-
-@app.post("/generate")
-def generate(req: GenerateRequest) -> Response:
- width, height = _normalize_dims(req.width, req.height)
- seed = req.seed if not req.randomize_seed else random.randint(0, MAX_SEED)
- image = _decode_image_data_url(req.image_data_url)
- png_bytes, used_seed = _generate_png_bytes(
- prompt=req.prompt,
- width=width,
- height=height,
- num_inference_steps=req.num_inference_steps,
- guidance_scale=req.guidance_scale,
- seed=seed,
- image=image,
- )
- return Response(
- content=png_bytes,
- media_type="image/png",
- headers={"X-Seed": str(used_seed)},
- )
-
-
-@app.post("/v1/images/generations")
-def compat_generate(req: CompatRequest) -> dict[str, Any]:
- width, height = _parse_size(req.size, req.width, req.height)
- seed_value = req.seed if req.seed is not None else 0
- seed = seed_value if not req.randomize_seed else random.randint(0, MAX_SEED)
- image = _decode_image_data_url(req.imageDataUrl)
-
- png_bytes, used_seed = _generate_png_bytes(
- prompt=req.prompt,
- width=width,
- height=height,
- num_inference_steps=req.num_inference_steps or DEFAULT_STEPS,
- guidance_scale=req.guidance_scale or DEFAULT_GUIDANCE,
- seed=seed,
- image=image,
- )
- b64_png = base64.b64encode(png_bytes).decode("ascii")
- return {
- "created": int(time.time()),
- "model": CHECKPOINT,
- "seed": used_seed,
- "data": [{"b64_json": b64_png}],
- }
-
-
-if __name__ == "__main__":
- import uvicorn
-
- parser = argparse.ArgumentParser(description="Private FLUX.2 Klein FastAPI server")
- parser.add_argument("--model", choices=["4b", "9b"], default=MODEL_SIZE)
- parser.add_argument("--host", default="127.0.0.1")
- parser.add_argument("--port", type=int, default=6006)
- args = parser.parse_args()
-
- MODEL_SIZE = args.model
- CHECKPOINT = _checkpoint_for(MODEL_SIZE)
- uvicorn.run(app, host=args.host, port=args.port)
diff --git a/fpup.py b/fpup.py
deleted file mode 100755
index 7263e87..0000000
--- a/fpup.py
+++ /dev/null
@@ -1,68 +0,0 @@
-#!/usr/bin/env -S uv run
-# /// script
-# dependencies = ["httpx"]
-# ///
-import argparse
-import concurrent.futures
-import json
-import re
-import sys
-import typing
-from pathlib import Path
-from typing import Optional
-import logging
-
-
-import httpx
-
-
-def upload_image(image_path: Path) -> dict:
- session = httpx.Client(timeout=20)
- logging.info(f"uploading {image_path.name}")
-
- with image_path.open("rb") as f:
- res = session.post(
- "https://fapping.empornium.sx/upload.php",
- files={
- "ImageUp": f,
- },
- )
-
- res = session.get("https://fapping.empornium.sx/uploaded/", follow_redirects=True)
- if m := re.search(r"var ImagesUp = (\[.+\]);", res.text, re.MULTILINE):
- data: list[dict] = json.loads(m.group(1))
- logging.info(f"upload result on the page={data}")
-
- return dict(
- image_url=data[0]["image_url"],
- thumbnail_url=data[0]["image_thumb_url"],
- )
-
- logging.error(f"failed to find upload result on the page. html={res.text}")
-
- raise ValueError(f"failed to parse upload result for {image_path.name}")
-
-
-def parse_args(argv: Optional[typing.Sequence[str]] = None) -> argparse.Namespace:
- arger = argparse.ArgumentParser()
- 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=2) as pool:
- results = list(pool.map(upload_image, image_paths))
- for it in results:
- print(it["image_url"])
-
-
-if __name__ == "__main__":
- logging.basicConfig(level=logging.INFO)
- main()
diff --git a/go.mod b/go.mod
deleted file mode 100644
index 8edb2f3..0000000
--- a/go.mod
+++ /dev/null
@@ -1,5 +0,0 @@
-module github.com/abdusco/playground
-
-go 1.23.0
-
-require golang.org/x/sync v0.10.0
diff --git a/go.sum b/go.sum
deleted file mode 100644
index cf16d91..0000000
--- a/go.sum
+++ /dev/null
@@ -1,2 +0,0 @@
-golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
-golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
diff --git a/html_popup.py b/html_popup.py
deleted file mode 100644
index 508b2b1..0000000
--- a/html_popup.py
+++ /dev/null
@@ -1,133 +0,0 @@
-import http.server
-import random
-import socketserver
-import sys
-import threading
-import time
-import typing
-import webbrowser
-
-# language=html
-inject_script = """
-
-"""
-
-
-def handler_for_html(html: str) -> typing.Type[http.server.SimpleHTTPRequestHandler]:
- class Handler(http.server.SimpleHTTPRequestHandler):
- def do_GET(self):
- if self.path == "/":
- self.send_response(200)
- self.send_header("Content-type", "text/html")
- self.end_headers()
- self.wfile.write(html.encode())
- else:
- self.send_error(404)
-
- def do_POST(self):
- if self.path == "/shutdown":
- self.send_response(200)
- self.end_headers()
- self.server.last_heartbeat = 0 # Force shutdown
- self.server.shutdown()
- elif self.path == "/heartbeat":
- self.send_response(200)
- self.end_headers()
- self.server.last_heartbeat = time.time()
- else:
- self.send_error(404)
-
- def log_message(self, format, *args):
- # Suppress logging
- pass
-
- return Handler
-
-
-class ServerWithHeartbeat(socketserver.TCPServer):
- def __init__(self, *args, **kwargs):
- super().__init__(*args, **kwargs)
- self.last_heartbeat = time.time()
-
-
-def monitor_heartbeat(server: ServerWithHeartbeat, timeout: int = 2):
- while True:
- time.sleep(0.5)
- if time.time() - server.last_heartbeat > timeout:
- print("No heartbeat received, shutting down...")
- server.shutdown()
- break
-
-
-def run_server(handler: http.server.SimpleHTTPRequestHandler, port: int = 8000):
- with ServerWithHeartbeat(("", port), handler) as httpd:
- print(f"Serving at port {port}")
-
- # Start heartbeat monitor in separate thread
- monitor_thread = threading.Thread(target=monitor_heartbeat, args=(httpd,))
- monitor_thread.daemon = True
- monitor_thread.start()
-
- httpd.serve_forever()
-
-
-def main():
- html = sys.stdin.read()
- if "" in html:
- html_to_display = html.replace("", f"{inject_script}{html}{inject_script}