test/dgiot
1
0
Fork 0
mirror of https://gitee.com/dgiiot/dgiot.git synced 2024-11-30 03:07:40 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:dgiot/dgiot

Browse Source
This commit is contained in:
U-JOHNLIU\jonhl 2024-03-23 02:25:42 +08:00
commit c92793d5e2
23 changed files with 733 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
apps/dgiot_device/src/dgiot_device.erl
View File

@ -77,7 +77,7 @@ get_productid(DeviceId) ->
{ok, #{<<"productid">> := ProductId}} ->
ProductId;
_ ->
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"product">> := #{<<"objectId">> := ProductId}}} ->
ProductId;
_ ->
@ -132,13 +132,13 @@ save_subdevice(DeviceId, DtuDeviceId, SlaveId) ->
<<"className">> => <<"Device">>,
<<"objectId">> => DtuDeviceId
},
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"route">> := OldRoute}} ->
dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"route">> => OldRoute#{DtuDeviceId => SlaveId}, <<"parentId">> => Parent});
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"route">> => OldRoute#{DtuDeviceId => SlaveId}, <<"parentId">> => Parent});
_ ->
dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"route">> => #{DtuDeviceId => SlaveId}, <<"parentId">> => Parent})
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"route">> => #{DtuDeviceId => SlaveId}, <<"parentId">> => Parent})
end.
get_sub_device(DtuDeviceId) ->
@ -146,7 +146,7 @@ get_sub_device(DtuDeviceId) ->
<<"where">> => #{<<"route.", DtuDeviceId/binary>> => #{<<"$regex">> => <<".+">>}},
<<"order">> => <<"devaddr">>, <<"limit">> => 1000,
<<"include">> => <<"product">>},
case dgiot_parse:query_object(<<"Device">>, Query) of
case dgiot_parsex:query_object(<<"Device">>, Query) of
{ok, #{<<"results">> := []}} -> [];
{ok, #{<<"results">> := List}} -> List;
_ -> []
@ -157,7 +157,7 @@ get_subdevices(DtuDeviceId, Keys) ->
Query = #{<<"keys">> => NewKeys,
<<"where">> => #{<<"route.", DtuDeviceId/binary>> => #{<<"$regex">> => <<".+">>}},
<<"order">> => <<"devaddr">>, <<"limit">> => 100},
case dgiot_parse:query_object(<<"Device">>, Query) of
case dgiot_parsex:query_object(<<"Device">>, Query) of
{ok, #{<<"results">> := []}} -> [];
{ok, #{<<"results">> := List}} -> List;
_ -> []
@ -168,7 +168,7 @@ get_sub_device(DtuAddr, SessionToken) ->
<<"where">> => #{<<"route.", DtuAddr/binary>> => #{<<"$regex">> => <<".+">>}},
<<"order">> => <<"devaddr">>, <<"limit">> => 1000,
<<"include">> => <<"product">>},
case dgiot_parse:query_object(<<"Device">>, Query, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
case dgiot_parsex:query_object(<<"Device">>, Query, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
{ok, #{<<"results">> := []}} -> [];
{ok, #{<<"results">> := List}} -> List;
_ -> []
@ -177,12 +177,12 @@ get_sub_device(DtuAddr, SessionToken) ->
create_device(#{<<"status">> := Status, <<"brand">> := Brand, <<"devModel">> := DevModel, <<"name">> := Name,
<<"devaddr">> := DevAddr, <<"product">> := ProductId, <<"ACL">> := Acl} = Device) ->
DeviceId = maps:get(<<"objectId">>, Device, dgiot_parse_id:get_deviceid(ProductId, DevAddr)),
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"ip">> := Ip} = Result} ->
Body = #{
<<"ip">> => maps:get(<<"ip">>, Device, Ip),
<<"status">> => Status},
dgiot_parse:update_object(<<"Device">>, DeviceId, Body),
dgiot_parsex:update_object(<<"Device">>, DeviceId, Body),
dgiot_device:put(#{<<"objectId">> => DeviceId, <<"status">> => Status}),
{ok, Result};
_R ->
@ -215,7 +215,7 @@ create_device(#{<<"status">> := Status, <<"brand">> := Brand, <<"devModel">> :=
}
},
case dgiot_parse:create_object(<<"Device">>, maps:without([<<"brand">>, <<"devModel">>], NewDevice)) of
case dgiot_parsex:create_object(<<"Device">>, maps:without([<<"brand">>, <<"devModel">>], NewDevice)) of
{ok, R} ->
dgiot_device:post(NewDevice#{<<"product">> => ProductId}),
{ok, R};
@ -230,7 +230,7 @@ create_device(ProductId, DeviceAddr, Ip) ->
{ok, _} ->
Body = #{<<"status">> => <<"ONLINE">>},
dgiot_device:online(DeviceId),
dgiot_parse:update_object(<<"Device">>, DeviceId, Body);
dgiot_parsex:update_object(<<"Device">>, DeviceId, Body);
_ ->
case dgiot_product:lookup_prod(ProductId) of
not_find ->
@ -258,7 +258,7 @@ create_device(ProductId, DeviceAddr, Ip) ->
get(ProductId, DevAddr) ->
Keys = [<<"objectId">>, <<"status">>, <<"isEnable">>],
DeviceId = dgiot_parse_id:get_deviceid(ProductId, DevAddr),
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, Device} ->
case maps:get(<<"isEnable">>, Device, false) of
false ->
@ -272,7 +272,7 @@ get(ProductId, DevAddr) ->
get_url(AppName) ->
Roleid = dgiot_parse_id:get_roleid(AppName),
case dgiot_parse:get_object(<<"_Role">>, Roleid) of
case dgiot_parsex:get_object(<<"_Role">>, Roleid) of
{ok, #{<<"tag">> := #{<<"appconfig">> := #{<<"file">> := Url}}}} ->
<<Url/binary>>;
_ -> <<"">>

2
apps/dgiot_device/src/dgiot_device_channel.erl
View File

@ -279,7 +279,7 @@ handle_message({sync_parse, Pid, 'before', put, Token, <<"Device">>, #{<<"locati
handle_message({sync_parse, _Pid, 'before', delete, _Token, <<"Channel">>, ObjectId}, State) ->
%% io:format("~s ~p ~p ~n", [?FILE, ?LINE, ObjectId]),
case dgiot_parse:get_object(<<"Channel">>, ObjectId) of
case dgiot_parsex:get_object(<<"Channel">>, ObjectId) of
{ok, #{<<"isEnable">> := true}} ->
dgiot_bridge:control_channel(ObjectId, <<"disable">>, <<>>);
_ -> pass

8
apps/dgiot_device/src/dgiot_device_hook.erl
View File

@ -35,7 +35,7 @@ put(_, _) ->
delete('after', DeviceId) ->
dgiot_device:delete(DeviceId),
case dgiot_parse:query_object(<<"Dict">>, #{<<"where">> => #{<<"key">> => DeviceId, <<"class">> => <<"Device">>}}) of
case dgiot_parsex:query_object(<<"Dict">>, #{<<"where">> => #{<<"key">> => DeviceId, <<"class">> => <<"Device">>}}) of
{ok, #{<<"results">> := Dicts}} ->
DictRequests =
lists:foldl(fun(#{<<"objectId">> := DictId}, Acc) ->
@ -45,11 +45,11 @@ delete('after', DeviceId) ->
<<"body">> => #{}
}]
end, [], Dicts),
dgiot_parse:batch(DictRequests);
dgiot_parsex:batch(DictRequests);
_ ->
pass
end,
case dgiot_parse:query_object(<<"View">>, #{<<"where">> => #{<<"key">> => DeviceId, <<"class">> => <<"Device">>}}) of
case dgiot_parsex:query_object(<<"View">>, #{<<"where">> => #{<<"key">> => DeviceId, <<"class">> => <<"Device">>}}) of
{ok, #{<<"results">> := Views}} ->
ViewRequests =
lists:foldl(fun(#{<<"objectId">> := ViewId}, Acc) ->
@ -59,7 +59,7 @@ delete('after', DeviceId) ->
<<"body">> => #{}
}]
end, [], Views),
dgiot_parse:batch(ViewRequests);
dgiot_parsex:batch(ViewRequests);
_ ->
pass
end;

6
apps/dgiot_device/src/dgiot_device_profile.erl
View File

@ -94,17 +94,17 @@ publish(ProductId, DeviceAddr, DeviceProfile, Delay) ->
end.
update_profile(DeviceId, NewProfile) ->
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, Device} ->
OldProfile = maps:get(<<"profile">>, Device, #{}),
dgiot_parse:update_object(<<"Device">>, DeviceId, #{
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{
<<"profile">> => dgiot_map:merge(OldProfile, NewProfile)});
_ ->
pass
end.
encode_profile(ProductId, Profile) ->
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, #{<<"name">> := ProductName, <<"thing">> := #{<<"properties">> := Properties}}} ->
lists:foldl(fun(X, Acc) ->
case X of

4
apps/dgiot_device/src/dgiot_maintenance.erl
View File

@ -15,7 +15,7 @@
%% 初始化工单巡检动态数据
init_inspection(#{<<"objectId">> := MaintenanceId, <<"info">> := Info, <<"status">> := 1, <<"product">> := #{<<"objectId">> := ProductId}, <<"device">> := #{<<"objectId">> := DeviceId}} = _QueryData) ->
InitData = get_initdata(<<"巡检"/utf8>>, ProductId),
dgiot_parse:update_object(<<"Maintenance">>, MaintenanceId, #{<<"info">> => Info#{<<"dynamicdata">> => InitData}}),
dgiot_parsex:update_object(<<"Maintenance">>, MaintenanceId, #{<<"info">> => Info#{<<"dynamicdata">> => InitData}}),
%% 下发巡检信息
%% $dg/device/{productId}/{deviceAddr}/init/response/inspection
case dgiot_device:lookup(DeviceId) of
@ -56,7 +56,7 @@ get_initdata(Type, ProductId) ->
end.
get_inspection(MaintenanceId) ->
case dgiot_parse:get_object(<<"Maintenance">>, MaintenanceId) of
case dgiot_parsex:get_object(<<"Maintenance">>, MaintenanceId) of
{ok, #{<<"number">> := Number, <<"status">> := Status, <<"info">> := Info}} ->
Dynamicdata = maps:get(<<"dynamicdata">>, Info, []),
Basic = #{

2
apps/dgiot_device/src/dgiot_notification_channel.erl
View File

@ -176,6 +176,6 @@ test(ProductId) ->
},
<<"type">> => <<ProductId/binary, "_alarm">>
},
dgiot_parse:create_object(<<"Notification">>, Body)
dgiot_parsex:create_object(<<"Notification">>, Body)
end, lists:seq(1, 6000))
end).

10
apps/dgiot_device/src/dgiot_ota_channel.erl
View File

@ -88,10 +88,10 @@ handle_message({dlink_ota_progress, ProductId, DevAddr, _Payload}, State) ->
%% 上报最新版本
handle_message({dlink_firmware_report, ProductId, DevAddr, #{<<"version">> := Version}}, State) ->
DeviceId = dgiot_parse_id:get_deviceid(ProductId, DevAddr),
dgiot_parse:update_object(<<"Device">>, DeviceId, #{}),
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{}),
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"basedata">> := Basedata} = _Device} ->
dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"basedata">> => Basedata#{<<"version">> => Version}});
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"basedata">> => Basedata#{<<"version">> => Version}});
_ ->
pass
end,
@ -113,7 +113,7 @@ stop(_ChannelType, _ChannelId, _State) ->
%% 全部设备
get_deviceids(#{<<"upgraderange">> := <<"0">>, <<"key">> := ProductId}) ->
case dgiot_parse:query_object(<<"Device">>, #{<<"keys">> => [<<"devaddr">>], <<"where">> => #{<<"product">> => ProductId}}) of
case dgiot_parsex:query_object(<<"Device">>, #{<<"keys">> => [<<"devaddr">>], <<"where">> => #{<<"product">> => ProductId}}) of
{ok, #{<<"results">> := Devices}} when length(Devices) > 0 ->
lists:foldl(fun(#{<<"objectId">> := DeviceId, <<"devaddr">> := Devaddr}, Acc) ->
Acc ++ [#{<<"objectId">> => DeviceId, <<"devaddr">> => Devaddr}]
@ -142,7 +142,7 @@ get_deviceids(#{<<"upgraderange">> := <<"1">>, <<"picker">> := Picker}) ->
get_deviceids(#{<<"upgraderange">> := <<"2">>, <<"danwei">> := RoleName}) ->
case dgiot_parse_auth:check_roles(RoleName) of
{200, #{<<"access_token">> := Depart_token}} ->
case dgiot_parse:query_object(<<"Device">>, #{}, [{"X-Parse-Session-Token", Depart_token}], [{from, rest}]) of
case dgiot_parsex:query_object(<<"Device">>, #{}, [{"X-Parse-Session-Token", Depart_token}], [{from, rest}]) of
{ok, #{<<"results">> := Devices}} when length(Devices) > 0 ->
lists:foldl(fun(#{<<"objectId">> := DeviceId, <<"devaddr">> := Devaddr}, Acc) ->
Acc ++ [#{<<"objectId">> => DeviceId, <<"devaddr">> => Devaddr}]

32
apps/dgiot_device/src/dgiot_product.erl
View File

@ -103,7 +103,7 @@ delete(ProductId) ->
get(ProductId) ->
Keys = [<<"ACL">>, <<"name">>, <<"devType">>, <<"status">>, <<"content">>, <<"profile">>, <<"nodeType">>, <<"dynamicReg">>, <<"topics">>, <<"productSecret">>],
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, Product} ->
{ok, maps:with(Keys, Product)};
{error, Reason} ->
@ -255,14 +255,14 @@ update_properties(ProductId, Product) ->
NewThing = Thing#{
<<"properties">> => NewProperties
},
dgiot_parse:update_object(<<"Product">>, ProductId, #{<<"thing">> => NewThing}),
dgiot_parsex:update_object(<<"Product">>, ProductId, #{<<"thing">> => NewThing}),
dgiot_data:insert(?DGIOT_PRODUCT, ProductId, Product#{<<"thing">> => NewThing});
_Error ->
[]
end.
update_properties() ->
case dgiot_parse:query_object(<<"Product">>, #{<<"skip">> => 0}) of
case dgiot_parsex:query_object(<<"Product">>, #{<<"skip">> => 0}) of
{ok, #{<<"results">> := Results}} ->
lists:foldl(fun(X, _Acc) ->
case X of
@ -278,16 +278,16 @@ update_properties() ->
%% 更新topics
update_topics() ->
{ok, #{<<"fields">> := Fields}} = dgiot_parse:get_schemas(<<"Product">>),
{ok, #{<<"fields">> := Fields}} = dgiot_parsex:get_schemas(<<"Product">>),
#{<<"type">> := Type} = maps:get(<<"topics">>, Fields),
case Type of
<<"Object">> ->
%% 判断目前topics 是那种类型，如果是object类型，就不更新
%% 删除原有topics字段
case dgiot_parse:del_filed_schemas(<<"topics">>, <<"Product">>) of
case dgiot_parsex:del_filed_schemas(<<"topics">>, <<"Product">>) of
{ok, _} ->
%% 新增topics字段
dgiot_parse:create_schemas(<<"Product">>, #{<<"topics">> => []});
dgiot_parsex:create_schemas(<<"Product">>, #{<<"topics">> => []});
Err ->
io:format("~s ~p ~p ~n", [?FILE, ?LINE, Err])
end;
@ -296,7 +296,7 @@ update_topics() ->
%% 产品字段新增
update_product_filed(_Filed) ->
case dgiot_parse:query_object(<<"Product">>, #{<<"skip">> => 0}) of
case dgiot_parsex:query_object(<<"Product">>, #{<<"skip">> => 0}) of
{ok, #{<<"results">> := Results}} ->
lists:foldl(fun(X, _Acc) ->
case X of
@ -369,9 +369,9 @@ format_product(#{<<"objectId">> := ProductId} = Product) ->
create_product(#{<<"name">> := ProductName, <<"devType">> := DevType, <<"category">> := #{
<<"objectId">> := CategoryId, <<"__type">> := <<"Pointer">>, <<"className">> := <<"Category">>}} = Product, SessionToken) ->
ProductId = dgiot_parse_id:get_productid(CategoryId, DevType, ProductName),
case dgiot_parse:get_object(<<"Product">>, ProductId, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
case dgiot_parsex:get_object(<<"Product">>, ProductId, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
{ok, #{<<"objectId">> := ObjectId}} ->
dgiot_parse:update_object(<<"Product">>, ObjectId, Product,
dgiot_parsex:update_object(<<"Product">>, ObjectId, Product,
[{"X-Parse-Session-Token", SessionToken}], [{from, rest}]);
_ ->
ACL = maps:get(<<"ACL">>, Product, #{}),
@ -386,7 +386,7 @@ create_product(#{<<"name">> := ProductName, <<"devType">> := DevType, <<"categor
}
},
<<"productSecret">> => dgiot_utils:random()},
dgiot_parse:create_object(<<"Product">>,
dgiot_parsex:create_object(<<"Product">>,
CreateProductArgs, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]);
Err ->
{400, Err}
@ -396,12 +396,12 @@ create_product(#{<<"name">> := ProductName, <<"devType">> := DevType, <<"categor
create_product(#{<<"name">> := ProductName, <<"devType">> := DevType, <<"category">> := #{
<<"objectId">> := CategoryId, <<"__type">> := <<"Pointer">>, <<"className">> := <<"Category">>}} = Product) ->
ProductId = dgiot_parse_id:get_productid(CategoryId, DevType, ProductName),
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, #{<<"objectId">> := ObjectId}} ->
dgiot_parse:update_object(<<"Product">>, ObjectId, Product),
dgiot_parsex:update_object(<<"Product">>, ObjectId, Product),
{ok, ObjectId};
_ ->
case dgiot_parse:create_object(<<"Product">>, Product) of
case dgiot_parsex:create_object(<<"Product">>, Product) of
{ok, #{<<"objectId">> := ObjectId}} ->
{ok, ObjectId};
{error, Reason} ->
@ -425,7 +425,7 @@ add_product_relation(ChannelIds, ProductId) ->
},
lists:map(fun
(ChannelId) when size(ChannelId) > 0 ->
dgiot_parse:update_object(<<"Channel">>, ChannelId, Map);
dgiot_parsex:update_object(<<"Channel">>, ChannelId, Map);
(_) ->
pass
end, ChannelIds).
@ -442,11 +442,11 @@ delete_product_relation(ProductId) ->
}
]}
},
case dgiot_parse:query_object(<<"Channel">>, #{<<"where">> => #{<<"product">> => #{
case dgiot_parsex:query_object(<<"Channel">>, #{<<"where">> => #{<<"product">> => #{
<<"__type">> => <<"Pointer">>, <<"className">> => <<"Product">>, <<"objectId">> => ProductId}}, <<"limit">> => 20}) of
{ok, #{<<"results">> := Results}} when length(Results) > 0 ->
lists:foldl(fun(#{<<"objectId">> := ChannelId}, _Acc) ->
dgiot_parse:update_object(<<"Channel">>, ChannelId, Map)
dgiot_parsex:update_object(<<"Channel">>, ChannelId, Map)
end, [], Results);
_ ->
[]

8
apps/dgiot_device/src/dgiot_product_hook.erl
View File

@ -33,13 +33,13 @@ post('after', #{<<"objectId">> := ProductId, <<"channel">> := Channel} = QueryDa
post('after', maps:without([<<"channel">>], QueryData));
%%post('after', #{<<"objectId">> := ProductId, <<"producttemplet">> := #{<<"objectId">> := ProducttempletId}} = _QueryData) ->
%% case dgiot_parse:query_object(<<"Dict">>, #{<<"where">> => #{<<"key">> => ProducttempletId, <<"class">> => <<"ProductTemplet">>}}) of
%% case dgiot_parsex:query_object(<<"Dict">>, #{<<"where">> => #{<<"key">> => ProducttempletId, <<"class">> => <<"ProductTemplet">>}}) of
%% {ok, #{<<"results">> := Dicts}} when length(Dicts) > 0 ->
%% dgiot_product_dict:post_batch(Dicts, ProductId);
%% _ ->
%% pass
%% end,
%% case dgiot_parse:query_object(<<"View">>, #{<<"where">> => #{<<"key">> => ProducttempletId, <<"class">> => <<"ProductTemplet">>}}) of
%% case dgiot_parsex:query_object(<<"View">>, #{<<"where">> => #{<<"key">> => ProducttempletId, <<"class">> => <<"ProductTemplet">>}}) of
%% {ok, #{<<"results">> := Views}} when length(Views) > 0 ->
%% dgiot_product_view:post_batch(Views, ProductId);
%% _ ->
@ -68,13 +68,13 @@ delete('before', _ProductId) ->
pass;
%%delete('after', ProductId) ->
%% case dgiot_parse:query_object(<<"Dict">>, #{<<"where">> => #{<<"key">> => ProductId, <<"class">> => <<"Product">>}}) of
%% case dgiot_parsex:query_object(<<"Dict">>, #{<<"where">> => #{<<"key">> => ProductId, <<"class">> => <<"Product">>}}) of
%% {ok, #{<<"results">> := Dicts}} ->
%% dgiot_product_dict:delete_batch(Dicts);
%% _ ->
%% pass
%% end,
%% case dgiot_parse:query_object(<<"View">>, #{<<"where">> => #{<<"key">> => ProductId, <<"class">> => <<"Product">>}}) of
%% case dgiot_parsex:query_object(<<"View">>, #{<<"where">> => #{<<"key">> => ProductId, <<"class">> => <<"Product">>}}) of
%% {ok, #{<<"results">> := Views}} ->
%% dgiot_product_view:delete_batch(Views);
%% _ ->

20
apps/dgiot_device/src/handler/dgiot_group_handler.erl
View File

@ -74,7 +74,7 @@ handle(OperationID, Args, Context, Req) ->
do_request(get_sports, _Args, _Context, _Req) ->
Items =
case dgiot_parse:query_object(<<"Device">>, #{<<"where">> => #{<<"product">> => <<"2dd1d8b42a">>}}) of
case dgiot_parsex:query_object(<<"Device">>, #{<<"where">> => #{<<"product">> => <<"2dd1d8b42a">>}}) of
{ok, #{<<"results">> := Devices}} ->
lists:foldl(fun(#{
<<"objectId">> := ObjectId,
@ -101,7 +101,7 @@ do_request(post_sports, #{<<"ids">> := Ids, <<"type">> := Type} = _Args, _Contex
%% io:format("Args ~p~n", [Args]),
lists:foldl(fun(DeviceId, _Acc) ->
dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"realstate">> => Type})
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"realstate">> => Type})
end, [], binary:split(Ids, <<$,>>, [global, trim])),
@ -116,7 +116,7 @@ do_request(post_sports, #{<<"ids">> := Ids, <<"type">> := Type} = _Args, _Contex
do_request(get_group_id, #{<<"id">> := Id},
#{<<"sessionToken">> := SessionToken} = _Context, _Req) ->
?LOG(info, "Id ~p", [Id]),
case dgiot_parse:query_object(<<"Device">>, #{<<"limit">> => 1, <<"where">> => #{<<"product">> => Id}},
case dgiot_parsex:query_object(<<"Device">>, #{<<"limit">> => 1, <<"where">> => #{<<"product">> => Id}},
[{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
{ok, #{<<"results">> := [Device | _]}} ->
#{<<"devaddr">> := Devaddr, <<"objectId">> := ParentId} = Device,
@ -161,7 +161,7 @@ do_request(_OperationId, _Args, _Context, _Req) ->
{error, <<"Not Allowed.">>}.
put_group(#{<<"productid">> := ProductId, <<"topoid">> := TopoId, <<"thingid">> := ThingId} = Payload, SessionToken) ->
case dgiot_parse:get_object(<<"Product">>, ProductId, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
case dgiot_parsex:get_object(<<"Product">>, ProductId, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
{ok, #{<<"config">> := Config, <<"thing">> := #{<<"properties">> := Properties}}} when length(Properties) > 0 ->
ControlList =
lists:foldl(fun(X, Acc) ->
@ -219,7 +219,7 @@ put_group(#{<<"productid">> := ProductId, <<"topoid">> := TopoId, <<"thingid">>
_ -> Acc ++ [X]
end
end, [], Properties),
case dgiot_parse:update_object(<<"Product">>, ProductId, #{
case dgiot_parsex:update_object(<<"Product">>, ProductId, #{
<<"config">> => Config#{<<"components">> => ControlList},
<<"thing">> => #{<<"properties">> => NewProperties}}) of
{ok, _} -> {ok, #{
@ -247,7 +247,7 @@ put_group(#{<<"productid">> := ProductId, <<"topoid">> := TopoId, <<"thingid">>
_ -> Acc ++ [X]
end
end, [], maps:get(<<"components">>, Config)),
case dgiot_parse:update_object(<<"Product">>, ProductId, #{
case dgiot_parsex:update_object(<<"Product">>, ProductId, #{
<<"config">> => Config#{<<"components">> => ControlList}}) of
{ok, _} -> {ok, #{
<<"config">> => Config#{<<"components">> => ControlList}}
@ -311,13 +311,13 @@ post_group(Body, SessionToken) ->
end.
delete_group(#{<<"name">> := _Name, <<"devType">> := _DevType} = Where, SessionToken) ->
case dgiot_parse:query_object(<<"Product">>, #{<<"where">> => Where, <<"limit">> => 1}) of
case dgiot_parsex:query_object(<<"Product">>, #{<<"where">> => Where, <<"limit">> => 1}) of
{ok, #{<<"results">> := [#{<<"objectId">> := ProductId} | _]}} ->
case dgiot_parse:query_object(<<"Device">>, #{<<"where">> => Where, <<"limit">> => 1}) of
case dgiot_parsex:query_object(<<"Device">>, #{<<"where">> => Where, <<"limit">> => 1}) of
{ok, #{<<"results">> := [#{<<"objectId">> := DeviceId} | _]}} ->
dgiot_parse:del_object(<<"Device">>, DeviceId, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]);
dgiot_parsex:del_object(<<"Device">>, DeviceId, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]);
_ -> pass
end,
dgiot_parse:del_object(<<"Product">>, ProductId, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]);
dgiot_parsex:del_object(<<"Product">>, ProductId, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]);
Error -> Error
end.

12
apps/dgiot_device/src/handler/dgiot_tdengine_handler.erl
View File

@ -120,7 +120,7 @@ do_request(get_device_deviceid, #{<<"deviceid">> := DeviceId} = Args, #{<<"sessi
{error, Error} -> {error, Error};
{ok, Channel} ->
?LOG(info, "DeviceId ~p", [DeviceId]),
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"objectId">> := DeviceId, <<"product">> := #{<<"objectId">> := ProductId}}} ->
dgiot_product_tdengine:get_product_data(Channel, ProductId, DeviceId, Args);
_ ->
@ -133,7 +133,7 @@ do_request(get_echart_deviceid, #{<<"deviceid">> := DeviceId, <<"style">> := Sty
case dgiot_product_tdengine:get_channel(SessionToken) of
{error, Error} -> {error, Error};
{ok, Channel} ->
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"objectId">> := DeviceId, <<"product">> := #{<<"objectId">> := ProductId}}} ->
case Style of
<<"amis_table">> ->
@ -156,7 +156,7 @@ do_request(get_devicecard_deviceid, #{<<"deviceid">> := DeviceId} = Args, #{<<"s
{error, Error};
{ok, Channel} ->
%% ?LOG(info,"DeviceId ~p", [DeviceId]),
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"objectId">> := DeviceId, <<"product">> := #{<<"objectId">> := ProductId}}} ->
dgiot_mqtt:subscribe_route_key([<<"$dg/user/realtimecard/", DeviceId/binary, "/#">>], <<"realtimecard">>, SessionToken),
dgiot_device_card:get_device_card(Channel, ProductId, DeviceId, Args);
@ -171,7 +171,7 @@ do_request(get_gps_track_deviceid, #{<<"deviceid">> := DeviceId, <<"starttime">>
Endtime = dgiot_datetime:get_today_stamp(dgiot_utils:to_int(dgiot_utils:to_int(End) / 1000)) * 1000,
case DeviceId of
<<"all">> ->
case dgiot_parse:query_object(<<"Device">>, #{}, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
case dgiot_parsex:query_object(<<"Device">>, #{}, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
{ok, #{<<"results">> := Results}} ->
NewResults =
lists:foldl(fun(#{<<"objectId">> := ObjectId, <<"name">> := Name}, Acc) ->
@ -209,11 +209,11 @@ do_request(get_gps_track_deviceid, #{<<"deviceid">> := DeviceId, <<"starttime">>
%% TDengine 概要: save_td
do_request(post_save_td, #{<<"productid">> := ProductId, <<"devaddr">> := DevAddr, <<"data">> := Ack} = _Args, _Context, _Req) ->
DeviceId = dgiot_parse_id:get_deviceid(ProductId, DevAddr),
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"devaddr">> := DevAddr}} ->
pass;
_ ->
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, #{<<"ACL">> := Acl, <<"devType">> := DevType}} ->
dgiot_device:create_device(#{
<<"devaddr">> => DevAddr,

18
apps/dgiot_device/src/utils/dgiot_device_cache.erl
View File

@ -54,7 +54,7 @@ save(ProductId, DevAddr) ->
save(#{<<"objectId">> := _DeviceId, <<"devaddr">> := _Devaddr, <<"product">> := _Product} = Device) ->
save_(Device);
save(#{<<"objectId">> := DeviceId}) ->
{ok, Device} = dgiot_parse:get_object(<<"Device">>, DeviceId),
{ok, Device} = dgiot_parsex:get_object(<<"Device">>, DeviceId),
save_(Device);
save(V) ->
io:format("~s ~p ~p ~n", [?FILE, ?LINE, V]),
@ -132,7 +132,7 @@ post(Device, SessionToken) ->
?LOG(error, "~s ~p DeviceId ~p Err = ~p.~n", [?FILE, ?LINE, DeviceId, Err]),
#{<<"role:admin">> => #{<<"read">> => true, <<"write">> => true}}
end,
dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"ACL">> => SetAcl}),
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"ACL">> => SetAcl}),
dgiot_device_cache:post(Device#{<<"ACL">> => SetAcl}).
put(Device) ->
@ -199,7 +199,7 @@ put_content(#{<<"product">> := Product, <<"objectId">> := DeviceId}) ->
ProductId = maps:get(<<"objectId">>, Product),
case dgiot_product:lookup_prod(ProductId) of
{ok, #{<<"content">> := Content}} ->
dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"content">> => Content});
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"content">> => Content});
_ ->
pass
end;
@ -213,7 +213,7 @@ put_profile(#{<<"product">> := Product, <<"objectId">> := DeviceId}) ->
ProductId = maps:get(<<"objectId">>, Product),
case dgiot_product:lookup_prod(ProductId) of
{ok, #{<<"profile">> := Profile}} ->
dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"profile">> => Profile});
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"profile">> => Profile});
_ ->
pass
end;
@ -227,7 +227,7 @@ put_location(#{<<"product">> := Product, <<"objectId">> := DeviceId}) ->
ProductId = maps:get(<<"objectId">>, Product),
case dgiot_product:lookup_prod(ProductId) of
{ok, #{<<"config">> := #{<<"location">> := Location, <<"address">> := Address}}} ->
dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"location">> => Location, <<"address">> => Address}),
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"location">> => Location, <<"address">> => Address}),
Location;
_ ->
#{<<"__type">> => <<"GeoPoint">>, <<"longitude">> => 120.065714, <<"latitude">> => 30.369491}
@ -273,7 +273,7 @@ get_address(DeviceId, DgLon, DgLat) ->
#{<<"baiduaddr">> := #{<<"formatted_address">> := Formatted_address}} ->
Formatted_address;
_ ->
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"address">> := Addr}} ->
Addr;
_ ->
@ -331,7 +331,7 @@ sync_parse(OffLine) ->
Now = dgiot_datetime:now_secs(),
case V of
['Device', Acl, _, State, Last, IsEnable, ProductId, Devaddr, DeviceSecret, Node, Longitude, Latitude] when (Now - Last) < 0 ->
case dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"status">> => <<"ONLINE">>, <<"isEnable">> => IsEnable, <<"lastOnlineTime">> => Last, <<"location">> => #{<<"__type">> => <<"GeoPoint">>, <<"longitude">> => Longitude, <<"latitude">> => Latitude}}) of
case dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"status">> => <<"ONLINE">>, <<"isEnable">> => IsEnable, <<"lastOnlineTime">> => Last, <<"location">> => #{<<"__type">> => <<"GeoPoint">>, <<"longitude">> => Longitude, <<"latitude">> => Latitude}}) of
{ok, _R} ->
insert_mnesia(DeviceId, Acl, true, State, Last, IsEnable, ProductId, Devaddr, DeviceSecret, Node, Longitude, Latitude);
_ ->
@ -339,7 +339,7 @@ sync_parse(OffLine) ->
end,
timer:sleep(50);
['Device', Acl, true, State, Last, IsEnable, ProductId, Devaddr, DeviceSecret, Node, Longitude, Latitude] when (Now - Last) > OffLine ->
case dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"status">> => <<"OFFLINE">>, <<"isEnable">> => IsEnable, <<"lastOnlineTime">> => Last, <<"location">> => #{<<"__type">> => <<"GeoPoint">>, <<"longitude">> => Longitude, <<"latitude">> => Latitude}}) of
case dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"status">> => <<"OFFLINE">>, <<"isEnable">> => IsEnable, <<"lastOnlineTime">> => Last, <<"location">> => #{<<"__type">> => <<"GeoPoint">>, <<"longitude">> => Longitude, <<"latitude">> => Latitude}}) of
{ok, _R} ->
insert_mnesia(DeviceId, Acl, false, State, Last, IsEnable, ProductId, Devaddr, DeviceSecret, Node, Longitude, Latitude);
_ ->
@ -347,7 +347,7 @@ sync_parse(OffLine) ->
end,
timer:sleep(50);
['Device', Acl, false, State, Last, IsEnable, ProductId, Devaddr, DeviceSecret, Node, Longitude, Latitude] when (Now - Last) < OffLine ->
case dgiot_parse:update_object(<<"Device">>, DeviceId, #{<<"status">> => <<"ONLINE">>, <<"isEnable">> => IsEnable, <<"lastOnlineTime">> => Last, <<"location">> => #{<<"__type">> => <<"GeoPoint">>, <<"longitude">> => Longitude, <<"latitude">> => Latitude}}) of
case dgiot_parsex:update_object(<<"Device">>, DeviceId, #{<<"status">> => <<"ONLINE">>, <<"isEnable">> => IsEnable, <<"lastOnlineTime">> => Last, <<"location">> => #{<<"__type">> => <<"GeoPoint">>, <<"longitude">> => Longitude, <<"latitude">> => Latitude}}) of
{ok, _R} ->
insert_mnesia(DeviceId, Acl, true, State, Last, IsEnable, ProductId, Devaddr, DeviceSecret, Node, Longitude, Latitude);
_ ->

2
apps/dgiot_device/src/utils/dgiot_device_static.erl
View File

@ -26,7 +26,7 @@
get_counter({Token, <<"product_counter">>}) ->
Query = #{<<"count">> => <<"objectId">>,
<<"keys">> => [<<"objectId">>], <<"where">> => #{}, <<"limit">> => 1},
case dgiot_parse:query_object(<<"Product">>, Query, [{"X-Parse-Session-Token", Token}], [{from, rest}]) of
case dgiot_parsex:query_object(<<"Product">>, Query, [{"X-Parse-Session-Token", Token}], [{from, rest}]) of
{ok, #{<<"count">> := Count}} ->
{ok, #{<<"lable">> => <<"产品数量"/utf8>>, <<"value">> => Count}};
_ ->

2
apps/dgiot_device/src/utils/dgiot_product_amis.erl
View File

@ -23,7 +23,7 @@
%%dgiot_product_amis:post(<<"d0cb711d3d">>).
post(ProductId) ->
dgiot_parse:create_object(<<"View">>, #{
dgiot_parsex:create_object(<<"View">>, #{
<<"title">> => ProductId,
<<"key">> => ProductId,
<<"type">> => <<"Amis">>,

12
apps/dgiot_device/src/utils/dgiot_product_csv.erl
View File

@ -57,7 +57,7 @@ create_product(ChannelId, FileName, Productmap, TdChannelId) ->
CategoryId = dgiot_product_csv:get_CategoryId(CategoryName),
ProductId = dgiot_parse_id:get_productid(CategoryId, DevType, ProductName),
Result =
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, #{<<"objectId">> := ProductId}} ->
{ok, ProductId};
_ ->
@ -99,7 +99,7 @@ create_device(FileName, Devicemap, ProductIds) ->
ProductName = maps:get(ProductId, ProductIds, '_'),
Devaddrs = dgiot_utils:unique_1(lists:flatten(ets:match(AtomName, {'_', [ProductName, '_', '_', DeviceName, '$1' | '_']}))),
NewAcl =
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, #{<<"ACL">> := Acl}} ->
Acl;
_ ->
@ -124,7 +124,7 @@ post_thing(FileName, ProductIds, Is_refresh) ->
maps:fold(fun(ProductId, ProductName, _) ->
Things = ets:match(AtomName, {'_', [ProductName, '_', '_', '_', '_', '$1', '$2', '$3', '$4', '$5', '$6', '$7', '$8', '$9', '$10', '$11', '$12' | '_']}),
NewProperties = post_properties(Things, AtomName, ProductId, ProductName),
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, #{<<"thing">> := Thing}} when Is_refresh ->
OldProperties =
lists:foldl(fun(#{<<"identifier">> := Identifier} = X, Acc) ->
@ -134,14 +134,14 @@ post_thing(FileName, ProductIds, Is_refresh) ->
maps:fold(fun(_, Prop, Acc) ->
Acc ++ [Prop]
end, [], dgiot_map:merge(OldProperties, NewProperties)),
dgiot_parse:update_object(<<"Product">>, ProductId, #{<<"thing">> => Thing#{<<"properties">> => Properties}});
dgiot_parsex:update_object(<<"Product">>, ProductId, #{<<"thing">> => Thing#{<<"properties">> => Properties}});
_ ->
pass
end
end, [], ProductIds).
get_CategoryId(CategoryName) ->
case dgiot_parse:query_object(<<"Category">>, #{<<"limit">> => 1, <<"where">> => #{<<"name">> => CategoryName}}) of
case dgiot_parsex:query_object(<<"Category">>, #{<<"limit">> => 1, <<"where">> => #{<<"name">> => CategoryName}}) of
{ok, #{<<"results">> := [#{<<"objectId">> := CategoryId} | _]}} ->
CategoryId;
_ ->
@ -157,7 +157,7 @@ get_CategoryId(CategoryName) ->
<<"parent">> => #{<<"objectId">> => <<"a60a85475a">>, <<"__type">> => <<"Pointer">>, <<"className">> => <<"Category">>},
<<"level">> => 1
},
case dgiot_parse:create_object(<<"Category">>, Body) of
case dgiot_parsex:create_object(<<"Category">>, Body) of
{ok, #{<<"objectId">> := ObjectId}} ->
ObjectId;
_ ->

6
apps/dgiot_device/src/utils/dgiot_product_dict.erl
View File

@ -37,7 +37,7 @@ post_batch(Dicts, ProductId) ->
<<"class">> => <<"Product">>}
}]
end, [], Dicts),
dgiot_parse:batch(DictRequests).
dgiot_parsex:batch(DictRequests).
delete_batch(Dicts) ->
DictRequests =
@ -48,7 +48,7 @@ delete_batch(Dicts) ->
<<"body">> => #{}
}]
end, [], Dicts),
dgiot_parse:batch(DictRequests).
dgiot_parsex:batch(DictRequests).
post(ProductId) ->
NewConfig = #{
@ -70,7 +70,7 @@ post(ProductId) ->
<<"height">> => <<"700">>,
<<"src">> => <<"//img7.ddove.com/upload/20181127/134600237598.jpg?timestamp=1635422987361">>},
<<"className">> => <<"Image">>}]}]}}},
dgiot_parse:create_object(<<"View">>, #{
dgiot_parsex:create_object(<<"View">>, #{
<<"title">> => ProductId,
<<"key">> => ProductId,
<<"type">> => <<"topo">>,

4
apps/dgiot_device/src/utils/dgiot_product_enum.erl
View File

@ -115,7 +115,7 @@ post_enum_value(ProductId, Identifier, Name) ->
upadte_thing(ProductId, Identifier, Name, -1)
end.
upadte_thing(ProductId, Identifier, Name, Max) ->
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, #{<<"thing">> := #{<<"properties">> := Properties} = Thing}} ->
NewProperties = lists:foldl(
fun(X, Acc) ->
@ -131,7 +131,7 @@ upadte_thing(ProductId, Identifier, Name, Max) ->
end
end, [], Properties),
NewThing = Thing#{<<"properties">> => NewProperties},
dgiot_parse:update_object(<<"Product">>, ProductId, #{<<"thing">> => NewThing}),
dgiot_parsex:update_object(<<"Product">>, ProductId, #{<<"thing">> => NewThing}),
#{Identifier => Max + 1};
_ ->

16
apps/dgiot_device/src/utils/dgiot_product_knova.erl
View File

@ -33,7 +33,7 @@
%%dgiot_product_knova:post(<<"d0cb711d3d">>).
post(ProductId) ->
dgiot_parse:create_object(<<"View">>, #{
dgiot_parsex:create_object(<<"View">>, #{
<<"title">> => ProductId,
<<"key">> => ProductId,
<<"type">> => <<"Topo">>,
@ -76,7 +76,7 @@ get_konva_thing(Arg, _Context) ->
ViewId
end,
Children =
case dgiot_parse:get_object(<<"View">>, NewViewId) of
case dgiot_parsex:get_object(<<"View">>, NewViewId) of
{ok, #{<<"data">> := #{<<"konva">> := #{<<"Stage">> := #{<<"children">> := Children2}}}}} when length(Children2) > 0 ->
Children2;
_ ->
@ -85,7 +85,7 @@ get_konva_thing(Arg, _Context) ->
case length(Children) > 0 of
true ->
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, #{<<"thing">> := #{<<"properties">> := Properties}}} ->
put({self(), shapeids}, []),
get_children(<<"web">>, ProductId, Children, ProductId, <<"KonvatId">>, <<"Shapeid">>, <<"Identifier">>, <<"Name">>),
@ -173,12 +173,12 @@ edit_konva(Arg, _Context) ->
{ok, ViewId} ->
ViewId
end,
case dgiot_parse:get_object(<<"View">>, NewViewId) of
case dgiot_parsex:get_object(<<"View">>, NewViewId) of
{ok, #{<<"data">> := #{<<"konva">> := #{<<"Stage">> := #{<<"children">> := Children2} = Stage} = Konva} = Data}} when length(Children2) > 0 ->
put({self(), shapeids}, []),
NewChildren = get_children(<<"web">>, ProductId, Children2, ProductId, ProductId, Shapeid, Identifier, Name),
NewData = Data#{<<"konva">> => Konva#{<<"Stage">> => Stage#{<<"children">> => NewChildren}}},
case dgiot_parse:update_object(<<"View">>, NewViewId, #{<<"data">> => NewData}) of
case dgiot_parsex:update_object(<<"View">>, NewViewId, #{<<"data">> => NewData}) of
{ok, Message} ->
{ok, #{<<"code">> => 200, <<"message">> => Message}};
{error, Message} ->
@ -356,7 +356,7 @@ save_Product_konva(ProductId) ->
dgiot_data:insert({product, <<ProductId/binary, Identifier/binary>>}, {Name, Type, Unit}),
dgiot_data:insert({thing, <<ProductId/binary, Identifier/binary>>}, P)
end, Properties),
case dgiot_parse:query_object(<<"View">>, #{<<"limit">> => 1, <<"where">> => #{<<"key">> => ProductId, <<"type">> => <<"topo">>, <<"class">> => <<"Product">>}}) of
case dgiot_parsex:query_object(<<"View">>, #{<<"limit">> => 1, <<"where">> => #{<<"key">> => ProductId, <<"type">> => <<"topo">>, <<"class">> => <<"Product">>}}) of
{ok, #{<<"results">> := Views}} when length(Views) > 0 ->
lists:foldl(fun(View, _Acc1) ->
#{<<"data">> := #{<<"konva">> := #{<<"Stage">> := #{<<"children">> := Children}}}} = View,
@ -371,7 +371,7 @@ save_Product_konva(ProductId) ->
get_stage(ProductId) ->
ViewId = dgiot_parse_id:get_viewid(ProductId, <<"Topo">>, <<"Product">>, ProductId),
case dgiot_parse:get_object(<<"View">>, ViewId) of
case dgiot_parsex:get_object(<<"View">>, ViewId) of
{ok, #{<<"data">> := #{<<"konva">> := #{<<"Stage">> := Stage}}}} ->
{ok, Stage};
_ ->
@ -380,7 +380,7 @@ get_stage(ProductId) ->
get_stage(ProductId, Tiltle) ->
ViewId = dgiot_parse_id:get_viewid(ProductId, <<"Topo">>, <<"Product">>, Tiltle),
case dgiot_parse:get_object(<<"View">>, ViewId) of
case dgiot_parsex:get_object(<<"View">>, ViewId) of
{ok, #{<<"data">> := #{<<"konva">> := #{<<"Stage">> := Stage}}}} ->
{ok, Stage};
_ ->

14
apps/dgiot_device/src/utils/dgiot_product_tdengine.erl
View File

@ -51,13 +51,13 @@ get_product(ProductId, Query) ->
end.
get_products(ProductId, ChannelId) ->
case dgiot_parse:get_object(<<"Product">>, ProductId) of
case dgiot_parsex:get_object(<<"Product">>, ProductId) of
{ok, #{<<"nodeType">> := 2}} ->
dgiot_data:insert({ProductId, ?TYPE}, ChannelId),
case dgiot_parse:query_object(<<"Device">>, #{<<"limit">> => 1, <<"where">> => #{<<"product">> => ProductId}}) of
case dgiot_parsex:query_object(<<"Device">>, #{<<"limit">> => 1, <<"where">> => #{<<"product">> => ProductId}}) of
{ok, #{<<"results">> := Results}} when length(Results) > 0 ->
[#{<<"objectId">> := Devid} | _] = Results,
case dgiot_parse:query_object(<<"Device">>, #{<<"limit">> => 1000, <<"keys">> => [<<"product">>], <<"where">> => #{<<"parentId">> => Devid}}) of
case dgiot_parsex:query_object(<<"Device">>, #{<<"limit">> => 1000, <<"keys">> => [<<"product">>], <<"where">> => #{<<"parentId">> => Devid}}) of
{ok, #{<<"results">> := R}} ->
lists:foldl(fun(#{<<"product">> := Product}, Acc) ->
#{<<"objectId">> := SubProductId} = Product,
@ -89,7 +89,7 @@ do_channel(ProductId, Session, Fun) ->
}
}
},
case dgiot_parse:query_object(<<"Channel">>, Body, [{"X-Parse-Session-Token", Session}], [{from, rest}]) of
case dgiot_parsex:query_object(<<"Channel">>, Body, [{"X-Parse-Session-Token", Session}], [{from, rest}]) of
{ok, #{<<"results">> := []}} ->
{404, <<"not find channel">>};
{ok, #{<<"results">> := [#{<<"objectId">> := ChannelId}]}} ->
@ -110,7 +110,7 @@ get_channel(Session) ->
},
case dgiot_data:get(?DGIOT_CHANNEL_SESSION, Session) of
not_find ->
case dgiot_parse:query_object(<<"Channel">>, Body, [{"X-Parse-Session-Token", Session}], [{from, rest}]) of
case dgiot_parsex:query_object(<<"Channel">>, Body, [{"X-Parse-Session-Token", Session}], [{from, rest}]) of
{ok, #{<<"results">> := []}} ->
{error, <<"not find channel">>};
{ok, #{<<"results">> := [#{<<"objectId">> := ChannelId}]}} ->
@ -192,11 +192,11 @@ check_field(<<"geopoint">>, V, #{<<"deviceid">> := DeviceId}) ->
[Longitude, Latitude] ->
case dgiot_gps:get_baidu_addr(Longitude, Latitude) of
#{<<"baiduaddr">> := #{<<"formatted_address">> := FormattedAddress}} ->
case dgiot_parse:get_object(<<"Device">>, DeviceId) of
case dgiot_parsex:get_object(<<"Device">>, DeviceId) of
{ok, #{<<"location">> := #{<<"__type">> := <<"GeoPoint">>, <<"longitude">> := Longitude, <<"latitude">> := Latitude}}} ->
pass;
{ok, #{<<"detail">> := Detail}} ->
dgiot_parse:update_object(<<"Device">>, DeviceId, #{
dgiot_parsex:update_object(<<"Device">>, DeviceId, #{
<<"location">> => #{<<"__type">> => <<"GeoPoint">>, <<"longitude">> => Longitude, <<"latitude">> => Latitude},
<<"detail">> => Detail#{<<"address">> => FormattedAddress}});
_ ->

6
apps/dgiot_device/src/utils/dgiot_product_view.erl
View File

@ -37,7 +37,7 @@ post_batch(Views, ProductId) ->
<<"class">> => <<"Product">>}
}]
end, [], Views),
dgiot_parse:batch(ViewRequests).
dgiot_parsex:batch(ViewRequests).
delete_batch(Views) ->
ViewRequests =
@ -48,11 +48,11 @@ delete_batch(Views) ->
<<"body">> => #{}
}]
end, [], Views),
dgiot_parse:batch(ViewRequests).
dgiot_parsex:batch(ViewRequests).
%%dgiot_product_view:post(<<"d0cb711d3d">>).
post(ProductId) ->
dgiot_parse:create_object(<<"View">>, #{
dgiot_parsex:create_object(<<"View">>, #{
<<"title">> => ProductId,
<<"key">> => ProductId,
<<"type">> => <<"topo">>,

59
apps/dgiot_parse/src/dgiot_parse_restx.erl
View File

@ -14,7 +14,7 @@
%% limitations under the License.
%%--------------------------------------------------------------------
-module(dgiot_parse_rest).
-module(dgiot_parse_restx).
-compile(nowarn_deprecated_function).
-author("kenneth").
-include("dgiot_parse.hrl").
@ -339,26 +339,17 @@ encode_body(_Path, _Method, Map, _) ->
dgiot_json:encode(Map).
do_request(Method, Path, Header, QueryData, Options) ->
NewQueryData =
case jsx:is_json(QueryData) of
true ->
jsx:decode(QueryData, [{labels, binary}, return_maps]);
false ->
QueryData
end,
do_request_before(Method, Path, Header, QueryData, Options),
case jsx:is_json(QueryData) of
true ->
jsx:decode(QueryData, [{labels, binary}, return_maps]);
false ->
QueryData
end,
case httpc_request(Method, Path, Header, QueryData, [], [], Options) of
{error, Reason} ->
{error, Reason};
{ok, StatusCode, Headers, ResBody} ->
case do_request_after(Method, Path, Header, NewQueryData, ResBody, Options) of
{ok, NewResBody} ->
{ok, StatusCode, Headers, NewResBody};
ignore ->
{ok, StatusCode, Headers, ResBody};
{error, Reason} ->
{error, Reason}
end
{ok, StatusCode, Headers, ResBody}
end.
httpc_request(Method, <<"/graphql">> = Path, Header, Body, HttpOptions, ReqOptions, Options) when Method == 'POST'; Method == 'PUT' ->
@ -394,40 +385,6 @@ httpc_request(Method, Request, HttpOptions, ReqOptions) ->
{error, Reason}
end.
do_request_before(Method0, Path, Header, QueryData, Options) ->
Method =
case proplists:get_value(from, Options) of
js when QueryData == <<>> ->
method(Method0, atom);
js ->
case maps:get(<<"_method">>, ?JSON_DECODE(QueryData), no) of
no -> method(Method0, atom);
Method1 -> method(Method1, atom)
end;
_ ->
method(Method0, atom)
end,
{match, PathList} = re:run(Path, <<"([^/]+)">>, [global, {capture, all_but_first, binary}]),
dgiot_parse_hook:do_request_hook('before', lists:concat(PathList), Method, dgiot_parse:get_token(Header), QueryData, Options).
do_request_after(Method0, Path, Header, NewQueryData, ResBody, Options) ->
Method =
case proplists:get_value(from, Options) of
js when NewQueryData == <<>> ->
method(Method0, atom);
js ->
case maps:get(<<"_method">>, ?JSON_DECODE(NewQueryData), no) of
no -> method(Method0, atom);
Method1 -> method(Method1, atom)
end;
_ ->
method(Method0, atom)
end,
{match, PathList} = re:run(Path, <<"([^/]+)">>, [global, {capture, all_but_first, binary}]),
%% io:format("~s ~p ~p ~p ~n",[?FILE, ?LINE, Path, NewQueryData]),
dgiot_parse_hook:do_request_hook('after', lists:concat(PathList), Method, dgiot_parse:get_token(Header), NewQueryData, ResBody).
list_join([], Sep) when is_list(Sep) -> [];
list_join([H | T], Sep) ->
to_list(H) ++ lists:append([Sep ++ to_list(X) || X <- T]).

619
apps/dgiot_parse/src/dgiot_parsex.erl Normal file
View File

@ -0,0 +1,619 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2020-2021 DGIOT Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------
-module(dgiot_parsex).
-author("kenneth").
-include("dgiot_parse.hrl").
-include_lib("dgiot/include/logger.hrl").
-define(DEFField, re:split(application:get_env(?MODULE, delete_field, ""), ",")).
%% API
-export([
health/0,
health/1,
init_database/2,
create_object/2,
create_object/3,
create_object/4,
get_object/2,
get_object/3,
get_object/4,
update_object/3,
update_object/4,
update_object/5,
del_object/2,
del_object/4,
del_table/1,
get_schemas/0,
get_schemas/1,
get_schemas/2,
create_schemas/1,
create_schemas/2,
update_schemas/1,
update_schemas/2,
del_schemas/1,
del_schemas/2,
set_class_level/2,
del_filed_schemas/2,
del_filed_schemas/3,
query_object/2,
query_object/3,
query_object/4,
aggregate_object/2,
aggregate_object/4,
read_page/4,
read_page/5,
format_data/2,
batch/1,
batch/2,
batch/3,
batch/4,
import/5,
import/6,
request/4,
request/5,
get_token/1,
get_qs/1
]).
-export([
request_rest/6,
get_header_token/2,
get_view_token/2
]).
health() ->
health(?DEFAULT).
health(Name) ->
Path = <<"/health">>,
request_rest(Name, 'GET', [], Path, #{}, [{from, rest}]).
%% 创建对象
create_object(Class, Map) ->
create_object(?DEFAULT, Class, Map).
create_object(Name, Class, Map) ->
create_object(Name, Class, Map, [], [{from, master}]).
create_object(Class, Map, Header, Options) ->
create_object(?DEFAULT, Class, Map, Header, Options).
create_object(Name, Class, #{<<"objectId">> := _ObjectId} = Map, Header, Options) ->
Path = <<"/classes/", Class/binary>>,
request_rest(Name, 'POST', Header, Path, Map, Options);
create_object(Name, Class, Map, Header, Options) ->
Path = <<"/classes/", Class/binary>>,
request_rest(Name, 'POST', Header, Path, dgiot_parse_id:get_objectid(Class, Map), Options).
%% 获取对象
get_object(Class, ObjectId) ->
get_object(?DEFAULT, Class, ObjectId).
get_object(Name, Class, ObjectId) ->
get_object(Name, Class, ObjectId, [], [{from, master}]).
get_object(Class, ObjectId, Header, Options) ->
get_object(?DEFAULT, Class, ObjectId, Header, Options).
get_object(Name, Class, ObjectId, Header, Options) ->
Path = <<"/classes/", Class/binary, "/", ObjectId/binary>>,
request_rest(Name, 'GET', Header, Path, #{}, Options).
%% 更新对象
update_object(Class, ObjectId, Map) ->
update_object(?DEFAULT, Class, ObjectId, Map).
update_object(Name, Class, ObjectId, Map) ->
update_object(Name, Class, ObjectId, Map, [], [{from, master}]).
update_object(Class, ObjectId, Map, Header, Options) ->
update_object(?DEFAULT, Class, ObjectId, Map, Header, Options).
update_object(Name, Class, ObjectId, Map, Header, Options) ->
Path = <<"/classes/", Class/binary, "/", ObjectId/binary>>,
request_rest(Name, 'PUT', Header, Path, Map, Options).
%% 批处理
batch(Requests) ->
batch(?DEFAULT, Requests).
batch(Name, Requests) ->
batch(Name, Requests, [], [{from, master}]).
batch(Requests, Header, Opts) ->
batch(?DEFAULT, Requests, Header, Opts).
batch(Name, Requests, Header, Opts) ->
request_rest(Name, 'POST', Header, <<"/batch">>, #{<<"requests">> => Requests}, Opts).
del_filed_schemas(Class, Fileds) ->
del_filed_schemas(?DEFAULT, Class, Fileds).
%%
del_filed_schemas(Name, Class, Fileds) ->
Path = <<"/schemas/", Class/binary>>,
NewFields = lists:foldl(
fun(X, Acc) ->
Acc#{X => #{<<"__op">> => <<"Delete">>}}
end, #{}, Fileds),
Method = <<"PUT">>,
Body = #{
<<"className">> => <<Class/binary>>,
<<"fields">> => NewFields,
<<"_method">> => Method
},
request_rest(Name, 'PUT', [], Path, Body, [{from, master}]).
%% 创建表结构
create_schemas(Fields) ->
create_schemas(?DEFAULT, Fields).
create_schemas(Name, #{<<"className">> := Class} = Fields) ->
Path = <<"/schemas/", Class/binary>>,
request_rest(Name, 'POST', [], Path, Fields, [{from, master}]).
%% 更新表结构
update_schemas(Fields) ->
update_schemas(?DEFAULT, Fields).
update_schemas(Name, #{<<"className">> := Class} = Fields) ->
Path = <<"/schemas/", Class/binary>>,
request_rest(Name, 'PUT', [], Path, Fields, [{from, master}]).
%% 删除表结构
del_schemas(Class) ->
del_schemas(?DEFAULT, Class).
del_schemas(Name, Class) ->
Path = <<"/schemas/", Class/binary>>,
request_rest(Name, 'DELETE', [], Path, #{}, [{from, master}]).
%% 获取表结构
get_schemas() ->
get_schemas(<<>>).
get_schemas(Class) ->
get_schemas(?DEFAULT, Class).
get_schemas(Name, Class) ->
Path = <<"/schemas/", Class/binary>>,
request_rest(Name, 'GET', [], Path, #{}, [{from, master}]).
%% 设置表权限
set_class_level(Class, Permissions) ->
set_class_level(?DEFAULT, Class, Permissions).
set_class_level(Name, Class, Permissions) ->
Path = <<"/schemas/", Class/binary>>,
Body = #{<<"classLevelPermissions">> => Permissions},
request_rest(Name, 'PUT', [], Path, Body, [{from, master}]).
%% limit和skip参数进行分页
%% 传递order逗号分隔列表按多个字段进行排序
%% http://docs.parseplatform.org/rest/guide/#query-constraints
query_object(Class, Args) ->
query_object(?DEFAULT, Class, Args).
query_object(Name, Class, Args) ->
query_object(Name, Class, Args, [], [{from, master}]).
query_object(Class, Args, Header, Options) ->
query_object(?DEFAULT, Class, Args, Header, Options).
query_object(Name, Class, Args, Header, Options) ->
Path = <<"/classes/", Class/binary>>,
request_rest(Name, 'GET', Header, Path, Args, Options).
%% limit和skip参数进行分页
%% 传递order逗号分隔列表按多个字段进行排序
%% http://docs.parseplatform.org/rest/guide/#query-constraints
aggregate_object(Class, Args) ->
aggregate_object(?DEFAULT, Class, Args).
aggregate_object(Name, Class, Args) ->
aggregate_object(Name, Class, Args, [], [{from, master}]).
aggregate_object(Class, Args, Header, Options) ->
aggregate_object(?DEFAULT, Class, Args, Header, Options).
aggregate_object(Name, Class, Args, Header, Options) ->
Path = <<"/aggregate/", Class/binary>>,
request_rest(Name, 'GET', Header, Path, Args, Options).
%% 删除对象
del_object(Class, ObjectId) ->
del_object(?DEFAULT, Class, ObjectId).
del_object(Name, Class, ObjectId) ->
del_object(Name, Class, ObjectId, [], [{from, master}]).
del_object(Class, ObjectId, Header, Options) ->
del_object(?DEFAULT, Class, ObjectId, Header, Options).
del_object(Name, Class, ObjectId, Header, Options) ->
Path = <<"/classes/", Class/binary, "/", ObjectId/binary>>,
request_rest(Name, 'DELETE', Header, Path, #{}, Options).
%% 删除表格
del_table(Class) ->
del_table(?DEFAULT, Class).
del_table(Name, Class) ->
Path = <<"/purge/", Class/binary>>,
request_rest(Name, 'DELETE', [], Path, #{}, [{from, master}]).
read_page(Class, Query, Skip, PageSize) ->
read_page(?DEFAULT, Class, Query, Skip, PageSize).
read_page(Name, Class, Query, Skip, PageSize) ->
%?LOG(info,"~p~n", [ Query#{<<"limit">> => PageSize, <<"skip">> => Skip}]),
case query_object(Name, Class, Query#{<<"limit">> => PageSize, <<"skip">> => Skip}) of
{error, Reason} ->
{error, Reason};
{ok, #{<<"results">> := Page}} ->
{ok, Page}
end.
import(Class, Datas, Count, Fun, Acc) ->
import(?DEFAULT, Class, Datas, Count, Fun, Acc).
import(Name, Class, {json, Path}, Count, Fun, Acc) ->
case file:read_file(Path) of
{ok, Bin} ->
case catch jsx:decode(Bin, [{labels, binary}, return_maps]) of
{'EXIT', Reason} ->
{error, Reason};
Datas ->
import(Name, Class, Datas, Count, Fun, Acc)
end;
{error, Reason} ->
{error, Reason}
end;
import(Name, Class, Datas, Count, Fun, Acc) ->
import(Name, Class, Datas, Count, [], Fun, Acc).
import(Name, Class, Datas, Count, Requests, Fun, Acc) when length(Requests) == Count; Datas == [] ->
case batch(Name, Requests) of
{error, Reason} ->
{error, Reason};
{ok, Results} ->
ResAcc = Fun(Results, Acc),
case Datas == [] of
true -> ResAcc;
false -> import(Name, Class, Datas, Count, Fun, ResAcc)
end
end;
import(Name, Class, [Data | Other], Count, Requests, Fun, Acc) when length(Requests) < Count ->
try
NewRequests = [#{
<<"method">> => <<"POST">>,
<<"path">> => <<"/classes/", Class/binary>>,
<<"body">> => Data
} | Requests],
import(Name, Class, Other, Count, NewRequests, Fun, Acc)
catch
_: {error, not_add} ->
import(Name, Class, Other, Count, Requests, Fun, Acc)
end.
%%%===================================================================
%%% Internal functions
%%%===================================================================
get_token(Header) ->
case proplists:get_value("X-Parse-Session-Token", Header) of
undefined ->
proplists:get_value(<<"X-Parse-Session-Token">>, Header);
Token1 ->
Token1
end.
get_header_token(<<"/classes/_Role", _/binary>>, Header) ->
Header;
get_header_token(<<"/classes/_User", _/binary>>, Header) ->
Header;
get_header_token(<<"/classes/_Session", _/binary>>, Header) ->
Header;
get_header_token(_, Header) ->
lists:foldl(
fun
({<<"X-Parse-Session-Token">>, Token}, Acc) ->
DepartToken = dgiot_parse_auth:get_depart_session(Token),
Acc ++ [{<<"X-Parse-Session-Token">>, DepartToken}];
({"X-Parse-Session-Token", Token}, Acc) ->
DepartToken = dgiot_parse_auth:get_depart_session(Token),
Acc ++ [{<<"X-Parse-Session-Token">>, DepartToken}];
({K, V}, Acc) ->
Acc ++ [{K, V}]
end, [], Header).
get_view_token(Header, ViewId) ->
UserToken =
case proplists:get_value("X-Parse-Session-Token", Header) of
undefined ->
proplists:get_value(<<"X-Parse-Session-Token">>, Header);
Token1 ->
Token1
end,
dgiot_parse_auth:put_view_session(UserToken, ViewId).
get_qs(Map) ->
lists:foldl(
fun
({N, V}, <<>>) ->
NewV = format_value(V),
<<"?", N/binary, "=", NewV/binary>>;
({N, V}, Acc) ->
NewV = format_value(V),
<<Acc/binary, "&", N/binary, "=", NewV/binary>>
end, <<>>, maps:to_list(Map)).
format_value(V) when is_binary(V) ->
dgiot_httpc:urlencode(V);
format_value(V) ->
Json = dgiot_json:encode(V),
%% V1 = re:replace(Json, <<"\[.*?\]">>, <<"">>, [global, {return, binary}, unicode]),
dgiot_httpc:urlencode(Json).
read_file(Path) ->
case file:read_file(Path) of
{ok, Bin} ->
case catch jsx:decode(Bin, [{labels, binary}, return_maps]) of
{'EXIT', Reason} ->
{error, Reason};
Data ->
{ok, Data}
end;
{error, Reason} ->
{error, Reason}
end.
%% 根据Map查找出关联数据
format_data(<<"Dict">>, Data) ->
NewData =
case Data of
#{<<"key">> := _} -> Data;
_ -> Data#{<<"key">> => dgiot_utils:to_md5(dgiot_json:encode(Data))}
end,
maps:fold(
fun(Key, Value, Acc) ->
case format_value(<<"Dict">>, Key, Value) of
not_add ->
throw({error, not_add});
{<<"Pointer">>, ParentId} ->
Acc#{Key => Value#{<<"objectId">> => ParentId}};
{<<"AddRelation">>, Objects} ->
Acc#{Key => Value#{<<"objects">> => Objects}};
NValue ->
Acc#{Key => NValue}
end
end, #{}, NewData);
format_data(Class, Data) ->
maps:fold(
fun(Key, Value, Acc) ->
case format_value(Class, Key, Value) of
not_add ->
throw({error, not_add});
{<<"Pointer">>, ParentId} ->
Acc#{Key => Value#{<<"objectId">> => ParentId}};
{<<"AddRelation">>, Objects} ->
Acc#{Key => Value#{<<"objects">> => Objects}};
NValue ->
Acc#{Key => NValue}
end
end, #{}, Data).
format_value(_Class, _Key, #{<<"__type">> := <<"Pointer">>, <<"className">> := ClassName, <<"objectId">> := #{<<"where">> := Where}}) ->
case query_object(ClassName, #{<<"where">> => Where}) of
{ok, #{<<"results">> := [#{<<"objectId">> := ObjectId} | _]}} ->
{<<"Pointer">>, ObjectId};
{ok, #{<<"results">> := []}} ->
{error, {Where, <<"object not find!">>}};
{error, Reason} ->
{error, Reason}
end;
format_value(Class, Key, #{<<"__op">> := <<"AddRelation">>, <<"objects">> := Objects}) ->
Fun =
fun(ClassName, ObjectId) ->
#{
<<"__type">> => <<"Pointer">>,
<<"className">> => ClassName,
<<"objectId">> => ObjectId
}
end,
{<<"AddRelation">>, lists:foldl(
fun(#{<<"className">> := ClassName} = Object, Acc) ->
case format_value(Class, Key, Object) of
{<<"Pointer">>, ObjectId} ->
[Fun(ClassName, ObjectId) | Acc];
{error, _Reason} ->
Acc
end
end, [], Objects)};
format_value(Class, Key, #{<<"iskey">> := IsKey, <<"value">> := Value}) ->
case IsKey of
false ->
Value;
true ->
case query_object(Class, #{<<"keys">> => Key, <<"where">> => #{Key => Value}}) of
{ok, #{<<"results">> := []}} ->
?LOG(info, "~p, ~p,~p", [Class, Key, Value]),
Value;
{ok, #{<<"results">> := _}} ->
not_add;
{error, Reason} ->
{error, Reason}
end
end;
format_value(_Class, _Key, Value) when is_binary(Value) ->
Node = atom_to_list(node()),
[NodeName, Host] = string:tokens(Node, "@"),
lists:foldl(
fun({RE, Replace}, New) ->
re:replace(New, RE, Replace, [global, {return, binary}])
end, Value, [{<<"\\{host\\}">>, Host}, {<<"\\{node\\}">>, Node}, {<<"\\{nodename\\}">>, NodeName}]);
format_value(_Class, _Key, Value) ->
Value.
%% Rest请求
request_rest(Name, Method, Header, Path, Body, Options) ->
Response = request(Name, Method, Header, Path, Body, Options),
handle_response(Response).
request(Method, Header, Path, Options) ->
request(Method, Header, Path, <<>>, Options).
request(Method, Header, Path0, Body, Options) ->
request(?DEFAULT, Method, Header, Path0, Body, Options).
request(Name, Method, Header, Path0, Body, Options) ->
case dgiot_parse_channel:get_config(Name) of
{ok, Cfg} ->
NewOpts = [{cfg, Cfg} | Options],
dgiot_parse_restx:request(Method, Header, Path0, Body, NewOpts);
{error, Reason} ->
{error, Reason}
end.
get_tables(Dirs) -> get_tables(Dirs, []).
get_tables([], Acc) -> Acc;
get_tables([Dir | Other], Acc) ->
Dir0 = Dir ++ "/tables/",
case file:list_dir(Dir0) of
{ok, Files} ->
Acc2 = lists:foldl(
fun(File, Acc1) ->
Path = Dir0 ++ File,
case filelib:is_file(Path) andalso read_file(Path) of
{ok, Data} ->
lists:concat([Data, Acc1]);
false ->
Acc1;
{error, Reason} ->
?LOG(error, "load error ~p,~p~n", [Path, Reason]),
Acc1
end
end, Acc, Files),
get_tables(Other, Acc2);
{error, enoent} ->
get_tables(Other, Acc)
end.
%% 初始化数据库，op : 如果已经存在是删除还是合并
init_database(Dirs, Op) ->
init_database(?DEFAULT, Dirs, Op).
init_database(Name, Dirs, Op) when Op == merge; Op == delete ->
case get_schemas(Name, <<>>) of
{ok, #{<<"results">> := OldSchemas1}} ->
OldSchemas =
lists:foldl(
fun(#{<<"className">> := Class} = Tab, Acc) ->
Fields = maps:without(?DEFField, maps:get(<<"fields">>, Tab, #{})),
case maps:size(Fields) == 0 of
true ->
Acc#{Class => Tab};
false ->
Acc#{Class => Tab#{<<"fields">> => Fields}}
end
end, #{}, OldSchemas1),
file:write_file("data/db.schema", dgiot_json:encode(maps:values(OldSchemas))),
Schemas = get_tables(Dirs),
init_tables(Name, OldSchemas, Schemas, Op);
{error, Reason} ->
?LOG(error, "~p~n", [Reason]),
{error, Reason}
end.
init_tables(Name, OldSchemas, Schemas, Op) ->
lists:foreach(
fun(#{<<"className">> := Class} = Schema) ->
NewFields = maps:get(<<"fields">>, Schema, #{}),
Result =
case maps:get(Class, OldSchemas, undefined) of
undefined ->
create_schemas(Name, Schema);
#{<<"fields">> := OldFields} when Op == merge ->
{Targets, Fields} = merge_table(Name, Class, maps:without(?DEFField, NewFields), maps:without(?DEFField, OldFields)),
TargetTab = maps:keys(Targets),
case length(TargetTab) > 0 of
true ->
% @todo 需要提前创建这些表
?LOG(info, "~p~n", [TargetTab]);
false ->
ok
end,
UpdateSchema = Schema#{<<"fields">> => Fields},
case maps:size(Fields) == 0 of
true ->
update_schemas(Name, maps:without([<<"fields">>], UpdateSchema));
false ->
update_schemas(Name, UpdateSchema)
end;
#{<<"fields">> := _OldFields} when Op == delete ->
case del_schemas(Name, Class) of
{ok, _} ->
create_schemas(Name, Schema);
Err ->
Err
end
end,
case Result of
{error, #{<<"message">> := Why}} ->
?LOG(error, "~p:~p~n", [Class, Why]);
{error, #{<<"error">> := Why}} ->
?LOG(error, "~p:~p~n", [Class, Why]);
ok ->
ok;
{ok, _Rtn} ->
%?LOG(info,"~p:create success -> ~p~n", [Class, Rtn]),
ok
end
end, Schemas).
merge_table(Name, Class, NewFields, OldFields) ->
maps:fold(
fun
(Key, Type, {Targets, Acc}) ->
case maps:get(Key, NewFields, no) of
no ->
{Targets, Acc#{Key => #{<<"__op">> => <<"Delete">>}}};
Type ->
{Targets, maps:without([Key], Acc)};
NewType ->
update_schemas(Name, #{<<"className">> => Class, <<"fields">> => #{Key => #{<<"__op">> => <<"Delete">>}}}),
case is_map(Type) andalso maps:get(<<"targetClass">>, NewType, false) of
false ->
{Targets, Acc};
TargetClass ->
{Targets#{TargetClass => true}, Acc}
end
end
end, {#{}, NewFields}, OldFields).
handle_response(Result) ->
Fun =
fun(Res, Body) ->
case jsx:is_json(Body) of
true ->
case catch jsx:decode(Body, [{labels, binary}, return_maps]) of
{'EXIT', Reason} ->
{error, Reason};
#{<<"code">> := Code, <<"error">> := #{<<"routine">> := Reason}} ->
{error, #{<<"code">> => Code, <<"error">> => Reason}};
Map when map_size(Map) == 0 ->
Res;
Map ->
{Res, Map}
end;
false ->
Res
end
end,
case Result of
{ok, HTTPCode, _Headers, Body} when HTTPCode == 200; HTTPCode == 201 ->
Fun(ok, Body);
{ok, HTTPCode, _Headers, Body} when HTTPCode == 404; HTTPCode == 400; HTTPCode == 500 ->
Fun(error, Body);
{ok, _HTTPCode, _Headers, Body} ->
Fun(error, Body);
{error, #{<<"code">> := Code, <<"routine">> := Reason}} ->
{error, #{<<"code">> => Code, <<"error">> => Reason}};
{error, Reason} ->
{error, #{<<"code">> => 1, <<"error">> => Reason}}
end.

2
apps/dgiot_tdengine/src/utils/dgiot_tdengine_dump.erl
View File

@ -37,7 +37,7 @@ export(ChannelId, #{<<"sessionToken">> := SessionToken} = _Body) ->
<<"keys">> => [<<"objectId">>, <<"product">>]
},
TdQuery = #{<<"limit">> => 10000, <<"function">> => <<"last">>, <<"interval">> => <<"1m">>},
case dgiot_parse:query_object(<<"Device">>, Query, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
case dgiot_parsex:query_object(<<"Device">>, Query, [{"X-Parse-Session-Token", SessionToken}], [{from, rest}]) of
{ok, #{<<"results">> := Data}} ->
lists:foldl(fun(Device, Acc) ->
export_device_data(ChannelId, Device, TdQuery, Acc)