Partial implementation of transfer
This commit is contained in:
Andrea Cavalli
parent
316604d3d5
commit
1c13945b8a
@ -1,9 +1,12 @@
|
||||
package it.cavallium;
|
||||
|
||||
import it.tdlight.jni.TdApi;
|
||||
import it.tdlight.tdlibsession.td.TdResult;
|
||||
import java.util.Optional;
|
||||
import java.util.function.Supplier;
|
||||
import javafx.application.Platform;
|
||||
import javafx.scene.control.Dialog;
|
||||
import org.slf4j.event.Level;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
public class MonoFxUtils {
|
||||
@ -46,4 +49,12 @@ public class MonoFxUtils {
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
public static <T extends TdApi.Object> Mono<T> orElseLogSkipError(TdResult<T> optional) {
|
||||
if (optional.failed()) {
|
||||
App.getLogService().append(Level.ERROR, "Received TDLib error: " + optional.cause());
|
||||
return Mono.empty();
|
||||
}
|
||||
return Mono.just(optional.result());
|
||||
}
|
||||
}
|
||||
|
||||
@ -20,7 +20,10 @@ import javafx.scene.control.ButtonType;
|
||||
import javafx.scene.control.ChoiceBox;
|
||||
import javafx.scene.control.ComboBox;
|
||||
import javafx.scene.control.Dialog;
|
||||
import javafx.scene.control.Label;
|
||||
import javafx.scene.control.ListView;
|
||||
import javafx.scene.control.ProgressBar;
|
||||
import javafx.scene.control.TableView;
|
||||
import javafx.scene.control.TextField;
|
||||
import javafx.scene.control.TextInputDialog;
|
||||
import javafx.scene.input.InputMethodEvent;
|
||||
@ -33,6 +36,10 @@ import reactor.core.scheduler.Schedulers;
|
||||
|
||||
public class PrimaryController {
|
||||
|
||||
public Label statusTxt;
|
||||
public ProgressBar statusBar;
|
||||
public Label statusPercentage;
|
||||
public TableView<UserStatus> statusTable;
|
||||
@FXML private Pane main;
|
||||
@FXML private TextField phoneNumber;
|
||||
@FXML private ListView<Text> userbotsList;
|
||||
@ -307,38 +314,63 @@ public class PrimaryController {
|
||||
public void onDestGroupShowing(Event event) {
|
||||
}
|
||||
|
||||
@SuppressWarnings("ConstantConditions")
|
||||
@FXML
|
||||
public void onDoTransfer(ActionEvent actionEvent) {
|
||||
if (sourceGroupCombo.getSelectionModel().getSelectedItem() == null || !(sourceGroupCombo.getSelectionModel().getSelectedItem() instanceof BaseChatInfo)) {
|
||||
if (sourceGroupCombo.getSelectionModel().getSelectedItem() == null
|
||||
|| !(sourceGroupCombo.getSelectionModel().getSelectedItem() instanceof BaseChatInfo)) {
|
||||
MonoFxUtils
|
||||
.showAndWait(new Alert(AlertType.ERROR, "You must select a source group", ButtonType.CLOSE))
|
||||
.subscribe();
|
||||
return;
|
||||
}
|
||||
if (destGroupCombo.getSelectionModel().getSelectedItem() == null || !(destGroupCombo.getSelectionModel().getSelectedItem() instanceof BaseChatInfo)) {
|
||||
if (destGroupCombo.getSelectionModel().getSelectedItem() == null
|
||||
|| !(destGroupCombo.getSelectionModel().getSelectedItem() instanceof BaseChatInfo)) {
|
||||
MonoFxUtils
|
||||
.showAndWait(new Alert(AlertType.ERROR, "You must select a destination group", ButtonType.CLOSE))
|
||||
.subscribe();
|
||||
return;
|
||||
}
|
||||
disableClicks();
|
||||
statusTable.getItems().clear();
|
||||
App
|
||||
.getTransferService()
|
||||
.transferMembers(sourceGroupCombo.getSelectionModel().getSelectedItem(),
|
||||
destGroupCombo.getSelectionModel().getSelectedItem(),
|
||||
userStatus -> {
|
||||
return Mono.empty();
|
||||
userStatus -> MonoFxUtils.runLater(() -> {
|
||||
if (statusTable.getItems().contains(userStatus)) {
|
||||
statusTable.getItems().remove(userStatus);
|
||||
}
|
||||
statusTable.getItems().add(userStatus);
|
||||
return Mono.empty();
|
||||
}),
|
||||
percentage -> MonoFxUtils.runLater(() -> {
|
||||
statusPercentage.setText(percentage + "%");
|
||||
statusBar.setProgress(((double) percentage / 100d));
|
||||
return Mono.empty();
|
||||
}),
|
||||
phaseDescription -> MonoFxUtils.runLater(() -> {
|
||||
App.getLogService().append(Level.INFO, phaseDescription);
|
||||
statusTxt.setText(phaseDescription);
|
||||
return Mono.empty();
|
||||
})
|
||||
)
|
||||
.onErrorResume(error -> MonoFxUtils.runLater(() -> {
|
||||
var alert = new Alert(AlertType.ERROR, "Error during transfer", ButtonType.CLOSE);
|
||||
alert.setContentText(error.getLocalizedMessage());
|
||||
return MonoFxUtils.showAndWait(alert);
|
||||
}).then())
|
||||
.subscribe(_v -> {}, _v -> {
|
||||
|
||||
}, () -> {
|
||||
.doOnTerminate(() -> {
|
||||
Platform.runLater(() -> {
|
||||
statusTxt.setText("");
|
||||
statusPercentage.setText("0%");
|
||||
statusBar.setProgress(0d);
|
||||
enableClicks();
|
||||
});
|
||||
})
|
||||
.subscribe(_v -> {
|
||||
}, _v -> {
|
||||
|
||||
}, () -> {});
|
||||
}
|
||||
}
|
||||
|
||||
@ -5,6 +5,7 @@ import it.tdlight.jni.TdApi.AuthorizationState;
|
||||
import it.tdlight.jni.TdApi.AuthorizationStateReady;
|
||||
import it.tdlight.jni.TdApi.Chat;
|
||||
import it.tdlight.jni.TdApi.ChatMemberStatusAdministrator;
|
||||
import it.tdlight.jni.TdApi.ChatMemberStatusCreator;
|
||||
import it.tdlight.jni.TdApi.GetSupergroupFullInfo;
|
||||
import it.tdlight.jni.TdApi.Object;
|
||||
import it.tdlight.jni.TdApi.Supergroup;
|
||||
@ -18,6 +19,7 @@ import it.tdlight.tdlibsession.td.TdResult;
|
||||
import it.tdlight.tdlibsession.td.easy.AsyncTdEasy;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.StringJoiner;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.stream.Collectors;
|
||||
import reactor.core.publisher.EmitterProcessor;
|
||||
@ -28,6 +30,7 @@ import reactor.core.scheduler.Schedulers;
|
||||
|
||||
public class TransferClient {
|
||||
|
||||
private final String alias;
|
||||
private final AsyncTdEasy client;
|
||||
private final ConcurrentHashMap<Integer, Supergroup> supergroupInfos = new ConcurrentHashMap<>();
|
||||
private final ConcurrentHashMap<Integer, SupergroupFullInfo> supergroupFullInfos = new ConcurrentHashMap<>();
|
||||
@ -35,7 +38,8 @@ public class TransferClient {
|
||||
private final EmitterProcessor<ItemUpdate<SupergroupInfo>> usefulSupergroups = EmitterProcessor.create();
|
||||
private final Scheduler scheduler;
|
||||
|
||||
public TransferClient(AsyncTdEasy client) {
|
||||
public TransferClient(String alias, AsyncTdEasy client) {
|
||||
this.alias = alias;
|
||||
this.client = client;
|
||||
|
||||
this.scheduler = Schedulers.boundedElastic();
|
||||
@ -81,6 +85,11 @@ public class TransferClient {
|
||||
}
|
||||
|
||||
private Mono<Void> onUpdateSupergroup(Supergroup supergroup) {
|
||||
// Fast checks to ignore most unwanted infos
|
||||
if (!supergroup.isChannel
|
||||
&&
|
||||
(supergroup.status.getConstructor() == ChatMemberStatusAdministrator.CONSTRUCTOR
|
||||
|| supergroup.status.getConstructor() == ChatMemberStatusCreator.CONSTRUCTOR)) {
|
||||
if (supergroupInfos.put(supergroup.id, supergroup) == null) {
|
||||
return this
|
||||
.<SupergroupFullInfo>send(new GetSupergroupFullInfo(supergroup.id))
|
||||
@ -88,6 +97,7 @@ public class TransferClient {
|
||||
.map(TdResult::result)
|
||||
.flatMap(supergroupFullInfo -> onUpdateSupergroupFullInfo(supergroup.id, supergroupFullInfo));
|
||||
}
|
||||
}
|
||||
return Mono.empty();
|
||||
}
|
||||
|
||||
@ -129,17 +139,26 @@ public class TransferClient {
|
||||
if (chatInfo == null || baseInfo == null) {
|
||||
return Optional.empty();
|
||||
} else {
|
||||
if (baseInfo.status.getConstructor() != ChatMemberStatusAdministrator.CONSTRUCTOR) {
|
||||
return Optional.empty();
|
||||
}
|
||||
var adminStatus = (ChatMemberStatusAdministrator) baseInfo.status;
|
||||
var baseChatInfo = new BaseChatInfo(id, chatInfo.title);
|
||||
if (baseInfo.status.getConstructor() == ChatMemberStatusCreator.CONSTRUCTOR) {
|
||||
var sgInfo = new SupergroupInfo(baseChatInfo, true, true);
|
||||
return Optional.of(sgInfo);
|
||||
} else if (baseInfo.status.getConstructor() == ChatMemberStatusAdministrator.CONSTRUCTOR) {
|
||||
var adminStatus = (ChatMemberStatusAdministrator) baseInfo.status;
|
||||
var sgInfo = new SupergroupInfo(baseChatInfo, adminStatus.canRestrictMembers, adminStatus.canInviteUsers);
|
||||
return Optional.of(sgInfo);
|
||||
} else {
|
||||
return Optional.empty();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public Flux<ItemUpdate<SupergroupInfo>> subscribeAdminSupergroups() {
|
||||
return usefulSupergroups.hide();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return alias;
|
||||
}
|
||||
}
|
||||
|
||||
@ -30,5 +30,9 @@ public interface TransferService {
|
||||
|
||||
Flux<ItemUpdate<SupergroupInfo>> subscribeAdminSupergroups();
|
||||
|
||||
Mono<Void> transferMembers(BaseChatInfo sourceGroup, BaseChatInfo destGroup, Function<UserStatus, Mono<Void>> userStatusConsumer);
|
||||
Mono<Void> transferMembers(BaseChatInfo sourceGroup,
|
||||
BaseChatInfo destGroup,
|
||||
Function<UserStatus, Mono<Void>> userStatusConsumer,
|
||||
Function<Integer, Mono<Void>> percentageConsumer,
|
||||
Function<String, Mono<Void>> phaseDescriptionConsumer);
|
||||
}
|
||||
|
||||
@ -2,41 +2,48 @@ package it.cavallium;
|
||||
|
||||
import static it.cavallium.PrimaryController.getUserbotPhoneNumber;
|
||||
|
||||
import com.google.common.collect.ConcurrentHashMultiset;
|
||||
import com.google.i18n.phonenumbers.NumberParseException;
|
||||
import com.google.i18n.phonenumbers.PhoneNumberUtil;
|
||||
import com.google.i18n.phonenumbers.PhoneNumberUtil.PhoneNumberFormat;
|
||||
import com.google.i18n.phonenumbers.Phonenumber.PhoneNumber;
|
||||
import com.hazelcast.cp.internal.util.Tuple2;
|
||||
import com.hazelcast.cp.internal.util.Tuple3;
|
||||
import it.tdlight.jni.TdApi;
|
||||
import it.tdlight.jni.TdApi.AuthorizationStateClosed;
|
||||
import it.tdlight.jni.TdApi.AuthorizationStateClosing;
|
||||
import it.tdlight.jni.TdApi.AuthorizationStateLoggingOut;
|
||||
import it.tdlight.jni.TdApi.AuthorizationStateReady;
|
||||
import it.tdlight.jni.TdApi.ChatMemberStatusAdministrator;
|
||||
import it.tdlight.jni.TdApi.ChatMemberStatusCreator;
|
||||
import it.tdlight.jni.TdApi.GetUserPrivacySettingRules;
|
||||
import it.tdlight.jni.TdApi.Supergroup;
|
||||
import it.tdlight.jni.TdApi.SupergroupFullInfo;
|
||||
import it.tdlight.jni.TdApi.Update;
|
||||
import it.tdlight.jni.TdApi.UpdateSupergroup;
|
||||
import it.tdlight.jni.TdApi.User;
|
||||
import it.tdlight.jni.TdApi.UserFullInfo;
|
||||
import it.tdlight.jni.TdApi.UserTypeRegular;
|
||||
import it.tdlight.tdlibsession.td.easy.AsyncTdEasy;
|
||||
import it.tdlight.tdlibsession.td.easy.ParameterInfoPasswordHint;
|
||||
import it.tdlight.tdlibsession.td.easy.TdEasySettings;
|
||||
import it.tdlight.tdlibsession.td.middle.TdClusterManager;
|
||||
import it.tdlight.tdlibsession.td.middle.direct.AsyncTdMiddleDirect;
|
||||
import it.tdlight.utils.MonoUtils;
|
||||
import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
|
||||
import it.unimi.dsi.fastutil.objects.ObjectSets;
|
||||
import java.io.File;
|
||||
import java.time.Duration;
|
||||
import java.util.HashSet;
|
||||
import java.util.Optional;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.function.BiFunction;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.stream.Collectors;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.reactivestreams.Publisher;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.slf4j.event.Level;
|
||||
import reactor.core.publisher.EmitterProcessor;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.publisher.ReplayProcessor;
|
||||
import reactor.core.scheduler.Scheduler;
|
||||
import reactor.core.scheduler.Schedulers;
|
||||
|
||||
@ -177,8 +184,28 @@ public class TransferServiceImpl implements TransferService {
|
||||
.then();
|
||||
}))
|
||||
.doOnSuccess((_v) -> {
|
||||
var newClient = new TransferClient(client);
|
||||
var newClient = new TransferClient(alias, client);
|
||||
clients.put(phoneNumberLong, newClient);
|
||||
|
||||
newClient.subscribeAdminSupergroups().doOnNext(supergroupInfoItemUpdate -> {
|
||||
if (supergroupInfoItemUpdate.isRemoved()) {
|
||||
supergroupClients.compute(supergroupInfoItemUpdate.getItem().getBaseChatInfo().getSupergroupIdInt(), (sgId, clients) -> {
|
||||
if (clients != null) {
|
||||
clients.remove(newClient);
|
||||
}
|
||||
return clients;
|
||||
});
|
||||
} else {
|
||||
supergroupClients.compute(supergroupInfoItemUpdate.getItem().getBaseChatInfo().getSupergroupIdInt(), (sgId, clients) -> {
|
||||
if (clients == null) {
|
||||
clients = ObjectSets.synchronize(new ObjectOpenHashSet<>());
|
||||
}
|
||||
clients.add(newClient);
|
||||
return clients;
|
||||
});
|
||||
}
|
||||
}).subscribe();
|
||||
|
||||
newClients.onNext(new ItemUpdate<>(false, newClient));
|
||||
}));
|
||||
})
|
||||
@ -239,23 +266,222 @@ public class TransferServiceImpl implements TransferService {
|
||||
.fromStream(clients.values().stream().map(client -> new ItemUpdate<>(false, client))))
|
||||
.filter(itemClient -> !itemClient.isRemoved())
|
||||
.map(ItemUpdate::getItem)
|
||||
.map(client1 -> client1.subscribeAdminSupergroups().doOnNext(supergroupInfoItemUpdate -> {
|
||||
supergroupClients.compute(supergroupInfoItemUpdate.getItem().getBaseChatInfo().getSupergroupIdInt(), (sgId, clients) -> {
|
||||
if (clients == null) {
|
||||
clients = new HashSet<>();
|
||||
}
|
||||
clients.add(client1);
|
||||
return clients;
|
||||
});
|
||||
}))
|
||||
.map(TransferClient::subscribeAdminSupergroups)
|
||||
.flatMap(f -> f);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<Void> transferMembers(BaseChatInfo sourceGroup,
|
||||
BaseChatInfo destGroup,
|
||||
Function<UserStatus, Mono<Void>> userStatusConsumer) {
|
||||
Function<UserStatus, Mono<Void>> userStatusConsumer,
|
||||
Function<Integer, Mono<Void>> percentageConsumer,
|
||||
Function<String, Mono<Void>> phaseDescriptionConsumer) {
|
||||
var sourceSupergroupClients = this.supergroupClients
|
||||
.getOrDefault(sourceGroup.getSupergroupIdInt(), Set.of())
|
||||
.stream()
|
||||
.filter(clients::containsValue)
|
||||
.collect(Collectors.toSet());
|
||||
if (sourceSupergroupClients.isEmpty()) {
|
||||
return Mono.error(new Exception("No userbot can remove members from the source group"));
|
||||
}
|
||||
|
||||
var destSupergroupClients = this.supergroupClients
|
||||
.getOrDefault(destGroup.getSupergroupIdInt(), Set.of())
|
||||
.stream()
|
||||
.filter(clients::containsValue)
|
||||
.collect(Collectors.toSet());
|
||||
if (destSupergroupClients.isEmpty()) {
|
||||
return Mono.error(new Exception("No userbot can add members to the destination group"));
|
||||
}
|
||||
|
||||
return percentageConsumer
|
||||
.apply(0)
|
||||
.then(phaseDescriptionConsumer.apply("Transfer from " + sourceGroup.getTitle() + " to " + destGroup.getTitle()))
|
||||
|
||||
// Check and get the set of userbots that can transfer users from group X to group Y
|
||||
.then(phaseDescriptionConsumer.apply("Checking available userbots for removing users in the source group"))
|
||||
.thenMany(Flux.fromIterable(sourceSupergroupClients))
|
||||
.flatMap(client -> client
|
||||
.send(new TdApi.GetMe())
|
||||
.timeout(Duration.ofSeconds(5))
|
||||
.then(client.<SupergroupFullInfo>send(new TdApi.GetSupergroupFullInfo(sourceGroup.getSupergroupIdInt())))
|
||||
.timeout(Duration.ofSeconds(5))
|
||||
.then(client.<Supergroup>send(new TdApi.GetSupergroup(sourceGroup.getSupergroupIdInt())))
|
||||
.timeout(Duration.ofSeconds(5))
|
||||
.filter(sourceGroupFullInfo -> {
|
||||
if (sourceGroupFullInfo.succeeded()) {
|
||||
if (sourceGroupFullInfo.result().status.getConstructor() == ChatMemberStatusAdministrator.CONSTRUCTOR) {
|
||||
var statusAdmin = (ChatMemberStatusAdministrator) sourceGroupFullInfo.result().status;
|
||||
if (statusAdmin.canRestrictMembers) {
|
||||
return true;
|
||||
} else {
|
||||
App.getLogService().append(Level.WARN, "Userbot " + client + " failed: Can't restrict members of group " + sourceGroup.getTitle());
|
||||
}
|
||||
} else if (sourceGroupFullInfo.result().status.getConstructor() == ChatMemberStatusCreator.CONSTRUCTOR) {
|
||||
return true;
|
||||
} else {
|
||||
App.getLogService().append(Level.WARN, "Userbot " + client + " failed: Can't administer group " + sourceGroup.getTitle());
|
||||
}
|
||||
} else {
|
||||
App.getLogService().append(Level.WARN, "Userbot " + client + " failed: " + sourceGroupFullInfo.cause());
|
||||
}
|
||||
return false;
|
||||
})
|
||||
.map(_v -> client)
|
||||
.onErrorResume(e -> {
|
||||
App.getLogService().append(Level.WARN, "Userbot " + client + " failed: " + e.getLocalizedMessage());
|
||||
return Mono.empty();
|
||||
}))
|
||||
.collect(Collectors.toSet())
|
||||
.flatMap(transferSourceClients -> {
|
||||
return phaseDescriptionConsumer.apply("Checking available userbots for adding users in the destination group")
|
||||
.thenMany(Flux.fromIterable(destSupergroupClients))
|
||||
.flatMap(client -> client
|
||||
.send(new TdApi.GetMe())
|
||||
.timeout(Duration.ofSeconds(5))
|
||||
.then(client.<Supergroup>send(new TdApi.GetSupergroup(destGroup.getSupergroupIdInt())))
|
||||
.timeout(Duration.ofSeconds(5))
|
||||
.filter(destGroupFullInfo -> {
|
||||
if (destGroupFullInfo.succeeded()) {
|
||||
if (destGroupFullInfo.result().status.getConstructor() == ChatMemberStatusAdministrator.CONSTRUCTOR) {
|
||||
var statusAdmin = (ChatMemberStatusAdministrator) destGroupFullInfo.result().status;
|
||||
if (statusAdmin.canInviteUsers) {
|
||||
return true;
|
||||
} else {
|
||||
App.getLogService().append(Level.WARN, "Userbot " + client + " failed: Can't invite members to group " + destGroup.getTitle());
|
||||
}
|
||||
} else if (destGroupFullInfo.result().status.getConstructor() == ChatMemberStatusCreator.CONSTRUCTOR) {
|
||||
return true;
|
||||
} else {
|
||||
App.getLogService().append(Level.WARN, "Userbot " + client + " failed: Can't administer group " + destGroup.getTitle());
|
||||
}
|
||||
} else {
|
||||
App.getLogService().append(Level.WARN, "Userbot " + client + " failed: " + destGroupFullInfo.cause());
|
||||
}
|
||||
return false;
|
||||
})
|
||||
.map(_v -> client)
|
||||
.onErrorResume(e -> {
|
||||
App.getLogService().append(Level.WARN, "Userbot " + client + " failed: " + e.getLocalizedMessage());
|
||||
return Mono.empty();
|
||||
}))
|
||||
.collect(Collectors.toSet())
|
||||
.map(transferDestClients -> Tuple2.of(transferSourceClients, transferDestClients));
|
||||
})
|
||||
.map(clientsTuple -> {
|
||||
var sourceClients = clientsTuple.element1;
|
||||
var destClients = clientsTuple.element2;
|
||||
App.getLogService().append(Level.INFO, "Found source userbots: " + sourceClients.stream().map(TransferClient::toString).collect(Collectors.joining(", ")));
|
||||
App.getLogService().append(Level.INFO, "Found destination userbots: " + destClients.stream().map(TransferClient::toString).collect(Collectors.joining(", ")));
|
||||
var chosenClients = new HashSet<TransferClient>(sourceClients);
|
||||
chosenClients.retainAll(destClients);
|
||||
return chosenClients;
|
||||
})
|
||||
.filter(chosenClients -> !chosenClients.isEmpty())
|
||||
.doOnNext(chosenClients -> {
|
||||
App.getLogService().append(Level.INFO, "Chosen userbots: " + chosenClients.stream().map(TransferClient::toString).collect(Collectors.joining(", ")));
|
||||
})
|
||||
.switchIfEmpty(Mono.defer(() -> {
|
||||
App.getLogService().append(Level.ERROR, "No userbots are admin in both groups!");
|
||||
return Mono.error(new Exception("No userbots are admin in both groups!"));
|
||||
}))
|
||||
// Now we have a set of userbots that can transfer the users
|
||||
|
||||
// Get the list of members of the first group from a bot
|
||||
.flatMap(clients -> {
|
||||
return phaseDescriptionConsumer.apply("Obtaining group members")
|
||||
.then(percentageConsumer.apply(5)).thenReturn(clients);
|
||||
}).flatMap(clients -> {
|
||||
// Get the first userbot
|
||||
var client = clients.stream().findAny().orElseThrow(() -> new NullPointerException("No userbots found"));
|
||||
|
||||
// Get the members of the source group
|
||||
return Mono.fromCallable(() -> {
|
||||
var members = TransferUtils.getSupergroupMembers(client, sourceGroup.getSupergroupIdInt());
|
||||
App.getLogService().append(Level.INFO, "Source group has " + members.size() + " members.");
|
||||
return members;
|
||||
}).subscribeOn(Schedulers.boundedElastic()).map(members -> Tuple3.of(client, clients, members));
|
||||
})
|
||||
// Finished getting the list of members of the source group
|
||||
|
||||
// Resolve users
|
||||
.flatMap(context -> {
|
||||
return phaseDescriptionConsumer.apply("Resolving users")
|
||||
.then(percentageConsumer.apply(10)).thenReturn(context);
|
||||
})
|
||||
.flatMap(context -> {
|
||||
var client = context.element1;
|
||||
var clients = context.element2;
|
||||
var unresolvedUsers = context.element3;
|
||||
return Flux
|
||||
.fromIterable(unresolvedUsers)
|
||||
.flatMap(userId -> client.<User>send(new TdApi.GetUser(userId)))
|
||||
.timeout(Duration.ofMinutes(2))
|
||||
.flatMap(MonoFxUtils::orElseLogSkipError)
|
||||
.collect(Collectors.toSet())
|
||||
.map(resolvedUsers -> Tuple3.of(client, clients, resolvedUsers));
|
||||
})
|
||||
// Finished resolving users
|
||||
|
||||
// Filter out unsuitable users
|
||||
.flatMap(context -> {
|
||||
return phaseDescriptionConsumer.apply("Filtering users")
|
||||
.then(percentageConsumer.apply(15)).thenReturn(context);
|
||||
})
|
||||
.flatMap(context -> {
|
||||
var client = context.element1;
|
||||
var clients = context.element2;
|
||||
var unfilteredUsers = context.element3;
|
||||
return Flux
|
||||
.fromIterable(unfilteredUsers)
|
||||
.filter(user -> {
|
||||
if (user.haveAccess) {
|
||||
return true;
|
||||
}
|
||||
userStatusConsumer.apply(new UserStatus(getName(user), user.id, UserStatusType.NO_ACCESS_HASH, ""));
|
||||
return false;
|
||||
})
|
||||
.filter(user -> {
|
||||
if (user.type.getConstructor() == UserTypeRegular.CONSTRUCTOR) {
|
||||
return true;
|
||||
}
|
||||
userStatusConsumer.apply(new UserStatus(getName(user), user.id, UserStatusType.NOT_REGULAR_USER, ""));
|
||||
return false;
|
||||
})
|
||||
.filter(user -> {
|
||||
if (!user.isScam) {
|
||||
return true;
|
||||
}
|
||||
userStatusConsumer.apply(new UserStatus(getName(user), user.id, UserStatusType.SCAM_USER, ""));
|
||||
return false;
|
||||
})
|
||||
.filter(user -> {
|
||||
if (user.restrictionReason == null || user.restrictionReason.isEmpty()) {
|
||||
return true;
|
||||
}
|
||||
userStatusConsumer.apply(new UserStatus(getName(user), user.id, UserStatusType.RESTRICTED_USER, "Restricted user: " + user.restrictionReason));
|
||||
return false;
|
||||
})
|
||||
.flatMap(user -> {
|
||||
return userStatusConsumer
|
||||
.apply(new UserStatus(getName(user), user.id, UserStatusType.JUST_FOUND, ""))
|
||||
.thenReturn(user);
|
||||
})
|
||||
.timeout(Duration.ofMinutes(2))
|
||||
.collect(Collectors.toSet())
|
||||
.map(resolvedUsers -> Tuple3.of(client, clients, resolvedUsers));
|
||||
})
|
||||
// Finished filtering unsuitable users
|
||||
|
||||
|
||||
.then(percentageConsumer.apply(100))
|
||||
.then(phaseDescriptionConsumer.apply("Done"))
|
||||
.then(Mono.delay(Duration.ofMillis(500)))
|
||||
.then();
|
||||
}
|
||||
|
||||
private static String getName(User user) {
|
||||
return String.join(" ", List.of("" + user.id, user.firstName, user.lastName));
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -263,7 +489,6 @@ public class TransferServiceImpl implements TransferService {
|
||||
return Flux
|
||||
.fromIterable(clients.values())
|
||||
.flatMap(client -> client.send(new TdApi.Close()))
|
||||
.log()
|
||||
.collectList()
|
||||
.then();
|
||||
}
|
||||
|
||||
@ -1,20 +1,26 @@
|
||||
package it.cavallium;
|
||||
|
||||
import it.tdlight.jni.TdApi;
|
||||
import it.tdlight.jni.TdApi.Chat;
|
||||
import it.tdlight.jni.TdApi.ChatListMain;
|
||||
import it.tdlight.jni.TdApi.ChatMember;
|
||||
import it.tdlight.jni.TdApi.ChatMemberStatus;
|
||||
import it.tdlight.jni.TdApi.ChatMembers;
|
||||
import it.tdlight.jni.TdApi.ChatPosition;
|
||||
import it.tdlight.jni.TdApi.Chats;
|
||||
import it.tdlight.jni.TdApi.GetChat;
|
||||
import it.tdlight.jni.TdApi.GetChats;
|
||||
import it.tdlight.jni.TdApi.GetSupergroupMembers;
|
||||
import it.tdlight.jni.TdApi.SupergroupFullInfo;
|
||||
import it.tdlight.utils.MonoUtils;
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.stream.Collectors;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.slf4j.event.Level;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
@ -142,6 +148,9 @@ public class TransferUtils {
|
||||
.subscriberContext(context);
|
||||
})
|
||||
.repeatWhen(nFlux -> nFlux.takeWhile(n -> n > 0))
|
||||
.doOnNext(chats -> {
|
||||
App.getLogService().append(Level.DEBUG, "Received " + chats.size() + " home chats");
|
||||
})
|
||||
.flatMap(Flux::fromIterable)
|
||||
.subscriberContext(ctx -> ctx.put("offsets",
|
||||
new AtomicReference<>(new ChatIdAndOrderOffsets(0L, 9223372036854775807L))
|
||||
@ -157,4 +166,86 @@ public class TransferUtils {
|
||||
}
|
||||
return Optional.empty();
|
||||
}
|
||||
|
||||
|
||||
public static final String[] searchTable = {"a", "b", "c", "d", "e", "f", "g", "h", "i", "l", "m", "o", "p", "p", "q",
|
||||
"r", "r", "s", "t", "u", "v", "x", "y", "z", "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "bot", "#"};
|
||||
|
||||
/**
|
||||
* Reimport supergroup or channel members fully
|
||||
*/
|
||||
public static Set<Integer> getSupergroupMembers(TransferClient client, int supergroupId) {
|
||||
var supergroupFullInfo = Objects.requireNonNull(
|
||||
client.<SupergroupFullInfo>send(new TdApi.GetSupergroupFullInfo(supergroupId))
|
||||
.flatMap(MonoUtils::orElseThrow)
|
||||
.block());
|
||||
|
||||
var fullChatMemberList = new HashMap<Integer, ChatMemberStatus>(supergroupFullInfo.memberCount + 20);
|
||||
|
||||
final var timeout = 24 * 60 * 60 * 1000; /* 1 day */
|
||||
final var limit = 200;
|
||||
|
||||
{
|
||||
var offset = 0;
|
||||
final var filter = new TdApi.SupergroupMembersFilterSearch("");
|
||||
|
||||
while (true) {
|
||||
TdApi.ChatMembers members = client.<ChatMembers>send(new TdApi.GetSupergroupMembers(supergroupId,
|
||||
filter,
|
||||
offset,
|
||||
limit
|
||||
)).timeout(Duration.ofMinutes(2)).flatMap(MonoUtils::orElseThrow).blockOptional().orElseThrow();
|
||||
for (ChatMember member : members.members) {
|
||||
fullChatMemberList.put(member.userId, member.status);
|
||||
}
|
||||
offset += members.members.length;
|
||||
if (members.members.length == 0) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// if upgradedFromBasicGroupId != 0 is a group
|
||||
// if upgradedFromMaxMessageId != 0 is a group
|
||||
if (supergroupFullInfo.upgradedFromBasicGroupId == 0 && supergroupFullInfo.upgradedFromMaxMessageId == 0
|
||||
&& supergroupFullInfo.memberCount > 500) {
|
||||
TdApi.Chat info = client
|
||||
.<Chat>send(new TdApi.GetChat(supergroupId))
|
||||
.flatMap(MonoUtils::orElseThrow)
|
||||
.timeout(Duration.ofMinutes(2))
|
||||
.blockOptional()
|
||||
.orElseThrow();
|
||||
if (info.type.getConstructor() == TdApi.ChatTypeSupergroup.CONSTRUCTOR) {
|
||||
var type = (TdApi.ChatTypeSupergroup) info.type;
|
||||
if (type.isChannel) {
|
||||
final var filterChannel = new TdApi.ChatMembersFilterMembers();
|
||||
for (String q : searchTable) {
|
||||
int offset = 0;
|
||||
while (true) {
|
||||
final var filter = new TdApi.SupergroupMembersFilterSearch(q);
|
||||
var members = client.<ChatMembers>send(new GetSupergroupMembers(supergroupId, filter, offset, limit))
|
||||
.flatMap(MonoUtils::orElseThrow)
|
||||
.timeout(Duration.ofMinutes(2))
|
||||
.blockOptional()
|
||||
.orElseThrow();
|
||||
for (int i = 0; i < members.members.length; i++) {
|
||||
var member = members.members[i];
|
||||
var userId = member.userId;
|
||||
if (!fullChatMemberList.containsKey(userId)) {
|
||||
fullChatMemberList.put(userId, member.status);
|
||||
}
|
||||
}
|
||||
offset += members.members.length;
|
||||
|
||||
if (members.members.length == 0) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return fullChatMemberList.keySet();
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,18 +1,42 @@
|
||||
package it.cavallium;
|
||||
|
||||
import java.util.StringJoiner;
|
||||
import javafx.beans.property.SimpleStringProperty;
|
||||
|
||||
public class UserStatus {
|
||||
private final String name;
|
||||
private final int id;
|
||||
private final UserStatusType statusType;
|
||||
private final String errorDescription;
|
||||
private final SimpleStringProperty user = new SimpleStringProperty("");
|
||||
private final SimpleStringProperty status = new SimpleStringProperty("");
|
||||
|
||||
public UserStatus(String name, int id, UserStatusType statusType, String errorDescription) {
|
||||
this.name = name;
|
||||
this.id = id;
|
||||
this.statusType = statusType;
|
||||
this.errorDescription = errorDescription;
|
||||
this.user.set(name);
|
||||
this.status.set(errorDescription.isBlank() ? translateStatusType(statusType) : errorDescription);
|
||||
}
|
||||
|
||||
private String translateStatusType(UserStatusType statusType) {
|
||||
switch (statusType) {
|
||||
case SCAM_USER:
|
||||
return "Scam user, skipped";
|
||||
case NO_ACCESS_HASH:
|
||||
return "No access hash, skipped";
|
||||
case RESTRICTED_USER:
|
||||
return "User restricted, skipped";
|
||||
case JUST_FOUND:
|
||||
return "";
|
||||
case UNKNOWN:
|
||||
return "Unknown";
|
||||
case NOT_REGULAR_USER:
|
||||
return "Not a regular user, skipped";
|
||||
default:
|
||||
return statusType.toString();
|
||||
}
|
||||
}
|
||||
|
||||
public int getId() {
|
||||
@ -31,6 +55,14 @@ public class UserStatus {
|
||||
return statusType;
|
||||
}
|
||||
|
||||
public String getUser() {
|
||||
return user.get();
|
||||
}
|
||||
|
||||
public String getStatus() {
|
||||
return status.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) {
|
||||
@ -45,21 +77,13 @@ public class UserStatus {
|
||||
if (id != that.id) {
|
||||
return false;
|
||||
}
|
||||
if (name != null ? !name.equals(that.name) : that.name != null) {
|
||||
return false;
|
||||
}
|
||||
if (statusType != that.statusType) {
|
||||
return false;
|
||||
}
|
||||
return errorDescription != null ? errorDescription.equals(that.errorDescription) : that.errorDescription == null;
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int result = name != null ? name.hashCode() : 0;
|
||||
result = 31 * result + id;
|
||||
result = 31 * result + (statusType != null ? statusType.hashCode() : 0);
|
||||
result = 31 * result + (errorDescription != null ? errorDescription.hashCode() : 0);
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
package it.cavallium;
|
||||
|
||||
public enum UserStatusType {
|
||||
UNKNOWN
|
||||
NO_ACCESS_HASH, NOT_REGULAR_USER, SCAM_USER, RESTRICTED_USER, JUST_FOUND, UNKNOWN
|
||||
}
|
||||
|
||||
@ -15,6 +15,7 @@
|
||||
<?import javafx.scene.control.TableView?>
|
||||
<?import javafx.scene.control.TextField?>
|
||||
<?import javafx.scene.control.ToolBar?>
|
||||
<?import javafx.scene.control.cell.PropertyValueFactory?>
|
||||
<?import javafx.scene.layout.BorderPane?>
|
||||
<?import javafx.scene.layout.ColumnConstraints?>
|
||||
<?import javafx.scene.layout.GridPane?>
|
||||
@ -38,7 +39,7 @@
|
||||
</Menu>
|
||||
</menus>
|
||||
</MenuBar>
|
||||
<SplitPane dividerPositions="0.8237442922374429" orientation="VERTICAL">
|
||||
<SplitPane dividerPositions="0.8237442922374429" orientation="VERTICAL" prefHeight="9999.0">
|
||||
<items>
|
||||
<SplitPane dividerPositions="0.5" focusTraversable="true" prefHeight="-1.0" prefWidth="-1.0">
|
||||
<items>
|
||||
@ -90,7 +91,6 @@
|
||||
<rowConstraints>
|
||||
<RowConstraints />
|
||||
<RowConstraints vgrow="SOMETIMES" />
|
||||
<RowConstraints vgrow="SOMETIMES" />
|
||||
</rowConstraints>
|
||||
<children>
|
||||
<VBox maxHeight="-Infinity" minHeight="-Infinity" spacing="10.0" GridPane.columnSpan="2">
|
||||
@ -105,8 +105,16 @@
|
||||
</VBox>
|
||||
<TableView fx:id="statusTable" GridPane.columnSpan="2" GridPane.rowIndex="1">
|
||||
<columns>
|
||||
<TableColumn minWidth="100.0" prefWidth="200.0" text="User" />
|
||||
<TableColumn minWidth="100.0" prefWidth="100.0" text="Status" />
|
||||
<TableColumn minWidth="100.0" prefWidth="200.0" text="User">
|
||||
<cellValueFactory>
|
||||
<PropertyValueFactory property="user" />
|
||||
</cellValueFactory>
|
||||
</TableColumn>
|
||||
<TableColumn minWidth="100.0" prefWidth="100.0" text="Status">
|
||||
<cellValueFactory>
|
||||
<PropertyValueFactory property="status" />
|
||||
</cellValueFactory>
|
||||
</TableColumn>
|
||||
</columns>
|
||||
<columnResizePolicy>
|
||||
<TableView fx:constant="CONSTRAINED_RESIZE_POLICY" />
|
||||
@ -127,7 +135,7 @@
|
||||
</Text>
|
||||
</top>
|
||||
<center>
|
||||
<ListView fx:id="log" style="-fx-font-size: 8" BorderPane.alignment="CENTER" />
|
||||
<ListView fx:id="log" style="-fx-font-size: 10" BorderPane.alignment="CENTER" />
|
||||
</center>
|
||||
</BorderPane>
|
||||
</items>
|
||||
|
||||
@ -1 +1 @@
|
||||
Subproject commit 85bac8670d201922ac695f3f086eb0c6cd526e8f
|
||||
Subproject commit 6fd5b099c3f90e1dbca28d37c9003541e47b4d51
|
||||
Loading…
Reference in New Issue
Block a user