Command: powershell -NoProfile -Command "$f='C:\ProgramData\AudioProcess\current\chirp_direct_server.py';Write-Output '---DB---';Get-Content $f | Select-Object -Skip 1060 -First 80;Write-Output '---UPSERT---';Get-Content $f | Select-Object -Skip 2660 -First 100"
Directory: projects
Status: SUCCESS
Exit code: 0
Cancel Job Rerun Command Refresh
---DB---
delay = max(0.05, min(30.0, earliest - now))
time.sleep(delay)
def note_translation_provider_error(self, provider: dict[str, Any], status: int | None = None) -> None:
# Temporary errors get a short backoff. Auth/config errors get a longer
# backoff so one bad Google/Argos provider does not keep blocking the queue.
if status in {400, 401, 403, 404, 408, 409, 429, 500, 502, 503, 504}:
with self.translation_pool_lock:
provider_id = str(provider.get("id", ""))
rpm = max(1, int(provider.get("rpm", self.config.get("translation_rpm", 25))))
if status in {400, 401, 403, 404}:
delay = max(600.0, 600.0 / rpm)
else:
delay = max(5.0, 120.0 / rpm)
self.translation_pool_next[provider_id] = max(
float(self.translation_pool_next.get(provider_id, 0.0)),
time.time() + delay,
)
def connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path, timeout=60)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
return conn
def init_db(self) -> None:
"""Create or migrate the local queue database safely.
v2.24 introduces a staged pipeline. Existing jobs remain in the jobs
table, while persistent audio chunks and Chirp operation names are kept
in audio_chunks so a restart can resume Google polling instead of
submitting the same long audio again.
"""
with self.connect() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS jobs (
job_id TEXT PRIMARY KEY,
video_path TEXT NOT NULL UNIQUE,
video_size INTEGER NOT NULL DEFAULT 0,
video_mtime REAL NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'pending',
progress REAL NOT NULL DEFAULT 0,
detail TEXT,
attempts INTEGER NOT NULL DEFAULT 0,
next_retry REAL NOT NULL DEFAULT 0,
original_srt TEXT,
english_srt TEXT,
error TEXT,
active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT '',
updated_at TEXT NOT NULL DEFAULT '',
status_changed_at TEXT NOT NULL DEFAULT ''
)
""")
existing = {str(row[1]) for row in conn.execute("PRAGMA table_info(jobs)")}
migrations = {
"video_size": "INTEGER NOT NULL DEFAULT 0",
"video_mtime": "REAL NOT NULL DEFAULT 0",
"status": "TEXT NOT NULL DEFAULT 'pending'",
"progress": "REAL NOT NULL DEFAULT 0",
"detail": "TEXT",
"attempts": "INTEGER NOT NULL DEFAULT 0",
"next_retry": "REAL NOT NULL DEFAULT 0",
"original_srt": "TEXT",
"english_srt": "TEXT",
"error": "TEXT",
"active": "INTEGER NOT NULL DEFAULT 1",
"created_at": "TEXT NOT NULL DEFAULT ''",
"updated_at": "TEXT NOT NULL DEFAULT ''",
"status_changed_at": "TEXT NOT NULL DEFAULT ''",
}
for column, declaration in migrations.items():
if column not in existing:
conn.execute(f"ALTER TABLE jobs ADD COLUMN {column} {declaration}")
conn.execute("""
CREATE TRIGGER IF NOT EXISTS trg_jobs_status_changed_at_insert
AFTER INSERT ON jobs
WHEN COALESCE(NULLIF(NEW.status_changed_at, ''), '') = ''
BEGIN
---UPSERT---
candidates: list[Path] = []
candidates.extend(Path(os.path.expandvars(str(x))).expanduser() for x in self.config.get("media_roots", []))
candidates.extend(self.roots_from_jellyfin_api())
for data_root in self.configured_jellyfin_data_roots():
for link in data_root.rglob("*.mblink"):
try:
for line in link.read_text(encoding="utf-8-sig", errors="ignore").splitlines():
raw = line.strip().strip('"')
if raw:
candidates.append(Path(os.path.expandvars(raw)).expanduser())
except OSError:
pass
unique: list[Path] = []
seen: set[str] = set()
for path in candidates:
try:
normalized = Path(os.path.abspath(str(path)))
except OSError:
continue
key = os.path.normcase(str(normalized))
if key not in seen:
seen.add(key)
unique.append(normalized)
available = [p.resolve() for p in unique if p.exists() and p.is_dir()]
unavailable = [p for p in unique if not (p.exists() and p.is_dir())]
return available, unavailable
def scan(self) -> dict[str, Any]:
with self.scan_lock:
roots, unavailable = self.discover_media_roots()
if not roots:
return {"ok": False, "jobs": 0, "roots": [], "unavailable": [str(x) for x in unavailable]}
found: list[Path] = []
for root in roots:
for current, dirs, files in os.walk(root):
dirs[:] = [d for d in dirs if not d.startswith(".chirp-")]
for name in files:
path = Path(current) / name
if path.suffix.lower() in VIDEO_EXTENSIONS:
found.append(path.resolve())
found.sort(key=lambda p: str(p).casefold())
now = utc_now()
active_ids: set[str] = set()
with self.connect() as conn:
for path in found:
try:
stat = path.stat()
except OSError:
continue
job_id = hashlib.sha256(os.path.normcase(str(path)).encode("utf-8")).hexdigest()[:24]
active_ids.add(job_id)
row = conn.execute("SELECT video_size,video_mtime FROM jobs WHERE job_id=?", (job_id,)).fetchone()
changed = bool(row and (int(row["video_size"]) != stat.st_size or abs(float(row["video_mtime"]) - stat.st_mtime) > 0.001))
conn.execute(
"""
INSERT INTO jobs(job_id,video_path,video_size,video_mtime,status,progress,detail,attempts,next_retry,active,created_at,updated_at)
VALUES(?,?,?,?, 'pending',0,'Discovered',0,0,1,?,?)
ON CONFLICT(job_id) DO UPDATE SET video_path=excluded.video_path,video_size=excluded.video_size,
video_mtime=excluded.video_mtime,active=1,updated_at=excluded.updated_at
""",
(job_id, str(path), stat.st_size, stat.st_mtime, now, now),
)
if changed:
conn.execute(
"UPDATE jobs SET status='pending',progress=0,detail='Source changed',attempts=0,next_retry=0,"
"original_srt=NULL,english_srt=NULL,error=NULL,updated_at=? WHERE job_id=?",
(now, job_id),
)
if active_ids:
placeholders = ",".join("?" for _ in active_ids)
conn.execute(f"UPDATE jobs SET active=0 WHERE job_id NOT IN ({placeholders})", tuple(active_ids))
else:
conn.execute("UPDATE jobs SET active=0")
return {"ok": True, "jobs": len(found), "roots": [str(x) for x in roots], "unavailable": [str(x) for x in unavailable]}
def update_job(self, job_id: str, **values: Any) -> None:
values["updated_at"] = utc_now()
columns = ",".join(f"{name}=?" for name in values)
with self.connect() as conn:
conn.execute(f"UPDATE jobs SET {columns} WHERE job_id=?", (*values.values(), job_id))
def _chunk_transcript_cues(self, chunk: sqlite3.Row) -> list[Cue]:
transcript = str(chunk["transcript_srt"] or "").strip()
if not transcript:
return []
cues = parse_srt(transcript)
if not cues:
return []
start_ms = int(chunk["start_ms"] or 0)
duration_ms = int(chunk["duration_ms"] or 0)
return apply_chunk_time_offset(cues, start_ms, duration_ms)
def _combined_chunk_cues(self, chunks: list[sqlite3.Row]) -> list[Cue]:
combined: list[Cue] = []
for chunk in chunks:
combined.extend(self._chunk_transcript_cues(chunk))
combined.sort(key=lambda c: (int(c.start_ms or 0), int(c.end_ms or 0), str(c.text or "")))
return combined