<commit_before>%%%-------------------------------------------------------------------
%%% @author
%%% @copyright (C) 2016
%%% @doc
%%%
%%% @end
%%% Created :
%%%-------------------------------------------------------------------
-module(privacy_redis_to_pika).

%% delete bad user session, must run in ejabberdctl debug
%% privacy_redis_to_pika:start().
%% privacy_redis_to_pika:stop().

%% API
-export([
         start/4,
         start/0,
         start/1,
         stop/0,
         scan/6,
         scan_worker/5,
         do_scan/6,
         start_from_codis/2,
         stop_all/1,
         is_running/0,
         is_good_exit/0
]).

-define(SCAN_COUNT, 500).
-define(REDIS_TIMEOUT, 60000).
-define(INFO_MSG(Format, Args), spawn(fun()->error_logger:info_msg("~s:~p "++Format++"~n",[?MODULE, ?LINE]++Args) end)).
-define(ERROR_MSG(Format, Args), error_logger:error_msg("~s:~p "++Format++"~n",[?MODULE, ?LINE]++Args)).

-define(RUN_MODE, privacy_redis_to_pika).
-define(PRIVACY_REDIS_KEY, "im:privacyv2:").
-define(PRIVACY_PIKA_KEY, "im:privacy:pika:").

%%%===================================================================
%%% API
%%%===================================================================

start() ->
    start(no_host, no_port, redis, 100).

start(SleepTime) ->
    start(no_host, no_port, redis, SleepTime).

start(Host, Port, CodisOrRedis, SleepTime) ->
    scan(Host, Port, CodisOrRedis, ?RUN_MODE, [SleepTime], 1000000000).

stop() ->
    stop_all(?RUN_MODE).

is_running() ->
    Pids = application:get_env(message_store, {codis_scan_pid_list, ?RUN_MODE}, []),
    case [Pid||Pid<-Pids, is_process_alive(Pid)] of
        [] ->
            false;
        _ ->
            true
    end.

is_good_exit() ->
    Pids = application:get_env(message_store, {codis_scan_pid_list, ?RUN_MODE}, []),
    case [Pid||Pid<-Pids, 
            not is_process_alive(Pid), 
            application:get_env(message_store, {codis_scan_result, ?RUN_MODE, Pid}, exit) /= ok] of
        [] -> true;
        _ ->
            false
    end.

scan(Host, Port, CodisOrRedis, RunMode, RunArgs, MaxScan) ->
    stop_all(RunMode),
    ServerList = server_list(Host, Port, CodisOrRedis),
    Pids = [scan_worker(RedisHost, RedisPort, RunMode, RunArgs, MaxScan)
            ||{RedisHost, RedisPort}<-ServerList],
    application:set_env(message_store, {codis_scan_pid_list, RunMode}, Pids),
    {ok,Pids}.

server_list(Host, Port, codis) ->
    {ok, ServerList} = start_from_codis(Host, Port),
    ServerList;
server_list(Host, Port, redis) ->
    [{Host,Port}].

stop_all(RunMode) ->
    [begin
         application:unset_env(message_store, {codis_scan_result, RunMode, OldPid}),
         exit(OldPid, kill)
     end||OldPid<-application:get_env(message_store, {codis_scan_pid_list, RunMode}, [])],
    application:unset_env(message_store, {codis_scan_pid_list, RunMode}),
    ok.

scan_worker(RedisHost, RedisPort, RunMode, RunArgs, MaxScan) ->
    proc_lib:spawn(?MODULE, do_scan, [RedisHost, RedisPort, RunMode, RunArgs, 0, MaxScan]).

do_scan(RedisHost, RedisPort, RunMode, RunArgs, Cursor, MaxScan) when MaxScan > 0 ->
    {Now,Max} = get_progress(),
    ?INFO_MSG("do scan:cursor=~p, progress=~p/~p(~.2f%)",[Cursor, Now, Max, (Now*100/max(1,Max))]),
    case scan_key_list(RedisHost, RedisPort, Cursor) of
        {ok, finish, List} ->
            do_handle(List, RunMode, RunArgs),
            ?INFO_MSG("Scan finish with finish:s=~p", [ {RedisHost,RedisPort}]),
            application:set_env(message_store, {codis_scan_result, RunMode, self()}, ok),
            ok;
        {ok, Next, List}->
            do_handle(List, RunMode, RunArgs),
            ?MODULE:do_scan(RedisHost, RedisPort, RunMode, RunArgs, Next, MaxScan-1);
        {error, Error} ->
            ?INFO_MSG("Scan finish with Error:~p,s=~p", [Error, {RedisHost,RedisPort}]),
            application:set_env(message_store, {codis_scan_result, RunMode, self()}, {error, Error}),
            {error, Error}
    end;
