local rspamd_logger = require "rspamd_logger"
local rspamd_fann = require "rspamd_fann"
local rspamd_util = require "rspamd_util"
-require "fun" ()
-local ucl = require "ucl"
+local fun = require "fun"
local redis_params
local classifier_config = {
local function maybe_load_fann(task, continue_cb, call_if_fail)
local function load_fann()
local function redis_fann_load_cb(err, data)
+ -- XXX: upstreams
if not err and type(data) == 'table' and type(data[2]) == 'string' then
local version = tonumber(data[1])
- local err,ann_data = rspamd_util.zstd_decompress(data[2])
+ local _err,ann_data = rspamd_util.zstd_decompress(data[2])
local ann
- if err or not ann_data then
- rspamd_logger.errx(task, 'cannot decompress ann: %s', err)
+ if _err or not ann_data then
+ rspamd_logger.errx(task, 'cannot decompress ann: %s', _err)
else
ann = rspamd_fann.load_data(ann_data)
end
end
local key = classifier_config.key
- local ret,_,_ = rspamd_redis_make_request(task,
+ local ret,_,upstream = rspamd_redis_make_request(task,
redis_params, -- connect params
key, -- hash key
false, -- is write
'HMGET', -- command
{key, 'version', 'data', 'spam', 'ham'} -- arguments
)
+ if not ret then
+ rspamd_logger.errx(task, 'redis error on host %s', upstream:get_addr())
+ upstream:fail()
+ end
end
local function check_fann()
+ local _, ret, upstream
local function redis_fann_check_cb(err, data)
+ if err then
+ rspamd_logger.errx(task, 'redis error on host %s: %s', upstream:get_addr(), err)
+ upstream:fail()
+ else
+ upstream:ok()
+ end
if not err and type(data) == 'string' then
local version = tonumber(data)
end
local key = classifier_config.key
- local ret,_,_ = rspamd_redis_make_request(task,
+ ret,_,upstream = rspamd_redis_make_request(task,
redis_params, -- connect params
key, -- hash key
false, -- is write
'HGET', -- command
{key, 'version'} -- arguments
)
+ if not ret then
+ rspamd_logger.errx(task, 'redis error on host %s', upstream:get_addr())
+ upstream:fail()
+ end
end
if not current_classify_ann.loaded then
end
local function tokens_to_vector(tokens)
- local vec = totable(map(function(tok) return tok[1] end, tokens))
+ local vec = fun.totable(fun.map(function(tok) return tok[1] end, tokens))
local ret = {}
- local ntok = #vec
local neurons = classifier_config.neurons
for i = 1,neurons do
ret[i] = 0
end
- each(function(e)
+ fun.each(function(e)
local n = (e % neurons) + 1
ret[n] = ret[n] + 1
end, vec)
end
local function save_fann(task, is_spam)
- local function redis_fann_save_cb(err, data)
+ local ret, conn, upstream
+ local function redis_fann_save_cb(err)
if err then
rspamd_logger.errx(task, "cannot save neural net to redis: %s", err)
+ upstream:fail()
+ else
+ upstream:ok()
end
end
else
current_classify_ann.ham_learned = current_classify_ann.ham_learned + 1
end
- local ret,conn,_ = rspamd_redis_make_request(task,
+ ret,conn,upstream = rspamd_redis_make_request(task,
redis_params, -- connect params
key, -- hash key
true, -- is write
'data', rspamd_util.zstd_compress(data),
}) -- arguments
- if conn then
+ if ret then
conn:add_cmd('HINCRBY', {key, 'version', 1})
if is_spam then
conn:add_cmd('HINCRBY', {key, 'spam', 1})
- rspamd_logger.errx(task, 'hui')
else
conn:add_cmd('HINCRBY', {key, 'ham', 1})
- rspamd_logger.errx(task, 'pezda')
end
+ else
+ rspamd_logger.errx(task, 'redis error on host %s: %s', upstream:get_addr())
+ upstream:fail()
end
end
if redis_params then
rspamd_classifiers['neural'] = {
classify = function(task, classifier, tokens)
- local function classify_cb(task)
+ local function classify_cb()
local min_learns = classifier:get_param('min_learns')
if min_learns then
rspamd_logger.infox(task, 'fann classifier score: %s', symscore)
if result > 0 then
- each(function(st)
+ fun.each(function(st)
task:insert_result(st:get_symbol(), result, symscore)
end,
- filter(function(st)
+ fun.filter(function(st)
return st:is_spam()
end, classifier:get_statfiles())
)
else
- each(function(st)
+ fun.each(function(st)
task:insert_result(st:get_symbol(), -result, symscore)
end,
- filter(function(st)
+ fun.filter(function(st)
return not st:is_spam()
end, classifier:get_statfiles())
)
maybe_load_fann(task, classify_cb, false)
end,
- learn = function(task, classifier, tokens, is_spam, is_unlearn)
- local function learn_cb(task, is_loaded)
+ learn = function(task, _, tokens, is_spam, _)
+ local function learn_cb(_, is_loaded)
if not is_loaded then
create_fann()
end
maybe_load_fann(task, learn_cb, true)
end,
}
-end
\ No newline at end of file
+end
local fann_symbol_spam = 'FANN_SPAM'
local fann_symbol_ham = 'FANN_HAM'
local fun = require "fun"
-local ucl = require "ucl"
local module_log_id = 0x100
-- Module vars
local matched_symbols = {}
local n = rspamd_config:get_symbols_count()
- each(function(s, score)
+ fun.each(function(s, score)
matched_symbols[s + 1] = rspamd_util.tanh(score)
- end, zip(syms, scores))
+ end, fun.zip(syms, scores))
for i=1,n do
if matched_symbols[i] then
local function load_fann(id)
local fname = gen_fann_file(id)
- local err,st = rspamd_util.stat(fname)
+ local err = rspamd_util.stat(fname)
if err then
return false
' is found in the cache; removing', data[id].fann:get_inputs(), n)
data[id].fann = nil
- local ret,err = rspamd_util.unlink(fname)
+ local ret,_err = rspamd_util.unlink(fname)
if not ret then
rspamd_logger.errx(rspamd_config, 'cannot remove invalid fann from %s: %s',
- fname, err)
+ fname, _err)
end
else
local layers = data[id].fann:get_layers()
rspamd_logger.infox(rspamd_config, 'fann has incorrect number of layers: %s, removing',
#layers)
data[id].fann = nil
- local ret,err = rspamd_util.unlink(fname)
+ local ret,_err = rspamd_util.unlink(fname)
if not ret then
rspamd_logger.errx(rspamd_config, 'cannot remove invalid fann from %s: %s',
- fname, err)
+ fname, _err)
end
else
rspamd_logger.infox(rspamd_config, 'loaded fann from %s', fname)
end
else
rspamd_logger.infox(rspamd_config, 'fann is invalid: "%s"; removing', fname)
- local ret,err = rspamd_util.unlink(fname)
+ local ret,_err = rspamd_util.unlink(fname)
if not ret then
rspamd_logger.errx(rspamd_config, 'cannot remove invalid fann from %s: %s',
- fname, err)
+ fname, _err)
end
end
-- Store fann on disk
local res = false
- local err,st = rspamd_util.stat(fname)
+ local err = rspamd_util.stat(fname)
+ local fd
if err then
- local fd,err = rspamd_util.create_file(fname)
+ fd,err = rspamd_util.create_file(fname)
if not fd then
rspamd_logger.errx(cf, 'cannot save fann in %s: %s', fname, err)
else
rspamd_util.unlock_file(fd) -- Closes fd as well
end
else
- local fd = rspamd_util.lock_file(fname)
+ fd = rspamd_util.lock_file(fname)
res = data[id].fann_train:save(fname)
rspamd_util.unlock_file(fd) -- Closes fd as well
end
else
if not data[id].checked then
data[id].checked = true
- local err,st = rspamd_util.stat(fname)
+ local err = rspamd_util.stat(fname)
if err then
data[id].exist = false
end
create_train_fann(n, id)
end
- local learn_spam, learn_ham = false, false
+ local learn_spam, learn_ham
if opts['spam_score'] then
learn_spam = score >= opts['spam_score']
else
if learn_spam or learn_ham then
local learn_data = symbols_to_fann_vector(
- map(function(r) return r[1] end, results),
- map(function(r) return r[2] end, results)
+ fun.map(function(r) return r[1] end, results),
+ fun.map(function(r) return r[2] end, results)
)
-- Add filtered meta tokens
- each(function(e) table.insert(learn_data, e) end, extra)
+ fun.each(function(e) table.insert(learn_data, e) end, extra)
if learn_spam then
data[id].fann_train:train(learn_data, {1.0})
max_epoch = opts['train']['max_epoch']
end
local ret = cfg:register_worker_script("log_helper",
- function(score, req_score, results, cf, id, extra)
+ function(score, req_score, results, cf, _id, extra)
-- map (snd x) (filter (fst x == module_id) extra)
- local extra_fann = map(function(e) return e[2] end,
- filter(function(e) return e[1] == module_log_id end, extra))
+ local extra_fann = fun.map(function(e) return e[2] end,
+ fun.filter(function(e) return e[1] == module_log_id end, extra))
if use_settings then
fann_train_callback(score, req_score, results, cf,
- tostring(id), opts['train'], extra_fann)
+ tostring(_id), opts['train'], extra_fann)
else
fann_train_callback(score, req_score, results, cf, '0',
opts['train'], extra_fann)
end)
rspamd_plugins["fann_score"] = {
log_callback = function(task)
- return totable(map(
+ return fun.totable(fun.map(
function(tok) return {module_log_id, tok} end,
rspamd_gen_metatokens(task)))
end