diff --git a/archive/threads/NOTimothyLottes/2023-08-18-the-subpix-version-is-definitely-better-because/index.md b/archive/threads/NOTimothyLottes/2023-08-18-the-subpix-version-is-definitely-better-because/index.md index 06096d1..aa174f1 100644 --- a/archive/threads/NOTimothyLottes/2023-08-18-the-subpix-version-is-definitely-better-because/index.md +++ b/archive/threads/NOTimothyLottes/2023-08-18-the-subpix-version-is-definitely-better-because/index.md @@ -1,11 +1,11 @@ --- -title: "The subpix version is definitely better." +title: "SPC: CRT shader on the Deck." type: archive source: twitter -source_url: "https://x.com/NOTimothyLottes/status/1692565070583136348" +source_url: "https://x.com/NOTimothyLottes/status/1692559239116406864" author: "NOTimothyLottes" handle: NOTimothyLottes -post_id: "1692565070583136348" +post_id: "1692559239116406864" date: 2023-08-18 archived: 2026-08-23 status: draft @@ -14,20 +14,44 @@ tags: - archive - twitter - NOTimothyLottes -description: "The subpix version is definitely better." +description: "SPC: CRT shader on the Deck." in_reply_to: "" --- ## Source -- URL: https://x.com/NOTimothyLottes/status/1692565070583136348 +- URL: https://x.com/NOTimothyLottes/status/1692559239116406864 - Author: NOTimothyLottes (@NOTimothyLottes) -- Posted: 2023-08-18 15:52:13 +- Posted: 2023-08-18 15:29:03 ## Thread **1/** +SPC: CRT shader on the Deck. This time not doing subpix render. Yet RGB vs BGR order matters still. One way it becomes possible to simulate even spaced aperture grill (removing scan fx), and the other more of the scan effect. + +![](https://files.catbox.moe/mve2kr.jpg) + +![](https://files.catbox.moe/vis95s.jpg) + +**2/** + +Comparing all methods (grille, bad convergence scan, subpix scan) + +![](https://files.catbox.moe/mve2kr.jpg) + +![](https://files.catbox.moe/vis95s.jpg) + +![](https://files.catbox.moe/34pnhf.jpg) + +**3/** + +The non-subpix scan has 1-pixel separation of RGB (intentional bad convergence) to help hide scaling. The subpix scan uses 1/3-pixel separation (less visible misaligned convergence). Pixel as in output pixel. + +**4/** + The subpix version is definitely better. Because of the almost 3x doubling of scanline width detail, it just perceptually feels a lot cleaner and easier to reconstruct in the mind. -Media (not lifted): `1692565070583136348_F30y9HPWUAAbmmw_orig.jpg` `1692565070583136348_F30y9h3WAAETuui_orig.jpg` +![](https://files.catbox.moe/ugcsjy.jpg) + +![](https://files.catbox.moe/g42t2i.jpg) diff --git a/scripts/twitter/README.md b/scripts/twitter/README.md index 8857951..ab2c55b 100644 --- a/scripts/twitter/README.md +++ b/scripts/twitter/README.md @@ -1,21 +1,66 @@ # twitter -Emit Threadwell archive notes from a `thread_data.json` dump. No fetch, no lift. +Emit Threadwell archive notes from a `thread_data.json` dump. No fetch. Lift and CRT snapshot are still manual. -## One dump +## Emit ``` python C:\projects\Threadwell\scripts\twitter\emit_archive.py --input C:\projects\manual_slop\docs\twitter\2064858927829745887 --vault C:\projects\Threadwell --slug steamos-init --archived 2026-08-23 ``` -`--input` is the dump directory. `--vault` is the Threadwell root. `--slug` overrides the directory slug. `--archived` is `YYYY-MM-DD` and defaults to today. +`--tip ` climbs same-handle `reply_to` and uses that chain as `index.md`. -## All dumps +`--all` skips a dump whose `post_id` already exists unless `--force`. + +`--force` rewrites `media.json` to orig-only. Restore CRT/OCR rows with `media_merge.py`. + +## Tip graph (no vault writes) ``` -python C:\projects\Threadwell\scripts\twitter\emit_archive.py --all C:\projects\manual_slop\docs\twitter --vault C:\projects\Threadwell --archived 2026-08-23 +python C:\projects\Threadwell\scripts\twitter\graph_dry_run.py --input C:\projects\manual_slop\docs\twitter\ ``` -Skips a dump whose `post_id` already exists in the vault unless `--force`. No fetch, no lift. +## CRT (stock slang only) -The converter writes the spine and branch notes under `archive/threads///` and a working copy under `assets/threads///` (`thread_data.json`, `media.json`, `gaps.md`, and any local media). Media is not published. +`crt_pass.py` nearest-scales, then runs **unpatched** `crt-lottes-multipass.slangp` / `crt-Cyclon.slangp` via `librashader-cli`. It does not edit shaders. ShaderGlass snapshot is still the look reference. + +``` +python C:\projects\Threadwell\scripts\twitter\crt_pass.py --in ORIG.png --out CRT.png --preset cyclon --nn 3 --shader-scale 100% --dry-run +``` + +`--nn 3` is Affinity 300%. `--shader-scale` is librashader `-d`. `--shaderglass` only launches the .sgp. + +If the plate is wrong, snapshot ShaderGlass yourself and drop `*_crt.png` next to the orig, then `media_merge.py`. + +## OCR + +``` +python C:\projects\Threadwell\scripts\twitter\ocr_pass.py --in ORIG.png --media-json THREAD\media.json +``` + +Prefers `umi-ocr` (`scoop install extras/umi-ocr`), then tesseract, then Windows.Media.Ocr. Writes `*_ocr.txt`. ShareX has no OCR CLI. + +## Select CRT vs orig in the note + +``` +python C:\projects\Threadwell\scripts\twitter\media_embed.py --thread ASSETS_THREAD --notes ARCHIVE_THREAD --media-id GBgHT_2WIAAKuNq --show crt +python C:\projects\Threadwell\scripts\twitter\media_embed.py --thread ASSETS_THREAD --notes ARCHIVE_THREAD --media-id GBgHT_2WIAAKuNq --attach-ocr +``` + +## Lift to catbox + +``` +python C:\projects\Threadwell\scripts\twitter\lift_catbox.py --thread ASSETS_THREAD --notes ARCHIVE_THREAD --orig +``` + +Reads `[catbox] userhash` from `secrets/credentials.toml`. Never prints it. Rewrites `Media (not lifted): \`file\`` to `![](https://files.catbox.moe/…)`. + +## Restore extra media.json rows + +``` +python C:\projects\Threadwell\scripts\twitter\media_merge.py --thread ASSETS_THREAD +``` + +## Frozen Onat + +`do_not_refetch.txt` — do not overwrite those three dumps. diff --git a/scripts/twitter/crt_pass.py b/scripts/twitter/crt_pass.py new file mode 100644 index 0000000..0582650 --- /dev/null +++ b/scripts/twitter/crt_pass.py @@ -0,0 +1,314 @@ +"""Apply a stock RetroArch slang preset to a still via librashader-cli. + +Does not patch shaders. Does not invent PixelSize. Your ShaderGlass .sgp files +are the look reference; this only runs the same .slangp with the listed params. + +Manual look you are matching: + 1. Nearest-upscale the orig (Affinity zoom). 300% => 3x. + 2. Overlay ShaderGlass with the .sgp (Cyclon PixelSize 6, Lottes PixelSize 4). + 3. Snapshot, then crop. + +This script can do step 1 (ffmpeg neighbor) and a stock slang render of the +result. It cannot reproduce ShaderGlass window capture. If the plate is wrong, +do it in ShaderGlass and drop the PNG next to the orig as *_crt.png. + +Presets: + lottes crt-lottes-multipass.slangp + LOTTES_CODE_(LOTTES_MULTIPASS 2X) V2.sgp + cyclon crt-Cyclon.slangp + LOTTES_CRT_CODE_(CYCLON 6X).sgp + +auto: cyclon if max(w,h) <= --auto-max (default 720), else lottes. +""" +from __future__ import annotations + +import argparse +import json +import shutil +import subprocess +import sys +from pathlib import Path + +LOTTES_SGP = Path( + r"C:\Users\Ed\Downloads\ShaderGlass-1.2.2-win-x64" + r"\LOTTES_CODE_(LOTTES_MULTIPASS 2X) V2.sgp" +) +CYCLON_SGP = Path( + r"C:\Users\Ed\Downloads\ShaderGlass-1.2.2-win-x64" + r"\LOTTES_CRT_CODE_(CYCLON 6X).sgp" +) +SLANG_LOTTES = Path( + r"C:\Users\Ed\scoop\apps\retroarch\1.22.2\shaders" + r"\shaders_slang\crt\crt-lottes-multipass.slangp" +) +SLANG_CYCLON = Path( + r"C:\Users\Ed\scoop\apps\retroarch\1.22.2\shaders" + r"\shaders_slang\crt\crt-Cyclon.slangp" +) +CLI_DEFAULT = Path(r"C:\Users\Ed\scoop\apps\librashader\0.12.0\librashader-cli.exe") + +# Knobs copied from the .sgp files. Not a substitute for PixelSize. +LOTTES_PARAMS = ( + "hardScan=-19,hardPix=-7,warpX=0,warpY=0,maskDark=1.3,maskLight=1.3," + "shadowMask=4,brightBoost=1.15,bloomAmount=0,shape=1.9" +) +CYCLON_PARAMS = ( + "SCANLINE=0.2,INTERLACE=0,M_TYPE=0,SLOT=0,SLOTW=2,Maskl=0.55,Maskh=1," + "bzl=0,ambient=0,REFLECT=0,WARPX=0,WARPY=0,BR_DEP=0,BRIGHTNESS_=1.05," + "BLACK=0.01,vig=0" +) + + +def probe_size(path: Path) -> tuple[int, int]: + raw = subprocess.check_output( + [ + "ffprobe", + "-v", + "error", + "-select_streams", + "v:0", + "-show_entries", + "stream=width,height", + "-of", + "json", + str(path), + ], + text=True, + ) + stream = json.loads(raw)["streams"][0] + return int(stream["width"]), int(stream["height"]) + + +def pick_preset(width: int, height: int, name: str, auto_max: int) -> tuple[str, Path, Path, str]: + if name == "auto": + name = "cyclon" if max(width, height) <= auto_max else "lottes" + if name == "lottes": + return name, LOTTES_SGP, SLANG_LOTTES, LOTTES_PARAMS + if name == "cyclon": + return name, CYCLON_SGP, SLANG_CYCLON, CYCLON_PARAMS + raise SystemExit(f"unknown preset {name}") + + +def find_cli(explicit: Path | None) -> Path: + if explicit is not None: + if not explicit.is_file(): + raise SystemExit(f"missing --cli {explicit}") + return explicit + which = shutil.which("librashader-cli") + if which: + return Path(which) + if CLI_DEFAULT.is_file(): + return CLI_DEFAULT + raise SystemExit( + "librashader-cli not on PATH. Install or pass --cli. " + "Stock binary lives at " + r"C:\Users\Ed\scoop\apps\librashader\0.12.0\librashader-cli.exe" + ) + + +def parse_crop(spec: str) -> str | None: + if spec in ("none", "", "off"): + return None + if spec == "auto": + return "auto" + parts = [int(p) for p in spec.replace(" ", "").split(",")] + if len(parts) != 4: + raise SystemExit("--crop wants auto, none, or L,T,R,B") + left, top, right, bottom = parts + return f"crop=iw-{left + right}:ih-{top + bottom}:{left}:{top}" + + +def detect_crop(path: Path) -> str | None: + proc = subprocess.run( + [ + "ffmpeg", + "-hide_banner", + "-i", + str(path), + "-vf", + "cropdetect=limit=24:round=2:reset=0", + "-frames:v", + "1", + "-f", + "null", + "-", + ], + capture_output=True, + text=True, + ) + crop = None + for line in (proc.stderr or "").splitlines(): + if "crop=" in line: + crop = line.rsplit("crop=", 1)[-1].strip() + return f"crop={crop}" if crop else None + + +def run_ffmpeg(src: Path, dest: Path, vf: str) -> None: + dest.parent.mkdir(parents=True, exist_ok=True) + subprocess.check_call( + [ + "ffmpeg", + "-y", + "-hide_banner", + "-loglevel", + "error", + "-i", + str(src), + "-vf", + vf, + "-frames:v", + "1", + "-update", + "1", + str(dest), + ] + ) + + +def run_librashader( + cli: Path, + slang: Path, + src: Path, + dest: Path, + *, + runtime: str, + dimensions: str, + params: str, +) -> None: + dest.parent.mkdir(parents=True, exist_ok=True) + cmd = [ + str(cli), + "render", + "-r", + runtime, + "-d", + dimensions, + "--params", + params, + "-p", + str(slang), + "-i", + str(src), + "-o", + str(dest), + ] + print(" ".join(cmd)) + subprocess.check_call(cmd) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--in", dest="src", type=Path, required=True) + parser.add_argument("--out", dest="dest", type=Path, required=True) + parser.add_argument("--preset", choices=("auto", "lottes", "cyclon"), default="auto") + parser.add_argument( + "--nn", + type=int, + default=1, + help="nearest integer pre-scale (Affinity zoom). 1 = none, 3 = 300%%", + ) + parser.add_argument( + "--shader-scale", + default="100%", + help="librashader -d on the (possibly nn-scaled) image. 100%%, 300%%, 600%%, or WIDTHxHEIGHT", + ) + parser.add_argument("--runtime", default="d3d11") + parser.add_argument("--auto-max", type=int, default=720) + parser.add_argument("--crop", default="none", help="auto | none | L,T,R,B") + parser.add_argument("--cli", type=Path, default=None) + parser.add_argument("--dry-run", action="store_true") + parser.add_argument( + "--shaderglass", + action="store_true", + help="launch ShaderGlass with the matching .sgp (manual snapshot)", + ) + parser.add_argument( + "--nn-only", + action="store_true", + help="only do the neighbor pre-scale; do not call librashader", + ) + args = parser.parse_args(argv) + + if not args.src.is_file(): + raise SystemExit(f"missing {args.src}") + if args.nn < 1: + raise SystemExit("--nn must be >= 1") + + width, height = probe_size(args.src) + name, sgp, slang, params = pick_preset(width, height, args.preset, args.auto_max) + print(f"size {width}x{height} preset={name} nn={args.nn} shader-scale={args.shader_scale}") + print(f"sgp {sgp}") + print(f"slang {slang}") + + work = args.src + tmp_nn: Path | None = None + if args.nn > 1: + tmp_nn = args.dest.with_name(args.dest.stem + f"_nn{args.nn}.png") + vf = f"scale=iw*{args.nn}:ih*{args.nn}:flags=neighbor,format=rgba" + print("ffmpeg", "-i", work, "-vf", vf, tmp_nn) + if not args.dry_run: + run_ffmpeg(work, tmp_nn, vf) + work = tmp_nn + + if args.nn_only: + if tmp_nn is None: + raise SystemExit("--nn-only needs --nn > 1") + if not args.dry_run: + shutil.copy2(tmp_nn, args.dest) + print(f"wrote {args.dest} (nn only)") + return 0 + + cli = None if args.dry_run else find_cli(args.cli) + if args.dry_run: + print( + "librashader-cli render", + "-r", + args.runtime, + "-d", + args.shader_scale, + "--params", + params, + "-p", + slang, + "-i", + work, + "-o", + args.dest, + ) + else: + assert cli is not None + if not slang.is_file(): + raise SystemExit(f"missing slang {slang}") + run_librashader( + cli, + slang, + work, + args.dest, + runtime=args.runtime, + dimensions=args.shader_scale, + params=params, + ) + + crop_spec = parse_crop(args.crop) + if crop_spec == "auto" and not args.dry_run: + crop_filter = detect_crop(args.dest) + if crop_filter: + cropped = args.dest.with_name(args.dest.stem + "_crop.png") + run_ffmpeg(args.dest, cropped, crop_filter) + shutil.move(str(cropped), str(args.dest)) + elif crop_spec and crop_spec != "auto" and not args.dry_run: + cropped = args.dest.with_name(args.dest.stem + "_crop.png") + run_ffmpeg(args.dest, cropped, crop_spec) + shutil.move(str(cropped), str(args.dest)) + + print(f"wrote {args.dest}") + if args.shaderglass: + exe = shutil.which("shaderglass") or shutil.which("ShaderGlass") + if not exe: + raise SystemExit("shaderglass not on PATH") + subprocess.Popen([exe, str(sgp)]) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/twitter/lift_catbox.py b/scripts/twitter/lift_catbox.py new file mode 100644 index 0000000..3db87fc --- /dev/null +++ b/scripts/twitter/lift_catbox.py @@ -0,0 +1,150 @@ +"""Upload curated stills to catbox and rewrite archive notes to HTTPS. + +Reads userhash from vault secrets/credentials.toml [catbox]. Never prints it. + + python lift_catbox.py --thread ASSETS_THREAD --notes ARCHIVE_THREAD --orig + +--orig marks every on-disk orig row publish=true then uploads those. +Default without --orig: only rows already publish=true. + +After upload, any note token that is the local filename (backticked or +bare) or a previous url for that row becomes ![](https://files.catbox.moe/…). +""" +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +import tomllib +from pathlib import Path + +API = "https://catbox.moe/user/api.php" +CREDS = Path("secrets/credentials.toml") + + +def load_userhash(vault: Path) -> str: + path = vault / "secrets" / "credentials.toml" + if not path.is_file(): + raise SystemExit(f"missing {path}") + text = path.read_text(encoding="utf-8") + try: + data = tomllib.loads(text) + h = str((data.get("catbox") or {}).get("userhash") or "").strip() + except tomllib.TOMLDecodeError: + h = "" + in_catbox = False + for raw in text.splitlines(): + line = raw.strip() + if line.startswith("[") and line.endswith("]"): + in_catbox = line[1:-1].strip().lower() == "catbox" + continue + if in_catbox and line.lower().startswith("userhash"): + _, _, rest = line.partition("=") + h = rest.strip().strip("\"'") + break + if not h: + raise SystemExit("credentials.toml [catbox] userhash is empty") + return h + + +def upload(path: Path, userhash: str) -> str: + cmd = [ + "curl.exe", + "-sS", + "-F", + "reqtype=fileupload", + "-F", + f"userhash={userhash}", + "-F", + f"fileToUpload=@{path}", + API, + ] + raw = subprocess.check_output(cmd, text=True, stderr=subprocess.STDOUT) + url = raw.strip() + if not url.startswith("https://"): + raise SystemExit(f"catbox refused {path.name}: {url[:200]}") + return url + + +def rewrite_notes(notes_dir: Path, old_tokens: list[str], url: str) -> int: + embed = f"![]({url})" + changed = 0 + for path in notes_dir.glob("*.md"): + text = path.read_text(encoding="utf-8") + new = text + for token in old_tokens: + if not token: + continue + new = new.replace(f"`{token}`", embed) + new = new.replace(f"![]({token})", embed) + new = re.sub(r"^Media \(not lifted\):\s*", "", new, flags=re.M) + while True: + split = re.sub( + r"(!\[[^\]]*\]\([^)]+\))\s+(!\[[^\]]*\]\([^)]+\))", + r"\1\n\n\2", + new, + ) + if split == new: + break + new = split + if new != text: + path.write_text(new, encoding="utf-8", newline="\n") + changed += 1 + print(f"rewrote {path.name}") + return changed + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--vault", type=Path, default=Path(r"C:\projects\Threadwell")) + parser.add_argument("--thread", type=Path, required=True) + parser.add_argument("--notes", type=Path, required=True) + parser.add_argument( + "--orig", + action="store_true", + help="publish+upload every orig file that exists on disk", + ) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args(argv) + + media_path = args.thread / "media.json" + data = json.loads(media_path.read_text(encoding="utf-8")) + items = data.setdefault("items", []) + if args.orig: + for item in items: + if item.get("role") == "orig": + item["publish"] = True + if "embed" not in item: + item["embed"] = True + + userhash = None if args.dry_run else load_userhash(args.vault) + uploaded = 0 + for item in items: + if not item.get("publish"): + continue + name = str(item.get("filename") or "") + src = args.thread / name + if not src.is_file(): + print(f"skip missing {name}") + continue + old = [name, str(item.get("url") or "")] + if args.dry_run: + print(f"would upload {name}") + continue + assert userhash is not None + url = upload(src, userhash) + item["url"] = url + uploaded += 1 + print(f"uploaded {name} -> {url}") + rewrite_notes(args.notes, old, url) + + if not args.dry_run: + media_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + print(f"uploaded {uploaded}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/twitter/media_embed.py b/scripts/twitter/media_embed.py new file mode 100644 index 0000000..ae5209c --- /dev/null +++ b/scripts/twitter/media_embed.py @@ -0,0 +1,172 @@ +"""Select which still a note cites, and optionally attach OCR under it. + +Notes currently say: + + Media (not lifted): `1736…_GBgHT…_orig.png` + +After lift they will say: + + ![](https://files.catbox.moe/xxxx.png) + +This script only rewrites those two shapes. It does not upload. + + python media_embed.py --thread --notes --media-id GBgHT_2WIAAKuNq --show crt + python media_embed.py --thread --notes --media-id GBgHT_2WIAAKuNq --attach-ocr +""" +from __future__ import annotations + +import argparse +import json +import re +from pathlib import Path + +IMAGE_MD = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)") +LIFTED_LINE = re.compile(r"^Media \(not lifted\): (.+)$", re.M) + + +def load_media(path: Path) -> dict: + return json.loads(path.read_text(encoding="utf-8")) + + +def save_media(path: Path, data: dict) -> None: + path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + + +def group_items(items: list[dict]) -> dict[tuple[str, str], dict[str, dict]]: + groups: dict[tuple[str, str], dict[str, dict]] = {} + for item in items: + key = (str(item.get("post_id") or ""), str(item.get("media_id") or "")) + groups.setdefault(key, {})[str(item.get("role") or "")] = item + return groups + + +def set_embed(items: list[dict], media_id: str, show: str) -> None: + found = False + for item in items: + if str(item.get("media_id") or "") != media_id: + continue + if item.get("role") in ("orig", "crt", "crt_outline"): + item["embed"] = item.get("role") == show + found = True + if not found: + raise SystemExit(f"no orig/crt row for media_id {media_id}") + + +def token_for(item: dict) -> str: + url = item.get("url") + if url: + return str(url) + return str(item.get("filename") or "") + + +def rewrite_notes(notes_dir: Path, old: str, new: str) -> int: + if not old or old == new: + return 0 + changed = 0 + for path in notes_dir.glob("*.md"): + text = path.read_text(encoding="utf-8") + if old not in text: + continue + path.write_text(text.replace(old, new), encoding="utf-8") + changed += 1 + print(f"rewrote {path.name}") + return changed + + +def ocr_block(text: str) -> str: + body = text.strip("\n") + return ( + "
\n" + "Text from still\n" + "\n" + "```\n" + f"{body}\n" + "```\n" + "\n" + "
" + ) + + +def attach_ocr(notes_dir: Path, image_token: str, ocr_text: str) -> int: + block = ocr_block(ocr_text) + changed = 0 + for path in notes_dir.glob("*.md"): + text = path.read_text(encoding="utf-8") + if image_token not in text: + continue + if "Text from still" in text and image_token in text: + print(f"ocr already near token in {path.name}") + continue + # After the line that cites the still. + lines = text.splitlines(keepends=True) + out: list[str] = [] + i = 0 + inserted = False + while i < len(lines): + out.append(lines[i]) + if image_token in lines[i] and not inserted: + if not lines[i].endswith("\n"): + out[-1] = lines[i] + "\n" + out.append("\n") + out.append(block + "\n") + inserted = True + i += 1 + if inserted: + path.write_text("".join(out), encoding="utf-8") + changed += 1 + print(f"attached ocr in {path.name}") + return changed + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--thread", type=Path, required=True, help="assets/threads/.../ folder") + parser.add_argument("--notes", type=Path, required=True, help="archive/threads/.../ folder") + parser.add_argument("--media-id", required=True) + parser.add_argument("--show", choices=("orig", "crt"), default=None) + parser.add_argument("--attach-ocr", action="store_true") + args = parser.parse_args(argv) + + media_path = args.thread / "media.json" + data = load_media(media_path) + items = data["items"] + groups = group_items(items) + match = None + pair = None + for key, roles in groups.items(): + if key[1] == args.media_id: + match = key + pair = roles + break + if pair is None or match is None: + raise SystemExit(f"media_id {args.media_id} not in {media_path}") + + if args.show: + if args.show not in pair: + raise SystemExit(f"no role={args.show} for {args.media_id}") + old_item = next((pair[r] for r in ("crt", "orig") if pair.get(r, {}).get("embed")), None) + if old_item is None: + old_item = pair.get("orig") or pair.get("crt") + set_embed(items, args.media_id, args.show) + save_media(media_path, data) + new_item = pair[args.show] + rewrite_notes(args.notes, token_for(old_item), token_for(new_item)) + print(f"embed {args.media_id} -> {args.show} ({token_for(new_item)})") + + if args.attach_ocr: + ocr = pair.get("ocr") + if ocr is None: + raise SystemExit(f"no role=ocr for {args.media_id}; run ocr_pass.py first") + ocr_path = args.thread / str(ocr["filename"]) + if not ocr_path.is_file(): + raise SystemExit(f"missing {ocr_path}") + shown = next((pair[r] for r in ("crt", "orig") if pair.get(r, {}).get("embed")), None) + if shown is None: + shown = pair.get("orig") or pair.get("crt") + attach_ocr(args.notes, token_for(shown), ocr_path.read_text(encoding="utf-8")) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/twitter/media_merge.py b/scripts/twitter/media_merge.py new file mode 100644 index 0000000..d801b38 --- /dev/null +++ b/scripts/twitter/media_merge.py @@ -0,0 +1,83 @@ +"""Merge non-orig media.json rows back after emit --force. + +emit collect_media rewrites media.json with orig only. CRT / OCR files stay +on disk. Run this on the thread assets dir to restore those rows. + + python media_merge.py --thread C:\\projects\\Threadwell\\assets\\threads\\HANDLE\\DATE-SLUG +""" +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +ROLE_SUFFIX = { + "_crt.png": "crt", + "_crt_outline.png": "crt_outline", + "_denoise.png": "denoise", + "_ocr.txt": "ocr", +} + + +def role_of(name: str) -> str | None: + for suffix, role in ROLE_SUFFIX.items(): + if name.endswith(suffix): + return role + return None + + +def parse_ids(name: str) -> tuple[str, str]: + stem = name + for suffix in ROLE_SUFFIX: + if stem.endswith(suffix): + stem = stem[: -len(suffix)] + break + if stem.endswith("_orig"): + stem = stem[: -len("_orig")] + parts = stem.split("_", 1) + if len(parts) != 2: + return "", stem + return parts[0], parts[1] + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--thread", type=Path, required=True) + args = parser.parse_args(argv) + media_path = args.thread / "media.json" + data = json.loads(media_path.read_text(encoding="utf-8")) + items = list(data.get("items") or []) + have = {(i.get("filename"), i.get("role")) for i in items} + added = 0 + for path in sorted(args.thread.iterdir()): + role = role_of(path.name) + if role is None: + continue + if (path.name, role) in have: + continue + post_id, media_id = parse_ids(path.name) + handle = "" + for item in items: + if item.get("media_id") == media_id: + handle = str(item.get("handle") or "") + break + items.append({ + "post_id": post_id, + "media_id": media_id, + "handle": handle, + "role": role, + "filename": path.name, + "publish": False, + "url": None, + "embed": False, + }) + added += 1 + print(f"restore {path.name} role={role}") + data["items"] = items + media_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + print(f"added {added} -> {media_path}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/twitter/ocr_pass.py b/scripts/twitter/ocr_pass.py new file mode 100644 index 0000000..a58942d --- /dev/null +++ b/scripts/twitter/ocr_pass.py @@ -0,0 +1,215 @@ +"""OCR a still. Prefer Umi-OCR HTTP, then tesseract, then Windows.Media.Ocr. + +Writes UTF-8 text. Optional: add a role=ocr row to media.json. + +Umi-OCR must be running (HTTP :1224). Start Umi-OCR.exe once. Then: + + python ocr_pass.py --in still.png --engine umi + +Uses tbpu.parser=single_code (keep indent) and English. ShareX has no OCR CLI. +""" +from __future__ import annotations + +import argparse +import base64 +import json +import shutil +import subprocess +import sys +import urllib.error +import urllib.request +from pathlib import Path + +UMI_URL = "http://127.0.0.1:1224/api/ocr" + + +def find_umi() -> str | None: + return shutil.which("umi-ocr") or shutil.which("Umi-OCR") + + +def find_tesseract() -> str | None: + return shutil.which("tesseract") + + +def umi_http_up() -> bool: + try: + urllib.request.urlopen("http://127.0.0.1:1224/api/ocr/get_options", timeout=2) + return True + except (urllib.error.URLError, TimeoutError, OSError): + return False + + +def run_umi(src: Path, dest: Path) -> None: + dest.parent.mkdir(parents=True, exist_ok=True) + if not umi_http_up(): + exe = find_umi() + if exe: + subprocess.Popen([exe]) + raise SystemExit( + "Umi-OCR HTTP not on 127.0.0.1:1224. Start Umi-OCR.exe and retry." + ) + payload = { + "base64": base64.b64encode(src.read_bytes()).decode("ascii"), + "options": { + "ocr.language": "English", + "tbpu.parser": "single_code", + "ocr.maxSideLen": 999999, + "data.format": "text", + }, + } + req = urllib.request.Request( + UMI_URL, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + ) + print(f"umi-http {src.name}") + raw = urllib.request.urlopen(req, timeout=180).read() + res = json.loads(raw.decode("utf-8")) + code = res.get("code") + data = res.get("data") + if code == 100 and isinstance(data, str): + text = data + elif code == 101: + text = "" + elif isinstance(data, list): + text = "".join( + str(block.get("text", "")) + str(block.get("end", "")) for block in data + ) + else: + raise SystemExit(f"umi-ocr failed code={code} data={data!r}") + dest.write_text(text.replace("\r\n", "\n"), encoding="utf-8") + + +def run_tesseract(src: Path, dest: Path) -> None: + exe = find_tesseract() + if not exe: + raise SystemExit("tesseract not on PATH. scoop install tesseract") + dest.parent.mkdir(parents=True, exist_ok=True) + out_base = dest.with_suffix("") + cmd = [exe, str(src), str(out_base), "-l", "eng"] + print(" ".join(cmd)) + subprocess.check_call(cmd) + txt = out_base.with_suffix(".txt") + if txt != dest: + dest.write_text(txt.read_text(encoding="utf-8", errors="replace"), encoding="utf-8") + + +def run_windows_ocr(src: Path, dest: Path) -> None: + ps = r""" +$ErrorActionPreference = 'Stop' +Add-Type -AssemblyName System.Runtime.WindowsRuntime +$null = [Windows.Storage.StorageFile, Windows.Storage, ContentType = WindowsRuntime] +$null = [Windows.Graphics.Imaging.BitmapDecoder, Windows.Graphics.Imaging, ContentType = WindowsRuntime] +$null = [Windows.Media.Ocr.OcrEngine, Windows.Media.Ocr, ContentType = WindowsRuntime] +function Await($op) { + $asTask = [System.WindowsRuntimeSystemExtensions].GetMethods() | Where-Object { + $_.Name -eq 'AsTask' -and $_.GetParameters().Count -eq 1 + } | Select-Object -First 1 + $task = $asTask.MakeGenericMethod($op.GetType().GenericTypeArguments[0]).Invoke($null, @($op)) + $task.Wait() + $task.Result +} +$src = $args[0] +$dest = $args[1] +$file = Await ([Windows.Storage.StorageFile]::GetFileFromPathAsync($src)) +$stream = Await ($file.OpenAsync([Windows.Storage.FileAccessMode]::Read)) +$decoder = Await ([Windows.Graphics.Imaging.BitmapDecoder]::CreateAsync($stream)) +$bitmap = Await ($decoder.GetSoftwareBitmapAsync()) +$engine = [Windows.Media.Ocr.OcrEngine]::TryCreateFromUserProfileLanguages() +if ($null -eq $engine) { throw 'Windows OCR engine unavailable' } +$result = Await ($engine.RecognizeAsync($bitmap)) +[System.IO.File]::WriteAllText($dest, $result.Text, [System.Text.UTF8Encoding]::new($false)) +""" + dest.parent.mkdir(parents=True, exist_ok=True) + cmd = [ + "powershell", + "-NoProfile", + "-Command", + ps, + str(src.resolve()), + str(dest.resolve()), + ] + print("windows.media.ocr", src.name) + subprocess.check_call(cmd) + + +def append_media_row(media_json: Path, src: Path, dest: Path) -> None: + data = json.loads(media_json.read_text(encoding="utf-8")) + items = data.setdefault("items", []) + stem = src.name + post_id = "" + media_id = "" + handle = "" + for item in items: + if item.get("filename") == src.name or item.get("filename") == dest.name: + post_id = str(item.get("post_id") or "") + media_id = str(item.get("media_id") or "") + handle = str(item.get("handle") or "") + break + if not post_id and "_" in src.stem: + parts = src.stem.split("_") + post_id = parts[0] + media_id = parts[1] if len(parts) > 1 else "" + row = { + "post_id": post_id, + "media_id": media_id, + "handle": handle, + "role": "ocr", + "filename": dest.name, + "publish": False, + "url": None, + "embed": False, + "source_filename": src.name, + } + items = [i for i in items if not (i.get("role") == "ocr" and i.get("source_filename") == src.name)] + items.append(row) + data["items"] = items + media_json.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + print(f"updated {media_json}") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--in", dest="src", type=Path, required=True) + parser.add_argument("--out", dest="dest", type=Path, default=None) + parser.add_argument( + "--engine", + choices=("auto", "umi", "tesseract", "windows"), + default="auto", + ) + parser.add_argument("--media-json", type=Path, default=None) + args = parser.parse_args(argv) + + if not args.src.is_file(): + raise SystemExit(f"missing {args.src}") + dest = args.dest + if dest is None: + dest = args.src.with_name(args.src.stem.replace("_orig", "") + "_ocr.txt") + if dest.name == args.src.stem + "_ocr.txt" and args.src.stem.endswith("_orig"): + dest = args.src.with_name(args.src.stem[: -len("_orig")] + "_ocr.txt") + + engine = args.engine + if engine == "auto": + if find_umi(): + engine = "umi" + elif find_tesseract(): + engine = "tesseract" + else: + engine = "windows" + print(f"engine={engine}") + if engine == "umi": + run_umi(args.src, dest) + elif engine == "tesseract": + run_tesseract(args.src, dest) + else: + run_windows_ocr(args.src, dest) + + text = dest.read_text(encoding="utf-8", errors="replace") + print(f"wrote {dest} ({len(text)} chars)") + if args.media_json: + append_media_row(args.media_json, args.src, dest) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())