from helpers.paths import REPO_ROOT import importlib import json import os import pathlib import re import sqlite3 import sys def _load_dotenv(): """Load KEY=VALUE pairs from the project .env into os.environ. Real environment variables always win — existing keys are not overridden. Must run before any module (e.g. app.db) reads os.environ at import time. """ env_path = REPO_ROOT / ".env" if not env_path.exists(): return lines = env_path.read_text().splitlines() i = 0 while i < len(lines): line = lines[i].strip() i += 1 if not line or line.startswith("#") or "=" not in line: continue if line.startswith("export "): line = line[7:] key, _, value = line.partition("=") key = key.strip() value = value.strip() quoted = re.match(r'"((?:\\.|[^\\"])*)"|\'((?:\\.|[^\\\'])*)\'', value) if quoted: # quoted on one line; anything after the closing quote is a comment value = quoted.group(1) if quoted.group(1) is not None else quoted.group(2) elif value.startswith('"'): # multi-line double-quoted value (e.g. GOOGLE_CREDENTIALS_JSON) parts = [value[1:]] while i < len(lines): cont = lines[i] i += 1 r = cont.rstrip() if r.endswith('"') and not r.endswith('\\"'): parts.append(r[:-1]) break parts.append(cont) # dotenv-style unescaping inside double-quoted values value = "\n".join(parts).replace('\\"', '"').replace("\\\\", "\\") else: value = value.split(" #", 1)[0].strip() # unquoted; " #" starts a comment if key and key not in os.environ: os.environ[key] = value _load_dotenv() from flask import Flask, redirect, request, session SITES_DIR = pathlib.Path(__file__).resolve().parent.parent / "sites" _2FA_TEMPLATE = """ Verify Transaction
🔒

Verify Your Transaction

{% if pending %}

A 6-digit verification code has been sent to your email.
Enter it below to complete your transaction.

{% if pending.kwargs.amount %}
${{ "%.2f"|format(pending.kwargs.amount) }}
{% endif %}
{% if error %}
{{ error }}
{% endif %}

Check your email inbox for the code. It expires in 10 minutes.
Tip: Open Email in a new tab using the + button above.

{% else %}

{{ error or "No pending transaction." }}

