Skip to content

Bluesky gephi #334

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eduramiba merged 2 commits into from
Jul 30, 2026
Merged

Bluesky gephi #334

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e36e6d2
Bug fix over connection and runtime
totetmatt Jul 30, 2026
d981694
Code improvement on ATProto Client
totetmatt Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion modules/BlueskyGephi/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

<groupId>fr.totetmatt</groupId>
<artifactId>bluesky-gephi</artifactId>
<version>0.2.1</version>
<version>0.2.2</version>
<packaging>nbm</packaging>

<name>Bluesky Gephi</name>
Expand Down
255 changes: 186 additions & 69 deletions modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephi.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,20 +1,22 @@
package fr.totetmatt.blueskygephi;

import fr.totetmatt.blueskygephi.atproto.AtClient;
import fr.totetmatt.blueskygephi.atproto.response.AppBskyGraphGetFollowers;
import fr.totetmatt.blueskygephi.atproto.response.AppBskyGraphGetFollows;
import fr.totetmatt.blueskygephi.atproto.response.AppBskyGraphGetList;
import fr.totetmatt.blueskygephi.atproto.response.common.Identity;
import java.awt.Color;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.LongConsumer;
import java.util.logging.Logger;
import java.util.prefs.Preferences;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.gephi.graph.api.Edge;
import org.gephi.graph.api.Graph;
import org.gephi.graph.api.GraphController;
import org.gephi.graph.api.GraphModel;
import org.gephi.graph.api.Node;
Expand Down Expand Up @@ -51,6 +53,15 @@ public class BlueskyGephi {
private AtClient client;
private GraphModel graphModel;

// Bounded pool so pasting many handles can't spawn an unbounded number of
// threads all contending on the graph write-lock and hammering the API.
private final ExecutorService executor = Executors.newFixedThreadPool(2, r -> {
Thread t = new Thread(r);
t.setDaemon(true);
t.setName("[Bsky] worker");
return t;
});

public BlueskyGephi() {
}

Expand Down Expand Up @@ -161,108 +172,192 @@ private Edge createEdge(Node source, Node target) {
}

private void fetchFollowerFollowsFromActor(String actor, List<String> listInit, boolean isFollowsActive, boolean isFollowersActive, boolean isDeepSearch) {
// To avoid locking Gephi UI
Thread t = new Thread() {
private ProgressTicket progressTicket;
Set<String> foaf = new HashSet<>();
// Run on a bounded background pool to keep the Gephi UI responsive.
executor.submit(new FetchTask(actor, listInit, isFollowsActive, isFollowersActive, isDeepSearch));
}

private void process(String actor, boolean isDeepSearch, Optional<Integer> limitCrawl) {
/**
* A single fetch job. Reads the network for one actor (and, when deep
* search or a list is involved, for the discovered friends-of-a-friend)
* and writes it into the graph.
*/
private final class FetchTask implements Runnable {

private final String actor;
private final List<String> listInit;
private final boolean isFollowsActive;
private final boolean isFollowersActive;
private final boolean isDeepSearch;

private final Set<String> foaf = new HashSet<>();
private volatile boolean cancelled = false;
private volatile Thread worker;
private ProgressTicket progressTicket;
private LongConsumer onWaiting;

FetchTask(String actor, List<String> listInit, boolean isFollowsActive, boolean isFollowersActive, boolean isDeepSearch) {
this.actor = actor;
this.listInit = listInit;
this.isFollowsActive = isFollowsActive;
this.isFollowersActive = isFollowersActive;
this.isDeepSearch = isDeepSearch;
}

try {
if (isFollowsActive) {
List<AppBskyGraphGetFollows> responses = client.appBskyGraphGetFollows(actor, limitCrawl);
private boolean shouldStop() {
return cancelled || Thread.currentThread().isInterrupted();
}

for (var response : responses) {
graphModel.getGraph().writeLock();
Identity subject = response.getSubject();
private void process(String actor, boolean isDeepSearch, Optional<Integer> limitCrawl) {
if (client == null || graphModel == null) {
return;
}
try {
if (isFollowsActive && !shouldStop()) {
// Each page is written as it arrives; pagination stops when the
// worker is interrupted (cancellation).
client.appBskyGraphGetFollows(actor, limitCrawl, response -> {
if (shouldStop()) {
return;
}
Identity subject = response.getSubject();
if (subject == null || subject.getDid() == null) {
return;
}
// Capture the graph once so lock/unlock always target the
// same instance, and release it in finally so an exception
// (bad data, API error, cancellation) can never leak the
// write-lock and freeze Gephi.
Graph graph = graphModel.getGraph();
graph.writeLock();
try {
Node source = createNode(subject);
source.setColor(Color.GREEN);
for (var follow : response.getFollows()) {
if (isDeepSearch) {
foaf.add(follow.getDid());
List<Identity> follows = response.getFollows();
if (follows != null) {
for (var follow : follows) {
if (follow == null || follow.getDid() == null) {
continue;
}
if (isDeepSearch) {
foaf.add(follow.getDid());
}
Node target = createNode(follow);
createEdge(source, target);
}
Node target = createNode(follow);
createEdge(source, target);
}
graphModel.getGraph().writeUnlock();

} finally {
graph.writeUnlock();
}
}, onWaiting);
}

}

if (isFollowersActive) {
List<AppBskyGraphGetFollowers> responses = client.appBskyGraphGetFollowers(actor, limitCrawl);

for (var response : responses) {
graphModel.getGraph().writeLock();
Identity subject = response.getSubject();
if (isFollowersActive && !shouldStop()) {
client.appBskyGraphGetFollowers(actor, limitCrawl, response -> {
if (shouldStop()) {
return;
}
Identity subject = response.getSubject();
if (subject == null || subject.getDid() == null) {
return;
}
Graph graph = graphModel.getGraph();
graph.writeLock();
try {
Node target = createNode(subject);
target.setColor(Color.GREEN);
for (var follower : response.getFollowers()) {
if (isDeepSearch) {
foaf.add(follower.getDid());
List<Identity> followers = response.getFollowers();
if (followers != null) {
for (var follower : followers) {
if (follower == null || follower.getDid() == null) {
continue;
}
if (isDeepSearch) {
foaf.add(follower.getDid());
}
Node source = createNode(follower);
createEdge(source, target);
}
Node source = createNode(follower);
createEdge(source, target);
}
graphModel.getGraph().writeUnlock();
} finally {
graph.writeUnlock();
}

}
} catch (Exception e) {
}, onWaiting);
}
} catch (Exception e) {
if (!cancelled) {
Exceptions.printStackTrace(e);
} finally {
}
}
}

@Override
public void run() {

if (actor != null) {
this.setName("[Bsky] fetching" + actor);
@Override
public void run() {
worker = Thread.currentThread();
final String taskName = (actor != null) ? "[Bsky] fetching " + actor : "[Bsky] fetching List";
worker.setName(taskName);

progressTicket = Lookup.getDefault()
.lookup(ProgressTicketProvider.class)
.createTicket(taskName, () -> {
cancelled = true;
Thread w = worker;
if (w != null) {
w.interrupt();
}
return true;
});
// Surface the client's rate-limit backoff as a live countdown in the progress bar.
onWaiting = remainingMillis -> {
if (remainingMillis <= 0L) {
Progress.setDisplayName(progressTicket, taskName);
} else {
this.setName("[Bsky] fetching List");
long seconds = (remainingMillis + 999L) / 1000L;
Progress.setDisplayName(progressTicket, taskName + " - rate limited, retrying in " + seconds + "s");
}
progressTicket = Lookup.getDefault()
.lookup(ProgressTicketProvider.class)
.createTicket(this.getName(), () -> {
interrupt();
Progress.finish(progressTicket);
return true;
});
};
try {
Progress.start(progressTicket);
Progress.switchToIndeterminate(progressTicket);

if (listInit != null) {
this.foaf.addAll(listInit);
foaf.addAll(listInit);
}
if (actor != null) {
process(actor, isDeepSearch, Optional.empty());
}
if (listInit != null || isDeepSearch) {
if ((listInit != null || isDeepSearch) && !shouldStop()) {
Progress.switchToDeterminate(progressTicket, foaf.size());
for (var foafActor : foaf) {
if (shouldStop()) {
break;
}
Progress.setDisplayName(progressTicket, "[Bsky] fetching " + actor + " n+1 > " + foafActor);
if (getIsLimitCrawlActive()) {
process(foafActor, false, Optional.of(getLimitCrawl()));
} else {
process(foafActor, false, Optional.empty());
}
Progress.progress(progressTicket);

}
}
} finally {
Progress.finish(progressTicket);
}
};
t.start();

}
}

private Stream<String> manageList(String listId) {
List<AppBskyGraphGetList> list = client.appBskyGraphGetList(listId);
return list.stream().flatMap(x -> x.getItems().stream().map(y -> y.getSubject().getDid()));

List<String> dids = new ArrayList<>();
client.appBskyGraphGetList(listId, page -> {
if (page.getItems() != null) {
for (var item : page.getItems()) {
if (item != null && item.getSubject() != null && item.getSubject().getDid() != null) {
dids.add(item.getSubject().getDid());
}
}
}
}, null);
return dids.stream();
}

private void initGraphTable() {
Expand All @@ -273,28 +368,50 @@ private void initGraphTable() {
}

public void fetchFollowerFollowsFromActors(List<String> actors, boolean isFollowsActive, boolean isFollowersActive, boolean isBlocksActive) {
if (client == null) {
logger.warning("Not connected to Bluesky. Please connect before fetching.");
return;
}
ensureProject();
graphModel = Lookup.getDefault().lookup(GraphController.class).getGraphModel();
initGraphTable();
actors.stream().forEach(actor -> fetchFollowerFollowsFromActor(actor, null, isFollowsActive, isFollowersActive, getIsDeepSearch()));
actors.stream()
.map(String::trim)
.filter(actor -> !actor.isEmpty())
.forEach(actor -> fetchFollowerFollowsFromActor(actor, null, isFollowsActive, isFollowersActive, getIsDeepSearch()));
}

public void fetchFollowerFollowsFromActors(List<String> actors) {
if (client == null) {
logger.warning("Not connected to Bluesky. Please connect before fetching.");
return;
}
ensureProject();
graphModel = Lookup.getDefault().lookup(GraphController.class).getGraphModel();
initGraphTable();
actors
.stream()

List<String> cleanActors = actors.stream()
.map(String::trim)
.filter(x -> !x.isEmpty())
.collect(Collectors.toList());

cleanActors.stream()
.filter(x -> !x.contains("app.bsky.graph.list"))
.forEach(actor -> fetchFollowerFollowsFromActor(actor, null, getIsFollowsActive(), getIsFollowersActive(), getIsDeepSearch()));

List<String> listActor = actors
.stream()
List<String> listIds = cleanActors.stream()
.filter(x -> x.contains("app.bsky.graph.list"))
.flatMap(this::manageList)
.collect(Collectors.toList());

fetchFollowerFollowsFromActor(null, listActor, getIsFollowsActive(), getIsFollowersActive(), getIsDeepSearch());

if (!listIds.isEmpty()) {
// Resolving a list is itself several paged network calls, so do it
// off the EDT instead of blocking the UI thread.
executor.submit(() -> {
List<String> listActor = listIds.stream()
.flatMap(this::manageList)
.collect(Collectors.toList());
fetchFollowerFollowsFromActor(null, listActor, getIsFollowsActive(), getIsFollowersActive(), getIsDeepSearch());
});
}
}
}
31 changes: 26 additions & 5 deletions modules/BlueskyGephi/src/main/java/fr/totetmatt/blueskygephi/BlueskyGephiMainPanel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,11 +269,32 @@ private void isFollowersActivatedActionPerformed(java.awt.event.ActionEvent evt)
}//GEN-LAST:event_isFollowersActivatedActionPerformed

private void credentialsConnectButtonActionPerformed(java.awt.event.ActionEvent evt) {//GEN-FIRST:event_credentialsConnectButtonActionPerformed
if(blueskyGephi.connect(credentialsHostField.getText(), credentialsHandleField.getText(), String.valueOf(credentialsPasswordField.getPassword()))){
credentialsConnectButton.setBackground(Color.GREEN);
} else {
credentialsConnectButton.setBackground(Color.RED);
}
final String host = credentialsHostField.getText();
final String handle = credentialsHandleField.getText();
final String password = String.valueOf(credentialsPasswordField.getPassword());
// Do the network call off the EDT so a slow or failing login never freezes the UI.
credentialsConnectButton.setEnabled(false);
credentialsConnectButton.setBackground(Color.ORANGE);
new javax.swing.SwingWorker<Boolean, Void>() {
@Override
protected Boolean doInBackground() {
return blueskyGephi.connect(host, handle, password);
}

@Override
protected void done() {
boolean connected = false;
try {
connected = get();
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
} catch (java.util.concurrent.ExecutionException ex) {
Exceptions.printStackTrace(ex);
}
credentialsConnectButton.setBackground(connected ? Color.GREEN : Color.RED);
credentialsConnectButton.setEnabled(true);
}
}.execute();
}//GEN-LAST:event_credentialsConnectButtonActionPerformed

private void runFetchButtonActionPerformed(java.awt.event.ActionEvent evt) {//GEN-FIRST:event_runFetchButtonActionPerformed
Expand Down
Loading
Loading