Add stage-4 language enrichment for text files
Text files sorted by stage 3 now flow onto a new work queue and worker pool that enrich them with natural language (Languages.jl LanguageDetector: name, ISO 639-3 code, confidence) and programming language (github-linguist), writing a .meta.json sidecar to data/text_done/ like the stage-2 known-file pipeline. github-linguist reads the git blob of a path inside a repo, so untracked data/ files are copied to /tmp (outside any repo, name preserved for extension heuristics) before detection. Programming-language lookup is best-effort (startup warning if missing, degraded/null on failure); natural-language failure yields a degraded sidecar, not a quarantine. Factored exiftool's timeout-kill into shared run_with_timeout and the durable sidecar-first commit into commit_enriched!, both reused by stage 4. Recovery re-drives data/text/; graceful drain closes the text queue after its stage-3 producers finish.
This commit is contained in:
@@ -7,6 +7,7 @@ using JSON3
|
||||
using Oxygen
|
||||
using Lux
|
||||
using JLD2
|
||||
using Languages
|
||||
|
||||
include("config.jl")
|
||||
include("job.jl")
|
||||
@@ -16,6 +17,7 @@ include("model.jl") # build_model() + read_features(); shared with bin/train
|
||||
include("classify.jl") # Classifier + load_classifier/classify (needs model.jl)
|
||||
include("metadata.jl") # exiftool extraction + sidecar enrichment (stage 2)
|
||||
include("content.jl") # binary-vs-text triage for unknown files (stage 3)
|
||||
include("language.jl") # natural + programming language enrichment for text (stage 4)
|
||||
include("worker.jl")
|
||||
|
||||
# Globals the HTTP handlers read at request time. Set once in `run`, before the
|
||||
@@ -25,7 +27,9 @@ const CONFIG = Ref{Config}()
|
||||
const QUEUE = Ref{ChannelQueue}() # stage-1 (classification) queue; HTTP intake enqueues here
|
||||
const KNOWN_QUEUE = Ref{ChannelQueue}() # stage-2 (enrichment) queue; stage-1 workers enqueue here
|
||||
const UNKNOWN_QUEUE = Ref{ChannelQueue}() # stage-3 (content triage) queue; stage-1 workers enqueue here
|
||||
const TEXT_QUEUE = Ref{ChannelQueue}() # stage-4 (language enrichment) queue; stage-3 workers enqueue here
|
||||
const CLASSIFIER = Ref{Classifier}() # loaded once at startup, shared read-only across workers
|
||||
const DETECTOR = Ref{LanguageDetector}() # natural-language detector; built once at startup, shared read-only
|
||||
|
||||
include("server.jl") # registers routes (references CONFIG/QUEUE at call time)
|
||||
|
||||
@@ -74,43 +78,60 @@ function run(; overrides...)
|
||||
|
||||
# All pools draw from the same OS threads. Warn on the *combined* size (still
|
||||
# allowed): oversubscription just means tasks share threads, not a failure.
|
||||
total_workers = cfg.worker_count + cfg.known_worker_count + cfg.unknown_worker_count
|
||||
total_workers = cfg.worker_count + cfg.known_worker_count + cfg.unknown_worker_count + cfg.text_worker_count
|
||||
if total_workers > Threads.nthreads()
|
||||
@warn "combined worker count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" classify_workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count total=total_workers nthreads=Threads.nthreads()
|
||||
@warn "combined worker count exceeds available threads; workers will share threads (start Julia with -t N for real parallelism)" classify_workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count text_workers=cfg.text_worker_count total=total_workers nthreads=Threads.nthreads()
|
||||
end
|
||||
|
||||
# exiftool is a hard prerequisite for stage-2 enrichment. Fail fast at
|
||||
# startup rather than discover it missing on the first known file.
|
||||
assert_exiftool()
|
||||
|
||||
# github-linguist powers stage-4 *programming*-language detection, but it's
|
||||
# best-effort (natural-language enrichment stands on its own), so a missing
|
||||
# binary is a warning, not a fatal error — per-file lookups degrade to none.
|
||||
linguist_available() || @warn "github-linguist not found on PATH; stage-4 text files will have no programming language (install it to enable)"
|
||||
|
||||
queue = ChannelQueue(cfg.queue_capacity)
|
||||
known_queue = ChannelQueue(cfg.known_queue_capacity)
|
||||
unknown_queue = ChannelQueue(cfg.unknown_queue_capacity)
|
||||
text_queue = ChannelQueue(cfg.text_queue_capacity)
|
||||
CONFIG[] = cfg
|
||||
QUEUE[] = queue
|
||||
KNOWN_QUEUE[] = known_queue
|
||||
UNKNOWN_QUEUE[] = unknown_queue
|
||||
TEXT_QUEUE[] = text_queue
|
||||
|
||||
# Load the classifier before serving. Fail fast: a server that silently
|
||||
# doesn't classify is a worse surprise than a clear startup error.
|
||||
CLASSIFIER[] = load_classifier(cfg.model_path)
|
||||
@info "loaded classifier" path=cfg.model_path
|
||||
|
||||
# Build the natural-language detector once (it loads the whatlang n-gram
|
||||
# model) and share it read-only across the stage-4 pool, like the classifier.
|
||||
DETECTOR[] = LanguageDetector()
|
||||
@info "loaded language detector"
|
||||
|
||||
# Stage-aware recovery: re-drive each stage's leftovers onto its own queue so
|
||||
# files resume where they were, not from scratch. spool/ → stage-1,
|
||||
# known/ → stage-2, unknown/ → stage-3.
|
||||
recovered = recover_dir!(cfg.spool_dir, queue)
|
||||
recovered_known = recover_dir!(cfg.known_dir, known_queue)
|
||||
recovered_unknown = recover_dir!(cfg.unknown_dir, unknown_queue)
|
||||
@info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity recovered=recovered recovered_known=recovered_known recovered_unknown=recovered_unknown
|
||||
recovered_text = recover_dir!(cfg.text_dir, text_queue)
|
||||
@info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count text_workers=cfg.text_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity text_capacity=cfg.text_queue_capacity recovered=recovered recovered_known=recovered_known recovered_unknown=recovered_unknown recovered_text=recovered_text
|
||||
|
||||
workers = [Threads.@spawn worker_loop(i, cfg, queue,
|
||||
(job, c, wid) -> handle_classify_job(job, c, wid, known_queue, unknown_queue))
|
||||
for i in 1:cfg.worker_count]
|
||||
known_workers = [Threads.@spawn worker_loop(i, cfg, known_queue, handle_known_job)
|
||||
for i in 1:cfg.known_worker_count]
|
||||
unknown_workers = [Threads.@spawn worker_loop(i, cfg, unknown_queue, handle_unknown_job)
|
||||
unknown_workers = [Threads.@spawn worker_loop(i, cfg, unknown_queue,
|
||||
(job, c, wid) -> handle_unknown_job(job, c, wid, text_queue))
|
||||
for i in 1:cfg.unknown_worker_count]
|
||||
text_workers = [Threads.@spawn worker_loop(i, cfg, text_queue,
|
||||
(job, c, wid) -> handle_text_job(job, c, wid, DETECTOR[]))
|
||||
for i in 1:cfg.text_worker_count]
|
||||
|
||||
register_routes()
|
||||
serve(; host = cfg.host, port = cfg.port, async = true, show_banner = false)
|
||||
@@ -130,10 +151,12 @@ function run(; overrides...)
|
||||
close!(queue) # 2. no new classify jobs; stage-1 drains buffered
|
||||
foreach(wait, workers) # 3. wait out stage-1 — the ONLY producer of BOTH the
|
||||
# known and unknown queues — so nothing else enqueues
|
||||
close!(known_queue) # 4. now safe to close the downstream queues
|
||||
close!(known_queue) # 4. now safe to close the queues stage-1 fed
|
||||
close!(unknown_queue)
|
||||
foreach(wait, known_workers) # 5. wait out stage-2 and stage-3
|
||||
foreach(wait, unknown_workers)
|
||||
foreach(wait, known_workers) # 5. wait out stage-2 (terminal) and stage-3 — stage-3 is
|
||||
foreach(wait, unknown_workers) # the ONLY producer of the text queue
|
||||
close!(text_queue) # 6. now safe to close the queue stage-3 fed
|
||||
foreach(wait, text_workers) # 7. wait out stage-4
|
||||
@info "shutdown complete"
|
||||
end
|
||||
atexit(drain)
|
||||
|
||||
@@ -17,15 +17,22 @@ Base.@kwdef struct Config
|
||||
# of the classify and enrich pools.
|
||||
unknown_worker_count::Int = Threads.nthreads()
|
||||
unknown_queue_capacity::Int = 1000
|
||||
# Stage 4 (language enrichment) has its own pool + queue too: detecting a text
|
||||
# file's natural language (Languages.jl) and programming language (shelling to
|
||||
# github-linguist) is a mix of CPU and process-spawn work, tuned independently.
|
||||
text_worker_count::Int = Threads.nthreads()
|
||||
text_queue_capacity::Int = 1000
|
||||
spool_dir::String = "data/spool" # files land here on intake (pending classification)
|
||||
known_dir::String = "data/known" # classified-known, awaiting enrichment (stage 2)
|
||||
unknown_dir::String = "data/unknown" # classified-unknown, awaiting content triage (stage 3)
|
||||
binary_dir::String = "data/binary" # stage-3 sink: unknown files that look like binary data
|
||||
text_dir::String = "data/text" # stage-3 sink: unknown files that look like text
|
||||
text_dir::String = "data/text" # classified-text, awaiting language enrichment (stage 4)
|
||||
done_dir::String = "data/done" # fully enriched known files (+ .meta.json sidecars)
|
||||
text_done_dir::String = "data/text_done" # fully enriched text files (+ .meta.json sidecars)
|
||||
failed_dir::String = "data/failed" # files move here if a worker throws
|
||||
model_path::String = "model/classifier.jld2" # committed classifier artifact, loaded at startup
|
||||
exiftool_timeout::Int = 30 # seconds before a stuck exiftool is killed → degraded sidecar
|
||||
linguist_timeout::Int = 30 # seconds before a stuck github-linguist is killed → no programming language
|
||||
end
|
||||
|
||||
"""
|
||||
@@ -39,16 +46,20 @@ Recognised variables:
|
||||
FS_HOST, FS_PORT, FS_WORKERS, FS_QUEUE_CAPACITY,
|
||||
FS_KNOWN_WORKERS, FS_KNOWN_QUEUE_CAPACITY,
|
||||
FS_UNKNOWN_WORKERS, FS_UNKNOWN_QUEUE_CAPACITY,
|
||||
FS_TEXT_WORKERS, FS_TEXT_QUEUE_CAPACITY,
|
||||
FS_SPOOL_DIR, FS_KNOWN_DIR, FS_UNKNOWN_DIR, FS_BINARY_DIR, FS_TEXT_DIR,
|
||||
FS_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH, FS_EXIFTOOL_TIMEOUT
|
||||
FS_DONE_DIR, FS_TEXT_DONE_DIR, FS_FAILED_DIR, FS_MODEL_PATH,
|
||||
FS_EXIFTOOL_TIMEOUT, FS_LINGUIST_TIMEOUT
|
||||
"""
|
||||
function config_from_env(; host=nothing, port=nothing, worker_count=nothing,
|
||||
queue_capacity=nothing, known_worker_count=nothing,
|
||||
known_queue_capacity=nothing, unknown_worker_count=nothing,
|
||||
unknown_queue_capacity=nothing, spool_dir=nothing,
|
||||
unknown_queue_capacity=nothing, text_worker_count=nothing,
|
||||
text_queue_capacity=nothing, spool_dir=nothing,
|
||||
known_dir=nothing, unknown_dir=nothing, binary_dir=nothing,
|
||||
text_dir=nothing, done_dir=nothing,
|
||||
failed_dir=nothing, model_path=nothing, exiftool_timeout=nothing)
|
||||
text_dir=nothing, done_dir=nothing, text_done_dir=nothing,
|
||||
failed_dir=nothing, model_path=nothing, exiftool_timeout=nothing,
|
||||
linguist_timeout=nothing)
|
||||
Config(
|
||||
host = something(host, get(ENV, "FS_HOST", "127.0.0.1")),
|
||||
port = something(port, parse(Int, get(ENV, "FS_PORT", "8080"))),
|
||||
@@ -58,22 +69,26 @@ function config_from_env(; host=nothing, port=nothing, worker_count=nothing,
|
||||
known_queue_capacity = something(known_queue_capacity, parse(Int, get(ENV, "FS_KNOWN_QUEUE_CAPACITY", "1000"))),
|
||||
unknown_worker_count = something(unknown_worker_count, parse(Int, get(ENV, "FS_UNKNOWN_WORKERS", string(Threads.nthreads())))),
|
||||
unknown_queue_capacity = something(unknown_queue_capacity, parse(Int, get(ENV, "FS_UNKNOWN_QUEUE_CAPACITY", "1000"))),
|
||||
text_worker_count = something(text_worker_count, parse(Int, get(ENV, "FS_TEXT_WORKERS", string(Threads.nthreads())))),
|
||||
text_queue_capacity = something(text_queue_capacity, parse(Int, get(ENV, "FS_TEXT_QUEUE_CAPACITY", "1000"))),
|
||||
spool_dir = something(spool_dir, get(ENV, "FS_SPOOL_DIR", "data/spool")),
|
||||
known_dir = something(known_dir, get(ENV, "FS_KNOWN_DIR", "data/known")),
|
||||
unknown_dir = something(unknown_dir, get(ENV, "FS_UNKNOWN_DIR", "data/unknown")),
|
||||
binary_dir = something(binary_dir, get(ENV, "FS_BINARY_DIR", "data/binary")),
|
||||
text_dir = something(text_dir, get(ENV, "FS_TEXT_DIR", "data/text")),
|
||||
done_dir = something(done_dir, get(ENV, "FS_DONE_DIR", "data/done")),
|
||||
text_done_dir = something(text_done_dir, get(ENV, "FS_TEXT_DONE_DIR", "data/text_done")),
|
||||
failed_dir = something(failed_dir, get(ENV, "FS_FAILED_DIR", "data/failed")),
|
||||
model_path = something(model_path, get(ENV, "FS_MODEL_PATH", "model/classifier.jld2")),
|
||||
exiftool_timeout = something(exiftool_timeout, parse(Int, get(ENV, "FS_EXIFTOOL_TIMEOUT", "30"))),
|
||||
linguist_timeout = something(linguist_timeout, parse(Int, get(ENV, "FS_LINGUIST_TIMEOUT", "30"))),
|
||||
)
|
||||
end
|
||||
|
||||
"Create all the pipeline-stage directories if they don't already exist."
|
||||
function ensure_dirs(cfg::Config)
|
||||
for d in (cfg.spool_dir, cfg.known_dir, cfg.unknown_dir, cfg.binary_dir,
|
||||
cfg.text_dir, cfg.done_dir, cfg.failed_dir)
|
||||
cfg.text_dir, cfg.done_dir, cfg.text_done_dir, cfg.failed_dir)
|
||||
mkpath(d)
|
||||
end
|
||||
return nothing
|
||||
|
||||
155
src/language.jl
Normal file
155
src/language.jl
Normal file
@@ -0,0 +1,155 @@
|
||||
# Stage-4 language enrichment for text files.
|
||||
#
|
||||
# A file that stage-3 sorted into `text/` is human-readable, but we don't yet
|
||||
# know *what* it is. This stage answers two questions and records them in a
|
||||
# `.meta.json` sidecar, exactly like the stage-2 known-file enrichment:
|
||||
#
|
||||
# * natural language — via Languages.jl's `LanguageDetector` (a Julia port of
|
||||
# the `whatlang` n-gram model): English vs. French vs. Japanese, plus a
|
||||
# confidence score. Pure Julia, no subprocess.
|
||||
# * programming language — via the `github-linguist` CLI, which recognizes
|
||||
# source and markup by extension + content heuristics. There is no
|
||||
# comparable native Julia library, so we shell out (mirroring stage-2's
|
||||
# exiftool dependency).
|
||||
#
|
||||
# Neither detector failing quarantines the file: a text file is wanted whether
|
||||
# or not we can name its language, so a failure yields a *degraded* sidecar
|
||||
# (what we know plus an `error` note), just like stage 2.
|
||||
#
|
||||
# The github-linguist quirk that shapes this code: run against a path *inside* a
|
||||
# git repository, linguist reads the file's committed git blob, not the bytes on
|
||||
# disk — and an untracked file (which every file under `data/` is) has no blob,
|
||||
# so it crashes. We sidestep this by copying the file to a fresh temp dir outside
|
||||
# any repo (preserving its name so linguist's extension heuristics still fire)
|
||||
# and pointing linguist there.
|
||||
|
||||
# How much of a text file to feed the natural-language detector. The whatlang
|
||||
# model saturates quickly, so a bounded prefix keeps memory flat on huge logs
|
||||
# while still giving the detector plenty of signal.
|
||||
const LANG_SAMPLE_BYTES = 65_536
|
||||
|
||||
"Return true if the `github-linguist` binary is on PATH."
|
||||
function linguist_available()
|
||||
try
|
||||
Base.run(pipeline(`github-linguist --version`; stdout=devnull, stderr=devnull))
|
||||
return true
|
||||
catch
|
||||
return false
|
||||
end
|
||||
end
|
||||
|
||||
"""
|
||||
read_text_sample(path) -> String
|
||||
|
||||
Read up to `LANG_SAMPLE_BYTES` of `path` as UTF-8 text, trimming a multi-byte
|
||||
character the window may have cut in half (reusing stage-3's `trim_truncated_utf8`)
|
||||
so the tail isn't misread as garbage.
|
||||
"""
|
||||
function read_text_sample(path::AbstractString)::String
|
||||
open(path, "r") do io
|
||||
chunk = read(io, LANG_SAMPLE_BYTES)
|
||||
return String(copy(trim_truncated_utf8(chunk)))
|
||||
end
|
||||
end
|
||||
|
||||
"""
|
||||
detect_natural_language(detector, text) -> (name, code, confidence)
|
||||
|
||||
Run the `LanguageDetector` on `text`, returning the language's English name
|
||||
(e.g. `"English"`), its ISO 639-3 code (e.g. `"eng"`), and the model's
|
||||
confidence in `[0,1]`. Returns `(nothing, nothing, nothing)` when there is no
|
||||
usable text (empty/whitespace) or the detector errors — the caller records that
|
||||
as a degraded result rather than failing the file.
|
||||
"""
|
||||
function detect_natural_language(detector, text::AbstractString)
|
||||
isempty(strip(text)) && return (nothing, nothing, nothing)
|
||||
try
|
||||
lang, _script, confidence = detector(text)
|
||||
return (Languages.english_name(lang), Languages.isocode(lang), confidence)
|
||||
catch
|
||||
return (nothing, nothing, nothing)
|
||||
end
|
||||
end
|
||||
|
||||
"""
|
||||
run_linguist(path, timeout) -> Union{String,Nothing}
|
||||
|
||||
Ask `github-linguist --json` for the programming/markup language of the file at
|
||||
`path`, returning the language name (e.g. `"Python"`, `"Markdown"`) or `nothing`
|
||||
when linguist can't identify one. Plain prose reports as `"Text"` and
|
||||
unrecognized content as JSON `null`; both collapse to `nothing` here, since only
|
||||
a real programming/markup language is worth recording.
|
||||
|
||||
`path` MUST be outside any git repository — see the module header for why.
|
||||
"""
|
||||
function run_linguist(path::AbstractString, timeout::Integer)
|
||||
bytes = run_with_timeout(`github-linguist --json $path`, timeout)
|
||||
bytes === nothing && return nothing
|
||||
|
||||
parsed = try
|
||||
JSON3.read(String(bytes))
|
||||
catch
|
||||
return nothing
|
||||
end
|
||||
# linguist --json emits a single object keyed by the file path; pull the one
|
||||
# entry rather than depend on the exact key spelling.
|
||||
isempty(parsed) && return nothing
|
||||
entry = first(values(parsed))
|
||||
lang = get(entry, :language, nothing)
|
||||
(lang === nothing || lang == "Text") && return nothing
|
||||
return String(lang)
|
||||
end
|
||||
|
||||
"""
|
||||
detect_programming_language(job, cfg) -> Union{String,Nothing}
|
||||
|
||||
Programming/markup language of a text file, or `nothing`. Copies the file to a
|
||||
throwaway temp dir *outside* the git repo — under its sanitized original name so
|
||||
linguist's extension heuristics still apply — runs linguist there, and cleans up.
|
||||
"""
|
||||
function detect_programming_language(job::Job, cfg::Config)
|
||||
lang = nothing
|
||||
mktempdir() do dir # tempdir() → /tmp, outside the repo
|
||||
safe = sanitize_filename(job.original_name)
|
||||
tmp = joinpath(dir, safe)
|
||||
cp(job.path, tmp; force=true)
|
||||
lang = run_linguist(tmp, cfg.linguist_timeout)
|
||||
end
|
||||
return lang
|
||||
end
|
||||
|
||||
"""
|
||||
build_text_metadata(detector, job, cfg) -> NamedTuple
|
||||
|
||||
Build the stage-4 sidecar payload for a text file: its natural language (name +
|
||||
ISO code + confidence) and programming/markup language, plus the Job's
|
||||
authoritative id/name/size. `error` is set only when natural-language detection
|
||||
produced nothing usable (the file is still enriched and committed); programming
|
||||
language is best-effort and its absence is normal, not an error.
|
||||
"""
|
||||
function build_text_metadata(detector, job::Job, cfg::Config)
|
||||
text = read_text_sample(job.path)
|
||||
name, code, confidence = detect_natural_language(detector, text)
|
||||
programming_language = detect_programming_language(job, cfg)
|
||||
|
||||
return (
|
||||
id = job.id,
|
||||
original_name = job.original_name,
|
||||
file_size = job.size, # authoritative, from intake
|
||||
content_type = "text",
|
||||
language = name,
|
||||
language_code = code,
|
||||
language_confidence = confidence,
|
||||
programming_language = programming_language,
|
||||
error = name === nothing ? "language detection produced no result" : nothing,
|
||||
)
|
||||
end
|
||||
|
||||
"""
|
||||
finalize_text!(cfg, job, meta) -> (file_dest, sidecar_dest)
|
||||
|
||||
Commit an enriched text file (stage 4) to `text_done/` via the shared
|
||||
sidecar-first `commit_enriched!`, giving text files the same crash-safe
|
||||
"file implies sidecar" guarantee as stage-2 known files.
|
||||
"""
|
||||
finalize_text!(cfg::Config, job::Job, meta) = commit_enriched!(cfg.text_done_dir, job, meta)
|
||||
@@ -68,20 +68,20 @@ function coalesce_tag(bytag::Dict{String,Any}, tags)
|
||||
end
|
||||
|
||||
"""
|
||||
run_exiftool(path, timeout) -> Union{Dict{String,Any},Nothing}
|
||||
run_with_timeout(cmd, timeout) -> Union{Vector{UInt8},Nothing}
|
||||
|
||||
Run `exiftool -json -G` on `path`, returning the parsed tag object, or `nothing`
|
||||
on non-zero exit, unparseable output, or timeout. The subprocess is killed after
|
||||
`timeout` seconds so one pathological file can't wedge a worker forever.
|
||||
Run `cmd`, capturing stdout, and return the captured bytes on clean exit, or
|
||||
`nothing` on non-zero exit or timeout. The subprocess is killed (SIGTERM, then
|
||||
SIGKILL after a grace period) once it overruns `timeout` seconds, so one
|
||||
pathological input can't wedge a worker forever. Shared by the exiftool (stage 2)
|
||||
and github-linguist (stage 4) shells.
|
||||
"""
|
||||
function run_exiftool(path::AbstractString, timeout::Integer)
|
||||
function run_with_timeout(cmd::Cmd, timeout::Integer)
|
||||
out = IOBuffer()
|
||||
# -json: machine output; -G: group-prefixed tags; -n: numeric (unformatted)
|
||||
# values so sizes/durations are numbers, not display strings.
|
||||
proc = Base.run(pipeline(`exiftool -json -G -n $path`; stdout=out, stderr=devnull); wait=false)
|
||||
proc = Base.run(pipeline(cmd; stdout=out, stderr=devnull); wait=false)
|
||||
|
||||
# Kill the process if it overruns the timeout. `t` polls rather than blocking
|
||||
# so we can `kill` a hung exiftool; the poll interval bounds shutdown latency.
|
||||
# so we can `kill` a hung child; the poll interval bounds shutdown latency.
|
||||
killed = Ref(false)
|
||||
t = Threads.@spawn begin
|
||||
waited = 0.0
|
||||
@@ -104,9 +104,23 @@ function run_exiftool(path::AbstractString, timeout::Integer)
|
||||
wait(t)
|
||||
|
||||
(killed[] || !success(proc)) && return nothing
|
||||
return take!(out)
|
||||
end
|
||||
|
||||
"""
|
||||
run_exiftool(path, timeout) -> Union{Dict{String,Any},Nothing}
|
||||
|
||||
Run `exiftool -json -G` on `path`, returning the parsed tag object, or `nothing`
|
||||
on non-zero exit, unparseable output, or timeout.
|
||||
"""
|
||||
function run_exiftool(path::AbstractString, timeout::Integer)
|
||||
# -json: machine output; -G: group-prefixed tags; -n: numeric (unformatted)
|
||||
# values so sizes/durations are numbers, not display strings.
|
||||
bytes = run_with_timeout(`exiftool -json -G -n $path`, timeout)
|
||||
bytes === nothing && return nothing
|
||||
|
||||
parsed = try
|
||||
JSON3.read(String(take!(out)))
|
||||
JSON3.read(String(bytes))
|
||||
catch
|
||||
return nothing
|
||||
end
|
||||
@@ -174,34 +188,44 @@ function normalize_metadata(job::Job, bytag::Dict{String,Any})
|
||||
end
|
||||
|
||||
"""
|
||||
finalize_known!(cfg, job, meta) -> (file_dest, sidecar_dest)
|
||||
commit_enriched!(dest_dir, job, meta) -> (file_dest, sidecar_dest)
|
||||
|
||||
Commit an enriched known file to `done/` with the sidecar-first ordering so the
|
||||
invariant *"a file in done/ implies its sidecar is already there"* always holds.
|
||||
Commit an enriched file to `dest_dir` with the sidecar-first ordering so the
|
||||
invariant *"a file in dest_dir implies its sidecar is already there"* always
|
||||
holds. Shared by every enrichment stage that emits a `.meta.json` sidecar
|
||||
(stage-2 known files → `done/`, stage-4 text files → `text_done/`).
|
||||
|
||||
Sequence: write `<name>.meta.json` to a temp name, fsync its bytes, rename it
|
||||
into place, fsync `done/` so the rename itself is durable, THEN move the file
|
||||
into `done/`. A crash between the two leaves only a harmless orphan sidecar in
|
||||
`done/` while the file stays in `known/`, so stage-aware recovery re-drives it
|
||||
and overwrites the sidecar — idempotent. The fsyncs make the ordering hold
|
||||
across power loss, not just process crashes.
|
||||
into place, fsync `dest_dir` so the rename itself is durable, THEN move the file
|
||||
into `dest_dir`. A crash between the two leaves only a harmless orphan sidecar in
|
||||
`dest_dir` while the file stays in its stage dir, so stage-aware recovery
|
||||
re-drives it and overwrites the sidecar — idempotent. The fsyncs make the
|
||||
ordering hold across power loss, not just process crashes.
|
||||
"""
|
||||
function finalize_known!(cfg::Config, job::Job, meta)
|
||||
function commit_enriched!(dest_dir::AbstractString, job::Job, meta)
|
||||
base = basename(job.path)
|
||||
sidecar = joinpath(cfg.done_dir, string(base, ".meta.json"))
|
||||
sidecar = joinpath(dest_dir, string(base, ".meta.json"))
|
||||
tmp_sidecar = string(sidecar, ".tmp")
|
||||
|
||||
# Write to a temp name then rename, so a reader in done/ never sees a partial
|
||||
# sidecar and a crash mid-write can't masquerade as a committed one.
|
||||
# Write to a temp name then rename, so a reader in dest_dir never sees a
|
||||
# partial sidecar and a crash mid-write can't masquerade as a committed one.
|
||||
open(tmp_sidecar, "w") do io
|
||||
write(io, JSON3.write(meta))
|
||||
flush(io)
|
||||
fsync_fd(fd(io)) # durably persist bytes before the rename
|
||||
end
|
||||
mv(tmp_sidecar, sidecar; force=true) # sidecar committed first
|
||||
fsync_dir(cfg.done_dir) # persist the rename itself, not just the bytes
|
||||
fsync_dir(dest_dir) # persist the rename itself, not just the bytes
|
||||
|
||||
file_dest = move_to(cfg.done_dir, job) # file arrival = commit point
|
||||
file_dest = move_to(dest_dir, job) # file arrival = commit point
|
||||
return (file_dest, sidecar)
|
||||
end
|
||||
|
||||
"""
|
||||
finalize_known!(cfg, job, meta) -> (file_dest, sidecar_dest)
|
||||
|
||||
Commit an enriched known file (stage 2) to `done/` via the shared sidecar-first
|
||||
`commit_enriched!`.
|
||||
"""
|
||||
finalize_known!(cfg::Config, job::Job, meta) = commit_enriched!(cfg.done_dir, job, meta)
|
||||
|
||||
|
||||
@@ -1,11 +1,12 @@
|
||||
# Worker tasks: pull jobs off a queue and process them. The loop scaffolding
|
||||
# (dequeue-until-drained, try/catch, quarantine-on-throw) is identical for every
|
||||
# stage, so `worker_loop` is parametrized with a `handler` and reused. Today
|
||||
# there are three stages:
|
||||
# there are four stages:
|
||||
#
|
||||
# stage 1 handle_classify_job spool/ → classify → known/ (+known queue) | unknown/ (+unknown queue)
|
||||
# stage 2 handle_known_job known/ → exiftool enrich → done/ (+ .meta.json)
|
||||
# stage 3 handle_unknown_job unknown/ → binary-vs-text sniff → binary/ | text/
|
||||
# stage 3 handle_unknown_job unknown/ → binary-vs-text sniff → binary/ | text/ (+text queue)
|
||||
# stage 4 handle_text_job text/ → language enrich → text_done/ (+ .meta.json)
|
||||
#
|
||||
# Adding a stage later is just another queue + pool + handler; the loop below
|
||||
# doesn't change.
|
||||
@@ -70,17 +71,42 @@ function handle_known_job(job::Job, cfg::Config, worker_id::Int)
|
||||
end
|
||||
|
||||
"""
|
||||
handle_unknown_job(job, cfg, worker_id)
|
||||
handle_unknown_job(job, cfg, worker_id, text_queue)
|
||||
|
||||
Stage 3. Sort an unrecognized file into a coarse content bucket by sniffing its
|
||||
first bytes: `binary/` if it looks like binary data, `text/` otherwise. Terminal
|
||||
— there is no further stage. Simple by design for now; richer handling can hang
|
||||
off either bucket later.
|
||||
first bytes: `binary/` (terminal — no further stage) if it looks like binary
|
||||
data, `text/` otherwise. A text file is then routed onward to the stage-4
|
||||
language-enrichment queue, retrying on a full queue rather than dropping the file
|
||||
(the same blocking backpressure stage 1 uses for its downstream queues).
|
||||
"""
|
||||
function handle_unknown_job(job::Job, cfg::Config, worker_id::Int)
|
||||
binary = is_binary(job.path)
|
||||
dest = move_to(binary ? cfg.binary_dir : cfg.text_dir, job)
|
||||
@info "sorted unknown" worker=worker_id id=job.id name=job.original_name kind=(binary ? :binary : :text) dest=dest
|
||||
function handle_unknown_job(job::Job, cfg::Config, worker_id::Int, text_queue::JobQueue)
|
||||
if is_binary(job.path)
|
||||
dest = move_to(cfg.binary_dir, job)
|
||||
@info "sorted unknown" worker=worker_id id=job.id name=job.original_name kind=:binary dest=dest
|
||||
else
|
||||
dest = move_to(cfg.text_dir, job)
|
||||
routed = Job(job.id, job.original_name, dest, job.size, job.received_at)
|
||||
while !enqueue!(text_queue, routed)
|
||||
sleep(ROUTE_ENQUEUE_RETRY_SECONDS) # text queue full → back off, don't drop
|
||||
end
|
||||
@info "routed to language enrichment" worker=worker_id id=job.id dest=dest
|
||||
end
|
||||
return nothing
|
||||
end
|
||||
|
||||
"""
|
||||
handle_text_job(job, cfg, worker_id, detector)
|
||||
|
||||
Stage 4. Enrich a text file with its natural language (via `detector`) and
|
||||
programming language (via github-linguist): build the sidecar and commit both to
|
||||
`text_done/` sidecar-first. Detection failure yields a *degraded* sidecar (the
|
||||
file is still wanted text), so the only way to land in `failed/` is a genuine I/O
|
||||
error committing — handled by `worker_loop`'s quarantine.
|
||||
"""
|
||||
function handle_text_job(job::Job, cfg::Config, worker_id::Int, detector)
|
||||
meta = build_text_metadata(detector, job, cfg)
|
||||
file_dest, sidecar = finalize_text!(cfg, job, meta)
|
||||
@info "enriched text" worker=worker_id id=job.id dest=file_dest sidecar=basename(sidecar) language=meta.language confidence=meta.language_confidence programming_language=meta.programming_language degraded=(meta.error !== nothing)
|
||||
return nothing
|
||||
end
|
||||
|
||||
|
||||
Reference in New Issue
Block a user