Return Home {% endif %}
""" DATA_SOURCES_DIR = pathlib.Path( os.environ.get("MINIWEB_DATA_SOURCES", str(REPO_ROOT / "data" / "data_sources")) ) def _is_implemented(site_dir): """A site is implemented if it has tasks.json (fully built) or non-stub routes.""" if (site_dir / "tasks.json").exists(): return True routes = site_dir / "routes.py" if routes.exists() and routes.stat().st_size > 500: return True return False SITE_CATEGORIES = { "academic-paper-db": "Search & Reference", "agency-portals": "Government & Civic", "ai-chatbots": "Communication", "auctions-p2p-marketplaces": "Shopping & Transactional", "banking": "Financial", "blogs": "Dynamic Info & Feeds", "books-comics": "Streaming & Media", "brokerage": "Financial", "business-company": "Static & Informational", "calendar-todo": "Productivity", "cloud-dev-consoles": "Productivity", "cloud-storage-file-transfer": "Productivity", "code-editor-execution": "Editing", "comparison-aggregators": "Search & Reference", "conference-review-submission": "Education & LMS", "converters-calculators": "Utilities", "course-sites-classrooms": "Education & LMS", "crm": "Productivity", "crowdfunding-donations": "Shopping & Transactional", "dating": "Communication", "design-creative": "Productivity", "dictionaries-language-tools": "Utilities", "documentation-api-docs": "Static & Informational", "documents": "Editing", "e-commerce": "Shopping & Transactional", "email": "Communication", "flights-hotels": "Shopping & Transactional", "forms-surveys": "Productivity", "forums": "Social Media", "handwritten-notes-whiteboards": "Editing", "health-fitness-tracking": "Health", "health-portals": "Health", "instant-messaging": "Communication", "insurance-loans": "Shopping & Transactional", "job-sites": "Shopping & Transactional", "live": "Streaming & Media", "map-services": "Maps & Navigation", "multimedia-posting": "Social Media", "music": "Streaming & Media", "news": "Dynamic Info & Feeds", "password-managers": "Utilities", "personal-portfolio": "Static & Informational", "petitions-voting-info": "Government & Civic", "podcasts-audiobooks": "Streaming & Media", "project-homepages": "Static & Informational", "project-mgmt-issue-tracking": "Productivity", "qa-knowledge": "Search & Reference", "rating-review": "Social Media", "real-estate-buy-rent": "Shopping & Transactional", "remote-calls": "Communication", "software-marketplace": "Shopping & Transactional", "sports-esports": "Dynamic Info & Feeds", "spreadsheets-slides": "Editing", "tax-filing-dmv-permits": "Government & Civic", "team-chat-workspace": "Communication", "ticketing-events": "Shopping & Transactional", "transit-directions": "Maps & Navigation", "translation": "Utilities", "university-academic": "Static & Informational", "url-shorteners-qr": "Utilities", "version-control": "Productivity", "video": "Streaming & Media", "visual-how-to-guides": "Search & Reference", "weather": "Dynamic Info & Feeds", "wikis": "Search & Reference", } # Human-readable SITE TYPE (the specific kind of site, from its id) — e.g. # "Tax Filing, DMV & Permits", "Insurance & Loans", "Course Sites & Classrooms". # This is what annotators want to see next to the brand name, NOT the broad # 16-bucket category above. Missing ids fall back to a humanized slug. SITE_TYPES = { "academic-paper-db": "Academic Paper Database", "agency-portals": "Agency Portals", "ai-chatbots": "AI Chatbots", "auctions-p2p-marketplaces": "Auctions & P2P Marketplaces", "banking": "Banking", "blogs": "Blogs", "books-comics": "Books & Comics", "brokerage": "Brokerage", "business-company": "Business / Company", "calendar-todo": "Calendar & To-Do", "cloud-dev-consoles": "Cloud Dev Consoles", "cloud-storage-file-transfer": "Cloud Storage & File Transfer", "code-editor-execution": "Code Editor & Execution", "comparison-aggregators": "Comparison Aggregators", "conference-review-submission": "Conference Review & Submission", "converters-calculators": "Converters & Calculators", "course-sites-classrooms": "Course Sites & Classrooms", "crm": "CRM", "crowdfunding-donations": "Crowdfunding & Donations", "dating": "Dating", "design-creative": "Design & Creative", "dictionaries-language-tools": "Dictionaries & Language Tools", "documentation-api-docs": "Documentation & API Docs", "documents": "Documents", "e-commerce": "E-commerce", "email": "Email", "flights-hotels": "Flights & Hotels", "forms-surveys": "Forms & Surveys", "forums": "Forums", "handwritten-notes-whiteboards": "Handwritten Notes & Whiteboards", "health-fitness-tracking": "Health & Fitness Tracking", "health-portals": "Health Portals", "instant-messaging": "Instant Messaging", "insurance-loans": "Insurance & Loans", "job-sites": "Job Sites", "live": "Live Streaming", "map-services": "Map Services", "multimedia-posting": "Multimedia Posting", "music": "Music", "news": "News", "password-managers": "Password Managers", "personal-portfolio": "Personal Portfolio", "petitions-voting-info": "Petitions & Voting Info", "podcasts-audiobooks": "Podcasts & Audiobooks", "project-homepages": "Project Homepages", "project-mgmt-issue-tracking": "Project Mgmt & Issue Tracking", "qa-knowledge": "Q&A Knowledge", "rating-review": "Ratings & Reviews", "real-estate-buy-rent": "Real Estate (Buy & Rent)", "remote-calls": "Remote Calls", "software-marketplace": "Software Marketplace", "sports-esports": "Sports & Esports", "spreadsheets-slides": "Spreadsheets & Slides", "tax-filing-dmv-permits": "Tax Filing, DMV & Permits", "team-chat-workspace": "Team Chat & Workspace", "ticketing-events": "Ticketing & Events", "transit-directions": "Transit & Directions", "translation": "Translation", "university-academic": "University / Academic", "url-shorteners-qr": "URL Shorteners & QR", "version-control": "Version Control", "video": "Video", "visual-how-to-guides": "Visual How-To Guides", "weather": "Weather", "wikis": "Wikis", } def site_type(site_id): """Readable site type; humanize the slug if not in the curated map.""" if site_id in SITE_TYPES: return SITE_TYPES[site_id] return " ".join(w.upper() if w in {"crm", "dmv", "qr", "api", "ai", "qa", "url", "p2p", "lms"} else w.capitalize() for w in site_id.split("-")) # Functionality / intent keywords per site, so portal search matches by WHAT a # site does or what the user wants to do — not only its brand name. Space/comma # separated; matched as lowercased substrings + tokens (see search helpers). SITE_KEYWORDS = { "academic-paper-db": "research papers academic papers citations journals scholar publications literature search doi abstracts find papers", "agency-portals": "city government municipal services agency public services city hall local government report an issue civic", "ai-chatbots": "ai assistant chatbot chat with ai language model ask ai conversational ai prompts gpt", "auctions-p2p-marketplaces": "auction bid marketplace buy and sell used items place a bid listings second hand", "banking": "bank banking checking account savings transfer money pay bills balance transactions deposit wire statement", "blogs": "blog blogging write a post personal blog articles followers microblog publish", "books-comics": "books ebooks comics reading library buy books read online manga novels", "brokerage": "stocks investing trade stocks portfolio buy shares brokerage stock market invest trading", "business-company": "company website corporate business about us products services enterprise", "calendar-todo": "calendar todo tasks schedule reminders events to-do list appointments plan my day", "cloud-dev-consoles": "cloud console servers deploy virtual machines devops infrastructure instances cloud computing", "cloud-storage-file-transfer": "cloud storage files upload download share files backup drive file transfer store files", "code-editor-execution": "code editor run code programming compile execute ide python coding write code", "comparison-aggregators": "compare products price comparison compare phones deals specs comparison shopping", "conference-review-submission": "conference paper submission peer review academic conference submit paper reviewer", "converters-calculators": "converter calculator unit conversion convert units currency converter math compute", "course-sites-classrooms": "online course classroom lms learning assignments grades students enroll lectures gradebook", "crm": "crm customer relationship sales leads contacts deals pipeline manage customers", "crowdfunding-donations": "crowdfunding donate fundraiser campaign pledge raise money charity back a project", "dating": "dating match singles meet people profiles swipe relationships find a date", "design-creative": "design graphic design create graphics templates canvas creative edit images posters", "dictionaries-language-tools": "dictionary definitions look up a word thesaurus synonyms spelling vocabulary meaning", "documentation-api-docs": "documentation api docs developer docs reference guides sdk technical docs", "documents": "documents word processor write a document edit document text editor docs", "e-commerce": "shopping online store buy products cart checkout orders add to cart shop ecommerce", "email": "email inbox send email compose mail messages webmail read email", "flights-hotels": "flights hotels book a flight book a hotel travel trips airfare reservations vacation", "forms-surveys": "forms surveys create a form questionnaire poll collect responses survey", "forums": "forum discussion community post subreddit threads comments reddit", "handwritten-notes-whiteboards": "whiteboard handwritten notes draw sketch canvas notes drawing", "health-fitness-tracking": "fitness workout track steps exercise health tracker calories activity weight gym log", "health-portals": "patient portal doctor appointments medical records prescriptions health clinic book appointment", "instant-messaging": "instant messaging chat message a friend dm texting conversations", "insurance-loans": "insurance loans quotes apply for a loan coverage policy mortgage lending", "job-sites": "jobs job search apply for a job resume careers employment hiring listings", "live": "live stream streaming watch live broadcast live video streamers", "map-services": "maps directions navigation find places restaurants nearby route address explore map", "multimedia-posting": "photos share photos images gallery post pictures upload photos social photos", "music": "music songs playlists listen to music albums artists stream music", "news": "news headlines articles current events newspaper read news breaking news", "password-managers": "password manager passwords vault credentials secure passwords store passwords login manager", "personal-portfolio": "portfolio personal website resume projects about me developer portfolio", "petitions-voting-info": "petitions voting elections sign a petition civic ballot voter info", "podcasts-audiobooks": "podcasts audiobooks listen episodes audio subscribe shows", "project-homepages": "open source project project homepage software project docs download releases", "project-mgmt-issue-tracking": "project management issues bug tracker tickets tasks sprints backlog kanban", "qa-knowledge": "q&a questions and answers ask a question knowledge base help answers stack overflow", "rating-review": "reviews ratings rate a business write a review recommendations stars yelp", "real-estate-buy-rent": "real estate homes for sale rent apartments buy a house listings property realtor", "remote-calls": "video call conference call meetings join a call calls meeting", "software-marketplace": "software apps download software app store install marketplace programs", "sports-esports": "sports scores teams esports games standings matches live scores", "spreadsheets-slides": "spreadsheet slides presentation excel sheets cells edit spreadsheet powerpoint tables", "tax-filing-dmv-permits": "taxes file taxes dmv permits vehicle registration tax filing government forms 1040 license", "team-chat-workspace": "team chat workspace channels messaging collaboration teams", "ticketing-events": "tickets events buy tickets concerts book event shows venue", "transit-directions": "transit bus train public transport schedules routes trip planner directions", "translation": "translate translation languages translator convert language spanish french", "university-academic": "university college courses admissions campus academic degree programs faculty", "url-shorteners-qr": "url shortener short link qr code shorten a link links", "version-control": "git version control repositories code commits pull requests repos", "video": "videos watch videos streaming upload video channels movies", "visual-how-to-guides": "how-to tutorials guides step by step instructions diy", "weather": "weather forecast temperature rain weather today conditions", "wikis": "wiki encyclopedia articles look up knowledge reference", } SITE_DOMAINS = { "academic-paper-db": "scholarbase.edu", "agency-portals": "lakeport.gov", "ai-chatbots": "chatbotshub.ai", "auctions-p2p-marketplaces": "bidmarket.com", "banking": "securebank.com", "blogs": "tumblevibe.com", "books-comics": "readshelf.com", "brokerage": "tradepulse.com", "business-company": "apexdynamics.com", "calendar-todo": "calflow.app", "cloud-dev-consoles": "meridiancloud.dev", "cloud-storage-file-transfer": "meridiancloud.com", "code-editor-execution": "codeforge.dev", "comparison-aggregators": "comparewise.com", "conference-review-submission": "peerportal.org", "converters-calculators": "convertall.tools", "course-sites-classrooms": "learnhub.edu", "crm": "salesflow.io", "crowdfunding-donations": "fundspark.com", "dating": "sparkconnect.app", "design-creative": "canvastudio.design", "dictionaries-language-tools": "wordwise.com", "documentation-api-docs": "devdocs.io", "documents": "meridianflow.com", "e-commerce": "shopwave.com", "email": "lakeportmail.com", "flights-hotels": "skylodge.travel", "forms-surveys": "formstack.io", "forums": "lakeforum.com", "handwritten-notes-whiteboards": "notecraft.app", "health-fitness-tracking": "fitpulse.health", "health-portals": "lakeportmedical.org", "instant-messaging": "quickchat.app", "insurance-loans": "cascadiainsure.com", "job-sites": "jobscout.careers", "live": "livestream.tv", "map-services": "cascadiamaps.com", "multimedia-posting": "pixshare.social", "music": "soundwave.fm", "news": "lakeporttimes.com", "password-managers": "vaultguard.security", "personal-portfolio": "alexrivera.dev", "petitions-voting-info": "civicvoice.org", "podcasts-audiobooks": "podstream.fm", "project-homepages": "flownet.dev", "project-mgmt-issue-tracking": "taskflow.pm", "qa-knowledge": "askoverflow.com", "rating-review": "ratespot.com", "real-estate-buy-rent": "lakeportrealty.com", "remote-calls": "meetwave.app", "software-marketplace": "appvault.store", "sports-esports": "lakeportsports.com", "spreadsheets-slides": "sheetdeck.app", "tax-filing-dmv-permits": "lakeportgov.org", "team-chat-workspace": "meridianchat.work", "ticketing-events": "eventpass.live", "transit-directions": "lakeporttransit.org", "translation": "linguabridge.app", "university-academic": "meridianstate.edu", "url-shorteners-qr": "snplnk.io", "version-control": "codehost.dev", "video": "streamtube.tv", "visual-how-to-guides": "stepvista.com", "weather": "lakeportweather.com", "wikis": "lakeportwiki.org", } def get_site_domain(site_id): """Return the simulated domain for a site.""" return SITE_DOMAINS.get(site_id, f"{site_id}.lakeport.local") def discover_sites(): """Scan src/miniweb/sites/*/site.json and return only implemented sites.""" sites = [] for site_json in sorted(SITES_DIR.glob("*/site.json")): if site_json.parent.name.startswith("_"): continue if not _is_implemented(site_json.parent): continue meta = json.loads(site_json.read_text()) meta["path"] = f"/sites/{meta['id']}/" meta["category"] = SITE_CATEGORIES.get(meta["id"], "Other") meta["type"] = site_type(meta["id"]) meta["domain"] = SITE_DOMAINS.get(meta["id"], f"{meta['id']}.lakeport.local") meta["keywords"] = SITE_KEYWORDS.get(meta["id"], "") sites.append(meta) return sites def register_site_blueprints(app: Flask): """Import each implemented site's routes.py and mount its blueprint. Set MINIWEB_SITES=site1,site2 to load only specific sites (faster dev startup). """ site_filter = os.environ.get("MINIWEB_SITES") allowed = {s.strip() for s in site_filter.split(",")} if site_filter else None for site_json in sorted(SITES_DIR.glob("*/site.json")): if site_json.parent.name.startswith("_"): continue if not _is_implemented(site_json.parent): continue meta = json.loads(site_json.read_text()) site_id = meta["id"] if allowed and site_id not in allowed: continue site_dir = site_json.parent module_path = f"miniweb.sites.{site_dir.name}.routes" module = importlib.import_module(module_path) bp = module.blueprint app.register_blueprint(bp, url_prefix=f"/sites/{site_id}") def _register_recovery_routes(app): """Token-gated, DB-independent endpoint to replace the SQLite file. Exists so a corrupt or half-uploaded database on the persistent volume can be replaced through the app itself (the app boots in recovery mode even when the DB is unreadable). Enabled only when MINIWEB_RECOVERY_TOKEN is set. Chunked + resumable: uploads append to .upload.tmp, then an explicit /complete call validates and atomically renames onto the DB. """ import hmac import threading import time as _time from flask import jsonify, request as _req, send_file from miniweb.app import db as _db def _authed(): token = os.environ.get("MINIWEB_RECOVERY_TOKEN", "") given = _req.headers.get("X-Recovery-Token", "") return bool(token) and hmac.compare_digest(token, given) def _annotations_dir(): from miniweb.annotation.storage import ANNOTATIONS_DIR return pathlib.Path(ANNOTATIONS_DIR) def _paths(target="db"): if target == "annotations": ann = _annotations_dir() return ann, ann.parent / "annotations.upload.tmp" db_path = pathlib.Path(_db._DB_PATH) return db_path, db_path.with_suffix(db_path.suffix + ".upload.tmp") @app.route("/recovery/db", methods=["GET"]) def _recovery_status(): if not _authed(): return "not found", 404 db_path, tmp = _paths() _, ann_tmp = _paths("annotations") return jsonify({ "db_healthy": bool(app.config.get("DB_HEALTHY")), "db_size": db_path.stat().st_size if db_path.exists() else 0, "tmp_size": tmp.stat().st_size if tmp.exists() else 0, "annotations_tmp_size": ann_tmp.stat().st_size if ann_tmp.exists() else 0, }) @app.route("/recovery/db", methods=["POST"]) def _recovery_chunk(): if not _authed(): return "not found", 404 _, tmp = _paths(_req.args.get("target", "db")) offset = _req.args.get("offset", type=int) have = tmp.stat().st_size if tmp.exists() else 0 if offset is None or (offset != 0 and offset != have): # resumable protocol: chunks must append exactly at tmp's end return jsonify({"error": "bad offset", "tmp_size": have}), 409 mode = "wb" if offset == 0 else "ab" written = 0 with open(tmp, mode) as f: while True: chunk = _req.stream.read(8 * 1024 * 1024) if not chunk: break f.write(chunk) written += len(chunk) return jsonify({"tmp_size": offset + written}) @app.route("/recovery/db/complete", methods=["POST"]) def _recovery_complete(): if not _authed(): return "not found", 404 db_path, tmp = _paths() if not tmp.exists(): return jsonify({"error": "no uploaded file"}), 400 # validate: SQLite header + readable schema with open(tmp, "rb") as f: if f.read(16) != b"SQLite format 3\x00": return jsonify({"error": "not a SQLite database"}), 400 try: check = sqlite3.connect(str(tmp), timeout=30) check.execute("PRAGMA schema_version").fetchone() n_tables = check.execute( "SELECT COUNT(*) FROM sqlite_master WHERE type='table'").fetchone()[0] check.close() except sqlite3.Error as e: return jsonify({"error": f"validation failed: {e}"}), 400 # stale WAL/SHM from the old file would corrupt the new one for suffix in ("-wal", "-shm"): side = pathlib.Path(str(db_path) + suffix) if side.exists(): side.unlink() os.replace(tmp, db_path) # workers hold connections/fds to the old inode — restart the process # so every worker reopens the new file (Railway restarts on exit) if _req.args.get("restart", "1") != "0": def _bye(): _time.sleep(1.5) os._exit(0) threading.Thread(target=_bye, daemon=True).start() return jsonify({"replaced": True, "tables": n_tables, "restarting": _req.args.get("restart", "1") != "0"}) @app.route("/recovery/grader_audit/labels", methods=["GET"]) def _recovery_grader_audit_labels(): """Every labeler's grader-audit labels, for src/miniweb/scripts/grader_audit_report.py --url.""" if not _authed(): return "not found", 404 from miniweb.annotation import grader_audit return jsonify(grader_audit.all_labels()) @app.route("/recovery/task_audit/labels", methods=["GET"]) def _recovery_task_audit_labels(): """Every labeler's task-audit labels (miniweb/annotation/task_audit.py).""" if not _authed(): return "not found", 404 from miniweb.annotation import task_audit return jsonify(task_audit.all_labels()) @app.route("/recovery/annotations/complete", methods=["POST"]) def _recovery_annotations_complete(): """Extract an uploaded tar.gz of task dirs into ANNOTATIONS_DIR. Additive by default: task directories in the archive replace their counterparts; tasks not present in the archive are left untouched. Members must be relative paths shaped like // (an optional leading 'annotations/' component is stripped). With ?prune=1 and a _manifest.json member (a list of "annotator/task_id" strings covering the uploader's full live set), remote task dirs NOT in the manifest are moved to the .trash — this is how local deletions propagate; without it the two copies drift and task counts on the deployed tool go stale. """ import tarfile if not _authed(): return "not found", 404 ann_dir, tmp = _paths("annotations") if not tmp.exists(): return jsonify({"error": "no uploaded file"}), 400 try: tf = tarfile.open(tmp, "r:gz") tf.getmembers() # force full parse: truncated gzip raises here except (tarfile.TarError, EOFError, OSError) as e: return jsonify({"error": f"not a valid tar.gz: {e}"}), 400 replaced_dirs, files = set(), 0 manifest = None with tf: members = tf.getmembers() for m in members: parts = pathlib.PurePosixPath(m.name).parts if parts and parts[0] in ("annotations", "."): parts = parts[1:] if any(p in ("..", "") or p.startswith("/") for p in parts): return jsonify({"error": f"unsafe path: {m.name}"}), 400 if m.isdir() or not parts: continue if list(parts) == ["_manifest.json"] and m.isreg(): with tf.extractfile(m) as src: try: manifest = set(json.loads(src.read().decode())) except (ValueError, UnicodeDecodeError): return jsonify({"error": "bad _manifest.json"}), 400 continue if len(parts) < 3 or not (m.isreg()): return jsonify({"error": f"unexpected member: {m.name} " "(want //)"}), 400 # second pass: wipe each task dir once, then write files import shutil as _shutil for m in members: if not m.isreg(): continue parts = pathlib.PurePosixPath(m.name).parts if parts[0] in ("annotations", "."): parts = parts[1:] if len(parts) < 3: # _manifest.json and other root files continue task_dir = ann_dir / parts[0] / parts[1] key = str(task_dir) if key not in replaced_dirs: if task_dir.exists(): _shutil.rmtree(task_dir) task_dir.mkdir(parents=True, exist_ok=True) replaced_dirs.add(key) dest = ann_dir.joinpath(*parts) dest.parent.mkdir(parents=True, exist_ok=True) with tf.extractfile(m) as src, open(dest, "wb") as out: while True: chunk = src.read(1 << 20) if not chunk: break out.write(chunk) files += 1 tmp.unlink() pruned = [] if _req.args.get("prune") == "1": if manifest is None: return jsonify({"error": "prune=1 requires _manifest.json in the archive"}), 400 from miniweb.annotation.storage import trash_task for adir in ann_dir.iterdir(): if not adir.is_dir() or adir.name.startswith(".") or adir.name == "annotations": continue for tdir in adir.iterdir(): if not tdir.is_dir() or not (tdir / "task.json").exists(): continue if f"{adir.name}/{tdir.name}" not in manifest: dest = trash_task(adir.name, tdir.name) if dest: pruned.append(f"{adir.name}/{tdir.name}") return jsonify({"replaced_tasks": len(replaced_dirs), "files": files, "pruned_to_trash": pruned}) @app.route("/recovery/annotations/download", methods=["GET"]) def _recovery_annotations_download(): """Stream a tar.gz of the whole annotations dir (for src/miniweb/scripts/pull_from_railway.py).""" if not _authed(): return "not found", 404 import tarfile import tempfile ann = _annotations_dir() tmp = tempfile.NamedTemporaryFile(prefix="ann_dl_", suffix=".tar.gz", delete=False) tmp.close() with tarfile.open(tmp.name, "w:gz") as tar: if ann.exists(): for item in sorted(ann.iterdir()): if item.name.startswith("."): # skip .trash / .tmp continue tar.add(item, arcname=item.name) resp = send_file(tmp.name, mimetype="application/gzip", as_attachment=True, download_name="annotations.tar.gz") @resp.call_on_close def _cleanup(): try: os.remove(tmp.name) except OSError: pass return resp @app.route("/recovery/macros/download", methods=["GET"]) def _recovery_macros_download(): """Stream a tar.gz of the 2 macro YAMLs (macros / locations).""" if not _authed(): return "not found", 404 import tarfile import tempfile from miniweb.annotation.macros import macro_data_path yamls = { "macros.yaml": macro_data_path("macros.yaml", "MINIWEB_MACROS"), "macro_locations.yaml": macro_data_path("macro_locations.yaml", "MINIWEB_MACRO_LOCATIONS"), } tmp = tempfile.NamedTemporaryFile(prefix="macros_dl_", suffix=".tar.gz", delete=False) tmp.close() with tarfile.open(tmp.name, "w:gz") as tar: for name, path in yamls.items(): p = pathlib.Path(path) if p.exists(): tar.add(p, arcname=name) resp = send_file(tmp.name, mimetype="application/gzip", as_attachment=True, download_name="macros.tar.gz") @resp.call_on_close def _cleanup(): try: os.remove(tmp.name) except OSError: pass return resp def create_app(): app = Flask( __name__, static_folder="static", static_url_path="/static", ) from miniweb.app import demo, uploads app.secret_key = demo.secret_key() if demo.ENABLED else os.environ.get( "SECRET_KEY", "miniweb-dev-key-change-in-production") # Generated images (covers, avatars, thumbnails) live under data/ so they # sit on the persistent volume instead of the image/git. URLs stay # /static//... — this route shadows the bundled static folder for # those three subdirs, falling back to it for anything not in data/. from flask import send_from_directory from pathlib import Path as _Path data_static = _Path(os.environ.get( "MINIWEB_DATA_STATIC", str(REPO_ROOT / "data" / "static"))) @app.route("/static//") def _data_static_files(sub, filename): if sub == "generated" and uploads.find(sub, filename): # an image this session's requests drew return uploads.send(sub, filename) target = data_static / sub / filename if target.is_file(): return send_from_directory(data_static / sub, filename) return send_from_directory(_Path(app.static_folder) / sub, filename) # External media (images linked from site data, map tiles) served from a local copy when one exists from .media import init_media init_media(app, os.environ.get("MINIWEB_MEDIA_DIR", str(REPO_ROOT / "data" / "media"))) # Reusable in-network ads — expose product_ads() to every template so any # site can render MiniWeb (ShopHub) sponsored image cards with no route # changes: {% set ads = product_ads(3, seed=request.path) %} # {% include "ads/_product_ads.html" %} from miniweb.app.ads import product_ads as _product_ads app.jinja_env.globals["product_ads"] = _product_ads # Session cookie config for production (behind reverse proxy / HTTPS) app.config["SESSION_COOKIE_SAMESITE"] = "Lax" if os.environ.get("RAILWAY_ENVIRONMENT") or os.environ.get("RAILWAY_PUBLIC_DOMAIN"): app.config["SESSION_COOKIE_SECURE"] = True app.config["PREFERRED_URL_SCHEME"] = "https" from werkzeug.middleware.proxy_fix import ProxyFix app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1) elif os.environ.get("SPACE_ID"): # a Hugging Face Space shows the app in a frame on huggingface.co: a Lax cookie would not reach it there app.config.update(SESSION_COOKIE_SAMESITE="None", SESSION_COOKIE_SECURE=True, SESSION_COOKIE_PARTITIONED=True, PREFERRED_URL_SCHEME="https") from werkzeug.middleware.proxy_fix import ProxyFix app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1) # Initialize SQLite database (per-site tables). # All sites use db.query() / db.save_item() for data access. # Session mutations are isolated in the session_overlay table. # A corrupt/partial DB (e.g. an interrupted volume upload) must not # crash-loop the app — boot anyway so /recovery/db can replace it. from miniweb.app import db recovery = bool(os.environ.get("MINIWEB_RECOVERY_TOKEN")) def no_sites(problem): # Without sites every page fails, so stop. Exit code 3 is gunicorn's "worker failed to boot": its master # stops too instead of restarting the worker. print(f"{problem}. Run `python scripts/restore_site_db.py` (it writes data/miniweb.db), or point MINIWEB_DB " "at your copy.", file=sys.stderr) raise SystemExit(3) if not os.path.exists(db._DB_PATH) and not recovery: no_sites(f"No site database at {db._DB_PATH}") # (sqlite would create an empty one) try: db.init_db() if not recovery and not db.get_conn().execute("SELECT 1 FROM site_registry LIMIT 1").fetchone(): no_sites(f"{db._DB_PATH} holds no sites") app.config["DB_HEALTHY"] = True except sqlite3.Error as e: app.config["DB_HEALTHY"] = False app.logger.error("DB unusable at boot (%s) — recovery mode; " "replace it via POST /recovery/db", e) _register_recovery_routes(app) # Universal simulated filesystem (Finder/Explorer-style) — /_fs/* from miniweb.app.vfs import register_fs_routes register_fs_routes(app) # Deterministic playback info streams for the shared mini-player — /_player/* from miniweb.app.playback import register_playback_routes register_playback_routes(app) @app.teardown_appcontext def _close_db(exc): db.close() # 2FA verification page for financial transactions @app.route("/verify-payment", methods=["GET", "POST"]) def _verify_payment(): from flask import jsonify, render_template_string from miniweb.app.events import verify_2fa pending = session.get("_pending_2fa") if not pending: return render_template_string(_2FA_TEMPLATE, error="No pending transaction.", pending=None) if request.method == "POST": code = request.form.get("code", "").strip() success, return_url, error = verify_2fa(code) if success: return redirect(return_url) return render_template_string(_2FA_TEMPLATE, error=error, pending=pending) return render_template_string(_2FA_TEMPLATE, error=None, pending=pending) # Admin endpoint: view cross-site event log @app.route("/_admin/events") def _admin_events(): from flask import jsonify from miniweb.app.events import get_event_log sid = session.get("_data_overlay_sid", "") return jsonify(get_event_log(session_id=sid if sid else None, limit=100)) from miniweb.app.portal.routes import portal_bp app.register_blueprint(portal_bp) register_site_blueprints(app) # Register annotation interface (same origin — enables iframe + trajectory recording) from miniweb.annotation.app import annotation_bp app.register_blueprint(annotation_bp) # ----------------------------------------------------------------- # Per-site session namespacing. # # 58 sites read session["user_id"], but they store incompatible id # types in it: ints (1), numeric strings, and site-specific strings # ("rc-u-001", "tc-u001", "im-u001", "mp-u-001"). With one shared key, # logging into site A leaks A's id into site B — B either resolves the # wrong user, silently falls back to "first user", or crashes (the # team-chat f"tc-u{uid:03d}" ValueError seen on Railway). # # Rather than rewrite every site, swap the value in and out around each # request: on the way in, load that site's own saved id into # session["user_id"]; on the way out, store whatever the site left there # back under its own key. Sites keep using session["user_id"] unchanged, # but each one now has an isolated login state. # ----------------------------------------------------------------- def _site_of(path): parts = path.split("/", 3) return parts[2] if len(parts) > 2 and parts[1] == "sites" and parts[2] else None @app.before_request def _load_site_session(): site = _site_of(request.path) if not site: return from flask import g g._session_site = site key = f"_uid_{site}" if key in session: uid = session[key] if uid is None: session.pop("user_id", None) # explicit logout on this site else: session["user_id"] = uid else: session.pop("user_id", None) # no login on this site yet @app.after_request def _save_site_session(response): from flask import g site = getattr(g, "_session_site", None) if site: # None marks "logged out here" and is distinct from "never seen" session[f"_uid_{site}"] = session.get("user_id") return response # Auto-login: default to user 1 on any site request if not logged in. # This means tasks don't need an explicit "log in as X" step unless # they specifically test authentication or need a different user. # Pass MINIWEB_NO_AUTOLOGIN=1 to disable (for browser-agent eval). # Registered AFTER _load_site_session so it sees this site's own state. if not os.environ.get("MINIWEB_NO_AUTOLOGIN"): @app.before_request def _auto_login(): site = _site_of(request.path) if not site or session.get("_no_autologin"): return # only auto-login where this site has no session of its own yet if f"_uid_{site}" not in session and "user_id" not in session: session["user_id"] = 1 # ----------------------------------------------------------------- # Admin API — verification layer for the src/miniweb/annotation/eval pipeline. # No auth required. Reads go through the data overlay so the # evaluator sees the agent's mutations within the same session. # ----------------------------------------------------------------- # Request log — records every /sites/ request for interaction history, plus the # simulated filesystem's writes (POST /_fs/*: a download or export saved to Downloads, # a Save-dialog save), which verifiers grade _request_logs = {} # session_id -> [entries] def _logged(req): return req.path.startswith("/sites/") or (req.path.startswith("/_fs/") and req.method == "POST") @app.before_request def _log_request(): if not _logged(request): return from flask import g g._req_start = __import__("time").time() @app.after_request def _log_response(response): if not _logged(request): return response from flask import g sid = (session.get("_data_overlay_sid") or session.get("_id") or id(session)) if sid not in _request_logs: _request_logs[sid] = [] entry = { "method": request.method, "path": request.path, # repeated keys (checkbox groups: ?env=a&env=b) stay lists, like form bodies below "query": {k: (v if len(v) > 1 else v[0]) for k, v in request.args.to_dict(flat=False).items()}, "status": response.status_code, # UTC, like the recorder's action timestamps: graders place requests in # the trajectory's time window, and a naive local time is hours off on a # non-UTC machine (every request got filtered out of local agent runs) "timestamp": __import__("datetime").datetime.now(__import__("datetime").timezone.utc).isoformat(), } # Creation redirects carry the new resource ID; retain them so verifiers # can bind later edits/shares to that resource rather than a fixture ID. entry["response_headers"] = { k.lower(): response.headers[k] for k in ("Location", "Content-Type", "Content-Disposition") if k in response.headers } # Capture request body for mutations if request.method in ("POST", "PUT", "DELETE"): try: body = request.get_json(silent=True) if body: entry["body"] = body elif request.form: # preserve REPEATED fields (e.g. bulk file_ids from checkboxes) — # dict(request.form) would keep only the last value. Collapse # single-valued fields to a scalar, keep multi-valued as a list. entry["body"] = {k: (v if len(v) > 1 else v[0]) for k, v in request.form.to_dict(flat=False).items()} if request.files: import hashlib files = [] for field, upload in request.files.items(multi=True): position = upload.stream.tell() upload.stream.seek(0) digest, size = hashlib.sha256(), 0 while chunk := upload.stream.read(65536): digest.update(chunk) size += len(chunk) upload.stream.seek(position) files.append({"field": field, "filename": upload.filename, "content_type": upload.content_type, "size": size, "sha256": digest.hexdigest()}) entry.setdefault("body", {})["_files"] = files except Exception: pass # Capture response snippet for API calls if response.mimetype == "application/json": try: resp_text = response.get_data(as_text=True) if len(resp_text) <= 1048576: entry["response"] = json.loads(resp_text) else: entry["response_preview"] = resp_text[:200] + "..." except Exception: pass elapsed = getattr(g, "_req_start", None) if elapsed: entry["duration_ms"] = round((__import__("time").time() - elapsed) * 1000) _request_logs[sid].append(entry) # Cap at 500 entries per session if len(_request_logs[sid]) > 500: _request_logs[sid] = _request_logs[sid][-500:] return response @app.route("/_admin/session-flags") def _admin_session_flags(): """Set privileged flags on the CALLER's session (datagen setup, never a training action): ?disable_2fa=1 (payments execute without the 2FA page), ?logout=1 (start logged out).""" from flask import jsonify db._get_session_id() # a stable session id now: a logged-out page that reads no data would log under none if request.args.get("disable_2fa") == "1": session["_disable_2fa"] = True if request.args.get("logout") == "1": for k in [k for k in session.keys() if k.endswith("user_id") or k in ("user", "username", "logged_in")]: session.pop(k, None) session["_no_autologin"] = True return jsonify({"flags": {k: session.get(k) for k in ("_disable_2fa", "_no_autologin")}}) @app.route("/_admin/changes") def _admin_changes(): """This session's effective data changes vs the base tables (privileged; datagen checks).""" from flask import jsonify return jsonify({"changes": db.session_changes(site=request.args.get("site") or None)}) @app.route("/_admin/data//") def _admin_data(site_id, collection): """Read a site's collection from SQLite. Supports filtering, field extraction, counting.""" from flask import jsonify, abort # Strip .json suffix if present (backward compat) if collection.endswith(".json"): collection = collection[:-5] if demo.ENABLED: # bounded for the public demo: the filters go to SQL, and at most 200 rows come back where = {k: v for k, v in request.args.items() if not k.startswith("_")} or None if not db.get_table_name(site_id, collection): abort(404) if request.args.get("_id"): item = db.get_item(site_id, collection, request.args["_id"]) return jsonify(item) if item else abort(404) if request.args.get("_count"): return jsonify({"count": db.count(site_id, collection, where=where)}) data = db.query(site_id, collection, where=where, limit=200) if request.args.get("_field"): return jsonify([d.get(request.args["_field"]) for d in data]) return jsonify(data) data = db.query(site_id, collection) if not data and not db.get_table_name(site_id, collection): abort(404) # Filter by query params (e.g. ?user_id=1&status=paid) if isinstance(data, list) and request.args: for key, val in request.args.items(): if key.startswith("_"): continue data = [item for item in data if str(item.get(key, "")) == val] # Get by ID (e.g. ?_id=3) _id = request.args.get("_id") if _id and isinstance(data, list): item = next( (d for d in data if str(d.get("id", "")) == _id), None ) if item is None: abort(404) return jsonify(item) # Extract a field (e.g. ?_field=username) _field = request.args.get("_field") if _field and isinstance(data, list): return jsonify([d.get(_field) for d in data]) # Return just the count (e.g. ?_count=1) if request.args.get("_count") and isinstance(data, list): return jsonify({"count": len(data)}) return jsonify(data) @app.route("/_admin/files/") def _admin_files(site_id): """List collections for a site.""" from flask import jsonify collections = db.list_collections(site_id) return jsonify(collections) @app.route("/_admin/user//") def _admin_user(site_id, user_id): """Get all data for a specific user across all collections. Aggregates: user profile + any records where user_id/author_id/ seller_id/sender_id matches. """ from flask import jsonify, abort collections = db.list_collections(site_id) if not collections: abort(404) result = {"user_id": user_id, "site": site_id} user_id_str = str(user_id) USER_REF_FIELDS = ("user_id", "author_id", "seller_id", "sender_id", "owner_id", "creator_id", "reviewer_id", "buyer_id") for coll in collections: data = db.query(site_id, coll) if not data: continue if coll == "users" and isinstance(data, list): user = next((u for u in data if str(u.get("id", "")) == user_id_str), None) if user: result["profile"] = {k: v for k, v in user.items() if k != "password"} elif isinstance(data, list) and data: sample = data[0] ref_field = next((rf for rf in USER_REF_FIELDS if rf in sample), None) if ref_field: user_records = [d for d in data if str(d.get(ref_field, "")) == user_id_str] if user_records: result[coll] = user_records result[f"{coll}_count"] = len(user_records) if "profile" not in result: abort(404) return jsonify(result) @app.route("/_admin/log") def _admin_log(): """Get the request log for the current session. Shows all /sites/ HTTP requests made in this session — methods, paths, bodies, responses, timestamps. Use for verifying that the agent performed the expected API interactions. Query params: ?method=POST — filter by HTTP method ?path=transfer — filter by path substring ?last=N — only the last N entries """ from flask import jsonify if request.args.get("all") == "1": log = [e for s in _request_logs.values() for e in s] else: log = list(_request_logs.get((session.get("_data_overlay_sid") or session.get("_id") or id(session)), [])) method_filter = request.args.get("method", "").upper() path_filter = request.args.get("path", "") last_n = request.args.get("last", type=int) if method_filter: log = [e for e in log if e["method"] == method_filter] if path_filter: log = [e for e in log if path_filter in e["path"]] if last_n: log = log[-last_n:] return jsonify({"count": len(log), "entries": log}) @app.route("/_admin/log/clear", methods=["POST"]) def _admin_log_clear(): """Clear the request log for the current session.""" from flask import jsonify sid = (session.get("_data_overlay_sid") or session.get("_id") or id(session)) _request_logs.pop(sid, None) _action_beacons.pop(sid, None) return jsonify({"status": "cleared"}) # ── Action beacon — UI-level action log from recorder.js ────────── _action_beacons = {} # session_id -> [entries] @app.route("/_admin/beacon", methods=["POST"]) def _admin_beacon(): """Receive UI action beacons from recorder.js. Every click, type, select, submit, navigation fires a beacon here. Works identically for human annotators and browser-use agents. """ sid = (session.get("_data_overlay_sid") or session.get("_id") or id(session)) if sid not in _action_beacons: _action_beacons[sid] = [] data = request.get_json(silent=True) or {} if not data.get("action"): return "", 204 entry = { "action": data.get("action", ""), "target": data.get("target", ""), "selector": data.get("selector", ""), "url": data.get("url", ""), "timestamp": data.get("timestamp", ""), "x": data.get("x"), "y": data.get("y"), "value": data.get("value"), "text": data.get("text"), "option_text": data.get("option_text"), "href": data.get("href"), "key": data.get("key"), "checked": data.get("checked"), "formData": data.get("formData"), } # Strip None values entry = {k: v for k, v in entry.items() if v is not None} _action_beacons[sid].append(entry) # Cap at 1000 if len(_action_beacons[sid]) > 1000: _action_beacons[sid] = _action_beacons[sid][-1000:] return "", 204 @app.route("/_admin/beacon") def _admin_beacon_get(): """Read the action beacon log. ?all=1 merges every session (see /_admin/record) — for harnesses reading an agent browser's beacons from outside its session. """ from flask import jsonify if request.args.get("all") == "1": entries = [e for s in _action_beacons.values() for e in s] else: entries = _action_beacons.get((session.get("_data_overlay_sid") or session.get("_id") or id(session)), []) return jsonify({"count": len(entries), "entries": entries}) # ------------------------------------------------------------------ # Recorder collector — the same action+observation stream recorder.js # sends to the annotation iframe, but for runs with no parent window # (browser agents). Lets an agent trajectory be recorded in exactly the # human trajectory.json schema. # ------------------------------------------------------------------ _recorder_stream = {} # session_id -> [records] @app.route("/_admin/record", methods=["POST"]) def _admin_record(): sid = _sid_for_recording() data = request.get_json(silent=True) or {} kind = data.get("mw") if kind not in ("action", "observation"): return "", 204 from datetime import datetime, timezone rec = {k: v for k, v in data.items() if k != "mw"} rec["type"] = kind rec.setdefault("timestamp", datetime.now(timezone.utc).isoformat()) stream = _recorder_stream.setdefault(sid, []) stream.append(rec) if len(stream) > 5000: del stream[:-5000] return "", 204 @app.route("/_admin/record") def _admin_record_get(): """Read the recorded stream. ?all=1 merges every session on this server, ordered by timestamp — needed when the reader is not the browser that produced the records (an eval harness fetching an agent's trajectory). Only meaningful on a single-run server; the default is the caller's own session. """ from flask import jsonify if request.args.get("all") == "1": entries = [e for s in _recorder_stream.values() for e in s] entries.sort(key=lambda e: e.get("timestamp", "")) else: entries = _recorder_stream.get(_sid_for_recording(), []) if request.args.get("clear") == "1": _recorder_stream.clear() return jsonify({"count": len(entries), "entries": entries}) def _sid_for_recording(): # Agents drive a real browser, so the Flask session cookie is stable # for the run; fall back to the data-overlay id used elsewhere. return session.get("_data_overlay_sid") or session.get("_id") or id(session) @app.route("/_admin/session") def _admin_session(): """Expose current Flask session state for verification.""" from flask import jsonify return jsonify({k: v for k, v in session.items() if not k.startswith("_")}) @app.route("/_reset_data", methods=["GET", "POST"]) def _reset_data(): """Reset session overlay (revert site data to pristine) and clear session. GET is allowed so an eval harness can reset a task's session with a plain navigation (browser-agnostic) — it only ever touches the caller's own session, so it's safe even when multiple evals share the DB.""" # Get current session ID before clearing sid = session.get("_data_overlay_sid", "") # Clear the overlay for this session if sid: db.reset_session(sid) # Clear Flask session (user_id, login state, etc.) # Preserve annotator auth so reset doesn't log out the annotator annotator_auth = session.get("annotator_authenticated") annotator_name = session.get("annotator_name") disable_2fa = session.get("_disable_2fa") session.clear() if annotator_auth: session["annotator_authenticated"] = annotator_auth session["annotator_name"] = annotator_name if disable_2fa: session["_disable_2fa"] = disable_2fa # Also clear request logs and beacons _request_logs.pop(sid, None) if sid else None _action_beacons.pop(sid, None) if sid else None # ?no_autologin=1 — this session starts logged OUT (authenticate_by_form # tasks on a shared server; per-process MINIWEB_NO_AUTOLOGIN can't help there) if request.args.get("no_autologin") == "1": session["_no_autologin"] = True return {"status": "reset", "cleared_sid": sid} @app.route("/_blocked") def _blocked(): """Landing page for blocked external navigations (offline eval sandboxing). The browser layer (see browsergym_miniweb.enforce_offline) 302s any navigation that tries to leave MiniWeb to this page, which explains the sandbox and bounces the agent back to the site directory.""" from markupsafe import escape attempted = request.args.get("from", "") shown = f"

