"""Explicit offline Chromium reader integration using mocked jobs and real MP3. No inference, corpus or production calls. Supply an already generated real MP3. Muting avoids this host's stalled physical audio sink; it is not audible/phone QA. """ from __future__ import annotations import argparse from datetime import datetime, timezone import json from pathlib import Path import re import sys from urllib.parse import urlparse from playwright.sync_api import sync_playwright ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from tts.text import playback_segments, utf16_offset BASE = "http://127.0.0.1:19876" def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--sample", type=Path, required=True) parser.add_argument("--output", type=Path, default=Path("/tmp/opencode/hitokoto-reader-browser.json")) args = parser.parse_args() audio = args.sample.read_bytes() results = {"at": datetime.now(timezone.utc).isoformat(), "scope": __doc__, "cases": []} with sync_playwright() as playwright: browser = playwright.chromium.launch(headless=True, args=["--no-sandbox"]) try: for mobile in (False, True): context = browser.new_context(viewport={"width": 390, "height": 844} if mobile else {"width": 1440, "height": 900}) jobs, requests, tokens, audio_requests, deleted = {}, [], {}, [], [] lost = [False] hold_text, held = [None], [] page_errors = [] def reply(route, data, status=200): route.fulfill(status=status, content_type="application/json", body=json.dumps(data), headers={"Cache-Control": "no-store"}) def describe(job): state = "expired" if job.get("expired") else "cancelled" if job.get("cancelled") else "ready" if job["polls"] >= 5 else "running" job["state"] = state text = job["body"]["text"] spans = [(utf16_offset(text, start), utf16_offset(text, end)) for start, end, _ in playback_segments(text)] count = len(spans) if job["polls"] >= 3 else 1 if job["polls"] else 0 value = {"id": job["id"], "state": state, "cache_hit": False, "expires_in": 900, "source_segments": [{"start": start, "end": end} for start, end in spans]} if count and state not in {"cancelled", "expired"}: value["progress"] = {"complete": state == "ready", "total_parts": len(spans), "parts": [ {"index": index, "total_parts": len(spans), "start": spans[index][0], "end": spans[index][1], "audio_start": index * 10.6, "audio_end": (index + 1) * 10.6, "audio_seconds": 10.6, "bytes": len(audio), "first_audio_seconds": .2, "pitch_semitones": job["body"].get("pitch_semitones", 0), "audio_url": f"/tts/audio/{job['id']}/parts/{index}.mp3"} for index in range(count)]} if state == "ready": value["result"] = {"audio_seconds": 21.2, "synthesis_seconds": 2.0, "audio_url": f"/tts/audio/{job['id']}.mp3", "pitch_semitones": job["body"].get("pitch_semitones", 0)} return value def handler(route): request = route.request path = urlparse(request.url).path names = {"/tts/": ("tts.html", "text/html"), "/tts/app.js": ("tts.js", "text/javascript"), "/tts/reader.js": ("reader.js", "text/javascript"), "/tts/style.css": ("tts.css", "text/css")} if path in names: name, mime = names[path] route.fulfill(content_type=mime, body=(ROOT / "static" / name).read_bytes()) elif path == "/tts/health": reply(route, {"ready": True, "engine": {"threads": 2}, "queued": 0}) elif path == "/tts/voices": reply(route, {"voices": [{"id": name, "name": name} for name in ("zf_001", "zf_017", "zm_010", "zm_029")]}) elif path == "/tts/jobs" and request.method == "POST": body = request.post_data_json requests.append(body) token = body["request_id"] if token not in tokens: id = f"{len(jobs) + 1:032d}" tokens[token] = id jobs[id] = {"id": id, "body": body, "polls": 0, "state": "running"} # Simulate an accepted POST whose response is lost. The # retry must reuse the token, not allocate another job. if mobile and not lost[0]: lost[0] = True route.abort("connectionfailed") elif body["text"] == hold_text[0]: hold_text[0] = None held.append((route, describe(jobs[tokens[token]]))) else: reply(route, describe(jobs[tokens[token]])) elif re.fullmatch(r"/tts/jobs/[0-9]{32}", path): job = jobs[path.rsplit("/", 1)[-1]] if request.method == "DELETE": job["cancelled"] = True deleted.append(job["id"]) else: job["polls"] += 1 reply(route, describe(job)) elif re.fullmatch(r"/tts/audio/[0-9]{32}/parts/[0-9]+\.mp3", path): audio_requests.append(path) data, status, headers = audio, 200, {"Cache-Control": "private, max-age=300"} match = re.fullmatch(r"bytes=(\d+)-(\d*)", request.headers.get("range", "")) if match: start = int(match[1]) end = min(len(audio) - 1, int(match[2]) if match[2] else len(audio) - 1) data, status = audio[start:end + 1], 206 headers["Content-Range"] = f"bytes {start}-{end}/{len(audio)}" route.fulfill(status=status, content_type="audio/mpeg", body=data, headers=headers) else: reply(route, {"error": {"code": "not_mocked", "message": path}}, 404) context.route(BASE + "/**", handler) page = context.new_page() page.set_default_timeout(60000) page.on("pageerror", lambda error: page_errors.append(str(error))) page.add_init_script('''document.addEventListener("DOMContentLoaded", () => { const audio=document.querySelector("#audio");audio.muted=true; });''') try: page.goto(BASE + "/tts/", wait_until="domcontentloaded") page.wait_for_function('!document.querySelector("#generate").disabled') assert page.evaluate('document.documentElement.scrollWidth <= innerWidth + 1') assert page.evaluate('localStorage.getItem(ReaderPlan.bookmarkKey)') is None text = "这是一篇长文，校长文章写得很好，我得亲自处理。😀\n" * 42 + "原文不是HTML" page.locator("#speech-text").fill(text) page.locator("#pitch").evaluate("node=>{node.value=-2;node.dispatchEvent(new Event('input',{bubbles:true}));}") assert page.locator("#pitch-value").text_content().startswith("-2") page.locator("#playback-rate").evaluate("node=>{node.value=2;node.dispatchEvent(new Event('input',{bubbles:true}));}") page.locator("#generate").click() page.wait_for_function('document.querySelector("#audio").currentTime > .2 && !document.querySelector("#audio").paused') assert page.evaluate('session.jobs.get(session.batch).state') == "running" assert page.locator("#pitch").is_disabled() assert all(body["pitch_semitones"] == -2 for body in requests) page.locator("#play").click() assert page.locator("#audio").evaluate("node => node.paused") assert page.locator("#reader-source").text_content() == text assert page.locator("#reader-source b").count() == 0 current = page.evaluate("({start:session.plan[session.current.batch].start+session.current.part.start,end:session.plan[session.current.batch].start+session.current.part.end})") assert page.locator("#current-source").text_content() == text.encode('utf-16-le')[current['start']*2:current['end']*2].decode('utf-16-le') assert page.locator('.source-part[aria-current="location"]').count() == 1 page.wait_for_function("session && !session.ticking") post_count, clip_count = len(requests), len(audio_requests) page.wait_for_timeout(1400) assert len(requests) == post_count and len(audio_requests) == clip_count assert page.evaluate("!preloader.hasAttribute('src') && !session.wantsPlay") count = len(requests) page.locator("#playback-rate").evaluate("node=>{node.value=1.75;node.dispatchEvent(new Event('input',{bubbles:true}));}") assert page.locator("#audio").evaluate("node=>node.playbackRate") == 1.75 assert len(requests) == count assert all(body['speed'] == 1 for body in requests) page.locator("#playback-rate").evaluate("node=>{node.value=2;node.dispatchEvent(new Event('input',{bubbles:true}));}") page.locator("#remember").check() page.evaluate("checkpoint(true)") saved = page.evaluate("JSON.parse(localStorage.getItem(ReaderPlan.bookmarkKey))") assert saved["text"] == text and saved["seconds"] > 0 assert saved["playbackRate"] == 2 assert saved["pitchSemitones"] == -2 assert "audio_url" not in saved and "id" not in saved if not mobile: expired_id = page.evaluate("session.jobs.get(session.batch).id") jobs[expired_id]["expired"] = True page.wait_for_function("session.jobs.get(session.batch).id === ''") page.wait_for_function("session && !session.ticking") count = len(requests) page.wait_for_timeout(1400) assert len(requests) == count, "Expired paused jobs must not automatically resubmit" assert page.locator("#audio").evaluate("node => node.paused"), "Recovery must respect a user pause" page.locator("#play").click() page.wait_for_function("session && session.batch >= 1 && session.current && !document.querySelector('#audio').paused") assert page.evaluate("session.jobs.size") <= 2 assert len({body["request_id"] for body in requests}) >= 2 if mobile: assert requests[0]["request_id"] == requests[1]["request_id"] # Cancel an incomplete speculative batch on native pause; # the current batch is retained and no third batch is queued. page.wait_for_function("session.jobs.get(session.batch+1)?.id && session.jobs.get(session.batch+1)?.state === 'running'") prefetched = page.evaluate("session.jobs.get(session.batch+1).id") page.locator("#audio").evaluate("node=>node.pause()") page.wait_for_function("!session.wantsPlay") page.wait_for_function("!session.jobs.has(session.batch+1)") page.wait_for_function("id=>!session.ticking", arg=prefetched) page.wait_for_timeout(200) assert prefetched in deleted assert page.evaluate("session.jobs.size") <= 1 # Previous/next and direct cross-batch source clicks preserve # a user pause and do not create tasks or fetch another MP3. navigation = page.evaluate("session.navigationIndex") page.locator("#next").click() assert page.evaluate("session.navigationIndex") == navigation + 1 page.locator("#previous").click() assert page.evaluate("session.navigationIndex") == navigation target = page.evaluate("session.locations.findIndex(part=>part.batch===session.batch+1)") assert target >= 0 page.locator(f'.source-part[data-location="{target}"]').click() page.wait_for_function("session && !session.ticking") count, clips = len(requests), len(audio_requests) page.wait_for_timeout(1400) assert len(requests) == count and len(audio_requests) == clips assert page.evaluate("session.navigationIndex") == target assert page.evaluate("!session.current && !session.wantsPlay") selected = page.evaluate("session.locations[session.navigationIndex].start") page.locator("#locate").click() page.locator("#play").click() page.wait_for_function("session && session.current && !document.querySelector('#audio').paused") assert page.evaluate("session.plan[session.current.batch].start+session.current.part.start") == selected # Rapid in-session seeks must not allow stale job responses # to change the selected source or continue obsolete work. page.locator("#next").click() page.locator("#previous").click() page.wait_for_function("session && session.current && !document.querySelector('#audio').paused") assert page.evaluate("session.plan[session.current.batch].start+session.current.part.start") == selected page.evaluate('Object.defineProperty(document,"hidden",{configurable:true,get:()=>true});document.dispatchEvent(new Event("visibilitychange"))') assert page.evaluate("session !== null") assert not page.locator("#audio").evaluate("node => node.paused") page.locator("#play").click() page.evaluate("checkpoint(true)") saved = page.evaluate("JSON.parse(localStorage.getItem(ReaderPlan.bookmarkKey))") post_count = len(requests) page.reload(wait_until="domcontentloaded") page.wait_for_function('!document.querySelector("#resume").disabled') assert len(requests) == post_count, "Restoring text must not automatically submit it" assert page.locator("#speech-text").input_value() == text assert page.locator("#playback-rate").input_value() == "2" assert page.locator("#pitch").input_value() == "-2" page.locator("#resume").click() page.wait_for_function("session && session.current && document.querySelector('#audio').readyState >= 2") assert page.evaluate("session.plan[session.current.batch].start + session.current.part.start") == saved["offset"] page.wait_for_function("document.querySelector('#audio').currentTime > .1") page.locator("#forget").click() assert page.evaluate("localStorage.getItem(ReaderPlan.bookmarkKey)") is None page.locator("#stop").click() assert not page.locator("#audio").evaluate("node => node.hasAttribute('src')") assert not page.locator("#pitch").is_disabled() # Hold a speculative POST response after server acceptance, # then pause before the capability comes back to the reader. # Its late response must be released, not adopted/replayed. hold_text[0] = page.evaluate("ReaderPlan.split(document.querySelector('#speech-text').value)[1].text") page.locator("#generate").click() page.wait_for_function("session && session.current && session.jobs.has(session.batch+1) && session.ticking && !session.jobs.get(session.batch+1).id") assert len(held) == 1 page.locator("#play").click() route, late = held.pop() reply(route, late) page.wait_for_function("session && !session.ticking") page.wait_for_timeout(200) assert late["id"] in deleted assert page.evaluate("!session.wantsPlay && session.jobs.size===1 && !preloader.hasAttribute('src')") page.locator("#stop").click() # Sleep mode stops at the exact current clip anchor, saves # the next original source position and releases owned work. page.locator("#generate").click() page.wait_for_function("session && session.current && document.querySelector('#audio').currentTime>.1") assert page.locator("#audio").evaluate("node=>node.playbackRate") == 2 page.locator("#remember").check() page.locator("#sleep-mode").select_option("clip") sleep_end = page.evaluate("session.plan[session.current.batch].start+session.current.part.end") page.locator("#audio").evaluate("node=>{node.currentTime=node.duration-.05;}") page.wait_for_function("session===null && document.querySelector('#status').textContent.includes('睡眠定时')") assert page.evaluate("bookmark.offset") == sleep_end assert page.evaluate("!preloader.hasAttribute('src') && sleep.mode==='off'") page.locator("#resume").click() page.wait_for_function("session && session.current && document.querySelector('#audio').currentTime>.1") assert page.evaluate("session.plan[session.current.batch].start+session.current.part.start") == sleep_end page.locator("#play").click() page.locator("#sleep-mode").select_option("5") assert page.evaluate("sleep.remaining()") <= 300 assert page.locator("#audio").evaluate("node=>node.paused") # Inject only the pure timer's clock; no actual five-minute # wait or fake inference is added to the default suite. page.evaluate("sleep.clock=()=>sleep.deadline;document.dispatchEvent(new Event('visibilitychange'))") page.wait_for_function("session===null") assert page.locator("#status").text_content().startswith("睡眠定时") assert page.evaluate("!document.querySelector('#audio').hasAttribute('src') && sleep.mode==='off'") assert page.evaluate("!('mediaSession' in navigator) || navigator.mediaSession.playbackState==='none'") page.locator("#forget").click() assert not page_errors, page_errors results["cases"].append({"mobile_viewport": mobile, "early_play_before_completion": True, "continuous_cross_batch": True, "bounded_prefetch": True, "bookmark_opt_in_reload_resume_clear": True, "lost_post_idempotent_retry": mobile, "expired_audio_recovery_while_paused": not mobile, "source_highlight_next_previous_click_seek": True, "pause_no_new_posts_or_preloads": True, "unfinished_prefetch_cancelled": True, "rapid_seek_stale_response_guard": True, "pause_during_post_late_capability_released": True, "instant_playback_rate_no_resynthesis_and_bookmark": True, "pitch_request_snapshot_and_opt_in_bookmark": True, "sleep_current_clip_stops_before_next_and_resumes": True, "sleep_deadline_while_paused_releases_work": True, "visibility_change_does_not_stop": True, "page_errors": page_errors}) print("Reader mobile" if mobile else "Reader desktop", "PASS", flush=True) finally: context.close() finally: browser.close() args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(results, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()