do_scan(RedisHost, RedisPort, RunMode, _RunArgs, _Cursor, _MaxScan) ->
    ?INFO_MSG("Scan finish with :~p,s=~p", [max_scan, {RedisHost,RedisPort}]),
    application:set_env(message_store, {codis_scan_result, RunMode, self()}, ok),
    ok.


do_handle(List, ?RUN_MODE, [SleepTime]) ->
    update_progress(length(List)),
    ?INFO_MSG("do_handle keys=~p",[List]),
    try
        UserList = [User||Key<-List, User<-[key_to_user(Key)], User /= skip],
        QueryPipline = [[get,?PRIVACY_REDIS_KEY ++ User]||User<-UserList],
        case easemob_redis:qp(privacy, QueryPipline) of
            PrivacyList when is_list(PrivacyList) ->
                WritePipline = [[set, ?PRIVACY_PIKA_KEY ++ User, Value]||
                                   {User,{ok, Value}}<-lists:zip(UserList, PrivacyList),
                                   Value /= <<"[]">>],
                ?INFO_MSG("WritePipline:~p",[WritePipline]),
                easemob_redis:qp(privacy_pika, WritePipline);
            _ ->
                ok
        end
     catch
         C:E ->
             ?ERROR_MSG("handle_key_list error:keys=~p, error=~p",[List, {C,E,erlang:get_stacktrace()}])
     end,
    if
        SleepTime > 0 ->
            timer:sleep(SleepTime);
        true ->
            skip
    end,
    ok.

key_to_user(Key) ->
    case binary_to_list(Key) of
        ?PRIVACY_REDIS_KEY ++ Rest when length(Rest) > 0 ->
            Rest;
        _ ->
            skip
    end.

get_progress() ->
    maybe_init_progress(),
    Now = get({now_progress,?MODULE}),<commit_msg>remove print<commit_after>%%%-------------------------------------------------------------------
%%% @author
%%% @copyright (C) 2016
%%% @doc
%%%
%%% @end
%%% Created :
%%%-------------------------------------------------------------------
-module(privacy_redis_to_pika).

%% delete bad user session, must run in ejabberdctl debug
%% privacy_redis_to_pika:start().
%% privacy_redis_to_pika:stop().

%% API
-export([
         start/4,
         start/0,
         start/1,
         stop/0,
         scan/6,
         scan_worker/5,
         do_scan/6,
         start_from_codis/2,
         stop_all/1,
         is_running/0,
         is_good_exit/0
]).

-define(SCAN_COUNT, 500).
-define(REDIS_TIMEOUT, 60000).
-define(INFO_MSG(Format, Args), spawn(fun()->error_logger:info_msg("~s:~p "++Format++"~n",[?MODULE, ?LINE]++Args) end)).
-define(ERROR_MSG(Format, Args), error_logger:error_msg("~s:~p "++Format++"~n",[?MODULE, ?LINE]++Args)).

-define(RUN_MODE, privacy_redis_to_pika).
-define(PRIVACY_REDIS_KEY, "im:privacyv2:").
-define(PRIVACY_PIKA_KEY, "im:privacy:pika:").

%%%===================================================================
%%% API
%%%===================================================================

start() ->
    start(no_host, no_port, redis, 100).

start(SleepTime) ->
    start(no_host, no_port, redis, SleepTime).

start(Host, Port, CodisOrRedis, SleepTime) ->
    scan(Host, Port, CodisOrRedis, ?RUN_MODE, [SleepTime], 1000000000).

stop() ->
    stop_all(?RUN_MODE).

is_running() ->
    Pids = application:get_env(message_store, {codis_scan_pid_list, ?RUN_MODE}, []),
    case [Pid||Pid<-Pids, is_process_alive(Pid)] of
        [] ->
            false;
        _ ->
            true
    end.

is_good_exit() ->
    Pids = application:get_env(message_store, {codis_scan_pid_list, ?RUN_MODE}, []),
    case [Pid||Pid<-Pids, 
            not is_process_alive(Pid), 
            application:get_env(message_store, {codis_scan_result, ?RUN_MODE, Pid}, exit) /= ok] of
        [] -> true;
        _ ->
            false
    end.