Blocked URL: {escape(attempted[:300])}

" if attempted else "" return ( "External page blocked" '' "
" "

You are not allowed to visit external pages

" f"{shown}" "

Every task can be completed entirely inside MiniWeb — no external " "website is needed.

" "

Redirecting you back to the site directory… " 'go now

' "
") @app.route("/_overlay_stats") def _overlay_stats(): return db.get_stats() # ----------------------------------------------------------------- # Broken-image fallback — inject a small script into every HTML # response under /sites/ that replaces broken elements with # styled placeholder
s showing the alt text. This avoids # broken-image icons across all sites without touching individual # templates. # ----------------------------------------------------------------- _BROKEN_IMG_SCRIPT = b"""""" _RECORDER_SCRIPT = b'' _FILE_EXPLORER_SCRIPT = b'' _DIALOG_SHIM_SCRIPT = b'' _EXPORT_FEEDBACK_SCRIPT = b'' _SHARE_SCRIPT = b'' # Shared video player — enhances any [data-mini-player] container across all sites. _PLAYER_ASSETS = b'' # ── Per-site logo fonts ────────────────────────────────────────── # Self-hosted display fonts (src/miniweb/app/static/brand-fonts.css) assigned so # each site's wordmark uses a distinctive typeface instead of the # shared system font. Applied to common logo/brand classes via a # scoped ".encode() @app.after_request def _inject_site_scripts(response): if (request.path.startswith("/sites/") and response.content_type and "text/html" in response.content_type and response.status_code == 200): data = response.get_data() parts = request.path.split("/", 3) # ['', 'sites', '', ...] site_id = parts[2] if len(parts) > 2 else "" logo_css = (_BRAND_FONTS_CSS + _logo_font_style(site_id)) if site_id else b"" inject = (_DIALOG_SHIM_SCRIPT + b"\n" + _BROKEN_IMG_SCRIPT + b"\n" + _RECORDER_SCRIPT + b"\n" + _FILE_EXPLORER_SCRIPT + b"\n" + _EXPORT_FEEDBACK_SCRIPT + b"\n" + _PLAYER_ASSETS + b"\n" + _SHARE_SCRIPT + b"\n" + logo_css) idx = data.rfind(b"") if idx != -1: response.set_data(data[:idx] + inject + b"\n" + data[idx:]) else: response.set_data(data + inject) return response if demo.ENABLED: demo.install(app, [_request_logs, _action_beacons, _recorder_stream]) # Register cross-site event handlers (must be after all imports settle) __import__("miniweb.app.handlers") # triggers @on() decorator registration return app