feat: adapt T-HUB features — folders schema, Instagram connector, outreach contacts, media RLS force
From the hacklab/T-HUB fork (diverged at a6843e9):
- 046_file_folders: folder trees (group + personal scope, RLS forced,
workspace-immutability trigger) + created_by on chat_media_archive
- Instagram connector wired into connect + publishing
- outreach contacts: per-collective DM tracking (todo/sent/replied/declined),
RLS forced, role-scoped permissions, audit without message text,
migration renumbered 041→047
- FORCE row-level security on media_assets
The J1 api agent's folder service/controller work continues separately.
This commit is contained in:
parent
57762f543c
commit
8f01765df3
25 changed files with 3054 additions and 4 deletions
61
apps/media-worker/scripts/install-windows-worker.ps1
Normal file
61
apps/media-worker/scripts/install-windows-worker.ps1
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
# Run in native Windows PowerShell as the user who will transcode on moxexleyhp1.
|
||||
[CmdletBinding()]
|
||||
param(
|
||||
[string]$JobRoot = 'C:\transcode-jobs',
|
||||
[string]$PythonExe
|
||||
)
|
||||
|
||||
$ErrorActionPreference = 'Stop'
|
||||
if ($env:OS -ne 'Windows_NT') {
|
||||
throw 'Run this installer on Windows directly, not inside WSL.'
|
||||
}
|
||||
|
||||
$WorkerScript = (Resolve-Path (Join-Path $PSScriptRoot 'windows-transcode-worker.py')).Path
|
||||
if (-not $PythonExe) {
|
||||
$PythonCommand = Get-Command python.exe -ErrorAction SilentlyContinue
|
||||
if (-not $PythonCommand) {
|
||||
$PythonCommand = Get-Command py.exe -ErrorAction SilentlyContinue
|
||||
}
|
||||
if (-not $PythonCommand) {
|
||||
throw 'Install Python 3.9+ for Windows, then rerun with -PythonExe C:\path\to\python.exe.'
|
||||
}
|
||||
$PythonExe = $PythonCommand.Source
|
||||
}
|
||||
# Resolve py.exe (if used) to the real interpreter for the scheduled task.
|
||||
$ResolvedPython = & $PythonExe -c 'import sys; assert sys.version_info >= (3, 9); print(sys.executable)'
|
||||
if ($LASTEXITCODE -ne 0 -or -not $ResolvedPython) {
|
||||
throw 'Could not run Python 3.9+. Pass -PythonExe with a working Windows interpreter.'
|
||||
}
|
||||
$PythonExe = ($ResolvedPython | Select-Object -Last 1).Trim()
|
||||
$FfmpegExe = (Get-Command ffmpeg.exe -ErrorAction Stop).Source
|
||||
$JobRoot = [System.IO.Path]::GetFullPath($JobRoot)
|
||||
$Inbox = Join-Path $JobRoot 'inbox'
|
||||
foreach ($Name in @('inbox', 'done', 'failed')) {
|
||||
New-Item -ItemType Directory -Force -Path (Join-Path $JobRoot $Name) | Out-Null
|
||||
}
|
||||
|
||||
$TaskName = 'TMBH-TranscodeWorker'
|
||||
$UserId = [System.Security.Principal.WindowsIdentity]::GetCurrent().Name
|
||||
$Arguments = '-u "{0}" --inbox "{1}" --ffmpeg "{2}"' -f $WorkerScript, $Inbox, $FfmpegExe
|
||||
$Action = New-ScheduledTaskAction -Execute $PythonExe -Argument $Arguments -WorkingDirectory $PSScriptRoot
|
||||
$AtLogon = New-ScheduledTaskTrigger -AtLogOn -User $UserId
|
||||
# No repetition duration means this trigger repeats indefinitely.
|
||||
$EveryFiveMinutes = New-ScheduledTaskTrigger -Once -At (Get-Date).AddMinutes(1) -RepetitionInterval (New-TimeSpan -Minutes 5)
|
||||
$Principal = New-ScheduledTaskPrincipal -UserId $UserId -LogonType Interactive -RunLevel Limited
|
||||
$Settings = New-ScheduledTaskSettingsSet -MultipleInstances IgnoreNew -ExecutionTimeLimit ([TimeSpan]::Zero) -StartWhenAvailable -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries
|
||||
|
||||
Register-ScheduledTask -TaskName $TaskName -Action $Action -Trigger @($AtLogon, $EveryFiveMinutes) -Principal $Principal -Settings $Settings -Description 'Poll JSON jobs and transcode with native Windows NVENC or Intel QSV.' -Force | Out-Null
|
||||
Start-ScheduledTask -TaskName $TaskName
|
||||
|
||||
Write-Host "Installed and started: $TaskName (logon + every 5 minutes; duplicate starts ignored)."
|
||||
Write-Host "Python: $PythonExe"
|
||||
Write-Host "FFmpeg: $FfmpegExe"
|
||||
Write-Host "Worker: $WorkerScript"
|
||||
Write-Host "Inbox: $Inbox"
|
||||
Write-Host "Done: $(Join-Path $JobRoot 'done'); failed: $(Join-Path $JobRoot 'failed') with <job>.json.err diagnostics."
|
||||
Write-Host 'Keep this Windows account logged on; the task uses its interactive GPU session.'
|
||||
Write-Host 'Keep the repository at this location, or rerun this installer after moving it.'
|
||||
Write-Host 'Publish jobs by writing a .tmp file, closing it, then renaming it to .json.'
|
||||
Write-Host "Status: Get-ScheduledTask -TaskName '$TaskName'"
|
||||
Write-Host "Stop: Stop-ScheduledTask -TaskName '$TaskName'"
|
||||
Write-Host "Manual tests: $(Join-Path $PSScriptRoot 'test-windows-worker.md')"
|
||||
116
apps/media-worker/scripts/test-windows-worker.md
Normal file
116
apps/media-worker/scripts/test-windows-worker.md
Normal file
|
|
@ -0,0 +1,116 @@
|
|||
# Native Windows transcode worker: manual checks on moxexleyhp1
|
||||
|
||||
Run these commands in **Windows PowerShell**, using Windows Python 3.9+ and
|
||||
the existing WinGet ffmpeg on PATH. Keep this Windows account logged on.
|
||||
|
||||
## Install
|
||||
|
||||
1. Open PowerShell in the Windows checkout of this repository.
|
||||
2. Run:
|
||||
|
||||
```powershell
|
||||
powershell.exe -NoProfile -ExecutionPolicy Bypass -File .\apps\media-worker\scripts\install-windows-worker.ps1
|
||||
Get-ScheduledTask -TaskName 'TMBH-TranscodeWorker'
|
||||
```
|
||||
|
||||
If Python is not found, add `-PythonExe 'C:\path\to\python.exe'` to the
|
||||
installer command. The optional `-JobRoot` changes all three queue folders.
|
||||
The task starts immediately, polls every 5 seconds, starts at logon, and
|
||||
is triggered every 5 minutes to recover if it exits. Duplicate starts are
|
||||
ignored; a per-inbox file lock also prevents competing manual workers.
|
||||
The installer can be rerun to update the task.
|
||||
|
||||
3. Verify `inbox`, `done`, and `failed` exist under `C:\transcode-jobs`.
|
||||
Generate a short input with video and audio:
|
||||
|
||||
```powershell
|
||||
ffmpeg -hide_banner -n -f lavfi -i 'testsrc2=size=1280x720:rate=30' -f lavfi -i 'sine=frequency=440:sample_rate=48000' -t 10 -c:v mpeg4 -q:v 3 -c:a aac 'C:\transcode-jobs\sample.mp4'
|
||||
if ($LASTEXITCODE -ne 0) { throw 'Sample generation failed' }
|
||||
```
|
||||
|
||||
Use fresh output/job names when repeating these checks. Existing output
|
||||
files are rejected and preserved. Relative input/output paths resolve
|
||||
against the inbox; absolute Windows paths are recommended. Outputs use
|
||||
their extension to choose the container. Jobs run sequentially.
|
||||
|
||||
## NVENC H.264 on the RTX 3090
|
||||
|
||||
- [ ] Publish a job atomically, so polling never reads a partly written JSON:
|
||||
|
||||
```powershell
|
||||
@{ input = 'C:\transcode-jobs\sample.mp4'; output = 'C:\transcode-jobs\nvenc.mp4'; preset = 'nvenc_h264'; crf = 23 } | ConvertTo-Json | Set-Content -Encoding UTF8 'C:\transcode-jobs\inbox\nvenc.tmp'
|
||||
Move-Item 'C:\transcode-jobs\inbox\nvenc.tmp' 'C:\transcode-jobs\inbox\nvenc.json'
|
||||
```
|
||||
|
||||
- [ ] Within a poll interval plus encode time, `done\nvenc.json` and a
|
||||
playable, nonempty `C:\transcode-jobs\nvenc.mp4` exist; the inbox job is gone.
|
||||
- [ ] Inspect the output:
|
||||
|
||||
```powershell
|
||||
ffprobe -v error -show_entries stream=codec_name,codec_type -of json 'C:\transcode-jobs\nvenc.mp4'
|
||||
```
|
||||
|
||||
Expect H.264 video and AAC audio. While encoding, observe NVIDIA's **Video
|
||||
Encode** graph in Task Manager (use a longer input if needed).
|
||||
- [ ] Repeat with `preset = 'nvenc_hevc'`, a new job name, and output
|
||||
`nvenc-hevc.mp4`; expect HEVC video and AAC audio. If the installed ffmpeg
|
||||
lacks `hevc_nvenc`, expect a failed job with encoder diagnostics instead.
|
||||
|
||||
## Intel QuickSync H.264
|
||||
|
||||
- [ ] Publish a QSV job, omitting `crf` to exercise the default quality of 23:
|
||||
|
||||
```powershell
|
||||
@{ input = 'C:\transcode-jobs\sample.mp4'; output = 'C:\transcode-jobs\qsv.mp4'; preset = 'qsv_h264' } | ConvertTo-Json | Set-Content -Encoding UTF8 'C:\transcode-jobs\inbox\qsv.tmp'
|
||||
Move-Item 'C:\transcode-jobs\inbox\qsv.tmp' 'C:\transcode-jobs\inbox\qsv.json'
|
||||
```
|
||||
|
||||
- [ ] Verify `done\qsv.json`, playable `qsv.mp4`, and H.264/AAC streams using
|
||||
the ffprobe command above with the QSV output path.
|
||||
- [ ] Observe the Intel GPU's **Video Encode** graph during the job. Encoder
|
||||
initialization failures belong in `failed` and must include ffmpeg's error;
|
||||
the worker never retries with a software video encoder.
|
||||
|
||||
## Failure and continued polling
|
||||
|
||||
- [ ] Queue a missing input:
|
||||
|
||||
```powershell
|
||||
@{ input = 'C:\transcode-jobs\does-not-exist.mp4'; output = 'C:\transcode-jobs\must-not-exist.mp4'; preset = 'nvenc_h264' } | ConvertTo-Json | Set-Content -Encoding UTF8 'C:\transcode-jobs\inbox\missing.tmp'
|
||||
Move-Item 'C:\transcode-jobs\inbox\missing.tmp' 'C:\transcode-jobs\inbox\missing.json'
|
||||
```
|
||||
|
||||
- [ ] Expect `failed\missing.json` and `failed\missing.json.err`, mentioning
|
||||
the missing input. No output is created and the inbox job is gone.
|
||||
- [ ] Queue a text file as input with a valid preset. Expect a failed JSON
|
||||
and `.json.err` containing ffmpeg's nonzero exit code and stderr. No final
|
||||
output or `.part.mp4` file should remain.
|
||||
- [ ] Queue malformed JSON, an unknown preset, and a string `crf` in separate
|
||||
jobs. Each must move to `failed` with a useful `.err` file.
|
||||
- [ ] Queue an existing output path (e.g. the sample). Expect failure without
|
||||
changing the existing file.
|
||||
- [ ] Queue another valid job with a fresh output name; verify completion
|
||||
after the failed jobs. Reuse a failed job filename and verify both archived
|
||||
jobs and their errors survive (a UUID is added on name collisions).
|
||||
|
||||
## Task lifecycle and foreground diagnostics
|
||||
|
||||
- [ ] Log off and back on; verify the task runs. Stop the task, then verify
|
||||
the repeating trigger starts it again within 5 minutes.
|
||||
- [ ] For foreground diagnostics or a single pass, stop **and disable** the
|
||||
task first so its repeating trigger cannot race the manual run:
|
||||
|
||||
```powershell
|
||||
Disable-ScheduledTask -TaskName 'TMBH-TranscodeWorker'
|
||||
Stop-ScheduledTask -TaskName 'TMBH-TranscodeWorker'
|
||||
python .\apps\media-worker\scripts\windows-transcode-worker.py --once
|
||||
# Exit status is 1 if any job in this pass failed; otherwise 0.
|
||||
Enable-ScheduledTask -TaskName 'TMBH-TranscodeWorker'
|
||||
Start-ScheduledTask -TaskName 'TMBH-TranscodeWorker'
|
||||
```
|
||||
|
||||
Omit `--once` for continuous foreground polling; press Ctrl+C to stop.
|
||||
A second worker on the same inbox must exit without executing jobs.
|
||||
Force-stopping during an encode can leave a hidden `.part` output; remove
|
||||
it after confirming no ffmpeg process is using it. Its job stays in the
|
||||
inbox for retry. Normal encode failures clean up the temporary output.
|
||||
185
apps/media-worker/scripts/windows-transcode-worker.py
Normal file
185
apps/media-worker/scripts/windows-transcode-worker.py
Normal file
|
|
@ -0,0 +1,185 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Native Windows hardware transcode worker; Python standard library only."""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
import uuid
|
||||
|
||||
|
||||
PRESETS = {
|
||||
"nvenc_h264": ["-c:v", "h264_nvenc", "-preset", "p4", "-rc", "vbr"],
|
||||
"qsv_h264": ["-c:v", "h264_qsv"],
|
||||
"nvenc_hevc": ["-c:v", "hevc_nvenc", "-preset", "p4", "-rc", "vbr"],
|
||||
}
|
||||
|
||||
|
||||
class WorkerLock:
|
||||
"""Windows releases this byte-range lock even if the worker crashes."""
|
||||
|
||||
def __init__(self, inbox):
|
||||
self.path = inbox / ".worker.lock"
|
||||
self.file = None
|
||||
|
||||
def __enter__(self):
|
||||
import msvcrt
|
||||
|
||||
self.file = self.path.open("a+b")
|
||||
self.file.seek(0, os.SEEK_END)
|
||||
if self.file.tell() == 0:
|
||||
self.file.write(b"\0")
|
||||
self.file.flush()
|
||||
self.file.seek(0)
|
||||
try:
|
||||
msvcrt.locking(self.file.fileno(), msvcrt.LK_NBLCK, 1)
|
||||
except OSError:
|
||||
self.file.close()
|
||||
self.file = None
|
||||
raise
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
# Closing the handle releases the lock; keep the file to avoid races.
|
||||
self.file.close()
|
||||
|
||||
|
||||
def load_job(job_path):
|
||||
# Windows PowerShell 5.1's UTF8 output includes a BOM.
|
||||
with job_path.open(encoding="utf-8-sig") as source:
|
||||
job = json.load(source)
|
||||
if not isinstance(job, dict):
|
||||
raise ValueError("Job must be a JSON object")
|
||||
for key in ("input", "output", "preset"):
|
||||
if not isinstance(job.get(key), str) or not job[key].strip():
|
||||
raise ValueError("Job requires a nonempty string: " + key)
|
||||
if job["preset"] not in PRESETS:
|
||||
raise ValueError("Unknown preset: " + job["preset"])
|
||||
quality = job.get("crf", 23)
|
||||
if not isinstance(quality, int) or isinstance(quality, bool):
|
||||
raise ValueError("crf must be an integer")
|
||||
|
||||
def job_file(value):
|
||||
path = Path(value).expanduser()
|
||||
return (job_path.parent / path).resolve()
|
||||
|
||||
input_path = job_file(job["input"])
|
||||
output_path = job_file(job["output"])
|
||||
if not input_path.is_file():
|
||||
raise ValueError("Input file does not exist: " + str(input_path))
|
||||
if input_path == output_path:
|
||||
raise ValueError("Input and output must be different files")
|
||||
if output_path.exists():
|
||||
raise ValueError("Output already exists: " + str(output_path))
|
||||
if not output_path.suffix:
|
||||
raise ValueError("Output needs a container extension, such as .mp4")
|
||||
return input_path, output_path, job["preset"], quality
|
||||
|
||||
|
||||
def archive_job(job_path, directory, error=None):
|
||||
target = directory / job_path.name
|
||||
while target.exists() or target.with_suffix(target.suffix + ".err").exists():
|
||||
target = directory / (job_path.stem + "-" + uuid.uuid4().hex + ".json")
|
||||
if error is not None:
|
||||
target.with_suffix(target.suffix + ".err").write_text(error, encoding="utf-8")
|
||||
job_path.rename(target)
|
||||
return target
|
||||
|
||||
|
||||
def process_job(job_path, done, failed, ffmpeg):
|
||||
temporary_output = None
|
||||
try:
|
||||
input_path, output_path, preset, quality = load_job(job_path)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
# Keep the container suffix so ffmpeg can infer the output format.
|
||||
temporary_output = output_path.with_name(
|
||||
"." + output_path.stem + "-" + uuid.uuid4().hex + ".part" + output_path.suffix
|
||||
)
|
||||
video_args = list(PRESETS[preset])
|
||||
if preset == "qsv_h264":
|
||||
video_args += ["-global_quality", str(quality)]
|
||||
else:
|
||||
video_args += ["-cq", str(quality), "-b:v", "0"]
|
||||
command = [
|
||||
ffmpeg, "-hide_banner", "-nostdin", "-n", "-i", str(input_path),
|
||||
"-map", "0:v:0", "-map", "0:a:0?",
|
||||
*video_args, "-pix_fmt", "nv12", "-c:a", "aac", "-b:a", "192k",
|
||||
str(temporary_output),
|
||||
]
|
||||
logging.info("Job %s: %s", job_path.name, subprocess.list2cmdline(command))
|
||||
with tempfile.TemporaryFile() as stderr:
|
||||
result = subprocess.run(command, stdin=subprocess.DEVNULL,
|
||||
stdout=subprocess.DEVNULL, stderr=stderr, check=False)
|
||||
if result.returncode != 0:
|
||||
# Bound diagnostic size even for a long-running failed encode.
|
||||
stderr.seek(0, os.SEEK_END)
|
||||
stderr.seek(max(0, stderr.tell() - 65536))
|
||||
detail = stderr.read().decode("utf-8", errors="replace")
|
||||
raise RuntimeError(
|
||||
"ffmpeg exited with code {}\nCommand: {}\n\n{}".format(
|
||||
result.returncode, subprocess.list2cmdline(command), detail
|
||||
)
|
||||
)
|
||||
if not temporary_output.is_file() or temporary_output.stat().st_size == 0:
|
||||
raise RuntimeError("ffmpeg returned success without a nonempty output")
|
||||
# On Windows rename refuses to replace an existing destination.
|
||||
temporary_output.rename(output_path)
|
||||
archive_job(job_path, done)
|
||||
logging.info("Completed %s -> %s", job_path.name, output_path)
|
||||
return True
|
||||
except Exception as error:
|
||||
logging.error("Failed %s: %s", job_path.name, error)
|
||||
archive_job(job_path, failed, "{}: {}\n".format(type(error).__name__, error))
|
||||
return False
|
||||
finally:
|
||||
if temporary_output is not None:
|
||||
temporary_output.unlink(missing_ok=True)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--inbox", type=Path, default=Path(r"C:\transcode-jobs\inbox"),
|
||||
help="JSON inbox; done/failed are sibling directories")
|
||||
parser.add_argument("--ffmpeg", default="ffmpeg", help="ffmpeg executable (default: PATH)")
|
||||
parser.add_argument("--poll-interval", type=float, default=5, help="Poll interval in seconds")
|
||||
parser.add_argument("--once", action="store_true", help="Process the current inbox and exit")
|
||||
args = parser.parse_args()
|
||||
if os.name != "nt":
|
||||
parser.error("Run this worker with Windows Python directly, not inside WSL")
|
||||
if not math.isfinite(args.poll_interval) or args.poll_interval <= 0:
|
||||
parser.error("--poll-interval must be a finite positive number")
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||||
inbox = args.inbox.resolve()
|
||||
done, failed = inbox.parent / "done", inbox.parent / "failed"
|
||||
for directory in (inbox, done, failed):
|
||||
directory.mkdir(parents=True, exist_ok=True)
|
||||
lock = WorkerLock(inbox)
|
||||
try:
|
||||
lock.__enter__()
|
||||
except OSError as error:
|
||||
logging.error("Cannot acquire worker lock (another worker may be running): %s", error)
|
||||
return 1
|
||||
try:
|
||||
logging.info("Polling %s", inbox)
|
||||
had_failure = False
|
||||
while True:
|
||||
for job_path in sorted(inbox.glob("*.json")):
|
||||
if job_path.is_file():
|
||||
had_failure = not process_job(job_path, done, failed, args.ffmpeg) or had_failure
|
||||
if args.once:
|
||||
return int(had_failure)
|
||||
time.sleep(args.poll_interval)
|
||||
except KeyboardInterrupt:
|
||||
logging.info("Worker stopped")
|
||||
return 0
|
||||
finally:
|
||||
lock.__exit__()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Loading…
Add table
Add a link
Reference in a new issue