ekfa/apps/efka/src/efka_downloader.erl

%%%-------------------------------------------------------------------
%%% @author anlicheng
%%% @copyright (C) 2025, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 19. 4月 2025 16:48
%%%-------------------------------------------------------------------
-module(efka_downloader).
-author("anlicheng").

-behaviour(gen_server).

%% API
-export([start_link/0, download/3]).
-export([test/0]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

-record(state, {

}).

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

test() ->
    {ok, Pid} = start_link(),
    Url = "http://118.178.229.213:3000/anlicheng/ekfa/archive/main.tar.gz",
    TargetDir = "/tmp/",
    Ref = download(Pid, Url, TargetDir),
    receive
        {download_response, Ref, Info} ->
            efka_logger:debug("info is: ~p", [Info])
    end,
    ok.

-spec download(Pid :: pid(), Url :: string(), TargetDir :: string()) -> reference().
download(Pid, Url, TargetDir) when is_pid(Pid), is_list(Url), is_list(TargetDir) ->
    Ref = make_ref(),
    ReceiverPid = self(),
    gen_server:cast(Pid, {download, ReceiverPid, Ref, Url, TargetDir}),
    Ref.

%% @doc Spawns the server and registers the local name (unique)
-spec(start_link() ->
    {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link() ->
    gen_server:start_link(?MODULE, [], []).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%% @private
%% @doc Initializes the server
-spec(init(Args :: term()) ->
    {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term()} | ignore).
init([]) ->
    {ok, #state{}}.

%% @private
%% @doc Handling call messages
-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
    State :: #state{}) ->
    {reply, Reply :: term(), NewState :: #state{}} |
    {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
    {noreply, NewState :: #state{}} |
    {noreply, NewState :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
    {stop, Reason :: term(), NewState :: #state{}}).
handle_call(_Request, _From, State = #state{}) ->
    {reply, ok, State}.

%% @private
%% @doc Handling cast messages
-spec(handle_cast(Request :: term(), State :: #state{}) ->
    {noreply, NewState :: #state{}} |
    {noreply, NewState :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term(), NewState :: #state{}}).
handle_cast({download, ReceiverPid, Ref, Url, TargetDir}, State = #state{}) ->
    SSLOpts = {ssl_options, [
        % 完全禁用证书验证
        {verify, verify_none}
    ]},

    TargetFile = get_filename_from_url(Url),
    StartTs = os:timestamp(),
    case hackney:request(get, Url, [], <<>>, [async, {stream_to, self()}, {pool, false}, SSLOpts]) of
        {ok, ClientRef} ->
            case receive_data(ClientRef, TargetDir, TargetFile) of
                ok ->
                    EndTs = os:timestamp(),
                    %% 计算操作的时间，单位为毫秒
                    CostMs = timer:now_diff(EndTs, StartTs) div 1000,
                    efka_logger:debug("download cost: ~p", [CostMs]),
                    ReceiverPid ! {download_response, Ref, {ok, CostMs}};
                {error, Reason} ->
                    ReceiverPid ! {download_response, Ref, {error, Reason}}
            end;
        {error, Reason} ->
            ReceiverPid ! {download_response, Ref, {error, Reason}}
    end,
    {noreply, State}.

%% @private
%% @doc Handling all non call/cast messages
-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
    {noreply, NewState :: #state{}} |
    {noreply, NewState :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term(), NewState :: #state{}}).
handle_info(_Info, State = #state{}) ->
    {noreply, State}.

%% @private
%% @doc This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_server terminates
%% with Reason. The return value is ignored.
-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
    State :: #state{}) -> term()).
terminate(_Reason, _State = #state{}) ->
    ok.

%% @private
%% @doc Convert process state when code is changed
-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
    Extra :: term()) ->
    {ok, NewState :: #state{}} | {error, Reason :: term()}).
code_change(_OldVsn, State = #state{}, _Extra) ->
    {ok, State}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% 读取请求 ResponseLine
receive_data(ClientRef, TargetDir, DefaultFile) when is_list(TargetDir), is_list(DefaultFile) ->
    receive
        {hackney_response, ClientRef, {status, 200, _Reason}} ->
            receive_data0(ClientRef, TargetDir, DefaultFile);
        {hackney_response, ClientRef, {status, StatusCode, _Reason}} ->
            {error, {http_status, StatusCode}}
    end.
%% 处理头部信息, 解析可能的文件名
receive_data0(ClientRef, TargetDir, DefaultFile) ->
    receive
        {hackney_response, ClientRef, {headers, Headers}} ->
            TargetFilename = extra_filename(Headers, DefaultFile),
            FullFilename = TargetDir ++ TargetFilename,
            efka_logger:debug("full name: ~p", [FullFilename]),
            {ok, File} = file:open(FullFilename, [write, binary]),
            receive_data1(ClientRef, File)
    end.
%% 接受文件数据
receive_data1(ClientRef, File) ->
    receive
        {hackney_response, ClientRef, {error, Reason}} ->
            ok = file:close(File),
            {error, Reason};
        {hackney_response, ClientRef, done} ->
            ok = file:close(File),
            hackney:close(ClientRef),
            ok;
        {hackney_response, ClientRef, Data} ->
            file:write(File, Data),
            receive_data1(ClientRef, File)
    end.

-spec extra_filename(Headers :: list(), Default :: string()) -> Filename :: string().
extra_filename(Headers, Default) when is_list(Headers), is_list(Default) ->
    case lists:filter(fun({K, _}) -> string:lowercase(K) =:= <<"content-disposition">> end, Headers) of
        [{_, <<"attachment; ", Rest/binary>>}|_] ->
            Params0 = binary:split(Rest, <<";">>, [global]),
            Params = lists:map(fun(P) -> list_to_tuple(binary:split(P, <<"=">>)) end, Params0),
            case proplists:get_value(<<"filename">>, Params) of
                undefined ->
                    Default;
                Filename0 ->
                    Filename = binary_to_list(Filename0),
                    string:trim(Filename, both, "\"")
            end;
        _ ->
            Default
    end.

-spec get_filename_from_url(Url :: string()) -> string().
get_filename_from_url(Url) when is_list(Url) ->
    URIMap = uri_string:parse(Url),
    Path = maps:get(path, URIMap),
    filename:basename(Path).