scan(Host, Port, CodisOrRedis, RunMode, RunArgs, MaxScan) ->
    stop_all(RunMode),
    ServerList = server_list(Host, Port, CodisOrRedis),
    Pids = [scan_worker(RedisHost, RedisPort, RunMode, RunArgs, MaxScan)
            ||{RedisHost, RedisPort}<-ServerList],
    application:set_env(message_store, {codis_scan_pid_list, RunMode}, Pids),
    {ok,Pids}.

server_list(Host, Port, codis) ->
    {ok, ServerList} = start_from_codis(Host, Port),
    ServerList;
server_list(Host, Port, redis) ->
    [{Host,Port}].

stop_all(RunMode) ->
    [begin
         application:unset_env(message_store, {codis_scan_result, RunMode, OldPid}),
         exit(OldPid, kill)
     end||OldPid<-application:get_env(message_store, {codis_scan_pid_list, RunMode}, [])],
    application:unset_env(message_store, {codis_scan_pid_list, RunMode}),
    ok.

scan_worker(RedisHost, RedisPort, RunMode, RunArgs, MaxScan) ->
    proc_lib:spawn(?MODULE, do_scan, [RedisHost, RedisPort, RunMode, RunArgs, 0, MaxScan]).

do_scan(RedisHost, RedisPort, RunMode, RunArgs, Cursor, MaxScan) when MaxScan > 0 ->
    {Now,Max} = get_progress(),
    ?INFO_MSG("do scan:cursor=~p, progress=~p/~p(~.2f%)",[Cursor, Now, Max, (Now*100/max(1,Max))]),
    case scan_key_list(RedisHost, RedisPort, Cursor) of
        {ok, finish, List} ->
            do_handle(List, RunMode, RunArgs),
            ?INFO_MSG("Scan finish with finish:s=~p", [ {RedisHost,RedisPort}]),
            application:set_env(message_store, {codis_scan_result, RunMode, self()}, ok),
            ok;
        {ok, Next, List}->
            do_handle(List, RunMode, RunArgs),
            ?MODULE:do_scan(RedisHost, RedisPort, RunMode, RunArgs, Next, MaxScan-1);
        {error, Error} ->
            ?INFO_MSG("Scan finish with Error:~p,s=~p", [Error, {RedisHost,RedisPort}]),
            application:set_env(message_store, {codis_scan_result, RunMode, self()}, {error, Error}),
            {error, Error}
    end;
do_scan(RedisHost, RedisPort, RunMode, _RunArgs, _Cursor, _MaxScan) ->
    ?INFO_MSG("Scan finish with :~p,s=~p", [max_scan, {RedisHost,RedisPort}]),
    application:set_env(message_store, {codis_scan_result, RunMode, self()}, ok),
    ok.


do_handle(List, ?RUN_MODE, [SleepTime]) ->
    update_progress(length(List)),
    ?INFO_MSG("do_handle keys=~p",[List]),
    try
        UserList = [User||Key<-List, User<-[key_to_user(Key)], User /= skip],
        QueryPipline = [[get,?PRIVACY_REDIS_KEY ++ User]||User<-UserList],
        case easemob_redis:qp(privacy, QueryPipline) of
            PrivacyList when is_list(PrivacyList) ->
                WritePipline = [[set, ?PRIVACY_PIKA_KEY ++ User, Value]||
                                   {User,{ok, Value}}<-lists:zip(UserList, PrivacyList),
                                   Value /= <<"[]">>],
                %?INFO_MSG("WritePipline:~p",[WritePipline]),
                easemob_redis:qp(privacy_pika, WritePipline);
            _ ->
                ok
        end
     catch
         C:E ->
             ?ERROR_MSG("handle_key_list error:keys=~p, error=~p",[List, {C,E,erlang:get_stacktrace()}])
     end,
    if
        SleepTime > 0 ->
            timer:sleep(SleepTime);
        true ->
            skip
    end,
    ok.

key_to_user(Key) ->
    case binary_to_list(Key) of
        ?PRIVACY_REDIS_KEY ++ Rest when length(Rest) > 0 ->
            Rest;
        _ ->
            skip
    end.

get_progress() ->
    maybe_init_progress(),
    Now = get({now_progress,?MODULE}),
{'fraction_non_alphanumeric': 0.20113903453263424, 'fraction_numerical': 0.008497559211715784, 'mean_word_length': 2.783515731874145, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 75, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4206390', 'n_tokens_mistral': 3977, 'n_tokens_neox': 3846, 'n_words': 650}